Review vehicle endpoints (#1536)

* Add full list of endpoints

* Adjust

* More

* Mypy

* Adjust

* Coverage

* Adjust

* Adjust

Author: epenet

src/renault_api/exceptions.py

@@ -1,5 +1,7 @@
 """Exceptions for Renault API."""
 
+from typing import Optional
+
 
 class RenaultException(Exception):  # noqa: N818
     """Base exception for Renault API errors."""
@@ -11,3 +13,14 @@ class NotAuthenticatedException(RenaultException):  # noqa: N818
     """You are not authenticated, or authentication has expired."""
 
     pass
+
+
+class EndpointNotAvailableError(RenaultException):
+    """The endpoint is not available for this model."""
+
+    def __init__(self, endpoint: str, model_code: Optional[str]) -> None:
+        self.endpoint = endpoint
+        self.model_code = model_code
+
+    def __str__(self) -> str:
+        return f"Endpoint '{self.endpoint}' not available for model '{self.model_code}'"

src/renault_api/kamereon/__init__.py

@@ -50,6 +50,8 @@
 DATA_ENDPOINTS = _KCA_GET_ENDPOINTS
 ACTION_ENDPOINTS = _KCA_POST_ENDPOINTS
 
+ACCOUNT_ENDPOINT_ROOT = "/commerce/v1/accounts/{account_id}/kamereon"
+
 
 def get_commerce_url(root_url: str) -> str:
     """Get the Kamereon base commerce url."""

src/renault_api/kamereon/models.py

@@ -2,6 +2,7 @@
 
 import json
 import logging
+from collections.abc import Mapping
 from dataclasses import dataclass
 from typing import Any
 from typing import Optional
@@ -84,6 +85,49 @@
     },
 }
 
+_DEFAULT_ENDPOINTS: dict[str, str] = {
+    "battery-status": "/kca/car-adapter/v2/cars/{vin}/battery-status",
+    "charge-history": "/kca/car-adapter/v1/cars/{vin}/charge-history",
+    "charge-mode": "/kca/car-adapter/v1/cars/{vin}/charge-mode",
+    "charges": "/kca/car-adapter/v1/cars/{vin}/charges",
+    "charging-settings": "/kca/car-adapter/v1/cars/{vin}/charging-settings",
+    "cockpit": "/kca/car-adapter/v1/cars/{vin}/cockpit",
+    "hvac-history": "/kca/car-adapter/v1/cars/{vin}/hvac-history",
+    "hvac-sessions": "/kca/car-adapter/v1/cars/{vin}/hvac-sessions",
+    "hvac-settings": "/kca/car-adapter/v1/cars/{vin}/hvac-settings",
+    "hvac-status": "/kca/car-adapter/v1/cars/{vin}/hvac-status",
+    "location": "/kca/car-adapter/v1/cars/{vin}/location",
+    "lock-status": "/kca/car-adapter/v1/cars/{vin}/lock-status",
+    "notification-settings": "/kca/car-adapter/v1/cars/{vin}/notification-settings",
+    "pressure": "/kca/car-adapter/v1/cars/{vin}/pressure",
+    "res-state": "/kca/car-adapter/v1/cars/{vin}/res-state",
+}
+
+_VEHICLE_ENDPOINTS: dict[str, dict[str, Optional[str]]] = {
+    "X101VE": {  # ZOE phase 1
+        "battery-status": _DEFAULT_ENDPOINTS["battery-status"],  # confirmed
+        "charge-mode": _DEFAULT_ENDPOINTS["charge-mode"],  # confirmed
+        "cockpit": _DEFAULT_ENDPOINTS["cockpit"],  # confirmed
+        "hvac-status": _DEFAULT_ENDPOINTS["hvac-status"],  # confirmed
+        "location": None,  # not supported
+        "lock-status": None,  # not supported
+        "pressure": None,  # not supported
+        "res-state": None,  # not supported
+    },
+    "XJA1VP": {  # CLIO V
+        "hvac-status": None,
+    },
+    "XJB1SU": {  # CAPTUR II
+        "hvac-status": None,
+    },
+    "XCB1VE": {  # MEGANE E-TECH
+        "lock-status": None,
+    },
+    "XCB1SE": {  # SCENIC E-TECH
+        "lock-status": None,
+    },
+}
+
 
 @dataclass
 class KamereonResponseError(BaseModel):
@@ -261,11 +305,7 @@ def reports_charging_power_in_watts(self) -> bool:
     def supports_endpoint(self, endpoint: str) -> bool:
         """Return True if model supports specified endpoint."""
         # Default to True for unknown vehicles
-        if self.model and self.model.code:
-            return VEHICLE_SPECIFICATIONS.get(  # type:ignore[no-any-return]
-                self.model.code, {}
-            ).get(f"support-endpoint-{endpoint}", True)
-        return True  # pragma: no cover
+        return self.get_endpoint(endpoint) is not None
 
     def warns_on_method(self, method: str) -> Optional[str]:
         """Return warning message if model trigger a warning on the method call."""
@@ -285,6 +325,38 @@ def controls_action_via_kcm(self, action: str) -> bool:
             ).get(f"control-{action}-via-kcm", False)
         return False  # pragma: no cover
 
+    def get_endpoints(self) -> Mapping[str, Optional[str]]:
+        """Return model endpoints."""
+        model_code = self.get_model_code()
+        if not model_code:
+            # Model code not available
+            return _DEFAULT_ENDPOINTS  # pragma: no cover
+
+        if model_code not in _VEHICLE_ENDPOINTS:
+            # Model not documented
+            _LOGGER.warning(
+                "Model %s is not documented, using default endpoints",
+                self.get_model_code(),
+            )
+            return _DEFAULT_ENDPOINTS
+
+        return _VEHICLE_ENDPOINTS[model_code]
+
+    def get_endpoint(self, endpoint: str) -> Optional[str]:
+        """Return model endpoint"""
+        endpoints = self.get_endpoints()
+
+        if endpoint not in endpoints:
+            # Endpoint not documented
+            _LOGGER.warning(
+                "Endpoint %s for model %s is not documented, using default endpoints",
+                endpoint,
+                self.get_model_code(),
+            )
+            return _DEFAULT_ENDPOINTS.get(endpoint)
+
+        return endpoints[endpoint]
+
 
 @dataclass
 class KamereonVehiclesLink(BaseModel):

src/renault_api/renault_session.py

@@ -6,6 +6,7 @@
 from typing import Optional
 
 import aiohttp
+from marshmallow.schema import Schema
 
 from . import gigya
 from . import kamereon
@@ -154,7 +155,12 @@ async def _get_jwt(self) -> str:
                 return jwt
 
     async def http_request(
-        self, method: str, endpoint: str, json: Optional[dict[str, Any]] = None
+        self,
+        method: str,
+        endpoint: str,
+        json: Optional[dict[str, Any]] = None,
+        *,
+        schema: Optional[Schema] = None,
     ) -> models.KamereonResponse:
         """GET to specified endpoint."""
         url = (await self._get_kamereon_root_url()) + endpoint
@@ -167,6 +173,7 @@ async def http_request(
             gigya_jwt=await self._get_jwt(),
             params=params,
             json=json,
+            schema=schema,
         )
 
     async def get_person(self) -> models.KamereonPersonResponse:

src/renault_api/renault_vehicle.py

@@ -10,7 +10,9 @@
 import aiohttp
 
 from .credential_store import CredentialStore
+from .exceptions import EndpointNotAvailableError
 from .exceptions import RenaultException
+from .kamereon import ACCOUNT_ENDPOINT_ROOT
 from .kamereon import models
 from .kamereon import schemas
 from .renault_session import RenaultSession
@@ -77,6 +79,25 @@ def vin(self) -> str:
         """Get vin."""
         return self._vin
 
+    async def _get_vehicle_data(
+        self, endpoint: str
+    ) -> models.KamereonVehicleDataResponse:
+        """GET to /v{endpoint_version}/cars/{vin}/{endpoint}."""
+        details = await self.get_details()
+        full_endpoint = details.get_endpoint(endpoint)
+        if full_endpoint is None:
+            raise EndpointNotAvailableError(endpoint, details.get_model_code())
+
+        full_endpoint = ACCOUNT_ENDPOINT_ROOT.replace(
+            "{account_id}", self.account_id
+        ) + full_endpoint.replace("{vin}", self.vin)
+
+        schema = schemas.KamereonVehicleDataResponseSchema
+        return cast(
+            models.KamereonVehicleDataResponse,
+            await self.session.http_request("GET", full_endpoint, schema=schema),
+        )
+
     async def get_details(self) -> models.KamereonVehicleDetails:
         """Get vehicle details."""
         if self._vehicle_details:
@@ -124,119 +145,79 @@ async def get_contracts(self) -> list[models.KamereonVehicleContract]:
 
     async def get_battery_status(self) -> models.KamereonVehicleBatteryStatusData:
         """Get vehicle battery status."""
-        response = await self.session.get_vehicle_data(
-            account_id=self.account_id,
-            vin=self.vin,
-            endpoint="battery-status",
-        )
+        response = await self._get_vehicle_data("battery-status")
         return cast(
             models.KamereonVehicleBatteryStatusData,
             response.get_attributes(schemas.KamereonVehicleBatteryStatusDataSchema),
         )
 
     async def get_tyre_pressure(self) -> models.KamereonVehicleTyrePressureData:
         """Get vehicle tyre pressure."""
-        response = await self.session.get_vehicle_data(
-            account_id=self.account_id,
-            vin=self.vin,
-            endpoint="pressure",
-        )
+        response = await self._get_vehicle_data("pressure")
         return cast(
             models.KamereonVehicleTyrePressureData,
             response.get_attributes(schemas.KamereonVehicleTyrePressureDataSchema),
         )
 
     async def get_location(self) -> models.KamereonVehicleLocationData:
         """Get vehicle location."""
-        response = await self.session.get_vehicle_data(
-            account_id=self.account_id,
-            vin=self.vin,
-            endpoint="location",
-        )
+        response = await self._get_vehicle_data("location")
         return cast(
             models.KamereonVehicleLocationData,
             response.get_attributes(schemas.KamereonVehicleLocationDataSchema),
         )
 
     async def get_hvac_status(self) -> models.KamereonVehicleHvacStatusData:
         """Get vehicle hvac status."""
-        response = await self.session.get_vehicle_data(
-            account_id=self.account_id,
-            vin=self.vin,
-            endpoint="hvac-status",
-        )
+        response = await self._get_vehicle_data("hvac-status")
         return cast(
             models.KamereonVehicleHvacStatusData,
             response.get_attributes(schemas.KamereonVehicleHvacStatusDataSchema),
         )
 
     async def get_hvac_settings(self) -> models.KamereonVehicleHvacSettingsData:
         """Get vehicle hvac settings (schedule+mode)."""
-        response = await self.session.get_vehicle_data(
-            account_id=self.account_id,
-            vin=self.vin,
-            endpoint="hvac-settings",
-        )
+        response = await self._get_vehicle_data("hvac-settings")
         return cast(
             models.KamereonVehicleHvacSettingsData,
             response.get_attributes(schemas.KamereonVehicleHvacSettingsDataSchema),
         )
 
     async def get_charge_mode(self) -> models.KamereonVehicleChargeModeData:
         """Get vehicle charge mode."""
-        response = await self.session.get_vehicle_data(
-            account_id=self.account_id,
-            vin=self.vin,
-            endpoint="charge-mode",
-        )
+        response = await self._get_vehicle_data("charge-mode")
         return cast(
             models.KamereonVehicleChargeModeData,
             response.get_attributes(schemas.KamereonVehicleChargeModeDataSchema),
         )
 
     async def get_cockpit(self) -> models.KamereonVehicleCockpitData:
         """Get vehicle cockpit."""
-        response = await self.session.get_vehicle_data(
-            account_id=self.account_id,
-            vin=self.vin,
-            endpoint="cockpit",
-        )
+        response = await self._get_vehicle_data("cockpit")
         return cast(
             models.KamereonVehicleCockpitData,
             response.get_attributes(schemas.KamereonVehicleCockpitDataSchema),
         )
 
     async def get_lock_status(self) -> models.KamereonVehicleLockStatusData:
         """Get vehicle lock status."""
-        response = await self.session.get_vehicle_data(
-            account_id=self.account_id,
-            vin=self.vin,
-            endpoint="lock-status",
-        )
+        response = await self._get_vehicle_data("lock-status")
         return cast(
             models.KamereonVehicleLockStatusData,
             response.get_attributes(schemas.KamereonVehicleLockStatusDataSchema),
         )
 
     async def get_res_state(self) -> models.KamereonVehicleResStateData:
         """Get vehicle res state."""
-        response = await self.session.get_vehicle_data(
-            account_id=self.account_id,
-            vin=self.vin,
-            endpoint="res-state",
-        )
+        response = await self._get_vehicle_data("res-state")
         return cast(
             models.KamereonVehicleResStateData,
             response.get_attributes(schemas.KamereonVehicleResStateDataSchema),
         )
 
     async def get_charging_settings(self) -> models.KamereonVehicleChargingSettingsData:
         """Get vehicle charging settings."""
-        response = await self.session.get_vehicle_data(
-            account_id=self.account_id,
-            vin=self.vin,
-            endpoint="charging-settings",
-        )
+        response = await self._get_vehicle_data("charging-settings")
         return cast(
             models.KamereonVehicleChargingSettingsData,
             response.get_attributes(schemas.KamereonVehicleChargingSettingsDataSchema),
@@ -246,11 +227,7 @@ async def get_notification_settings(
         self,
     ) -> models.KamereonVehicleNotificationSettingsData:
         """Get vehicle notification settings."""
-        response = await self.session.get_vehicle_data(
-            account_id=self.account_id,
-            vin=self.vin,
-            endpoint="notification-settings",
-        )
+        response = await self._get_vehicle_data("notification-settings")
         return cast(
             models.KamereonVehicleNotificationSettingsData,
             response.get_attributes(