Skip to content

Commit

Permalink
use pydantic and move to APIv5
Browse files Browse the repository at this point in the history
  • Loading branch information
jschlyter committed Jun 26, 2024
1 parent 08386c8 commit 966f28d
Show file tree
Hide file tree
Showing 5 changed files with 190 additions and 172 deletions.
18 changes: 11 additions & 7 deletions chargeamps/external.py
Original file line number Diff line number Diff line change
Expand Up @@ -20,7 +20,7 @@
)

API_BASE_URL = "https://eapi.charge.space"
API_VERSION = "v4"
API_VERSION = "v5"


class ChargeAmpsExternalClient(ChargeAmpsClient):
Expand Down Expand Up @@ -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(
Expand All @@ -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(
Expand All @@ -114,21 +114,24 @@ 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"""
Expand All @@ -143,7 +146,8 @@ 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
Expand Down
74 changes: 37 additions & 37 deletions chargeamps/models.py
Original file line number Diff line number Diff line change
@@ -1,25 +1,39 @@
"""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,
)

def to_dict(self) -> dict:
return self.dict(by_alias=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
Expand All @@ -30,46 +44,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
Expand All @@ -78,21 +82,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
Expand Down
28 changes: 0 additions & 28 deletions chargeamps/utils.py

This file was deleted.

Loading

0 comments on commit 966f28d

Please sign in to comment.