mirror of
https://github.com/home-assistant/core.git
synced 2026-02-23 03:17:06 +00:00
129 lines
4.2 KiB
Python
129 lines
4.2 KiB
Python
"""The Volvo integration."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
|
|
from volvocarsapi.api import VolvoCarsApi
|
|
from volvocarsapi.models import VolvoApiException, VolvoAuthException, VolvoCarsVehicle
|
|
|
|
from homeassistant.const import CONF_API_KEY
|
|
from homeassistant.core import HomeAssistant
|
|
from homeassistant.exceptions import (
|
|
ConfigEntryAuthFailed,
|
|
ConfigEntryError,
|
|
ConfigEntryNotReady,
|
|
)
|
|
from homeassistant.helpers.aiohttp_client import async_get_clientsession
|
|
from homeassistant.helpers.config_entry_oauth2_flow import (
|
|
ImplementationUnavailableError,
|
|
OAuth2Session,
|
|
async_get_config_entry_implementation,
|
|
)
|
|
|
|
from .api import VolvoAuth
|
|
from .const import CONF_VIN, DOMAIN, PLATFORMS
|
|
from .coordinator import (
|
|
VolvoConfigEntry,
|
|
VolvoContext,
|
|
VolvoFastIntervalCoordinator,
|
|
VolvoMediumIntervalCoordinator,
|
|
VolvoRuntimeData,
|
|
VolvoSlowIntervalCoordinator,
|
|
VolvoVerySlowIntervalCoordinator,
|
|
)
|
|
|
|
|
|
async def async_setup_entry(hass: HomeAssistant, entry: VolvoConfigEntry) -> bool:
|
|
"""Set up Volvo from a config entry."""
|
|
|
|
api = await _async_auth_and_create_api(hass, entry)
|
|
context = await _async_create_context(api)
|
|
|
|
# Order is important! Faster intervals must come first.
|
|
# Different interval coordinators are in place to keep the number
|
|
# of requests under 5000 per day. This lets users use the same
|
|
# API key for two vehicles (as the limit is 10000 per day).
|
|
coordinators = (
|
|
VolvoFastIntervalCoordinator(hass, entry, context),
|
|
VolvoMediumIntervalCoordinator(hass, entry, context),
|
|
VolvoSlowIntervalCoordinator(hass, entry, context),
|
|
VolvoVerySlowIntervalCoordinator(hass, entry, context),
|
|
)
|
|
await asyncio.gather(*(c.async_config_entry_first_refresh() for c in coordinators))
|
|
|
|
entry.runtime_data = VolvoRuntimeData(coordinators, context)
|
|
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."""
|
|
return await hass.config_entries.async_unload_platforms(entry, PLATFORMS)
|
|
|
|
|
|
async def _async_auth_and_create_api(
|
|
hass: HomeAssistant, entry: VolvoConfigEntry
|
|
) -> VolvoCarsApi:
|
|
try:
|
|
implementation = await async_get_config_entry_implementation(hass, entry)
|
|
except ImplementationUnavailableError as err:
|
|
raise ConfigEntryNotReady(
|
|
translation_domain=DOMAIN,
|
|
translation_key="oauth2_implementation_unavailable",
|
|
) from err
|
|
oauth_session = OAuth2Session(hass, entry, implementation)
|
|
web_session = async_get_clientsession(hass)
|
|
auth = VolvoAuth(web_session, oauth_session)
|
|
api = VolvoCarsApi(
|
|
web_session,
|
|
auth,
|
|
entry.data[CONF_API_KEY],
|
|
entry.data[CONF_VIN],
|
|
)
|
|
|
|
try:
|
|
await api.async_get_access_token()
|
|
except VolvoAuthException as err:
|
|
raise ConfigEntryAuthFailed from err
|
|
except VolvoApiException as err:
|
|
raise ConfigEntryNotReady from err
|
|
|
|
return api
|
|
|
|
|
|
async def _async_create_context(api: VolvoCarsApi) -> VolvoContext:
|
|
vehicle = await _async_load_vehicle(api)
|
|
supported_commands = await _async_load_supported_commands(api)
|
|
return VolvoContext(api, vehicle, supported_commands)
|
|
|
|
|
|
async def _async_load_vehicle(api: VolvoCarsApi) -> VolvoCarsVehicle:
|
|
try:
|
|
vehicle = await api.async_get_vehicle_details()
|
|
except VolvoAuthException as ex:
|
|
raise ConfigEntryAuthFailed(
|
|
translation_domain=DOMAIN,
|
|
translation_key="unauthorized",
|
|
translation_placeholders={"message": ex.message},
|
|
) from ex
|
|
|
|
if vehicle is None:
|
|
raise ConfigEntryError(translation_domain=DOMAIN, translation_key="no_vehicle")
|
|
|
|
return vehicle
|
|
|
|
|
|
async def _async_load_supported_commands(api: VolvoCarsApi) -> list[str]:
|
|
try:
|
|
commands = await api.async_get_commands()
|
|
except VolvoAuthException as ex:
|
|
raise ConfigEntryAuthFailed(
|
|
translation_domain=DOMAIN,
|
|
translation_key="unauthorized",
|
|
translation_placeholders={"message": ex.message},
|
|
) from ex
|
|
|
|
return [c.command for c in commands if c is not None]
|