Add new Volvo integration

.strict-typing

@@ -536,6 +536,7 @@ homeassistant.components.valve.*
 homeassistant.components.velbus.*
 homeassistant.components.vlc_telnet.*
 homeassistant.components.vodafone_station.*
+homeassistant.components.volvo.*
 homeassistant.components.wake_on_lan.*
 homeassistant.components.wake_word.*
 homeassistant.components.wallbox.*

homeassistant/components/volvo/__init__.py

@@ -0,0 +1,63 @@
+"""The Volvo integration."""
+
+from __future__ import annotations
+
+import logging
+
+from aiohttp import ClientResponseError
+from volvocarsapi.api import VolvoCarsApi
+
+from homeassistant.const import CONF_API_KEY
+from homeassistant.core import HomeAssistant
+from homeassistant.exceptions import ConfigEntryAuthFailed, ConfigEntryNotReady
+from homeassistant.helpers.aiohttp_client import async_get_clientsession
+from homeassistant.helpers.config_entry_oauth2_flow import (
+    OAuth2Session,
+    async_get_config_entry_implementation,
+)
+
+from .api import VolvoAuth
+from .const import CONF_VIN, PLATFORMS
+from .coordinator import VolvoConfigEntry, VolvoData, VolvoDataCoordinator
+
+_LOGGER = logging.getLogger(__name__)
+
+
+async def async_setup_entry(hass: HomeAssistant, entry: VolvoConfigEntry) -> bool:
+    """Set up Volvo from a config entry."""
+    _LOGGER.debug("%s - Loading entry", entry.entry_id)
+
+    # Create APIs
+    implementation = await async_get_config_entry_implementation(hass, entry)
+    oauth_session = OAuth2Session(hass, entry, implementation)
+    web_session = async_get_clientsession(hass)
+    auth = VolvoAuth(web_session, oauth_session)
+
+    try:
+        await auth.async_get_access_token()
+    except ClientResponseError as err:
+        if err.status == 401:
+            raise ConfigEntryAuthFailed from err
+
+        raise ConfigEntryNotReady from err
+
+    api = VolvoCarsApi(
+        web_session,
+        auth,
+        entry.data[CONF_VIN],
+        entry.data[CONF_API_KEY],
+    )
+
+    # Setup entry
+    coordinator = VolvoDataCoordinator(hass, entry, api)
+    await coordinator.async_config_entry_first_refresh()
+    entry.runtime_data = VolvoData(coordinator)
+    await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
+
+    return True
+
+
+async def async_unload_entry(hass: HomeAssistant, entry: VolvoConfigEntry) -> bool:
+    """Unload a config entry."""
+    _LOGGER.debug("%s - Unloading entry", entry.entry_id)
+    return await hass.config_entries.async_unload_platforms(entry, PLATFORMS)

homeassistant/components/volvo/api.py

@@ -0,0 +1,38 @@
+"""API for Volvo bound to Home Assistant OAuth."""
+
+from typing import cast
+
+from aiohttp import ClientSession
+from volvocarsapi.auth import AccessTokenManager
+
+from homeassistant.helpers.config_entry_oauth2_flow import OAuth2Session
+
+
+class VolvoAuth(AccessTokenManager):
+    """Provide Volvo authentication tied to an OAuth2 based config entry."""
+
+    def __init__(self, websession: ClientSession, oauth_session: OAuth2Session) -> None:
+        """Initialize Volvo auth."""
+        super().__init__(websession)
+        self._oauth_session = oauth_session
+
+    async def async_get_access_token(self) -> str:
+        """Return a valid access token."""
+        await self._oauth_session.async_ensure_token_valid()
+        return cast(str, self._oauth_session.token["access_token"])
+
+
+class ConfigFlowVolvoAuth(AccessTokenManager):
+    """Provide Volvo authentication before a ConfigEntry exists.
+
+    This implementation directly provides the token without supporting refresh.
+    """
+
+    def __init__(self, websession: ClientSession, token: str) -> None:
+        """Initialize ConfigFlowVolvoAuth."""
+        super().__init__(websession)
+        self._token = token
+
+    async def async_get_access_token(self) -> str:
+        """Return the token for the Volvo API."""
+        return self._token

homeassistant/components/volvo/application_credentials.py

@@ -0,0 +1,58 @@
+"""Application credentials platform for the Volvo integration."""
+
+from volvocarsapi.auth import AUTHORIZE_URL, TOKEN_URL
+
+from homeassistant.components.application_credentials import (
+    AuthorizationServer,
+    ClientCredential,
+)
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers.config_entry_oauth2_flow import (
+    AbstractOAuth2Implementation,
+    LocalOAuth2ImplementationWithPkce,
+)
+
+from .const import SCOPES
+
+
+async def async_get_auth_implementation(
+    hass: HomeAssistant, auth_domain: str, credential: ClientCredential
+) -> AbstractOAuth2Implementation:
+    """Return auth implementation for a custom auth implementation."""
+    return VolvoOAuth2Implementation(
+        hass,
+        auth_domain,
+        credential,
+        authorization_server=AuthorizationServer(
+            authorize_url=AUTHORIZE_URL,
+            token_url=TOKEN_URL,
+        ),
+    )
+
+
+class VolvoOAuth2Implementation(LocalOAuth2ImplementationWithPkce):
+    """Volvo oauth2 implementation."""
+
+    def __init__(
+        self,
+        hass: HomeAssistant,
+        auth_domain: str,
+        credential: ClientCredential,
+        authorization_server: AuthorizationServer,
+    ) -> None:
+        """Initialize."""
+        super().__init__(
+            hass,
+            auth_domain,
+            credential.client_id,
+            authorization_server.authorize_url,
+            authorization_server.token_url,
+            credential.client_secret,
+        )
+
+    @property
+    def extra_authorize_data(self) -> dict:
+        """Extra data that needs to be appended to the authorize url."""
+        return super().extra_authorize_data | {
+            "scope": " ".join(SCOPES),
+        }

homeassistant/components/volvo/config_flow.py

@@ -0,0 +1,176 @@
+"""Config flow for Volvo."""
+
+from __future__ import annotations
+
+from collections.abc import Mapping
+import logging
+from typing import Any
+
+import voluptuous as vol
+from volvocarsapi.api import VolvoCarsApi
+from volvocarsapi.models import VolvoApiException
+
+from homeassistant.config_entries import (
+    SOURCE_REAUTH,
+    SOURCE_RECONFIGURE,
+    ConfigFlowResult,
+)
+from homeassistant.const import CONF_ACCESS_TOKEN, CONF_API_KEY, CONF_NAME, CONF_TOKEN
+from homeassistant.helpers import aiohttp_client
+from homeassistant.helpers.config_entry_oauth2_flow import AbstractOAuth2FlowHandler
+from homeassistant.helpers.selector import SelectSelector, SelectSelectorConfig
+
+from .api import ConfigFlowVolvoAuth
+from .const import CONF_VIN, DOMAIN, MANUFACTURER
+
+_LOGGER = logging.getLogger(__name__)
+
+
+class VolvoOAuth2FlowHandler(AbstractOAuth2FlowHandler, domain=DOMAIN):
+    """Config flow to handle Volvo OAuth2 authentication."""
+
+    DOMAIN = DOMAIN
+
+    def __init__(self) -> None:
+        """Initialize Volvo config flow."""
+        super().__init__()
+
+        self._vins: list[str] = []
+        self._config_data: dict = {}
+
+    @property
+    def logger(self) -> logging.Logger:
+        """Return logger."""
+        return _LOGGER
+
+    # Overridden method
+    async def async_oauth_create_entry(self, data: dict) -> ConfigFlowResult:
+        """Create an entry for the flow."""
+        self._config_data |= data
+        return await self.async_step_api_key()
+
+    # By convention method
+    async def async_step_reauth(self, _: Mapping[str, Any]) -> ConfigFlowResult:
+        """Perform reauth upon an API authentication error."""
+        return await self.async_step_reauth_confirm()
+
+    # By convention method
+    async def async_step_reconfigure(
+        self, _: dict[str, Any] | None = None
+    ) -> ConfigFlowResult:
+        """Reconfigure the entry."""
+        return await self.async_step_api_key()
+
+    async def async_step_reauth_confirm(
+        self, user_input: dict[str, Any] | None = None
+    ) -> ConfigFlowResult:
+        """Confirm reauth dialog."""
+        if user_input is None:
+            return self.async_show_form(
+                step_id="reauth_confirm",
+                description_placeholders={CONF_NAME: self._get_reauth_entry().title},
+            )
+        return await self.async_step_user()
+
+    async def async_step_api_key(
+        self, user_input: dict[str, Any] | None = None
+    ) -> ConfigFlowResult:
+        """Handle the API key step."""
+        errors: dict[str, str] = {}
+
+        if user_input is not None:
+            web_session = aiohttp_client.async_get_clientsession(self.hass)
+            token = self._config_data[CONF_TOKEN][CONF_ACCESS_TOKEN]
+            auth = ConfigFlowVolvoAuth(web_session, token)
+            api = VolvoCarsApi(web_session, auth, "", user_input[CONF_API_KEY])
+
+            try:
+                self._vins = await api.async_get_vehicles()
+            except VolvoApiException:
+                _LOGGER.exception("Unable to retrieve vehicles")
+                errors["base"] = "cannot_load_vehicles"
+
+            if not errors:
+                self._config_data |= user_input
+
+                if len(self._vins) == 1:
+                    # If there is only one VIN, take that as value and
+                    # immediately create the entry. No need to show
+                    # additional step.
+                    self._config_data[CONF_VIN] = self._vins[0]
+                    return await self._async_create_or_update()
+
+                if self.source in (SOURCE_REAUTH, SOURCE_RECONFIGURE):
+                    # Don't let users change the VIN. The entry should be
+                    # recreated if they want to change the VIN.
+                    return await self._async_create_or_update()
+
+                return await self.async_step_vin()
+
+        if user_input is None:
+            if self.source == SOURCE_REAUTH:
+                user_input = self._config_data = dict(self._get_reauth_entry().data)
+            elif self.source == SOURCE_RECONFIGURE:
+                user_input = self._config_data = dict(
+                    self._get_reconfigure_entry().data
+                )
+            else:
+                user_input = {}
+
+        schema = vol.Schema(
+            {
+                vol.Required(
+                    CONF_API_KEY, default=user_input.get(CONF_API_KEY, "")
+                ): str,
+            },
+        )
+
+        return self.async_show_form(
+            step_id="api_key", data_schema=schema, errors=errors
+        )
+
+    async def async_step_vin(
+        self, user_input: dict[str, Any] | None = None
+    ) -> ConfigFlowResult:
+        """Handle the VIN step."""
+        if user_input is not None:
+            self._config_data |= user_input
+            return await self._async_create_or_update()
+
+        schema = vol.Schema(
+            {
+                vol.Required(CONF_VIN): SelectSelector(
+                    SelectSelectorConfig(
+                        options=self._vins,
+                        multiple=False,
+                    )
+                ),
+            },
+        )
+
+        return self.async_show_form(step_id="vin", data_schema=schema)
+
+    async def _async_create_or_update(self) -> ConfigFlowResult:
+        vin = self._config_data[CONF_VIN]
+        await self.async_set_unique_id(vin)
+
+        if self.source == SOURCE_REAUTH:
+            self._abort_if_unique_id_mismatch()
+            return self.async_update_reload_and_abort(
+                self._get_reauth_entry(),
+                data_updates=self._config_data,
+            )
+
+        if self.source == SOURCE_RECONFIGURE:
+            self._abort_if_unique_id_mismatch()
+            return self.async_update_reload_and_abort(
+                self._get_reconfigure_entry(),
+                data_updates=self._config_data,
+                reload_even_if_entry_is_unchanged=False,
+            )
+
+        self._abort_if_unique_id_configured()
+        return self.async_create_entry(
+            title=f"{MANUFACTURER} {vin}",
+            data=self._config_data,
+        )

homeassistant/components/volvo/const.py

@@ -0,0 +1,43 @@
+"""Constants for the Volvo integration."""
+
+from homeassistant.const import Platform
+
+DOMAIN = "volvo"
+PLATFORMS: list[Platform] = [Platform.SENSOR]
+
+ATTR_API_TIMESTAMP = "api_timestamp"
+
+CONF_VIN = "vin"
+
+DATA_BATTERY_CAPACITY = "battery_capacity_kwh"
+
+MANUFACTURER = "Volvo"
+
+SCOPES = [
+    "openid",
+    "conve:battery_charge_level",
+    "conve:brake_status",
+    "conve:climatization_start_stop",
+    "conve:command_accessibility",
+    "conve:commands",
+    "conve:diagnostics_engine_status",
+    "conve:diagnostics_workshop",
+    "conve:doors_status",
+    "conve:engine_status",
+    "conve:fuel_status",
+    "conve:lock_status",
+    "conve:odometer_status",
+    "conve:trip_statistics",
+    "conve:tyre_status",
+    "conve:vehicle_relation",
+    "conve:warnings",
+    "conve:windows_status",
+    "energy:battery_charge_level",
+    "energy:charging_connection_status",
+    "energy:charging_current_limit",
+    "energy:charging_system_status",
+    "energy:electric_range",
+    "energy:estimated_charging_time",
+    "energy:recharge_status",
+    "energy:target_battery_level",
+]