rewrite

chargeamps/chargeamps2mqtt.py

@@ -5,8 +5,7 @@
 import aiohttp
 import aiomqtt
 
-from chargeamps.local import ChargeAmpsLocalClient, LocalChargePoint
-from chargeamps.models import ChargePointType
+from chargeamps.websocket import parse_websocket_message
 
 mqtt_reconnect_interval = 30
 mqtt_broker = "127.0.0.1"
@@ -22,44 +21,20 @@
 ws_receive_timeout = 1
 
 
-async def mqtt_results(
-    mqtt_client: aiomqtt.Client, chargeamps_client: ChargeAmpsLocalClient
-):
+async def mqtt_results(mqtt_client: aiomqtt.Client) -> None:
     logger = logging.getLogger("mqtt_results")
     while True:
         logger.debug("Waiting for results")
         result = await mqtt_results_queue.get()
         logger.debug("Got result: %s", result)
 
         await mqtt_client.publish(
-            topic=f"{mqtt_base_topic}/results", payload=result.encode()
+            topic=f"{mqtt_base_topic}/{str(result.__class__.__name__)}",
+            payload=result.model_dump_json(exclude={"message"}).encode(),
         )
 
-        chargeamps_client.process_message(result)
 
-        await mqtt_client.publish(
-            topic=f"{mqtt_base_topic}/status",
-            payload=chargeamps_client.state.model_dump_json(
-                exclude={"connector_settings", "connector_status"}
-            ).encode(),
-        )
-
-        for n, settings in enumerate(chargeamps_client.state.connector_settings):
-            await mqtt_client.publish(
-                topic=f"{mqtt_base_topic}/connector/{n}/settings",
-                payload=settings.model_dump_json().encode(),
-            )
-
-        for n, status in enumerate(chargeamps_client.state.connector_status):
-            await mqtt_client.publish(
-                topic=f"{mqtt_base_topic}/connector/{n}/status",
-                payload=status.model_dump_json().encode(),
-            )
-
-
-async def mqtt_commands(
-    mqtt_client: aiomqtt.Client, chargeamps_client: ChargeAmpsLocalClient
-):
+async def mqtt_commands(mqtt_client: aiomqtt.Client) -> None:
     logger = logging.getLogger("mqtt_commands")
     await mqtt_client.subscribe(mqtt_commands_topic)
     logger.info("Subscribed to %s", mqtt_commands_topic)
@@ -69,16 +44,16 @@ async def mqtt_commands(
         await mqtt_commands_queue.put(message.payload.decode())
 
 
-async def mqtt_handler(chargeamps_client: ChargeAmpsLocalClient):
+async def mqtt_handler() -> None:
     logger = logging.getLogger("mqtt_handler")
     logger.debug("Starting")
     while True:
         try:
             async with aiomqtt.Client(mqtt_broker) as mqtt_client:
                 logger.info("Connected to %s", mqtt_broker)
                 async with asyncio.TaskGroup() as tg:
-                    tg.create_task(mqtt_commands(mqtt_client, chargeamps_client))
-                    tg.create_task(mqtt_results(mqtt_client, chargeamps_client))
+                    tg.create_task(mqtt_commands(mqtt_client))
+                    tg.create_task(mqtt_results(mqtt_client))
                 logger.debug("Tasks done")
         except aiomqtt.MqttError:
             logger.error(
@@ -88,7 +63,7 @@ async def mqtt_handler(chargeamps_client: ChargeAmpsLocalClient):
             await asyncio.sleep(mqtt_reconnect_interval)
 
 
-async def ws_handler(ws_server: str):
+async def ws_handler(ws_server: str) -> None:
     logger = logging.getLogger("ws_handler")
     logger.debug("Starting")
     while True:
@@ -105,7 +80,16 @@ async def ws_handler(ws_server: str):
                             logger.debug(
                                 "Received result from websocket: %s", result.data
                             )
-                            await mqtt_results_queue.put(result.data)
+                            try:
+                                message = parse_websocket_message(result.data)
+                                await mqtt_results_queue.put(message)
+                            except ValueError as exc:
+                                logger.error(
+                                    "Error parsing message. %s",
+                                    result.data,
+                                    exc_info=exc,
+                                )
+                                pass
                         except asyncio.TimeoutError:
                             pass
                         logger.debug("Checking commands queue")
@@ -125,16 +109,13 @@ async def ws_handler(ws_server: str):
             await asyncio.sleep(ws_reconnect_interval)
 
 
-async def server():
-    chargepoint = LocalChargePoint(url=ws_server, type=ChargePointType.HALO)
-    chargeamps_client = ChargeAmpsLocalClient(chargepoint)
-
+async def server() -> None:
     async with asyncio.TaskGroup() as tg:
-        tg.create_task(mqtt_handler(chargeamps_client))
+        tg.create_task(mqtt_handler())
         tg.create_task(ws_handler(ws_server))
 
 
-def main():
+def main() -> None:
     logging.basicConfig(level=logging.DEBUG)
     asyncio.run(server())
 

chargeamps/websocket.py

@@ -0,0 +1,211 @@
+"""Charge-Amps Local API Client"""
+
+from abc import abstractmethod
+from datetime import datetime
+from typing import Self
+
+from pydantic import BaseModel
+
+
+class WebsocketMessage(BaseModel):
+    message: str
+
+    @classmethod
+    @abstractmethod
+    def from_message(cls, message: str) -> Self:
+        pass
+
+
+class HaloConnectorSettings(WebsocketMessage):
+    connector_one_enabled: bool
+    connector_one_rfid_lock: bool
+    connector_one_max_current: int
+    connector_one_charge_mode: int
+
+    connector_two_enabled: bool
+
+    dimmer: int
+    downlight_enabled: bool
+
+    @classmethod
+    def from_message(cls, message: str) -> Self:
+        parameters = message.split(",")
+        assert len(parameters) == 8
+        assert int(parameters[0]) == 1
+        return cls(
+            message=message,
+            connector_one_enabled=bool(int(parameters[1])),
+            connector_two_enabled=bool(int(parameters[2])),
+            connector_one_rfid_lock=bool(int(parameters[3])),
+            downlight_enabled=bool(int(parameters[4])),
+            connector_one_max_current=int(parameters[5]) // 10,
+            dimmer=int(parameters[6]),
+            connector_one_charge_mode=int(parameters[7]),
+        )
+
+
+class AuraChargePointSettings(WebsocketMessage):
+    charge_point_total_max_current: int
+    three_phase_installation: bool
+    min_current: int
+    max_current: int
+    phase_order: int
+
+    connector_one_max_current: int
+    connector_one_l1_voltage: int
+    connector_one_l2_voltage: int
+    connector_one_l3_voltage: int
+
+    connector_two_max_current: int
+    connector_two_l1_voltage: int
+    connector_two_l2_voltage: int
+    connector_two_l3_voltage: int
+
+    @classmethod
+    def from_message(cls, message: str) -> Self:
+        parameters = message.split(",")
+        assert len(parameters) == 15
+        assert int(parameters[0]) == 104
+        return cls(
+            message=message,
+            charge_point_total_max_current=int(parameters[1]),
+            three_phase_installation=bool(int(parameters[2])),
+            min_current=int(parameters[3]),
+            max_current=int(parameters[4]),
+            phase_order=int(parameters[6]),
+            connector_one_max_current=int(parameters[7]),
+            connector_one_l1_voltage=int(parameters[8]),
+            connector_one_l2_voltage=int(parameters[9]),
+            connector_one_l3_voltage=int(parameters[10]),
+            connector_two_max_current=int(parameters[11]),
+            connector_two_l1_voltage=int(parameters[12]),
+            connector_two_l2_voltage=int(parameters[13]),
+            connector_two_l3_voltage=int(parameters[14]),
+        )
+
+
+class AuraConnectorSettings(WebsocketMessage):
+    connector_one_enabled: bool
+    connector_one_cable_lock: bool
+    connector_one_rfid_lock: bool
+    connector_one_max_current: int
+    connector_one_charge_mode: int
+
+    connector_two_enabled: bool
+    connector_two_cable_lock: bool
+    connector_two_rfid_lock: bool
+    connector_two_max_current: int
+    connector_two_charge_mode: int
+
+    dimmer: int
+
+    @classmethod
+    def from_message(cls, message: str) -> Self:
+        parameters = message.split(",")
+        assert len(parameters) == 12
+        assert int(parameters[0]) == 101
+        return cls(
+            message=message,
+            connector_one_enabled=bool(int(parameters[1])),
+            connector_one_cable_lock=bool(int(parameters[2])),
+            connector_one_rfid_lock=bool(int(parameters[3])),
+            connector_one_max_current=int(parameters[4]) // 10,
+            connector_one_charge_mode=int(parameters[5]),
+            connector_two_enabled=bool(int(parameters[6])),
+            connector_two_cable_lock=bool(int(parameters[7])),
+            connector_two_rfid_lock=bool(int(parameters[8])),
+            connector_two_max_current=int(parameters[9]) // 10,
+            connector_two_charge_mode=int(parameters[10]),
+            dimmer=int(parameters[11]),
+        )
+
+
+class ChargePointStatus(WebsocketMessage):
+    charge_point_id: str
+    last_update: datetime
+
+    @classmethod
+    def from_message(cls, message: str) -> Self:
+        parameters = message.split(",")
+        assert len(parameters) == 3
+        assert int(parameters[0]) == 6
+        return cls(
+            message=message, charge_point_id=parameters[1], last_update=parameters[2]
+        )
+
+
+class WifiConnectionStatus(WebsocketMessage):
+    ssid: str
+    rssi: int
+    connected: bool
+
+    @classmethod
+    def from_message(cls, message: str) -> Self:
+        parameters = message.split(",")
+        assert len(parameters) == 4
+        assert int(parameters[0]) == 7
+        return cls(
+            message=message,
+            ssid=parameters[1],
+            rssi=int(parameters[2]),
+            connected=bool(int(parameters[3])),
+        )
+
+
+class OcppStatus(WebsocketMessage):
+    connected: bool
+    endpoint: str
+
+    @classmethod
+    def from_message(cls, message: str) -> Self:
+        parameters = message.split(",")
+        assert len(parameters) == 3
+        assert int(parameters[0]) == 14
+        return cls(
+            message=message, connected=bool(int(parameters[1])), endpoint=parameters[2]
+        )
+
+
+class ConnectorStatus(WebsocketMessage):
+    power: int | None = None
+    current_l1: float | None = None
+    current_l2: float | None = None
+    current_l3: float | None = None
+    charge_session_energy: int | None = None
+    status: str | None = None
+
+    @classmethod
+    def from_message(cls, message: str) -> Self:
+        parameters = message.split(",")
+        assert len(parameters) == 7
+        assert int(parameters[0]) in [8, 108]
+        return cls(
+            message=message,
+            power=int(parameters[1]),
+            current_l1=int(parameters[2]) / 1000,
+            current_l2=int(parameters[3]) / 1000,
+            current_l3=int(parameters[4]) / 1000,
+            charge_session_energy=int(parameters[5]),
+            status=parameters[6],
+        )
+
+
+WEBSOCKET_MESSAGE_CLASSES = {
+    1: HaloConnectorSettings,
+    104: AuraChargePointSettings,
+    101: AuraConnectorSettings,
+    6: ChargePointStatus,
+    7: WifiConnectionStatus,
+    8: ConnectorStatus,
+    14: OcppStatus,
+    108: ConnectorStatus,
+}
+
+
+def parse_websocket_message(message: str) -> WebsocketMessage:
+    parameters = message.split(",")
+    preamble = int(parameters[0])
+    try:
+        return WEBSOCKET_MESSAGE_CLASSES[preamble].from_message(message)
+    except KeyError as exc:
+        raise ValueError("Unknown message type") from exc

tests/test_local.py

@@ -1,4 +1,5 @@
 from chargeamps.local import ChargeAmpsLocalClient, ChargePointType, LocalChargePoint
+from chargeamps.websocket import parse_websocket_message
 
 MESSAGES_HALO = [
     "1,1,1,0,0,160,0,4",
@@ -13,21 +14,20 @@
 MESSAGES_AURA = ["101,1,1,0,160,4,1,1,0,160,4,100"]
 
 
-def test_local_halo():
-    halo = LocalChargePoint(url="ws://127.0.0.1:8002", type=ChargePointType.HALO)
-    client = ChargeAmpsLocalClient(chargepoint=halo)
-
+def test_messages():
     for message in MESSAGES_AURA:
-        client.state.process_message(message)
+        m = parse_websocket_message(message)
+        print(m)
+    for message in MESSAGES_HALO:
+        m = parse_websocket_message(message)
+        print(m)
 
-    print(client.state)
+
+def test_local_halo():
+    halo = LocalChargePoint(url="ws://127.0.0.1:8002", type=ChargePointType.HALO)
+    _ = ChargeAmpsLocalClient(chargepoint=halo)
 
 
 def test_local_aura():
     aura = LocalChargePoint(url="ws://127.0.0.1:8002", type=ChargePointType.AURA)
-    client = ChargeAmpsLocalClient(chargepoint=aura)
-
-    for message in MESSAGES_AURA:
-        client.state.process_message(message)
-
-    print(client.state)
+    _ = ChargeAmpsLocalClient(chargepoint=aura)