sync

chargeamps/local.py

@@ -1,7 +1,135 @@
 """Charge-Amps Local API Client"""
 
-from .base import ChargeAmpsClient
+import logging
+import uuid
+from typing import Self
+from urllib.parse import urlparse
 
+from pydantic import BaseModel, ConfigDict
 
-class ChargeAmpsLocalClient(ChargeAmpsClient):
-    pass
+from .models import (
+    ChargePointConnectorMode,
+    ChargePointConnectorSettings,
+    ChargePointType,
+)
+
+DEFAULT_HOST = "localhost"
+DEFAULT_PORT = 8080
+
+
+class LocalChargePoint(BaseModel):
+    name: str
+    host: str
+    port: int
+    pin: str
+
+    model_config = ConfigDict(frozen=True)
+
+    @classmethod
+    def from_url(cls, url: str, pin: str, name: str | None = None) -> Self:
+        res = urlparse(url)
+        if res.scheme != "ws":
+            raise ValueError("Only ws URLs supported")
+        return cls(
+            name=name or str(uuid.uuid4()),
+            host=res.hostname or DEFAULT_HOST,
+            port=res.port or DEFAULT_PORT,
+            pin=pin,
+        )
+
+
+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
+
+    @staticmethod
+    def get_default_halo_connectors() -> list[ChargePointConnectorSettings]:
+        return [
+            ChargePointConnectorSettings(
+                charge_point_id="",
+                connector_id=0,
+                mode=str(ChargePointConnectorMode.OFF),
+                rfid_lock=False,
+                cable_lock=False,
+                max_current=None,
+            ),
+            ChargePointConnectorSettings(
+                charge_point_id="",
+                connector_id=1,
+                mode=str(ChargePointConnectorMode.OFF),
+                rfid_lock=False,
+                cable_lock=False,
+                max_current=None,
+            ),
+        ]
+
+    @staticmethod
+    def get_default_aura_connectors() -> list[ChargePointConnectorSettings]:
+        return [
+            ChargePointConnectorSettings(
+                charge_point_id="",
+                connector_id=0,
+                mode=str(ChargePointConnectorMode.OFF),
+                rfid_lock=False,
+                cable_lock=False,
+                max_current=None,
+            ),
+            ChargePointConnectorSettings(
+                charge_point_id="",
+                connector_id=1,
+                mode=str(ChargePointConnectorMode.OFF),
+                rfid_lock=False,
+                cable_lock=False,
+                max_current=None,
+            ),
+        ]
+
+    def update_settings(self, message: str) -> None:
+        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
+
+
+class ChargeAmpsLocalClient:  # (ChargeAmpsClient):
+    def __init__(self, chargepoints: list[LocalChargePoint]) -> None:
+        self.chargepoints: list[LocalChargePointStatus] = [
+            LocalChargePointStatus(chargepoint=cp) for cp in chargepoints
+        ]

chargeamps/models.py

@@ -1,6 +1,7 @@
 """Data models for ChargeAmps API"""
 
 from datetime import datetime
+from enum import Enum
 from typing import Optional
 
 from pydantic import BaseModel, ConfigDict, PlainSerializer
@@ -15,22 +16,32 @@
 ]
 
 
-class FrozenBaseSchema(BaseModel):
+class ChargePointType(Enum):
+    HALO = 1
+    AURA = 2
+
+
+class ChargePointConnectorMode(Enum):
+    OFF = "Off"
+    ON = "On"
+    SCHEDULE = "Schedule"
+
+
+class BaseSchema(BaseModel):
     model_config = ConfigDict(
         alias_generator=to_camel,
         populate_by_name=True,
         from_attributes=True,
-        frozen=True,
     )
 
 
-class ChargePointConnector(FrozenBaseSchema):
+class ChargePointConnector(BaseSchema):
     charge_point_id: str
     connector_id: int
     type: str
 
 
-class ChargePoint(FrozenBaseSchema):
+class ChargePoint(BaseSchema):
     id: str
     name: str
     password: str
@@ -41,13 +52,13 @@ class ChargePoint(FrozenBaseSchema):
     connectors: list[ChargePointConnector]
 
 
-class ChargePointMeasurement(FrozenBaseSchema):
+class ChargePointMeasurement(BaseSchema):
     phase: str
     current: float
     voltage: float
 
 
-class ChargePointConnectorStatus(FrozenBaseSchema):
+class ChargePointConnectorStatus(BaseSchema):
     charge_point_id: str
     connector_id: int
     total_consumption_kwh: float
@@ -58,19 +69,19 @@ class ChargePointConnectorStatus(FrozenBaseSchema):
     session_id: Optional[int] = None
 
 
-class ChargePointStatus(FrozenBaseSchema):
+class ChargePointStatus(BaseSchema):
     id: str
     status: str
     connector_statuses: list[ChargePointConnectorStatus]
 
 
-class ChargePointSettings(FrozenBaseSchema):
+class ChargePointSettings(BaseSchema):
     id: str
     dimmer: str
     down_light: bool
 
 
-class ChargePointConnectorSettings(FrozenBaseSchema):
+class ChargePointConnectorSettings(BaseSchema):
     charge_point_id: str
     connector_id: int
     mode: str
@@ -79,7 +90,7 @@ class ChargePointConnectorSettings(FrozenBaseSchema):
     max_current: Optional[float] = None
 
 
-class ChargingSession(FrozenBaseSchema):
+class ChargingSession(BaseSchema):
     id: int
     charge_point_id: str
     connector_id: int
@@ -89,7 +100,7 @@ class ChargingSession(FrozenBaseSchema):
     end_time: Optional[CustomDateTime] = None
 
 
-class StartAuth(FrozenBaseSchema):
+class StartAuth(BaseSchema):
     rfid_length: int
     rfid_format: str
     rfid: str

pyproject.toml

@@ -27,12 +27,16 @@ classifiers = [
 chargeamps = "chargeamps.cli:main"
 
 [tool.poetry.dependencies]
-python = "^3.9"
+python = "^3.11"
 aiohttp = "^3.9.5"
 PyJWT = "^2.1.0"
 isoduration = "^20.11.0"
 ciso8601 = "^2.2.0"
 pydantic = "^2.7.4"
+aiomqtt = {version = "^2.1.0", optional = true}
+
+[tool.poetry.extras]
+mqtt = ["aiomqtt"]
 
 [tool.poetry.group.dev.dependencies]
 pytest = "^7.4.3"