Renault code quality improvements (#55454)

homeassistant/components/renault/renault_coordinator.py

@@ -11,11 +11,12 @@ from renault_api.kamereon.exceptions import (
     KamereonResponseException,
     NotSupportedException,
 )
+from renault_api.kamereon.models import KamereonVehicleDataAttributes
 
 from homeassistant.core import HomeAssistant
 from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
 
-T = TypeVar("T")
+T = TypeVar("T", bound=KamereonVehicleDataAttributes)
 
 
 class RenaultDataUpdateCoordinator(DataUpdateCoordinator[T]):

homeassistant/components/renault/renault_entities.py

@@ -3,14 +3,13 @@ from __future__ import annotations
 
 from collections.abc import Mapping
 from dataclasses import dataclass
-from typing import Any, Optional, TypeVar, cast
+from typing import Any, Optional, cast
-
-from renault_api.kamereon.models import KamereonVehicleDataAttributes
 
 from homeassistant.helpers.entity import Entity, EntityDescription
 from homeassistant.helpers.typing import StateType
 from homeassistant.helpers.update_coordinator import CoordinatorEntity
 
+from .renault_coordinator import T
 from .renault_vehicle import RenaultVehicleProxy
 
 
@@ -28,8 +27,6 @@ class RenaultEntityDescription(EntityDescription, RenaultRequiredKeysMixin):
 
 ATTR_LAST_UPDATE = "last_update"
 
-T = TypeVar("T", bound=KamereonVehicleDataAttributes)
-
 
 class RenaultDataEntity(CoordinatorEntity[Optional[T]], Entity):
     """Implementation of a Renault entity with a data coordinator."""

homeassistant/components/renault/renault_vehicle.py

@@ -2,9 +2,11 @@
 from __future__ import annotations
 
 import asyncio
+from collections.abc import Awaitable
+from dataclasses import dataclass
 from datetime import timedelta
 import logging
-from typing import cast
+from typing import Callable, cast
 
 from renault_api.kamereon import models
 from renault_api.renault_vehicle import RenaultVehicle
@@ -25,6 +27,20 @@ from .renault_coordinator import RenaultDataUpdateCoordinator
 LOGGER = logging.getLogger(__name__)
 
 
+@dataclass
+class RenaultCoordinatorDescription:
+    """Class describing Renault coordinators."""
+
+    endpoint: str
+    key: str
+    update_method: Callable[
+        [RenaultVehicle],
+        Callable[[], Awaitable[models.KamereonVehicleDataAttributes]],
+    ]
+    # Optional keys
+    requires_electricity: bool = False
+
+
 class RenaultVehicleProxy:
     """Handle vehicle communication with Renault servers."""
 
@@ -61,48 +77,23 @@ class RenaultVehicleProxy:
         return self._device_info
 
     async def async_initialise(self) -> None:
-        """Load available sensors."""
+        """Load available coordinators."""
-        if await self.endpoint_available("cockpit"):
+        self.coordinators = {
-            self.coordinators["cockpit"] = RenaultDataUpdateCoordinator(
+            coord.key: RenaultDataUpdateCoordinator(
                 self.hass,
                 LOGGER,
                 # Name of the data. For logging purposes.
-                name=f"{self.details.vin} cockpit",
+                name=f"{self.details.vin} {coord.key}",
-                update_method=self.get_cockpit,
+                update_method=coord.update_method(self._vehicle),
                 # Polling interval. Will only be polled if there are subscribers.
                 update_interval=self._scan_interval,
             )
-        if await self.endpoint_available("hvac-status"):
+            for coord in COORDINATORS
-            self.coordinators["hvac_status"] = RenaultDataUpdateCoordinator(
+            if (
-                self.hass,
+                self.details.supports_endpoint(coord.endpoint)
-                LOGGER,
+                and (not coord.requires_electricity or self.details.uses_electricity())
-                # Name of the data. For logging purposes.
-                name=f"{self.details.vin} hvac_status",
-                update_method=self.get_hvac_status,
-                # Polling interval. Will only be polled if there are subscribers.
-                update_interval=self._scan_interval,
-            )
-        if self.details.uses_electricity():
-            if await self.endpoint_available("battery-status"):
-                self.coordinators["battery"] = RenaultDataUpdateCoordinator(
-                    self.hass,
-                    LOGGER,
-                    # Name of the data. For logging purposes.
-                    name=f"{self.details.vin} battery",
-                    update_method=self.get_battery_status,
-                    # Polling interval. Will only be polled if there are subscribers.
-                    update_interval=self._scan_interval,
-                )
-            if await self.endpoint_available("charge-mode"):
-                self.coordinators["charge_mode"] = RenaultDataUpdateCoordinator(
-                    self.hass,
-                    LOGGER,
-                    # Name of the data. For logging purposes.
-                    name=f"{self.details.vin} charge_mode",
-                    update_method=self.get_charge_mode,
-                    # Polling interval. Will only be polled if there are subscribers.
-                    update_interval=self._scan_interval,
             )
+        }
         # Check all coordinators
         await asyncio.gather(
             *(
@@ -130,24 +121,28 @@ class RenaultVehicleProxy:
                 )
                 del self.coordinators[key]
 
-    async def endpoint_available(self, endpoint: str) -> bool:
-        """Ensure the endpoint is available to avoid unnecessary queries."""
-        return await self._vehicle.supports_endpoint(
-            endpoint
-        ) and await self._vehicle.has_contract_for_endpoint(endpoint)
 
-    async def get_battery_status(self) -> models.KamereonVehicleBatteryStatusData:
+COORDINATORS: tuple[RenaultCoordinatorDescription, ...] = (
-        """Get battery status information from vehicle."""
+    RenaultCoordinatorDescription(
-        return await self._vehicle.get_battery_status()
+        endpoint="cockpit",
-
+        key="cockpit",
-    async def get_charge_mode(self) -> models.KamereonVehicleChargeModeData:
+        update_method=lambda x: x.get_cockpit,
-        """Get charge mode information from vehicle."""
+    ),
-        return await self._vehicle.get_charge_mode()
+    RenaultCoordinatorDescription(
-
+        endpoint="hvac-status",
-    async def get_cockpit(self) -> models.KamereonVehicleCockpitData:
+        key="hvac_status",
-        """Get cockpit information from vehicle."""
+        update_method=lambda x: x.get_hvac_status,
-        return await self._vehicle.get_cockpit()
+    ),
-
+    RenaultCoordinatorDescription(
-    async def get_hvac_status(self) -> models.KamereonVehicleHvacStatusData:
+        endpoint="battery-status",
-        """Get hvac status information from vehicle."""
+        key="battery",
-        return await self._vehicle.get_hvac_status()
+        requires_electricity=True,
+        update_method=lambda x: x.get_battery_status,
+    ),
+    RenaultCoordinatorDescription(
+        endpoint="charge-mode",
+        key="charge_mode",
+        requires_electricity=True,
+        update_method=lambda x: x.get_charge_mode,
+    ),
+)

homeassistant/components/renault/sensor.py

@@ -42,7 +42,8 @@ from .const import (
     DEVICE_CLASS_PLUG_STATE,
     DOMAIN,
 )
-from .renault_entities import RenaultDataEntity, RenaultEntityDescription, T
+from .renault_coordinator import T
+from .renault_entities import RenaultDataEntity, RenaultEntityDescription
 from .renault_hub import RenaultHub
 
 
@@ -61,7 +62,7 @@ class RenaultSensorEntityDescription(
     """Class describing Renault sensor entities."""
 
     icon_lambda: Callable[[RenaultSensor[T]], str] | None = None
-    requires_fuel: bool | None = None
+    requires_fuel: bool = False
     value_lambda: Callable[[RenaultSensor[T]], StateType] | None = None