Adds organisation API support

README.md

@@ -4,12 +4,24 @@ This repository contains a Python module for the Charge Amps' electric vehicle c
 
 The module is developed by [Kirei AB](https://www.kirei.se) and is not supported by [Charge Amps AB](https://chargeamps.com).
 
+## How to use
+
+The simplest way is to install via `pip`, as in:
+
+```
+pip install chargeamps
+```
+
+If you need access to the organisation API calls, you need to specify this feature upon installation:
+
+```
+pip install chargeamps["organisations"]
+```
 
 ## External API Key
 
 You need an API key to use the Charge Amps external API. Contact [Charge Amps Support](mailto:[email protected]) for extensive API documentation and API key.
 
-
 ## References
 
 - [Charge Amps External REST API](https://eapi.charge.space/swagger/)

chargeamps/external.py

@@ -19,6 +19,8 @@
     ChargePointStatus,
     ChargingSession,
     StartAuth,
+    User,
+    Partner,
 )
 
 API_BASE_URL = "https://eapi.charge.space"
@@ -44,6 +46,7 @@ def __init__(
         self._token = None
         self._token_expire = 0
         self._refresh_token = None
+        self._user: User = None
 
     async def shutdown(self) -> None:
         await self._session.close()
@@ -72,6 +75,7 @@ async def _ensure_token(self) -> None:
             except HTTPException:
                 self._logger.warning("Token refresh failed")
                 self._token = None
+                self._user = None
                 self._refresh_token = None
         else:
             self._token = None
@@ -89,6 +93,7 @@ async def _ensure_token(self) -> None:
             except HTTPException as exc:
                 self._logger.error("Login failed")
                 self._token = None
+                self._user = None
                 self._refresh_token = None
                 self._token_expire = 0
                 raise exc
@@ -100,6 +105,7 @@ async def _ensure_token(self) -> None:
         response_payload = await response.json()
 
         self._token = response_payload["token"]
+        self._user = response_payload["user"]
         self._refresh_token = response_payload["refreshToken"]
 
         token_payload = jwt.decode(self._token, options={"verify_signature": False})
@@ -209,6 +215,13 @@ async def set_chargepoint_connector_settings(
         request_uri = f"/api/{API_VERSION}/chargepoints/{charge_point_id}/connectors/{connector_id}/settings"
         await self._put(request_uri, json=payload)
 
+    async def get_partner(self, charge_point_id: str) -> Partner:
+        """Get partner details"""
+        request_uri = f"/api/{API_VERSION}/chargepoints/{charge_point_id}/partner"
+        response = await self._get(request_uri)
+        payload = await response.json()
+        return Partner.model_validate(payload)
+
     async def remote_start(
         self, charge_point_id: str, connector_id: int, start_auth: StartAuth
     ) -> None:
@@ -226,3 +239,16 @@ async def reboot(self, charge_point_id) -> None:
         """Reboot chargepoint"""
         request_uri = f"/api/{API_VERSION}/chargepoints/{charge_point_id}/reboot"
         await self._put(request_uri, json="{}")
+
+    async def get_logged_in_user(self) -> User:
+        """Get authenticated user info"""
+        if not self._user or not isinstance(self._user, dict):
+            raise ValueError("No user is currently logged in")
+
+        user_id = self._user["id"]
+
+        request_uri = f"/api/{API_VERSION}/users/{user_id}"
+        response = await self._get(request_uri)
+        payload = await response.json()
+
+        return User.model_validate(payload)

chargeamps/models.py

@@ -15,6 +15,16 @@
 ]
 
 
+def feature_required(feature_flag):
+    def decorator(cls):
+        if feature_flag == "organisations":
+            return cls
+        else:
+            raise ImportError(f"Feature '{feature_flag}' is not enabled")
+
+    return decorator
+
+
 class FrozenBaseSchema(BaseModel):
     model_config = ConfigDict(
         alias_generator=to_camel,
@@ -94,3 +104,61 @@ class StartAuth(FrozenBaseSchema):
     rfid_format: str
     rfid: str
     external_transaction_id: str
+
+
+class RfidTag(FrozenBaseSchema):
+    active: bool
+    rfid: str | None
+    rfidDec: str | None
+    rfidDecReverse: str | None
+
+
+class User(FrozenBaseSchema):
+    id: str
+    first_name: str | None
+    last_name: str | None
+    email: str | None
+    mobile: str | None
+    rfid_tags: Optional[list[RfidTag]]
+    user_status: str
+
+
+class Partner(FrozenBaseSchema):
+    id: int
+    name: str
+    description: str
+    email: str
+    phone: str
+
+
+# Only way to register new RFID seems to be through an Organization
+@feature_required("organisations")
+class Rfid(FrozenBaseSchema):
+    rfid: str | None
+    rfidDec: str | None
+    rfidDecReverse: str | None
+
+
+@feature_required("organisations")
+class Organisation(FrozenBaseSchema):
+    id: str
+    name: str
+    description: str
+
+
+@feature_required("organisations")
+class OrganisationChargingSession(FrozenBaseSchema):
+    id: int
+    charge_point_id: Optional[str]
+    connector_id: int
+    user_id: str
+    rfid: Optional[str]
+    rfidDec: Optional[str]
+    rfidDecReverse: Optional[str]
+    organisation_id: Optional[str]
+    session_type: str
+    start_time: Optional[CustomDateTime] = None
+    end_time: Optional[CustomDateTime] = None
+    external_transaction_id: Optional[str]
+    total_consumption_kwh: float
+    external_id: Optional[str]