send commands to device via command topics

main.py

@@ -10,39 +10,52 @@ from bleak.exc import BleakError, BleakDeviceNotFoundError
 
 from src.homeassistant import MqttSensor
 from src.bleclient import BleClient, Result
-from src.variables import variables
+from src.variables import variables, VariableContainer
 
-send_config = True
 request_interval = 20   # In seconds
 reconnect_interval = 5  # In seconds
 
 async def request_and_publish_details(sensor: MqttSensor, mppt: BleClient) -> None:
-    global send_config
     details = await mppt.request_details()
     if details:
         print(f"Battery: {details['battery_percentage'].value}% ({details['battery_voltage'].value}V)")
-        if send_config:
-            print(f"configuring {len(details)}/{len(variables)} entities")
-            await sensor.store_config(details)
-            send_config = False
-        
+        await sensor.store_config(details)
         await sensor.publish(details)
     else:
         print("No values recieved")
 
+async def subscribe_and_watch_switches(sensor: MqttSensor, mppt: BleClient):
+    variable = variables['manual_control_switch']
+    variable_container = VariableContainer([variable])
+    await sensor.subscribe(variable_container)
+    await sensor.store_config(variable_container)
+    for command in await sensor.get_commands():
+        results = mppt.write([command])
+        await sensor.publish(results)
+
 async def run_mppt(sensor: MqttSensor, address: str):
+    task = None
+    loop = asyncio.get_event_loop()
     try:
         async with BleClient(address) as mppt:
+            task = loop.create_task(subscribe_and_watch_switches(sensor, mppt))
             while True:
-                await request_and_publish_details()
+                await request_and_publish_details(sensor, mppt)
                 await asyncio.sleep(request_interval)
+                if not task.cancelled() and task.exception:
+                    break
 
     except BleakDeviceNotFoundError:
         print(f"BLE device with address {address} was not found")
-        await asyncio.sleep(reconnect_interval)
     except BleakError as e:
         print(f"BLE error occurred: {e}")
-        await asyncio.sleep(reconnect_interval)
+    finally:
+        if task:
+            task.cancel()
+            try:
+                await task
+            except asyncio.CancelledError:
+                pass
 
 async def run_mqtt(address, host, port, username, password):
     while True:
@@ -51,14 +64,14 @@ async def run_mqtt(address, host, port, username, password):
                 print(f"Connected to MQTT broker at {host}:{port}")
                 while True:
                     await run_mppt(sensor, address)
+                    await asyncio.sleep(reconnect_interval)
         except aiomqtt.MqttError as error:
             print(f'Error "{error}". Reconnecting in {reconnect_interval} seconds.')
-            await asyncio.sleep(reconnect_interval)
         except asyncio.CancelledError:
             raise  # Re-raise the CancelledError to stop the task
         except Exception as e:
             print(f"An error occurred during BLE communication: {e}")
-            await asyncio.sleep(reconnect_interval)
+        await asyncio.sleep(reconnect_interval)
 
 async def main(*args):
     try:

src/bleclient.py

@@ -4,7 +4,7 @@ from bleak import BleakClient, BleakScanner
 from bleak.backends.characteristic import BleakGATTCharacteristic
 
 from src.crc import crc16
-from src.protocol import LumiaxClient, ResultContainer
+from src.protocol import LumiaxClient, ResultContainer, Result
 
 class BleClient(LumiaxClient):
     DEVICE_NAME_UUID = "00002a00-0000-1000-8000-00805f9b34fb"
@@ -55,6 +55,23 @@ class BleClient(LumiaxClient):
 
     async def request_details(self) -> ResultContainer:
         return await self.read(0x3030, 43)
+    
+    async def write(self, results: list[Result], repeat = 10, timeout = 5) -> ResultContainer:
+        async with self.lock:
+            start_address, command = self.get_write_command(0xFE, results)
+            self.response_queue = asyncio.Queue() # Clear the queue
+            i = 0
+            # send the command multiple times
+            while self.response_queue.empty() and i < repeat:
+                i += 1
+                await self.client.write_gatt_char(self.WRITE_UUID, command)
+                try:
+                    # Wait for either a response or timeout
+                    await asyncio.wait_for(self.response_queue.get(), timeout=timeout)
+                    return results
+                except asyncio.TimeoutError:
+                    pass
+            return None
 
     async def get_device_name(self):
         device_name = await self.client.read_gatt_char(self.DEVICE_NAME_UUID)  # Read the device name from the BLE device

src/homeassistant.py

@@ -1,9 +1,10 @@
 import json
+import re
 
 from aiomqtt import Client
 
-from src.protocol import ResultContainer, VariableContainer, Variable, FunctionCodes
-
+from src.protocol import ResultContainer, Result, FunctionCodes
+from src.variables import VariableContainer, Variable, variables
 
 class MqttSensor(Client):
     # Define the base topic for MQTT Discovery
@@ -19,6 +20,10 @@ class MqttSensor(Client):
         "manufacturer": "Solarlife",
     }
 
+    def __init__(self, *args, **kwargs):
+        super(MqttSensor, self).__init__(*args, **kwargs)
+        self.known_names = set()
+
     # https://www.home-assistant.io/integrations/#search/mqtt
     def get_platform(self, variable: Variable) -> str:
         is_writable = FunctionCodes.WRITE_MEMORY_SINGLE in value.function_codes or \
@@ -51,54 +56,63 @@ class MqttSensor(Client):
         platform = self.get_platform(variable)
         return f"{self.base_topic}/{platform}/{self.sensor_name}/{variable.name}/command"
 
-    async def store_config(self, results: ResultContainer) -> None:
+    async def store_config(self, variables: VariableContainer) -> None:
         # Publish each item in the results to its own MQTT topic
-        for key, value in results.items():
-            config_topic = self.get_config_topic(value)
-            state_topic = self.get_state_topic(value)
-            command_topic = self.get_command_topic(value)
+        for key, variable in variables.items():
+            if key in self.known_names:
+                continue
+            self.known_names.add(key)
+            print(f"publishing homeassistant config for {key}")
+            config_topic = self.get_config_topic(variable)
+            state_topic = self.get_state_topic(variable)
+            command_topic = self.get_command_topic(variable)
 
             # Create the MQTT Discovery payload
             payload = {
-                "name": value.friendly_name,
+                "name": variable.friendly_name,
                 "device": self.device_info,
                 "unique_id": f"solarlife_{key}",
                 "state_topic": state_topic,
             }
 
-            if value.multiplier != 0:
-                payload["unit_of_measurement"] = value.unit
+            if variable.multiplier != 0:
+                payload["unit_of_measurement"] = variable.unit
                 
-            if "daily_energy" in key:
+            if "daily" in key and "Wh" in variable.unit:
                 payload['device_class'] = "energy"
                 payload['state_class'] = "total_increasing"
-            elif "total_energy" in key:
+            elif "total" in key and "Wh" in variable.unit:
                 payload['device_class'] = "energy"
                 payload['state_class'] = "total"
-            elif "voltage" in key:
+            elif "Wh" in variable.unit:
+                payload['device_class'] = "energy"
+                payload['state_class'] = "measurement"
+            elif "V" in variable.unit:
                 payload['device_class'] = "voltage"
                 payload['state_class'] = "measurement"
-            elif "current" in key:
+            elif "A" in variable.unit:
                 payload['device_class'] = "current"
                 payload['state_class'] = "measurement"
-            elif "power" in key:
+            elif "W" in variable.unit:
                 payload['device_class'] = "power"
                 payload['state_class'] = "measurement"
-            elif "temperature" in key:
+            elif "°C" in variable.unit:
                 payload['device_class'] = "temperature"
                 payload['state_class'] = "measurement"
             elif key == "battery_percentage":
                 payload['device_class'] = "battery"
                 payload['state_class'] = "measurement"
+            elif "timing_period" in key or "delay" in key:
+                payload['device_class'] = "duration"
 
-            if value.binary_payload:
-                on, off = value.binary_payload
+            if variable.binary_payload:
+                on, off = variable.binary_payload
                 payload["payload_on"] = on
                 payload["payload_off"] = off
 
             # Handle writable entities
-            if FunctionCodes.WRITE_MEMORY_SINGLE in value.function_codes or \
-               FunctionCodes.WRITE_STATUS_REGISTER in value.function_codes:
+            if FunctionCodes.WRITE_MEMORY_SINGLE in variable.function_codes or \
+               FunctionCodes.WRITE_STATUS_REGISTER in variable.function_codes:
                 payload["command_topic"] = command_topic
 
             # Publish the MQTT Discovery payload
@@ -110,4 +124,18 @@ class MqttSensor(Client):
             state_topic = self.get_state_topic(value)
 
             # Publish the entity state
-            await super(MqttSensor, self).publish(state_topic, payload=str(value.value))
+            await super(MqttSensor, self).publish(state_topic, payload=str(value.value))
+
+    async def subscribe(self, variables: VariableContainer):
+        for key, variable in variables.items():
+            if FunctionCodes.WRITE_MEMORY_SINGLE in value.function_codes or \
+               FunctionCodes.WRITE_STATUS_REGISTER in value.function_codes:
+                command_topic = self.get_command_topic(value)
+                await super(MqttSensor, self).subscribe(topic=command_topic, qos=2)
+    
+    async def get_commands(self) -> ResultContainer:
+        for message in await self.messages:
+            match = re.match(rf"^{self.base_topic}/\w+/{self.sensor_name}/(\w+)/", message.topic)
+            variable_name = match.group(1)
+            variable = variables[variable_name]
+            yield Result(**vars(variable), value=message.payload)

src/protocol.py

@@ -1,6 +1,6 @@
 import struct
 from dataclasses import dataclass
-from typing import Any, List, Union, Tuple
+from typing import Any, List, Union, Tuple, Optional
 
 from .variables import variables, Variable, FunctionCodes
 from .crc import crc16
@@ -33,6 +33,12 @@ class ResultContainer:
     def items(self):
         return self._result_map.items()
 
+    def get(self, key: str) -> Optional[Result]:
+        if isinstance(key, str):
+            return self._result_map.get(key)
+        else:
+            raise TypeError("Key must be a variable name string.")
+
 class LumiaxClient:
     def __init__(self):
         self.device_id = 0xFE
@@ -198,7 +204,9 @@ class LumiaxClient:
 
         return ResultContainer(results)
 
-    def _find_raw_value_by_brute_force(self, variable: Variable, value):
+    def _find_raw_value_by_brute_force(self, variable: Variable, value: str):
+        if variable.multiplier:
+            value = float(value)
         n_bits = 32 if variable.is_32_bit else 16
         if variable.is_signed:
             for i in range(0, 2**(n_bits-1) + 1):

src/variables.py

@@ -1,6 +1,6 @@
 from dataclasses import dataclass
 from enum import Enum
-from typing import Callable, List, Tuple, Union
+from typing import Callable, List, Tuple, Union, Optional
 
 
 class FunctionCodes(Enum):
@@ -47,6 +47,12 @@ class VariableContainer:
 
     def items(self):
         return self._variable_map.items()
+    
+    def get(self, key: str) -> Optional[Variable]:
+        if isinstance(key, str):
+            return self._variable_map.get(key)
+        else:
+            raise TypeError("Key must be a variable name string.")
 
 def _get_functional_status_registers(function_codes: list[int], offset: int):
     return [
@@ -218,7 +224,7 @@ variables = VariableContainer([
     Variable(0x3053, True,  False, [0x04], "kWh", 100, "solar_panel_total_energy", "Total charging capacity", None, None),
     Variable(0x3055, True,  False, [0x04], "kWh", 100, "load_daily_energy", "The electricity consumption of the day", None, None),
     Variable(0x3056, True,  False, [0x04], "kWh", 100, "load_total_energy", "Total electricity consumption", None, None),
-    Variable(0x3058, False, False, [0x04], "Min", 1, "total_light_time_during_the_day", "Total light time during the day", None, None),
+    Variable(0x3058, False, False, [0x04], "min", 1, "total_light_time_during_the_day", "Total light time during the day", None, None),
     Variable(0x309D, False, False, [0x04], "", 1, "run_days", "The number of running days", None, None),
     Variable(0x30A0, False, False, [0x04], "V", 100, "battery_voltage", "Battery voltage", None, None),
     Variable(0x30A1, False, True,  [0x04], "A", 100, "battery_current", "Battery current", None, None),
@@ -311,44 +317,44 @@ variables = VariableContainer([
                     ["Manual", "T0T", "Timing switch"])[x], None),
     Variable(0x902C, False, False, [0x03, 0x06, 0x10], "", 0, "mt_series_manual_control_default", "MT Series manual control mode default setting",
         lambda x: ["On", "Off"][x], ("On", "Off")),
-    Variable(0x902D, False, False, [0x03, 0x06, 0x10], "Min", 1, "mt_series_timing_period_1", "MT Series timing opening period 1",
+    Variable(0x902D, False, False, [0x03, 0x06, 0x10], "min", 1, "mt_series_timing_period_1", "MT Series timing opening period 1",
         lambda x: ((x >> 8) & 0xFF) * 60 + max(x & 0xFF, 59), None),
-    Variable(0x902E, False, False, [0x03, 0x06, 0x10], "Min", 1, "mt_series_timing_period_2", "MT Series timing opening period 2",
+    Variable(0x902E, False, False, [0x03, 0x06, 0x10], "min", 1, "mt_series_timing_period_2", "MT Series timing opening period 2",
         lambda x: ((x >> 8) & 0xFF) * 60 + max(x & 0xFF, 59), None),
     Variable(0x902F, False, False, [0x03, 0x06, 0x10], "sec", 1, "timed_start_time_1_seconds", "Timed start time 1-seconds", None, None),
-    Variable(0x9030, False, False, [0x03, 0x06, 0x10], "Min", 1, "timed_start_time_1_minutes", "Timed start time 1-minute", None, None),
+    Variable(0x9030, False, False, [0x03, 0x06, 0x10], "min", 1, "timed_start_time_1_minutes", "Timed start time 1-minute", None, None),
     Variable(0x9031, False, False, [0x03, 0x06, 0x10], "hour", 1, "timed_start_time_1_hours", "Timed start time 1-hour", None, None),
     Variable(0x9032, False, False, [0x03, 0x06, 0x10], "sec", 1, "timed_off_time_1_seconds", "Timed off time 1-seconds", None, None),
-    Variable(0x9033, False, False, [0x03, 0x06, 0x10], "Min", 1, "timed_off_time_1_minutes", "Timed off time 1-minute", None, None),
+    Variable(0x9033, False, False, [0x03, 0x06, 0x10], "min", 1, "timed_off_time_1_minutes", "Timed off time 1-minute", None, None),
     Variable(0x9034, False, False, [0x03, 0x06, 0x10], "hour", 1, "timed_off_time_1_hours", "Timed off time 1-hour", None, None),
     Variable(0x9035, False, False, [0x03, 0x06, 0x10], "sec", 1, "timed_start_time_2_seconds", "Timed start time 2-seconds", None, None),
-    Variable(0x9036, False, False, [0x03, 0x06, 0x10], "Min", 1, "timed_start_time_2_minutes", "Timed start time 2-minute", None, None),
+    Variable(0x9036, False, False, [0x03, 0x06, 0x10], "min", 1, "timed_start_time_2_minutes", "Timed start time 2-minute", None, None),
     Variable(0x9037, False, False, [0x03, 0x06, 0x10], "hour", 1, "timed_start_time_2_hours", "Timed start time 2-hour", None, None),
     Variable(0x9038, False, False, [0x03, 0x06, 0x10], "sec", 1, "timed_off_time_2_seconds", "Timed off time 2-seconds", None, None),
-    Variable(0x9039, False, False, [0x03, 0x06, 0x10], "Min", 1, "timed_off_time_2_minutes", "Timed off time 2-minute", None, None),
+    Variable(0x9039, False, False, [0x03, 0x06, 0x10], "min", 1, "timed_off_time_2_minutes", "Timed off time 2-minute", None, None),
     Variable(0x903A, False, False, [0x03, 0x06, 0x10], "hour", 1, "timed_off_time_2_hours", "Timed off time 2-hour", None, None),
     Variable(0x903B, False, False, [0x03, 0x06, 0x10], "", 0, "time_control_period_selection", "Time control period selection",
         lambda x: ["1 period", "2 periods"][x], None),
     Variable(0x903C, False, False, [0x03, 0x06, 0x10], "V", 100, "light_controlled_dark_voltage", "Light controlled dark voltage", None, None),
-    Variable(0x903D, False, False, [0x03, 0x06, 0x10], "Min", 1, "day_night_delay_time", "Day/Night delay time", None, None),
+    Variable(0x903D, False, False, [0x03, 0x06, 0x10], "min", 1, "day_night_delay_time", "Day/Night delay time", None, None),
     Variable(0x903E, False, False, [0x03, 0x06, 0x10], "%", 0.1, "dc_series_timing_control_time_1_dimming", "DC series timing control time 1 dimming", None, None),
     Variable(0x903F, False, False, [0x03, 0x06, 0x10], "%", 0.1, "dc_series_timing_control_time_2_dimming", "DC series timing control time 2 dimming", None, None),
-    Variable(0x9040, False, False, [0x03, 0x06, 0x10], "Min", 1.0/30, "dc_series_time_1", "DC Series time 1", None, None),
+    Variable(0x9040, False, False, [0x03, 0x06, 0x10], "min", 1.0/30, "dc_series_time_1", "DC Series time 1", None, None),
     Variable(0x9041, False, False, [0x03, 0x06, 0x10], "%", 0.1, "dc_series_time_1_dimming", "DC Series the time 1 dimming", None, None),
-    Variable(0x9042, False, False, [0x03, 0x06, 0x10], "Min", 1.0/30, "dc_series_time_2", "DC Series time 2", None, None),
+    Variable(0x9042, False, False, [0x03, 0x06, 0x10], "min", 1.0/30, "dc_series_time_2", "DC Series time 2", None, None),
     Variable(0x9043, False, False, [0x03, 0x06, 0x10], "%", 0.1, "dc_series_time_2_dimming", "DC Series the time 2 dimming", None, None),
-    Variable(0x9044, False, False, [0x03, 0x06, 0x10], "Sec", 1.0/30, "dc_series_time_3", "DC Series time 3", None, None),
+    Variable(0x9044, False, False, [0x03, 0x06, 0x10], "sec", 1.0/30, "dc_series_time_3", "DC Series time 3", None, None),
     Variable(0x9045, False, False, [0x03, 0x06, 0x10], "%", 0.1, "dc_series_time_3_dimming", "DC Series the time 3 dimming", None, None),
-    Variable(0x9046, False, False, [0x03, 0x06, 0x10], "Sec", 1.0/30, "dc_series_time_4", "DC Series time 4", None, None),
+    Variable(0x9046, False, False, [0x03, 0x06, 0x10], "sec", 1.0/30, "dc_series_time_4", "DC Series time 4", None, None),
     Variable(0x9047, False, False, [0x03, 0x06, 0x10], "%", 0.1, "dc_series_time_4_dimming", "DC Series the time 4 dimming", None, None),
-    Variable(0x9048, False, False, [0x03, 0x06, 0x10], "Sec", 1.0/30, "dc_series_time_5", "DC Series time 5", None, None),
+    Variable(0x9048, False, False, [0x03, 0x06, 0x10], "sec", 1.0/30, "dc_series_time_5", "DC Series time 5", None, None),
     Variable(0x9049, False, False, [0x03, 0x06, 0x10], "%", 0.1, "dc_series_time_5_dimming", "DC Series the time 5 dimming", None, None),
     Variable(0x904A, False, False, [0x03, 0x06, 0x10], "A", 100, "dc_series_load_current_limit", "DC Series load current limit", None, None),
     Variable(0x904B, False, False, [0x03, 0x06, 0x10], "", 0, "dc_series_auto_dimming", "DC Series auto dimming",
         lambda x: ["Auto dimming", "365 mode", "No dimming", "No dimming"][x & 0xF], None),
     Variable(0x904C, False, False, [0x03, 0x06, 0x10], "V", 100, "dc_series_dimming_voltage", "DC Series dimming voltage", None, None),
     Variable(0x904D, False, False, [0x03, 0x06, 0x10], "%", 1, "dc_series_dimming_percentage", "DC Series dimming percentage", None, None),
-    Variable(0x904E, False, False, [0x03, 0x06, 0x10], "Sec", 0.1, "sensing_delay_off_time", "Sensing delay off time", None, None),
+    Variable(0x904E, False, False, [0x03, 0x06, 0x10], "sec", 0.1, "sensing_delay_off_time", "Sensing delay off time", None, None),
     Variable(0x904F, False, False, [0x03, 0x06, 0x10], "%", 0.1, "infrared_dimming_when_no_people", "Dimming of Infrared Series controller when no people", None, None),
     Variable(0x9052, False, False, [0x03, 0x06, 0x10], "", 0, "light_controlled_switch", "Light controlled switch", None, ("On", "Off")),
     Variable(0x9053, False, False, [0x03, 0x06, 0x10], "V", 100, "light_controlled_daybreak_voltage", "Light-control led daybreak voltage", None, None),

tests/variable_test.py

@@ -24,6 +24,8 @@ class TestVariables(unittest.TestCase):
     def test_indexer(self):
         variable = variables['battery_percentage']
         self.assertEqual('battery_percentage', variable.name)
+        variable = variables.get('battery_percentage')
+        self.assertEqual('battery_percentage', variable.name)
         self.assertIsNotNone(variables.items())
 if __name__ == "__main__":
     unittest.main()