controller-data-analysis/requests.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

class Requests():
  def __init__(self, mqtt_client) -> None:
    self._mqtt_client = mqtt_client

  def get_device(self) -> None:
    self._mqtt_client.publish({ 
      'header': 'get_device_info_all/0' 
    })
    return

  def device_parameter_array_set(self, device: str, name: str, data: str) -> None:
    dataArray = [0 for i in range(16)]
    for index in range(0, len(data), 2):
      dataArray[int(index / 2)] = int(data[index:index+2],16)
    
    self._mqtt_client.publish({ 
     'header': 'device_parameter/0',
     'device': device,
     'parameter': name,
     'content': '0:16=' + ','.join([str(ele) for ele in dataArray])
    })
    return

  def device_parameter_set(self, device: str, name: str, data: str) -> None:
    self._mqtt_client.publish({ 
     'header': 'device_parameter/0',
     'device': device,
     'parameter': name,
     'content': data
    })
    return

  def device_parameter_get(self, device: str, name: str) -> None:
    self._mqtt_client.publish({ 
     'header': 'device_parameter_value/0',
     'device': device,
     'parameter': name,
    })

  def device_instruction_send(self, device: str, instruction: str) -> None:
    self._mqtt_client.publish({
      'header': 'device_instruction/0',
      'device': device,
      'instruction': instruction
    })

  def update_device_calibration(self, device: str) -> None:
      self._mqtt_client.publish({
        'header': 'device_instruction/0',
        'device': device,
      })
    
  def hardware_scan(self) -> None:
      self._mqtt_client.publish({
        'header': 'hardware_scan/0',
      })
  
  def hardware_device(self) -> None:
    self._mqtt_client.publish({
      'header': 'hardware_device/0',
    })
  
  def hardware_device_connect(self, device_name, device_address, serial_number) -> None:
    self._mqtt_client.publish({
      'header': 'hardware_device_connect/0',
      'content': {
        'device_name': device_name,
        'device_address': self.mac_tuple_to_str(device_address),
        'serial_number': serial_number
      }
    })
  
  def mac_tuple_to_str(self, mac_array):
    return ':'.join('{:02x}'.format(b) for b in mac_array)