Gracefully handle unsupported features (#1)

custom_components/volvo_cars/binary_sensor.py

@@ -452,7 +452,11 @@ async def async_setup_entry(
 ) -> None:
     """Set up sensors."""
     coordinator = entry.runtime_data.coordinator
-    sensors = [VolvoCarsSensor(coordinator, description) for description in SENSORS]
+    sensors = [
+        VolvoCarsSensor(coordinator, description)
+        for description in SENSORS
+        if description.api_field not in coordinator.unsupported_keys
+    ]
 
     async_add_entities(sensors)
 

custom_components/volvo_cars/button.py

@@ -121,12 +121,14 @@ async def async_press(self) -> None:
                 self.entity_description.api_command
             )
 
+            status = result.invoke_status if result else "<none>"
+
             _LOGGER.debug(
                 "Command %s result: %s",
                 self.entity_description.api_command,
-                result.invoke_status,
+                status,
             )
-            self._attr_extra_state_attributes[ATTR_LAST_RESULT] = result.invoke_status
+            self._attr_extra_state_attributes[ATTR_LAST_RESULT] = status
             self._attr_extra_state_attributes[ATTR_API_TIMESTAMP] = datetime.now(
                 UTC
             ).isoformat()

custom_components/volvo_cars/coordinator.py

@@ -22,7 +22,12 @@
 from .const import CONF_REFRESH_TOKEN, CONF_VCC_API_KEY, CONF_VIN, DOMAIN, MANUFACTURER
 from .volvo.api import VolvoCarsApi
 from .volvo.auth import VolvoCarsAuthApi
-from .volvo.models import VolvoAuthException, VolvoCarsApiBaseModel, VolvoCarsVehicle
+from .volvo.models import (
+    VolvoAuthException,
+    VolvoCarsApiBaseModel,
+    VolvoCarsValueField,
+    VolvoCarsVehicle,
+)
 
 _LOGGER = logging.getLogger(__name__)
 
@@ -56,6 +61,7 @@ def __init__(
 
         self.entry = entry
         self._auth_api = auth_api
+
         self.api = VolvoCarsApi(
             async_get_clientsession(hass),
             entry.data[CONF_VIN],
@@ -71,7 +77,14 @@ def __init__(
 
         self.vehicle: VolvoCarsVehicle
         self.device: DeviceInfo
-        self.commands: list[str]
+        self.commands: list[str] = []
+
+        self.supports_location: bool = False
+        self.supports_doors: bool = False
+        self.supports_tyres: bool = False
+        self.supports_warnings: bool = False
+        self.supports_windows: bool = False
+        self.unsupported_keys: list[str] = []
 
     async def _async_setup(self) -> None:
         """Set up the coordinator.
@@ -88,47 +101,85 @@ async def _async_setup(self) -> None:
             serial_number=self.vehicle.vin,
         )
 
-        commands = await self.api.async_get_commands()
-        self.commands = [command.command for command in commands]
-
         self.hass.config_entries.async_update_entry(
             self.entry,
             title=f"{MANUFACTURER} {self.vehicle.description.model} ({self.vehicle.vin})",
         )
 
+        # Check supported commands
+        commands = await self.api.async_get_commands()
+        self.commands = [command.command for command in commands if command]
+
+        # Check if location is supported
+        location = await self.api.async_get_location()
+        self.supports_location = location.get("location") is not None
+
+        # Check if doors are supported
+        doors = await self.api.async_get_doors_status()
+        self.supports_doors = not self._is_all_unspecified(doors)
+
+        # Check if tyres are supported
+        tyres = await self.api.async_get_tyre_states()
+        self.supports_tyres = not self._is_all_unspecified(tyres)
+
+        # Check if warnings are supported
+        warnings = await self.api.async_get_warnings()
+        self.supports_warnings = not self._is_all_unspecified(warnings)
+
+        # Check if windows are supported
+        windows = await self.api.async_get_window_states()
+        self.supports_windows = not self._is_all_unspecified(windows)
+
+        # Keep track of unsupported keys
+        self.unsupported_keys.append("location")
+        self.unsupported_keys += [
+            key
+            for key, value in (doors | tyres | warnings | windows).items()
+            if value is None or value.value == "UNSPECIFIED"
+        ]
+
     async def _async_update_data(self) -> dict[str, Any]:
         """Fetch data from API."""
+        future: asyncio.Future[dict[str, VolvoCarsApiBaseModel | None]] = (
+            asyncio.Future()
+        )
+        future.set_result({})
+
         try:
             # Note: asyncio.TimeoutError and aiohttp.ClientError are already
             # handled by the data update coordinator.
             async with asyncio.timeout(30):
-                data: dict[str, VolvoCarsApiBaseModel] = {}
+                data: dict[str, VolvoCarsApiBaseModel | None] = {}
 
                 results = await asyncio.gather(
                     self.api.async_get_api_status(),
                     self.api.async_get_availability_status(),
                     self.api.async_get_brakes_status(),
                     self.api.async_get_diagnostics(),
-                    self.api.async_get_doors_status(),
+                    self.api.async_get_doors_status()
+                    if self.supports_doors
+                    else future,
                     self.api.async_get_engine_status(),
                     self.api.async_get_engine_warnings(),
-                    self.api.async_get_location(),
+                    self.api.async_get_fuel_status()
+                    if self.vehicle.has_combustion_engine()
+                    else future,
+                    self.api.async_get_location() if self.supports_location else future,
                     self.api.async_get_odometer(),
+                    self.api.async_get_recharge_status()
+                    if self.vehicle.has_battery_engine()
+                    else future,
                     self.api.async_get_statistics(),
-                    self.api.async_get_tyre_states(),
-                    self.api.async_get_warnings(),
-                    self.api.async_get_window_states(),
+                    self.api.async_get_tyre_states() if self.supports_tyres else future,
+                    self.api.async_get_warnings() if self.supports_warnings else future,
+                    self.api.async_get_window_states()
+                    if self.supports_windows
+                    else future,
                 )
 
                 for result in results:
                     data |= result
 
-                if self.vehicle.has_combustion_engine():
-                    data |= await self.api.async_get_fuel_status()
-
-                if self.vehicle.has_battery_engine():
-                    data |= await self.api.async_get_recharge_status()
-
         except VolvoAuthException as ex:
             # Raising ConfigEntryAuthFailed will cancel future updates
             # and start a config flow with SOURCE_REAUTH (async_step_reauth)
@@ -161,3 +212,8 @@ async def async_refresh_token(self, _: datetime | None = None) -> None:
             }
             self.hass.config_entries.async_update_entry(self.entry, data=data)
             self.api.update_access_token(result.token.access_token)
+
+    def _is_all_unspecified(self, items: dict[str, VolvoCarsValueField | None]) -> bool:
+        return all(
+            item is None or item.value == "UNSPECIFIED" for item in items.values()
+        )

custom_components/volvo_cars/device_tracker.py

@@ -42,7 +42,9 @@ async def async_setup_entry(
     """Set up tracker."""
     coordinator = entry.runtime_data.coordinator
     trackers = [
-        VolvoCarsDeviceTracker(coordinator, description) for description in TRACKERS
+        VolvoCarsDeviceTracker(coordinator, description)
+        for description in TRACKERS
+        if coordinator.supports_location
     ]
 
     async_add_entities(trackers)

custom_components/volvo_cars/lock.py

@@ -109,13 +109,15 @@ async def _async_handle_command(self, command: str, locked: bool) -> None:
             self.async_write_ha_state()
 
             result = await self.coordinator.api.async_execute_command(command)
-            _LOGGER.debug("Lock '%s' result: %s", command, result.invoke_status)
-            self._attr_extra_state_attributes[ATTR_LAST_RESULT] = result.invoke_status
+            status = result.invoke_status if result else "<none>"
+
+            _LOGGER.debug("Lock '%s' result: %s", command, status)
+            self._attr_extra_state_attributes[ATTR_LAST_RESULT] = status
             self._attr_extra_state_attributes[ATTR_API_TIMESTAMP] = datetime.now(
                 UTC
             ).isoformat()
 
-            if result.invoke_status not in ("COMPLETED", "DELIVERED"):
+            if status not in ("COMPLETED", "DELIVERED"):
                 self._attr_is_locking = False
                 self._attr_is_unlocking = False
                 self.async_write_ha_state()
@@ -125,8 +127,8 @@ async def _async_handle_command(self, command: str, locked: bool) -> None:
                     translation_key="lock_failure",
                     translation_placeholders={
                         "command": command,
-                        "status": result.invoke_status,
-                        "message": result.message,
+                        "status": status,
+                        "message": result.message if result else "",
                     },
                 )
 

custom_components/volvo_cars/volvo/api.py

@@ -69,59 +69,61 @@ async def async_get_api_status(self) -> dict[str, VolvoCarsValue]:
             _LOGGER.debug("Request [API status] error: %s", ex.message)
             raise VolvoApiException from ex
 
-    async def async_get_availability_status(self) -> dict[str, VolvoCarsValueField]:
+    async def async_get_availability_status(
+        self,
+    ) -> dict[str, VolvoCarsValueField | None]:
         """Get availability status."""
         return await self._async_get_field(
             _API_CONNECTED_ENDPOINT, "command-accessibility"
         )
 
-    async def async_get_brakes_status(self) -> dict[str, VolvoCarsValueField]:
+    async def async_get_brakes_status(self) -> dict[str, VolvoCarsValueField | None]:
         """Get brakes status."""
         return await self._async_get_field(_API_CONNECTED_ENDPOINT, "brakes")
 
-    async def async_get_commands(self) -> list[VolvoCarsAvailableCommand]:
+    async def async_get_commands(self) -> list[VolvoCarsAvailableCommand | None]:
         """Get available commands."""
         items = await self._async_get_data_list(_API_CONNECTED_ENDPOINT, "commands")
         return [VolvoCarsAvailableCommand.from_dict(item) for item in items]
 
-    async def async_get_diagnostics(self) -> dict[str, VolvoCarsValueField]:
+    async def async_get_diagnostics(self) -> dict[str, VolvoCarsValueField | None]:
         """Get diagnostics."""
         return await self._async_get_field(_API_CONNECTED_ENDPOINT, "diagnostics")
 
-    async def async_get_doors_status(self) -> dict[str, VolvoCarsValueField]:
+    async def async_get_doors_status(self) -> dict[str, VolvoCarsValueField | None]:
         """Get doors status."""
         return await self._async_get_field(_API_CONNECTED_ENDPOINT, "doors")
 
-    async def async_get_engine_status(self) -> dict[str, VolvoCarsValueField]:
+    async def async_get_engine_status(self) -> dict[str, VolvoCarsValueField | None]:
         """Get engine status."""
         return await self._async_get_field(_API_CONNECTED_ENDPOINT, "engine-status")
 
-    async def async_get_engine_warnings(self) -> dict[str, VolvoCarsValueField]:
+    async def async_get_engine_warnings(self) -> dict[str, VolvoCarsValueField | None]:
         """Get engine warnings."""
         return await self._async_get_field(_API_CONNECTED_ENDPOINT, "engine")
 
-    async def async_get_fuel_status(self) -> dict[str, VolvoCarsValueField]:
+    async def async_get_fuel_status(self) -> dict[str, VolvoCarsValueField | None]:
         """Get fuel status."""
         return await self._async_get_field(_API_CONNECTED_ENDPOINT, "fuel")
 
-    async def async_get_location(self) -> dict[str, VolvoCarsLocation]:
+    async def async_get_location(self) -> dict[str, VolvoCarsLocation | None]:
         """Get location."""
         data = await self._async_get_data_dict(_API_LOCATION_ENDPOINT, "location")
         return {"location": VolvoCarsLocation.from_dict(data)}
 
-    async def async_get_odometer(self) -> dict[str, VolvoCarsValueField]:
+    async def async_get_odometer(self) -> dict[str, VolvoCarsValueField | None]:
         """Get odometer."""
         return await self._async_get_field(_API_CONNECTED_ENDPOINT, "odometer")
 
-    async def async_get_recharge_status(self) -> dict[str, VolvoCarsValueField]:
+    async def async_get_recharge_status(self) -> dict[str, VolvoCarsValueField | None]:
         """Get recharge status."""
         return await self._async_get_field(_API_ENERGY_ENDPOINT, "recharge-status")
 
-    async def async_get_statistics(self) -> dict[str, VolvoCarsValueField]:
+    async def async_get_statistics(self) -> dict[str, VolvoCarsValueField | None]:
         """Get statistics."""
         return await self._async_get_field(_API_CONNECTED_ENDPOINT, "statistics")
 
-    async def async_get_tyre_states(self) -> dict[str, VolvoCarsValueField]:
+    async def async_get_tyre_states(self) -> dict[str, VolvoCarsValueField | None]:
         """Get tyre states."""
         return await self._async_get_field(_API_CONNECTED_ENDPOINT, "tyres")
 
@@ -130,24 +132,26 @@ async def async_get_vehicle_details(self) -> VolvoCarsVehicle:
         data = await self._async_get_data_dict(_API_CONNECTED_ENDPOINT, "")
         return VolvoCarsVehicle.parse_obj(data)
 
-    async def async_get_warnings(self) -> dict[str, VolvoCarsValueField]:
+    async def async_get_warnings(self) -> dict[str, VolvoCarsValueField | None]:
         """Get warnings."""
         return await self._async_get_field(_API_CONNECTED_ENDPOINT, "warnings")
 
-    async def async_get_window_states(self) -> dict[str, VolvoCarsValueField]:
+    async def async_get_window_states(self) -> dict[str, VolvoCarsValueField | None]:
         """Get window states."""
         return await self._async_get_field(_API_CONNECTED_ENDPOINT, "windows")
 
-    async def async_execute_command(self, command: str) -> VolvoCarsCommandResult:
+    async def async_execute_command(
+        self, command: str
+    ) -> VolvoCarsCommandResult | None:
         """Execute a command."""
         body = await self._async_post(_API_CONNECTED_ENDPOINT, f"commands/{command}")
         data: dict = body.get("data", {})
-        data["invoke_status"] = data.pop("invokeStatus")
+        data["invoke_status"] = data.pop("invokeStatus", None)
         return VolvoCarsCommandResult.from_dict(data)
 
     async def _async_get_field(
         self, endpoint: str, operation: str
-    ) -> dict[str, VolvoCarsValueField]:
+    ) -> dict[str, VolvoCarsValueField | None]:
         body = await self._async_get(endpoint, operation)
         data: dict = body.get("data", {})
         return {
@@ -206,6 +210,9 @@ async def _async_request(
                 response.raise_for_status()
                 return data
         except ClientResponseError as ex:
+            if ex.status == 404:
+                return {}
+
             _LOGGER.debug("Request [%s] error: %s", operation, ex.message)
             if ex.status in (401, 403):
                 raise VolvoAuthException from ex

custom_components/volvo_cars/volvo/models.py

@@ -3,11 +3,13 @@
 from dataclasses import KW_ONLY, dataclass, field, is_dataclass
 from datetime import datetime
 import inspect
-from typing import Any
+from typing import Any, TypeVar
 
 # pylint: disable-next=no-name-in-module
 from pydantic import BaseModel, Field
 
+T = TypeVar("T", bound="VolvoCarsApiBaseModel")
+
 
 class VolvoCarsModel(BaseModel):
     """Representation of a Volvo Cars model."""
@@ -53,7 +55,7 @@ class VolvoCarsApiBaseModel:
     extra_data: dict[str, Any] = field(default_factory=dict[str, Any])
 
     @classmethod
-    def from_dict(cls, data: dict[str, Any]):
+    def from_dict(cls: type[T], data: dict[str, Any]) -> T | None:
         """Create instance from json dict."""
         parameters = inspect.signature(cls).parameters
         class_data: dict[str, Any] = {}
@@ -76,6 +78,9 @@ def from_dict(cls, data: dict[str, Any]):
             else:
                 extra_data[key] = value
 
+        if len(class_data) == 0:
+            return None
+
         class_data["extra_data"] = extra_data
         return cls(**class_data)