Pydantic

chargeamps/cli.py

@@ -33,7 +33,7 @@ async def command_list_chargepoints(
 ) -> None:
     res = []
     for cp in await client.get_chargepoints():
-        res.append(cp.to_dict())
+        res.append(cp.model_dump(by_alias=True))
     print(json.dumps(res, indent=4))
 
 
@@ -45,9 +45,9 @@ async def command_get_chargepoint_status(
     if args.connector_id:
         for c in cp.connector_statuses:
             if c.connector_id == args.connector_id:
-                print(json.dumps(c.to_dict(), indent=4))
+                print(json.dumps(c.model_dump(by_alias=True), indent=4))
     else:
-        print(json.dumps(cp.to_dict(), indent=4))
+        print(json.dumps(cp.model_dump(by_alias=True), indent=4))
 
 
 async def command_get_chargepoint_sessions(
@@ -68,7 +68,7 @@ async def command_get_chargepoint_sessions(
             charge_point_id, start_time, end_time
         ):
             if args.connector_id is None or args.connector_id == session.connector_id:
-                res.append(session.to_dict())
+                res.append(session.model_dump(by_alias=True))
         res = sorted(res, key=lambda i: i["id"])
         print(json.dumps(res, indent=4))
 
@@ -78,7 +78,7 @@ async def command_get_chargepoint_settings(
 ) -> None:
     charge_point_id = await get_chargepoint_id(client, args)
     settings = await client.get_chargepoint_settings(charge_point_id)
-    print(json.dumps(settings.to_dict(), indent=4))
+    print(json.dumps(settings.model_dump(by_alias=True), indent=4))
 
 
 async def command_set_chargepoint_settings(
@@ -92,7 +92,7 @@ async def command_set_chargepoint_settings(
         settings.down_light = args.downlight
     await client.set_chargepoint_settings(settings)
     settings = await client.get_chargepoint_settings(charge_point_id)
-    print(json.dumps(settings.to_dict(), indent=4))
+    print(json.dumps(settings.model_dump(by_alias=True), indent=4))
 
 
 async def command_get_connector_settings(
@@ -109,7 +109,7 @@ async def command_get_connector_settings(
         settings = await client.get_chargepoint_connector_settings(
             charge_point_id, connector_id
         )
-        res.append(settings.to_dict())
+        res.append(settings.model_dump(by_alias=True))
     print(json.dumps(res, indent=4))
 
 
@@ -133,7 +133,7 @@ async def command_set_connector_settings(
     settings = await client.get_chargepoint_connector_settings(
         charge_point_id, connector_id
     )
-    print(json.dumps(settings.to_dict(), indent=4))
+    print(json.dumps(settings.model_dump(by_alias=True), indent=4))
 
 
 def add_arg_chargepoint(parser, required=False) -> None:

chargeamps/external.py

@@ -20,7 +20,7 @@
 )
 
 API_BASE_URL = "https://eapi.charge.space"
-API_VERSION = "v4"
+API_VERSION = "v5"
 
 
 class ChargeAmpsExternalClient(ChargeAmpsClient):
@@ -84,7 +84,7 @@ async def get_chargepoints(self) -> list[ChargePoint]:
         response = await self._get(request_uri)
         res = []
         for chargepoint in await response.json():
-            res.append(ChargePoint.from_dict(chargepoint))
+            res.append(ChargePoint.model_validate(chargepoint))
         return res
 
     async def get_all_chargingsessions(
@@ -105,7 +105,7 @@ async def get_all_chargingsessions(
         response = await self._get(request_uri, params=query_params)
         res = []
         for session in await response.json():
-            res.append(ChargingSession.from_dict(session))
+            res.append(ChargingSession.model_validate(session))
         return res
 
     async def get_chargingsession(
@@ -114,25 +114,28 @@ async def get_chargingsession(
         """Get charging session"""
         request_uri = f"/api/{API_VERSION}/chargepoints/{charge_point_id}/chargingsessions/{session}"
         response = await self._get(request_uri)
-        return ChargingSession.from_dict(await response.json())
+        payload = await response.json()
+        return ChargingSession.model_validate(payload)
 
     async def get_chargepoint_status(self, charge_point_id: str) -> ChargePointStatus:
         """Get charge point status"""
         request_uri = f"/api/{API_VERSION}/chargepoints/{charge_point_id}/status"
         response = await self._get(request_uri)
-        return ChargePointStatus.from_dict(await response.json())
+        payload = await response.json()
+        return ChargePointStatus.model_validate(payload)
 
     async def get_chargepoint_settings(
         self, charge_point_id: str
     ) -> ChargePointSettings:
         """Get chargepoint settings"""
         request_uri = f"/api/{API_VERSION}/chargepoints/{charge_point_id}/settings"
         response = await self._get(request_uri)
-        return ChargePointSettings.from_dict(await response.json())
+        payload = await response.json()
+        return ChargePointSettings.model_validate(payload)
 
     async def set_chargepoint_settings(self, settings: ChargePointSettings) -> None:
         """Set chargepoint settings"""
-        payload = settings.to_dict()
+        payload = settings.model_dump(by_alias=True)
         charge_point_id = settings.id
         request_uri = f"/api/{API_VERSION}/chargepoints/{charge_point_id}/settings"
         await self._put(request_uri, json=payload)
@@ -143,13 +146,14 @@ async def get_chargepoint_connector_settings(
         """Get all owned chargepoints"""
         request_uri = f"/api/{API_VERSION}/chargepoints/{charge_point_id}/connectors/{connector_id}/settings"
         response = await self._get(request_uri)
-        return ChargePointConnectorSettings.from_dict(await response.json())
+        payload = await response.json()
+        return ChargePointConnectorSettings.model_validate(payload)
 
     async def set_chargepoint_connector_settings(
         self, settings: ChargePointConnectorSettings
     ) -> None:
         """Get all owned chargepoints"""
-        payload = settings.to_dict()
+        payload = settings.model_dump(by_alias=True)
         charge_point_id = settings.charge_point_id
         connector_id = settings.connector_id
         request_uri = f"/api/{API_VERSION}/chargepoints/{charge_point_id}/connectors/{connector_id}/settings"
@@ -159,7 +163,7 @@ async def remote_start(
         self, charge_point_id: str, connector_id: int, start_auth: StartAuth
     ) -> None:
         """Remote start chargepoint"""
-        payload = start_auth.to_dict()
+        payload = start_auth.model_dump(by_alias=True)
         request_uri = f"/api/{API_VERSION}/chargepoints/{charge_point_id}/connectors/{connector_id}/remotestart"
         await self._put(request_uri, json=payload)
 

chargeamps/models.py

@@ -1,25 +1,36 @@
-"""Base class and data classes for ChargeAmps API"""
+"""Data models for ChargeAmps API"""
 
-from dataclasses import dataclass
 from datetime import datetime
 from typing import Optional
 
-from dataclasses_json import LetterCase, dataclass_json
+from pydantic import BaseModel, ConfigDict, PlainSerializer
+from pydantic.alias_generators import to_camel
+from typing_extensions import Annotated
 
-from .utils import datetime_field
+CustomDateTime = Annotated[
+    datetime,
+    PlainSerializer(
+        lambda _datetime: _datetime.strftime("%Y-%m-%dT%H:%M:%SZ"), return_type=str
+    ),
+]
 
 
-@dataclass_json(letter_case=LetterCase.CAMEL)
-@dataclass(frozen=True)
-class ChargePointConnector:
+class FrozenBaseSchema(BaseModel):
+    model_config = ConfigDict(
+        alias_generator=to_camel,
+        populate_by_name=True,
+        from_attributes=True,
+        frozen=True,
+    )
+
+
+class ChargePointConnector(FrozenBaseSchema):
     charge_point_id: str
     connector_id: int
     type: str
 
 
-@dataclass_json(letter_case=LetterCase.CAMEL)
-@dataclass(frozen=True)
-class ChargePoint:
+class ChargePoint(FrozenBaseSchema):
     id: str
     name: str
     password: str
@@ -30,46 +41,36 @@ class ChargePoint:
     connectors: list[ChargePointConnector]
 
 
-@dataclass_json(letter_case=LetterCase.CAMEL)
-@dataclass(frozen=True)
-class ChargePointMeasurement:
+class ChargePointMeasurement(FrozenBaseSchema):
     phase: str
     current: float
     voltage: float
 
 
-@dataclass_json(letter_case=LetterCase.CAMEL)
-@dataclass(frozen=True)
-class ChargePointConnectorStatus:
+class ChargePointConnectorStatus(FrozenBaseSchema):
     charge_point_id: str
     connector_id: int
     total_consumption_kwh: float
     status: str
     measurements: Optional[list[ChargePointMeasurement]]
-    start_time: Optional[datetime] = datetime_field()
-    end_time: Optional[datetime] = datetime_field()
-    session_id: Optional[str] = None
+    start_time: Optional[CustomDateTime] = None
+    end_time: Optional[CustomDateTime] = None
+    session_id: Optional[int] = None
 
 
-@dataclass_json(letter_case=LetterCase.CAMEL)
-@dataclass(frozen=True)
-class ChargePointStatus:
+class ChargePointStatus(FrozenBaseSchema):
     id: str
     status: str
     connector_statuses: list[ChargePointConnectorStatus]
 
 
-@dataclass_json(letter_case=LetterCase.CAMEL)
-@dataclass(frozen=False)
-class ChargePointSettings:
+class ChargePointSettings(FrozenBaseSchema):
     id: str
     dimmer: str
     down_light: bool
 
 
-@dataclass_json(letter_case=LetterCase.CAMEL)
-@dataclass(frozen=False)
-class ChargePointConnectorSettings:
+class ChargePointConnectorSettings(FrozenBaseSchema):
     charge_point_id: str
     connector_id: int
     mode: str
@@ -78,21 +79,17 @@ class ChargePointConnectorSettings:
     max_current: Optional[float] = None
 
 
-@dataclass_json(letter_case=LetterCase.CAMEL)
-@dataclass(frozen=True)
-class ChargingSession:
-    id: str
+class ChargingSession(FrozenBaseSchema):
+    id: int
     charge_point_id: str
     connector_id: int
     session_type: str
     total_consumption_kwh: float
-    start_time: Optional[datetime] = datetime_field()
-    end_time: Optional[datetime] = datetime_field()
+    start_time: Optional[CustomDateTime] = None
+    end_time: Optional[CustomDateTime] = None
 
 
-@dataclass_json(letter_case=LetterCase.CAMEL)
-@dataclass(frozen=True)
-class StartAuth:
+class StartAuth(FrozenBaseSchema):
     rfid_length: int
     rfid_format: str
     rfid: str