mirror of
https://github.com/home-assistant/core.git
synced 2026-08-24 10:13:52 -05:00
152 lines
4.7 KiB
Python
152 lines
4.7 KiB
Python
"""The Volvo integration."""
|
|
|
|
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 import config_validation as cv
|
|
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 homeassistant.helpers.typing import ConfigType
|
|
|
|
from .api import VolvoAuth
|
|
from .const import CONF_VIN, DOMAIN, PLATFORMS
|
|
from .coordinator import (
|
|
VolvoConfigEntry,
|
|
VolvoContext,
|
|
VolvoFastIntervalCoordinator,
|
|
VolvoMediumIntervalCoordinator,
|
|
VolvoRuntimeData,
|
|
VolvoSlowIntervalCoordinator,
|
|
VolvoVerySlowIntervalCoordinator,
|
|
)
|
|
from .services import async_setup_services
|
|
|
|
CONFIG_SCHEMA = cv.config_entry_only_config_schema(DOMAIN)
|
|
|
|
|
|
async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
|
|
"""Set up Volvo integration."""
|
|
|
|
await async_setup_services(hass)
|
|
return True
|
|
|
|
|
|
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)
|
|
|
|
# Different interval coordinators are in place to keep the number
|
|
# of requests under 10000 per day.
|
|
fast_coordinator = VolvoFastIntervalCoordinator(hass, entry, context)
|
|
medium_coordinator = VolvoMediumIntervalCoordinator(hass, entry, context)
|
|
slow_coordinator = VolvoSlowIntervalCoordinator(hass, entry, context)
|
|
very_slow_coordinator = VolvoVerySlowIntervalCoordinator(hass, entry, context)
|
|
|
|
await asyncio.gather(
|
|
*(
|
|
c.async_config_entry_first_refresh()
|
|
for c in (
|
|
fast_coordinator,
|
|
medium_coordinator,
|
|
slow_coordinator,
|
|
very_slow_coordinator,
|
|
)
|
|
)
|
|
)
|
|
|
|
entry.runtime_data = VolvoRuntimeData(
|
|
fast_coordinator,
|
|
medium_coordinator,
|
|
slow_coordinator,
|
|
very_slow_coordinator,
|
|
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]
|