core/homeassistant/components/tibber/__init__.py

"""Support for Tibber."""

import asyncio
from dataclasses import dataclass, field
import logging
from typing import Final

import aiohttp
from aiohttp.client_exceptions import ClientError
import tibber

from homeassistant.const import CONF_ACCESS_TOKEN, EVENT_HOMEASSISTANT_STOP, Platform
from homeassistant.core import Event, HomeAssistant
from homeassistant.exceptions import (
    ConfigEntryAuthFailed,
    ConfigEntryNotReady,
    OAuth2TokenRequestError,
    OAuth2TokenRequestReauthError,
)
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 homeassistant.util import dt as dt_util, ssl as ssl_util

from .const import AUTH_IMPLEMENTATION, DATA_HASS_CONFIG, DOMAIN, TibberConfigEntry
from .coordinator import (
    TibberDataAPICoordinator,
    TibberDataCoordinator,
    TibberFetchPriceCoordinator,
    TibberPriceCoordinator,
)
from .services import async_setup_services

PLATFORMS = [Platform.BINARY_SENSOR, Platform.NOTIFY, Platform.SENSOR]

DISCONNECT_TIMEOUT: Final = 10

CONFIG_SCHEMA = cv.config_entry_only_config_schema(DOMAIN)

_LOGGER = logging.getLogger(__name__)


@dataclass
class TibberRuntimeData:
    """Runtime data for Tibber API entries."""

    session: OAuth2Session
    data_api_coordinator: TibberDataAPICoordinator | None = field(default=None)
    data_coordinator: TibberDataCoordinator | None = field(default=None)
    fetch_price_coordinator: TibberFetchPriceCoordinator | None = field(default=None)
    price_coordinator: TibberPriceCoordinator | None = field(default=None)
    _client: tibber.Tibber | None = None

    async def _async_get_access_token(self) -> str:
        """Return a valid Tibber access token."""
        await self.session.async_ensure_token_valid()
        token = self.session.token
        access_token: str | None = token.get(CONF_ACCESS_TOKEN)
        if not access_token:
            raise ConfigEntryAuthFailed("Access token missing from OAuth session")
        return access_token

    async def async_get_client(self, hass: HomeAssistant) -> tibber.Tibber:
        """Return an authenticated Tibber client."""
        access_token = await self._async_get_access_token()
        if self._client is None:
            self._client = tibber.Tibber(
                access_token=access_token,
                websession=async_get_clientsession(hass),
                time_zone=dt_util.get_default_time_zone(),
                ssl=ssl_util.get_default_context(),
                refresh_access_token=self._async_get_access_token,
            )
        else:
            await self._client.set_access_token(access_token)
        return self._client

    async def async_disconnect(self) -> None:
        """Disconnect the cached realtime connection without raising."""
        if self._client is None:
            return
        try:
            async with asyncio.timeout(DISCONNECT_TIMEOUT):
                await self._client.rt_disconnect()
        except Exception:
            _LOGGER.warning(
                "Error disconnecting the Tibber realtime connection", exc_info=True
            )


async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
    """Set up the Tibber component."""

    hass.data[DATA_HASS_CONFIG] = config

    async_setup_services(hass)

    return True


async def async_setup_entry(hass: HomeAssistant, entry: TibberConfigEntry) -> bool:
    """Set up a config entry."""

    # Added in 2026.1 to migrate existing users to OAuth2 (Tibber Data API).
    # Can be removed after 2026.7
    if AUTH_IMPLEMENTATION not in entry.data:
        raise ConfigEntryAuthFailed(
            translation_domain=DOMAIN,
            translation_key="data_api_reauth_required",
        )

    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

    session = OAuth2Session(hass, entry, implementation)
    try:
        await session.async_ensure_token_valid()
    except OAuth2TokenRequestReauthError as err:
        raise ConfigEntryAuthFailed(
            "OAuth session is not valid, reauthentication required"
        ) from err
    except (OAuth2TokenRequestError, ClientError) as err:
        raise ConfigEntryNotReady from err

    entry.runtime_data = TibberRuntimeData(
        session=session,
    )

    tibber_connection = await entry.runtime_data.async_get_client(hass)

    async def _close(event: Event) -> None:
        await entry.runtime_data.async_disconnect()

    entry.async_on_unload(hass.bus.async_listen_once(EVENT_HOMEASSISTANT_STOP, _close))

    try:
        await tibber_connection.update_info()
    except (
        TimeoutError,
        aiohttp.ClientError,
        tibber.RetryableHttpExceptionError,
    ) as err:
        raise ConfigEntryNotReady("Unable to connect") from err
    except tibber.InvalidLoginError as err:
        raise ConfigEntryAuthFailed("Invalid login credentials") from err
    except tibber.FatalHttpExceptionError as err:
        raise ConfigEntryNotReady("Fatal HTTP error from Tibber API") from err

    if tibber_connection.get_homes(only_active=True):
        fetch_price_coordinator = TibberFetchPriceCoordinator(hass, entry)
        await fetch_price_coordinator.async_config_entry_first_refresh()
        entry.runtime_data.fetch_price_coordinator = fetch_price_coordinator

        price_coordinator = TibberPriceCoordinator(hass, entry, fetch_price_coordinator)
        await price_coordinator.async_config_entry_first_refresh()
        entry.runtime_data.price_coordinator = price_coordinator

        data_coordinator = TibberDataCoordinator(hass, entry, tibber_connection)
        await data_coordinator.async_config_entry_first_refresh()
        entry.runtime_data.data_coordinator = data_coordinator

    coordinator = TibberDataAPICoordinator(hass, entry)
    await coordinator.async_config_entry_first_refresh()
    entry.runtime_data.data_api_coordinator = coordinator

    await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
    return True


async def async_unload_entry(
    hass: HomeAssistant, config_entry: TibberConfigEntry
) -> bool:
    """Unload a config entry."""
    if unload_ok := await hass.config_entries.async_unload_platforms(
        config_entry, PLATFORMS
    ):
        await config_entry.runtime_data.async_disconnect()
    return unload_ok