Add renault integration (home-assistant#39605)

.strict-typing

@@ -81,6 +81,7 @@ homeassistant.components.recorder.purge
 homeassistant.components.recorder.repack
 homeassistant.components.recorder.statistics
 homeassistant.components.remote.*
+homeassistant.components.renault.*
 homeassistant.components.rituals_perfume_genie.*
 homeassistant.components.scene.*
 homeassistant.components.select.*

CODEOWNERS

@@ -411,6 +411,7 @@ homeassistant/components/rainmachine/* @bachya
 homeassistant/components/random/* @fabaff
 homeassistant/components/recollect_waste/* @bachya
 homeassistant/components/rejseplanen/* @DarkFox
+homeassistant/components/renault/* @epenet
 homeassistant/components/repetier/* @MTrab
 homeassistant/components/rflink/* @javicalle
 homeassistant/components/rfxtrx/* @danielhiversen @elupus @RobBie1221

homeassistant/components/renault/__init__.py

@@ -0,0 +1,45 @@
+"""Support for Renault devices."""
+import aiohttp
+
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.const import CONF_PASSWORD, CONF_USERNAME
+from homeassistant.core import HomeAssistant
+from homeassistant.exceptions import ConfigEntryNotReady
+
+from .const import CONF_LOCALE, DOMAIN, PLATFORMS
+from .renault_hub import RenaultHub
+
+
+async def async_setup_entry(hass: HomeAssistant, config_entry: ConfigEntry) -> bool:
+    """Load a config entry."""
+    renault_hub = RenaultHub(hass, config_entry.data[CONF_LOCALE])
+    try:
+        login_success = await renault_hub.attempt_login(
+            config_entry.data[CONF_USERNAME], config_entry.data[CONF_PASSWORD]
+        )
+    except aiohttp.ClientConnectionError as exc:
+        raise ConfigEntryNotReady() from exc
+
+    if not login_success:
+        return False
+
+    hass.data.setdefault(DOMAIN, {})
+    await renault_hub.async_initialise(config_entry)
+
+    hass.data[DOMAIN][config_entry.unique_id] = renault_hub
+
+    hass.config_entries.async_setup_platforms(config_entry, PLATFORMS)
+
+    return True
+
+
+async def async_unload_entry(hass: HomeAssistant, config_entry: ConfigEntry) -> bool:
+    """Unload a config entry."""
+    unload_ok = await hass.config_entries.async_unload_platforms(
+        config_entry, PLATFORMS
+    )
+
+    if unload_ok:
+        hass.data[DOMAIN].pop(config_entry.unique_id)
+
+    return unload_ok

homeassistant/components/renault/config_flow.py

@@ -0,0 +1,92 @@
+"""Config flow to configure Renault component."""
+from __future__ import annotations
+
+from typing import Any
+
+from renault_api.const import AVAILABLE_LOCALES
+import voluptuous as vol
+
+from homeassistant import config_entries
+from homeassistant.const import CONF_PASSWORD, CONF_USERNAME
+from homeassistant.data_entry_flow import FlowResult
+
+from .const import CONF_KAMEREON_ACCOUNT_ID, CONF_LOCALE, DOMAIN
+from .renault_hub import RenaultHub
+
+
+class RenaultFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
+    """Handle a Renault config flow."""
+
+    VERSION = 1
+
+    def __init__(self) -> None:
+        """Initialize the Renault config flow."""
+        self.renault_config: dict[str, Any] = {}
+        self.renault_hub: RenaultHub | None = None
+
+    async def async_step_user(
+        self, user_input: dict[str, Any] | None = None
+    ) -> FlowResult:
+        """Handle a Renault config flow start.
+
+        Ask the user for API keys.
+        """
+        if user_input:
+            locale = user_input[CONF_LOCALE]
+            self.renault_config.update(user_input)
+            self.renault_config.update(AVAILABLE_LOCALES[locale])
+            self.renault_hub = RenaultHub(self.hass, locale)
+            if not await self.renault_hub.attempt_login(
+                user_input[CONF_USERNAME], user_input[CONF_PASSWORD]
+            ):
+                return self._show_user_form({"base": "invalid_credentials"})
+            return await self.async_step_kamereon()
+        return self._show_user_form()
+
+    def _show_user_form(self, errors: dict[str, Any] | None = None) -> FlowResult:
+        """Show the API keys form."""
+        return self.async_show_form(
+            step_id="user",
+            data_schema=vol.Schema(
+                {
+                    vol.Required(CONF_LOCALE): vol.In(AVAILABLE_LOCALES.keys()),
+                    vol.Required(CONF_USERNAME): str,
+                    vol.Required(CONF_PASSWORD): str,
+                }
+            ),
+            errors=errors or {},
+        )
+
+    async def async_step_kamereon(
+        self, user_input: dict[str, Any] | None = None
+    ) -> FlowResult:
+        """Select Kamereon account."""
+        if user_input:
+            await self.async_set_unique_id(user_input[CONF_KAMEREON_ACCOUNT_ID])
+            self._abort_if_unique_id_configured()
+
+            self.renault_config.update(user_input)
+            return self.async_create_entry(
+                title=user_input[CONF_KAMEREON_ACCOUNT_ID], data=self.renault_config
+            )
+
+        assert self.renault_hub
+        accounts = await self.renault_hub.get_account_ids()
+        if len(accounts) == 0:
+            return self.async_abort(reason="kamereon_no_account")
+        if len(accounts) == 1:
+            await self.async_set_unique_id(accounts[0])
+            self._abort_if_unique_id_configured()
+
+            self.renault_config[CONF_KAMEREON_ACCOUNT_ID] = accounts[0]
+            return self.async_create_entry(
+                title=self.renault_config[CONF_KAMEREON_ACCOUNT_ID],
+                data=self.renault_config,
+            )
+
+        return self.async_show_form(
+            step_id="kamereon",
+            data_schema=vol.Schema(
+                {vol.Required(CONF_KAMEREON_ACCOUNT_ID): vol.In(accounts)}
+            ),
+        )

homeassistant/components/renault/const.py

@@ -0,0 +1,15 @@
+"""Constants for the Renault component."""
+DOMAIN = "renault"
+
+CONF_LOCALE = "locale"
+CONF_KAMEREON_ACCOUNT_ID = "kamereon_account_id"
+
+DEFAULT_SCAN_INTERVAL = 300  # 5 minutes
+
+PLATFORMS = [
+    "sensor",
+]
+
+DEVICE_CLASS_PLUG_STATE = "renault__plug_state"
+DEVICE_CLASS_CHARGE_STATE = "renault__charge_state"
+DEVICE_CLASS_CHARGE_MODE = "renault__charge_mode"

homeassistant/components/renault/manifest.json

@@ -0,0 +1,13 @@
+{
+  "domain": "renault",
+  "name": "Renault",
+  "config_flow": true,
+  "documentation": "https://www.home-assistant.io/integrations/renault",
+  "requirements": [
+    "renault-api==0.1.4"
+  ],
+  "codeowners": [
+    "@epenet"
+  ],
+  "iot_class": "cloud_polling"
+}

homeassistant/components/renault/renault_coordinator.py

@@ -0,0 +1,72 @@
+"""Proxy to handle account communication with Renault servers."""
+from __future__ import annotations
+
+from collections.abc import Awaitable
+from datetime import timedelta
+import logging
+from typing import Callable, TypeVar
+
+from renault_api.kamereon.exceptions import (
+    AccessDeniedException,
+    KamereonResponseException,
+    NotSupportedException,
+)
+
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
+
+T = TypeVar("T")
+
+
+class RenaultDataUpdateCoordinator(DataUpdateCoordinator[T]):
+    """Handle vehicle communication with Renault servers."""
+
+    def __init__(
+        self,
+        hass: HomeAssistant,
+        logger: logging.Logger,
+        *,
+        name: str,
+        update_interval: timedelta,
+        update_method: Callable[[], Awaitable[T]],
+    ) -> None:
+        """Initialise coordinator."""
+        super().__init__(
+            hass,
+            logger,
+            name=name,
+            update_interval=update_interval,
+            update_method=update_method,
+        )
+        self.access_denied = False
+        self.not_supported = False
+
+    async def _async_update_data(self) -> T:
+        """Fetch the latest data from the source."""
+        if self.update_method is None:
+            raise NotImplementedError("Update method not implemented")
+        try:
+            return await self.update_method()
+        except AccessDeniedException as err:
+            # Disable because the account is not allowed to access this Renault endpoint.
+            self.update_interval = None
+            self.access_denied = True
+            raise UpdateFailed(f"This endpoint is denied: {err}") from err
+
+        except NotSupportedException as err:
+            # Disable because the vehicle does not support this Renault endpoint.
+            self.update_interval = None
+            self.not_supported = True
+            raise UpdateFailed(f"This endpoint is not supported: {err}") from err
+
+        except KamereonResponseException as err:
+            # Other Renault errors.
+            raise UpdateFailed(f"Error communicating with API: {err}") from err
+
+    async def async_config_entry_first_refresh(self) -> None:
+        """Refresh data for the first time when a config entry is setup.
+
+        Contrary to base implementation, we are not raising ConfigEntryNotReady
+        but only updating the `access_denied` and `not_supported` flags.
+        """
+        await self._async_refresh(log_failures=False, raise_on_auth_failed=True)

homeassistant/components/renault/renault_entities.py

@@ -0,0 +1,103 @@
+"""Base classes for Renault entities."""
+from __future__ import annotations
+
+from typing import Any, Generic, Optional, TypeVar
+
+from renault_api.kamereon.enums import ChargeState, PlugState
+from renault_api.kamereon.models import (
+    KamereonVehicleBatteryStatusData,
+    KamereonVehicleChargeModeData,
+    KamereonVehicleCockpitData,
+    KamereonVehicleHvacStatusData,
+)
+
+from homeassistant.helpers.entity import Entity
+from homeassistant.helpers.update_coordinator import CoordinatorEntity
+from homeassistant.util import slugify
+
+from .renault_vehicle import RenaultVehicleProxy
+
+ATTR_LAST_UPDATE = "last_update"
+
+T = TypeVar("T")
+
+
+class RenaultDataEntity(Generic[T], CoordinatorEntity[Optional[T]], Entity):
+    """Implementation of a Renault entity with a data coordinator."""
+
+    def __init__(
+        self, vehicle: RenaultVehicleProxy, entity_type: str, coordinator_key: str
+    ) -> None:
+        """Initialise entity."""
+        super().__init__(vehicle.coordinators[coordinator_key])
+        self.vehicle = vehicle
+        self._entity_type = entity_type
+        self._attr_device_info = self.vehicle.device_info
+        self._attr_name = entity_type
+        self._attr_unique_id = slugify(
+            f"{self.vehicle.details.vin}-{self._entity_type}"
+        )
+
+    @property
+    def available(self) -> bool:
+        """Return if entity is available."""
+        # Data can succeed, but be empty
+        return super().available and self.coordinator.data is not None
+
+    @property
+    def data(self) -> T | None:
+        """Return collected data."""
+        return self.coordinator.data
+
+
+class RenaultBatteryDataEntity(RenaultDataEntity[KamereonVehicleBatteryStatusData]):
+    """Implementation of a Renault entity with battery coordinator."""
+
+    def __init__(self, vehicle: RenaultVehicleProxy, entity_type: str) -> None:
+        """Initialise entity."""
+        super().__init__(vehicle, entity_type, "battery")
+
+    @property
+    def extra_state_attributes(self) -> dict[str, Any]:
+        """Return the state attributes of this entity."""
+        last_update = self.data.timestamp if self.data else None
+        return {ATTR_LAST_UPDATE: last_update}
+
+    @property
+    def is_charging(self) -> bool:
+        """Return charge state as boolean."""
+        return (
+            self.data is not None
+            and self.data.get_charging_status() == ChargeState.CHARGE_IN_PROGRESS
+        )
+
+    @property
+    def is_plugged_in(self) -> bool:
+        """Return plug state as boolean."""
+        return (
+            self.data is not None and self.data.get_plug_status() == PlugState.PLUGGED
+        )
+
+
+class RenaultChargeModeDataEntity(RenaultDataEntity[KamereonVehicleChargeModeData]):
+    """Implementation of a Renault entity with charge_mode coordinator."""
+
+    def __init__(self, vehicle: RenaultVehicleProxy, entity_type: str) -> None:
+        """Initialise entity."""
+        super().__init__(vehicle, entity_type, "charge_mode")
+
+
+class RenaultCockpitDataEntity(RenaultDataEntity[KamereonVehicleCockpitData]):
+    """Implementation of a Renault entity with cockpit coordinator."""
+
+    def __init__(self, vehicle: RenaultVehicleProxy, entity_type: str) -> None:
+        """Initialise entity."""
+        super().__init__(vehicle, entity_type, "cockpit")
+
+
+class RenaultHVACDataEntity(RenaultDataEntity[KamereonVehicleHvacStatusData]):
+    """Implementation of a Renault entity with hvac_status coordinator."""
+
+    def __init__(self, vehicle: RenaultVehicleProxy, entity_type: str) -> None:
+        """Initialise entity."""
+        super().__init__(vehicle, entity_type, "hvac_status")