sync

chargeamps/local.py

@@ -1,6 +1,5 @@
 """Charge-Amps Local API Client"""
 
-import logging
 import uuid
 from typing import Self
 from urllib.parse import urlparse
@@ -22,11 +21,14 @@ class LocalChargePoint(BaseModel):
     host: str
     port: int
     pin: str
+    type: ChargePointType
 
     model_config = ConfigDict(frozen=True)
 
     @classmethod
-    def from_url(cls, url: str, pin: str, name: str | None = None) -> Self:
+    def from_url(
+        cls, url: str, pin: str, type: ChargePointType, name: str | None = None
+    ) -> Self:
         res = urlparse(url)
         if res.scheme != "ws":
             raise ValueError("Only ws URLs supported")
@@ -35,19 +37,21 @@ def from_url(cls, url: str, pin: str, name: str | None = None) -> Self:
             host=res.hostname or DEFAULT_HOST,
             port=res.port or DEFAULT_PORT,
             pin=pin,
+            type=type,
         )
 
 
 class LocalChargePointStatus(BaseModel):
     chargepoint: LocalChargePoint
-    type: ChargePointType | None = None
     charge_point_id: str | None = None
     connector_settings: list[ChargePointConnectorSettings] | None = None
     dimmer: int = 0
     light: bool = False
 
+
+class LocalChargePointStatusHalo(LocalChargePointStatus):
     @staticmethod
-    def get_default_halo_connectors() -> list[ChargePointConnectorSettings]:
+    def get_default_connectors() -> list[ChargePointConnectorSettings]:
         return [
             ChargePointConnectorSettings(
                 charge_point_id="",
@@ -67,8 +71,24 @@ def get_default_halo_connectors() -> list[ChargePointConnectorSettings]:
             ),
         ]
 
+    def update_settings(self, message: str):
+        parameters = [int(x) for x in message.split(",")]
+
+        assert parameters[0] == 1
+        assert len(parameters) == 8
+
+        self.connector_settings[0].mode = "On" if parameters[1] else "Off"
+        self.connector_settings[1].mode = "On" if parameters[2] else "Off"
+        self.connector_settings[0].rfid_lock = bool(parameters[3])
+        self.light = bool(parameters[4])
+        self.connector_settings[0].max_current = parameters[5] // 10
+        self.dimmer = bool(parameters[6])
+        assert parameters[7] == 4
+
+
+class LocalChargePointStatusAura(LocalChargePointStatus):
     @staticmethod
-    def get_default_aura_connectors() -> list[ChargePointConnectorSettings]:
+    def get_default_connectors() -> list[ChargePointConnectorSettings]:
         return [
             ChargePointConnectorSettings(
                 charge_point_id="",
@@ -88,48 +108,43 @@ def get_default_aura_connectors() -> list[ChargePointConnectorSettings]:
             ),
         ]
 
-    def update_settings(self, message: str) -> None:
+    def update_settings(self, message: str):
         parameters = [int(x) for x in message.split(",")]
 
-        match parameters[0]:
-            case 1:
-                self.type = ChargePointType.HALO
-                if self.connector_settings is None:
-                    self.connector_settings = self.get_default_halo_connectors()
-                assert len(parameters) == 8
-                self.connector_settings[0].mode = "On" if parameters[1] else "Off"
-                self.connector_settings[1].mode = "On" if parameters[2] else "Off"
-                self.connector_settings[0].rfid_lock = bool(parameters[3])
-                self.light = bool(parameters[4])
-                self.connector_settings[0].max_current = parameters[5] // 10
-                self.dimmer = bool(parameters[6])
-                assert parameters[7] == 4
-
-            case 101:
-                self.type = ChargePointType.AURA
-                if self.connector_settings is None:
-                    self.connector_settings = self.get_default_aura_connectors()
-                assert len(parameters) == 12
-                self.connector_settings[0].mode = "On" if parameters[1] else "Off"
-                self.connector_settings[0].cable_lock = bool(parameters[2])
-                self.connector_settings[0].rfid_lock = bool(parameters[3])
-                self.connector_settings[0].max_current = parameters[4] // 10
-                assert parameters[5] == 4
-                self.connector_settings[1].mode = "On" if parameters[6] else "Off"
-                self.connector_settings[1].cable_lock = bool(parameters[7])
-                self.connector_settings[1].rfid_lock = bool(parameters[8])
-                self.connector_settings[1].max_current = parameters[9] // 10
-                assert parameters[10] == 4
-                self.dimmer = parameters[11]
-            case _:
-                logging.warning("Unknown settings preamble: %d", parameters[0])
-
-    def update_status(self, message: str) -> None:
-        pass
+        assert parameters[0] == 101
+        assert len(parameters) == 12
+
+        self.connector_settings[0].mode = "On" if parameters[1] else "Off"
+        self.connector_settings[0].cable_lock = bool(parameters[2])
+        self.connector_settings[0].rfid_lock = bool(parameters[3])
+        self.connector_settings[0].max_current = parameters[4] // 10
+        assert parameters[5] == 4
+        self.connector_settings[1].mode = "On" if parameters[6] else "Off"
+        self.connector_settings[1].cable_lock = bool(parameters[7])
+        self.connector_settings[1].rfid_lock = bool(parameters[8])
+        self.connector_settings[1].max_current = parameters[9] // 10
+        assert parameters[10] == 4
+        self.dimmer = parameters[11]
 
 
 class ChargeAmpsLocalClient:  # (ChargeAmpsClient):
     def __init__(self, chargepoints: list[LocalChargePoint]) -> None:
-        self.chargepoints: list[LocalChargePointStatus] = [
-            LocalChargePointStatus(chargepoint=cp) for cp in chargepoints
-        ]
+        self.chargepoints: list[LocalChargePoint] = chargepoints
+        self.status: list[LocalChargePointStatus] = []
+
+        for cp in self.chargepoints:
+            match cp.type:
+                case ChargePointType.HALO:
+                    self.status.append(
+                        LocalChargePointStatusHalo(
+                            chargepoint=cp,
+                            connector_settings=LocalChargePointStatusHalo.get_default_connectors(),
+                        )
+                    )
+                case ChargePointType.AURA:
+                    self.status.append(
+                        LocalChargePointStatusAura(
+                            chargepoint=cp,
+                            connector_settings=LocalChargePointStatusAura.get_default_connectors(),
+                        )
+                    )

tests/test_local.py

@@ -1,13 +1,17 @@
-from chargeamps.local import ChargeAmpsLocalClient, LocalChargePoint
+from chargeamps.local import ChargeAmpsLocalClient, ChargePointType, LocalChargePoint
 
 SETTINGS_AURA = "101,1,1,0,160,4,1,1,0,160,4,100"
 SETTINGS_HALO = "1,1,1,0,0,160,0,4"
 
 
 def test_local():
-    aura = LocalChargePoint.from_url("ws://127.0.0.1:8001", pin="12345678")
-    halo = LocalChargePoint.from_url("ws://127.0.0.1:8002", pin="12345678")
+    aura = LocalChargePoint.from_url(
+        "ws://127.0.0.1:8001", pin="12345678", type=ChargePointType.AURA
+    )
+    halo = LocalChargePoint.from_url(
+        "ws://127.0.0.1:8002", pin="12345678", type=ChargePointType.HALO
+    )
     client = ChargeAmpsLocalClient(chargepoints=[aura, halo])
 
-    client.chargepoints[0].update_settings(SETTINGS_AURA)
-    client.chargepoints[1].update_settings(SETTINGS_HALO)
+    client.status[0].update_settings(SETTINGS_AURA)
+    client.status[1].update_settings(SETTINGS_HALO)