Drop and log unknown capabilities, log responses, deserialize with Pydantic

myskoda/models/info.py

@@ -1,9 +1,12 @@
 """Models for responses of api/v2/garage/vehicles/{vin}."""
 
+import logging
 from datetime import date
 from enum import StrEnum
 
-from pydantic import BaseModel, Field
+from pydantic import BaseModel, Field, validator
+
+_LOGGER = logging.getLogger(__name__)
 
 
 class CapabilityId(StrEnum):
@@ -80,6 +83,14 @@ class Capability(BaseModel):
 class Capabilities(BaseModel):
     capabilities: list[Capability]
 
+    @validator("capabilities", pre=True, always=True)
+    def drop_unknown_capabilities(cls, value: list[dict]) -> list[dict]:  # noqa: N805
+        """Drop any unknown capabilities and log a message."""
+        unknown_capabilities = [c for c in value if c["id"] not in CapabilityId]
+        if unknown_capabilities:
+            _LOGGER.info(f"Dropping unknown capabilities: {unknown_capabilities}")
+        return [c for c in value if c["id"] in CapabilityId]
+
 
 class Battery(BaseModel):
     capacity: int = Field(None, alias="capacityInKWh")

myskoda/models/mqtt.py

@@ -4,7 +4,7 @@
 from enum import StrEnum
 from typing import Generic, TypeVar
 
-from pydantic import BaseModel, Field, validator
+from pydantic import BaseModel, Field
 
 
 class OperationStatus(StrEnum):
@@ -108,22 +108,8 @@ class ServiceEventChargingData(ServiceEventData):
     mode: ServiceEventChargeMode
     state: ServiceEventChargingState
     soc: int
-    charged_range: str = Field(None, alias="chargedRange")
-    time_to_finish: str | None = Field(None, alias="timeToFinish")
-
-    @validator("soc")
-    def _parse_soc(cls, value: str) -> int:  # noqa: N805
-        return int(value)
-
-    @validator("charged_range")
-    def _parse_charged_range(cls, value: str) -> int:  # noqa: N805
-        return int(value)
-
-    @validator("time_to_finish")
-    def _parse_time_to_finish(cls, value: str) -> int | None:  # noqa: N805
-        if value == "null":
-            return None
-        return int(value)
+    charged_range: int = Field(None, alias="chargedRange")
+    time_to_finish: int | None = Field(None, alias="timeToFinish")
 
 
 class ServiceEventCharging(ServiceEvent):

myskoda/rest_api.py

@@ -74,89 +74,99 @@ async def get_info(self, vin: str) -> Info:
             f"{BASE_URL_SKODA}/api/v2/garage/vehicles/{vin}?connectivityGenerations=MOD1&connectivityGenerations=MOD2&connectivityGenerations=MOD3&connectivityGenerations=MOD4",
             headers=await self._headers(),
         ) as response:
-            _LOGGER.debug("vin %s: Received basic info", vin)
-            return Info(**await response.json())
+            response_text = await response.text()
+            _LOGGER.debug(f"vin {vin}: received basic info: {response_text}")
+            return Info.parse_raw(response_text)
 
     async def get_charging(self, vin: str) -> Charging:
         """Retrieve information related to charging for the specified vehicle."""
         async with self.session.get(
             f"{BASE_URL_SKODA}/api/v1/charging/{vin}", headers=await self._headers()
         ) as response:
-            _LOGGER.debug("Received charging info")
+            response_text = await response.text()
+            _LOGGER.debug(f"vin {vin}: received charging info: {response_text}")
             print(await response.json())
-            return Charging(**await response.json())
+            return Charging.parse_raw(response_text)
 
     async def get_status(self, vin: str) -> Status:
         """Retrieve the current status for the specified vehicle."""
         async with self.session.get(
             f"{BASE_URL_SKODA}/api/v2/vehicle-status/{vin}",
             headers=await self._headers(),
         ) as response:
-            _LOGGER.debug("vin %s: Received status")
-            return Status(**await response.json())
+            response_text = await response.text()
+            _LOGGER.debug(f"vin {vin}: received status: {response_text}")
+            return Status.parse_raw(response_text)
 
     async def get_air_conditioning(self, vin: str) -> AirConditioning:
         """Retrieve the current air conditioning status for the specified vehicle."""
         async with self.session.get(
             f"{BASE_URL_SKODA}/api/v2/air-conditioning/{vin}",
             headers=await self._headers(),
         ) as response:
-            _LOGGER.debug("vin %s: Received air conditioning")
-            return AirConditioning(**await response.json())
+            response_text = await response.text()
+            _LOGGER.debug(f"vin {vin}: received air conditioning: {response_text}")
+            return AirConditioning.parse_raw(response_text)
 
     async def get_positions(self, vin: str) -> Positions:
         """Retrieve the current position for the specified vehicle."""
         async with self.session.get(
             f"{BASE_URL_SKODA}/api/v1/maps/positions?vin={vin}",
             headers=await self._headers(),
         ) as response:
-            _LOGGER.debug("vin %s: Received position")
-            return Positions(**await response.json())
+            response_text = await response.text()
+            _LOGGER.debug(f"vin {vin}: received position: {response_text}")
+            return Positions.parse_raw(response_text)
 
     async def get_driving_range(self, vin: str) -> DrivingRange:
         """Retrieve estimated driving range for combustion vehicles."""
         async with self.session.get(
             f"{BASE_URL_SKODA}/api/v2/vehicle-status/{vin}/driving-range",
             headers=await self._headers(),
         ) as response:
-            _LOGGER.debug("vin %s: Received driving range")
-            return DrivingRange(**await response.json())
+            response_text = await response.text()
+            _LOGGER.debug(f"vin {vin}: received driving range: {response_text}")
+            return DrivingRange.parse_raw(response_text)
 
     async def get_trip_statistics(self, vin: str) -> TripStatistics:
         """Retrieve statistics about past trips."""
         async with self.session.get(
             f"{BASE_URL_SKODA}/api/v1/trip-statistics/{vin}?offsetType=week&offset=0&timezone=Europe%2FBerlin",
             headers=await self._headers(),
         ) as response:
-            _LOGGER.debug("vin %s: Received trip statistics")
-            return TripStatistics(**await response.json())
+            response_text = await response.text()
+            _LOGGER.debug(f"vin {vin}: received trip statistics: {response_text}")
+            return TripStatistics.parse_raw(response_text)
 
     async def get_maintenance(self, vin: str) -> Maintenance:
         """Retrieve maintenance report."""
         async with self.session.get(
             f"{BASE_URL_SKODA}/api/v3/vehicle-maintenance/vehicles/{vin}",
             headers=await self._headers(),
         ) as response:
-            _LOGGER.debug("vin %s: Received maintenance report")
-            return Maintenance(**await response.json())
+            response_text = await response.text()
+            _LOGGER.debug(f"vin {vin}: received maintenance report: {response_text}")
+            return Maintenance.parse_raw(response_text)
 
     async def get_health(self, vin: str) -> Health:
         """Retrieve health information for the specified vehicle."""
         async with self.session.get(
             f"{BASE_URL_SKODA}/api/v1/vehicle-health-report/warning-lights/{vin}",
             headers=await self._headers(),
         ) as response:
-            _LOGGER.debug("vin %s: Received health")
-            return Health(**await response.json())
+            response_text = await response.text()
+            _LOGGER.debug(f"vin {vin}: received health: {response_text}")
+            return Health.parse_raw(response_text)
 
     async def get_user(self) -> User:
         """Retrieve user information about logged in user."""
         async with self.session.get(
             f"{BASE_URL_SKODA}/api/v1/users",
             headers=await self._headers(),
         ) as response:
-            _LOGGER.debug("Received user")
-            return User(**await response.json())
+            response_text = await response.text()
+            _LOGGER.debug(f"received user: {response_text}")
+            return User.parse_raw(response_text)
 
     async def list_vehicles(self) -> list[str]:
         """List all vehicles by their vins."""