Try different operator ids to get results for more stations

Author: functionpointer

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "pychargecloud"
-version = "0.0.5"
+version = "0.1.0"
 dependencies = [
     "aiohttp",
     "yarl",

src/chargecloudapi/api.py

@@ -1,24 +1,86 @@
-import aiohttp.client_exceptions
+import datetime
+
 import pydantic
 from aiohttp import ClientSession, ClientResponse
 from yarl import URL
 from .models import *
 import logging
 
-DEFAULT_URL = "https://app.chargecloud.de/emobility:ocpi/606a0da0dfdd338ee4134605653d4fd8/app/2.0/locations"
+operatorIDs = [
+    OperatorID(name="maingau", op_id="606a0da0dfdd338ee4134605653d4fd8"),
+    OperatorID(name="SW Kiel", op_id="6336fe713f2eb7fa04b97ff6651b76f8"),
+    OperatorID(name="Rheinenergie", op_id="c4ce9bb82a86766833df8a4818fa1b5c"),
+]
 
 
 class Api:
     """Class to make authenticated requests."""
 
-    def __init__(self, websession: ClientSession, base_url: str | None = None):
+    def __init__(self, websession: ClientSession):
         """Initialize the auth."""
         self.websession = websession
-        self.base_url = URL(base_url or DEFAULT_URL)
         self.logger = logging.getLogger("chargecloudapi")
 
-    async def location_by_evse_id(self, evse_id: str, **kwargs) -> list[Location]:
-        response = await self.request("GET", self.base_url % {"evse": evse_id})
+    async def perform_smart_api_call(
+        self, evse_id: str, call_data: SmartCallData | None
+    ) -> tuple[Location | None, SmartCallData]:
+        """
+        Perform API request with automatic operatorId choice.
+
+        When calling for a specific evse for the first time, set call_data = None.
+        This function will automatically try some operatorIDs and find the best one.
+        It will then return a SmartCallData along with the result.
+
+        If you want to repeat the API request, specify the SmartCallData object for improved efficiency.
+        """
+
+        if call_data is None:
+            call_data = SmartCallData(
+                evse_id=evse_id,
+                last_call=datetime.datetime.now().isoformat(),
+                operator_id=None,
+            )
+        if call_data.evse_id != evse_id:
+            raise ValueError("call_data does not fit to evse_id")
+
+        ret = None
+        if call_data.operator_id is not None:
+            ret = await self.location_by_evse_id(evse_id, call_data.operator_id)
+            if len(ret) == 0:
+                self.logger.warning(f"empty resp from {call_data.operator_id.name}")
+                call_data.operator_id = None
+
+        if call_data.operator_id is None:
+            self.logger.debug(f"trying all operators")
+            for opid in operatorIDs:
+                ret = await self.location_by_evse_id(evse_id=evse_id, operator_id=opid)
+                if len(ret) != 0:
+                    call_data.operator_id = opid
+                    self.logger.info(f"found operator {opid.name}")
+                    break
+        if ret is None:
+            return None, call_data
+        else:
+            return ret[0], call_data
+
+    async def location_by_evse_id(
+        self, evse_id: str, operator_id: str | OperatorID, **kwargs
+    ) -> list[Location]:
+        """
+        Perform API request.
+        Usually yields just one Location object.
+
+        for the operator id, see chargecloudapi.api.operatorIDs for examples
+        """
+        if isinstance(operator_id, OperatorID):
+            operator_id = operator_id.op_id
+        elif not operator_id:
+            raise ValueError("no operator_id given")
+        url_template = "https://app.chargecloud.de/emobility:ocpi/{}/app/2.0/locations"
+
+        url_obj = URL(url_template.format(operator_id))
+        response = await self.request("GET", url_obj % {"evse": evse_id})
+
         response.raise_for_status()
         json = await response.json()
         self.logger.debug(f"got json {json}")

src/chargecloudapi/models.py

@@ -15,7 +15,9 @@ class Coordinates(BaseModel):
 class Connector(BaseModel):
     id: int
     status: Status
-    standard: Literal["IEC_62196_T2", "IEC_62196_T2_COMBO", "DOMESTIC_F", "CHADEMO", "TESLA"]
+    standard: Literal[
+        "IEC_62196_T2", "IEC_62196_T2_COMBO", "DOMESTIC_F", "CHADEMO", "TESLA"
+    ]
     format: Literal["CABLE", "SOCKET"]
     power_type: Literal["AC_1_PHASE", "AC_3_PHASE", "DC"]
     ampere: float
@@ -79,3 +81,14 @@ class Response(BaseModel):
     status_code: str
     status_message: str
     timestamp: DateTimeISO8601
+
+
+class OperatorID(BaseModel):
+    name: str
+    op_id: str
+
+
+class SmartCallData(BaseModel):
+    evse_id: str
+    last_call: DateTimeISO8601
+    operator_id: OperatorID | None

src/main.py

@@ -1,17 +1,13 @@
 import sys
 
 import chargecloudapi
+from chargecloudapi.api import operatorIDs
 import aiohttp
 import asyncio
 import logging
 
 
 async def main():
-    operator_IDs = {
-        "maingau": "606a0da0dfdd338ee4134605653d4fd8",
-        "SW Kiel": "6336fe713f2eb7fa04b97ff6651b76f8",
-        "Rheinenergie": "c4ce9bb82a86766833df8a4818fa1b5c",
-    }
     evse_ids = [
         "DE*REK*E100241*002",
         "DE*REK*E100196*001",
@@ -21,18 +17,20 @@ async def main():
         "DE*UFC*E210004",
         "DE*ION*E207101",
         "DE*EDR*E11000150*2",
+        "DESWME052601",
+        "DE*SWM*E052601",
+        "DE*TNK*E00136*02",
     ]
     results: dict[str, dict[str, list[chargecloudapi.Location]]] = {
         evse_id: {} for evse_id in evse_ids
     }
-    for operator_name, operator_id in operator_IDs.items():
+    for operator_id in operatorIDs:
         async with aiohttp.ClientSession() as session:
-            base_url = f"https://app.chargecloud.de/emobility:ocpi/{operator_id}/app/2.0/locations"
-            api = chargecloudapi.Api(session, base_url=base_url)
+            api = chargecloudapi.Api(session)
 
             for evse_id in evse_ids:
-                locations = await api.location_by_evse_id(evse_id)
-                results[evse_id][operator_name] = locations
+                locations = await api.location_by_evse_id(evse_id, operator_id)
+                results[evse_id][operator_id.name] = locations
 
     for evse_id, res in results.items():
         print(f"evse_id {evse_id}:")