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:support@chargeamps.com) for extensive API documentation and API key.
 
-
 ## References
 
 - [Charge Amps External REST API](https://eapi.charge.space/swagger/)

chargeamps/external.py

@@ -17,6 +17,8 @@
     ChargePointStatus,
     ChargingSession,
     StartAuth,
+    User,
+    Partner,
 )
 
 API_BASE_URL = "https://eapi.charge.space"
@@ -47,6 +49,7 @@ def __init__(
         self._refresh_token = None
         self._token_skew = 30
         self._token_lock = asyncio.Lock()
+        self._user: User | None = None
 
     async def shutdown(self) -> None:
         if self._owns_client:
@@ -80,6 +83,7 @@ async def _exclusive_ensure_token(self) -> None:
             except (httpx.HTTPStatusError, httpx.RequestError):
                 self._logger.warning("Token refresh failed")
                 self._token = None
+                self._user = None
                 self._refresh_token = None
         else:
             self._token = None
@@ -97,6 +101,7 @@ async def _exclusive_ensure_token(self) -> None:
             except (httpx.HTTPStatusError, httpx.RequestError) as exc:
                 self._logger.error("Login failed")
                 self._token = None
+                self._user = None
                 self._refresh_token = None
                 self._token_expire = 0
                 raise exc
@@ -107,6 +112,9 @@ async def _exclusive_ensure_token(self) -> None:
 
         response_payload = response.json()
         self._token = response_payload["token"]
+        user_payload = response_payload.get("user")
+        if user_payload is not None:
+            self._user = User.model_validate(user_payload)
         self._refresh_token = response_payload.get("refreshToken", self._refresh_token)
         token_payload = jwt.decode(self._token, options={"verify_signature": False})
         self._token_expire = int(token_payload.get("exp", 0))
@@ -229,6 +237,13 @@ async def set_chargepoint_connector_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 = response.json()
+        return Partner.model_validate(payload)
+
     async def remote_start(
         self, charge_point_id: str, connector_id: int, start_auth: StartAuth
     ) -> None:
@@ -246,3 +261,9 @@ async def reboot(self, charge_point_id: str) -> 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 self._user is None:
+            raise ValueError("No user is currently logged in")
+        return self._user

chargeamps/models.py

@@ -12,6 +12,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,
@@ -91,3 +101,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: list[RfidTag] | None
+    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: str | None
+    connector_id: int
+    user_id: str
+    rfid: str | None
+    rfidDec: str | None
+    rfidDecReverse: str | None
+    organisation_id: str | None
+    session_type: str
+    start_time: CustomDateTime | None = None
+    end_time: CustomDateTime | None = None
+    external_transaction_id: str | None
+    total_consumption_kwh: float
+    external_id: str | None