peterlu14
77 lines
2.1 KiB
Python
77 lines
2.1 KiB
Python
#!/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) |