diff --git a/homeassistant/components/tibber/coordinator.py b/homeassistant/components/tibber/coordinator.py index 1edb5e932692..1110f81f621d 100644 --- a/homeassistant/components/tibber/coordinator.py +++ b/homeassistant/components/tibber/coordinator.py @@ -25,7 +25,6 @@ from homeassistant.components.recorder.statistics import ( ) from homeassistant.const import UnitOfEnergy from homeassistant.core import HomeAssistant -from homeassistant.exceptions import ConfigEntryAuthFailed from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed from homeassistant.util import dt as dt_util from homeassistant.util.unit_conversion import EnergyConverter @@ -92,11 +91,40 @@ def _build_home_data(home: tibber.TibberHome) -> TibberHomeData: return result -class TibberDataCoordinator(DataUpdateCoordinator[None]): - """Handle Tibber data and insert statistics.""" +class TibberCoordinator[_DataT](DataUpdateCoordinator[_DataT]): + """Base Tibber coordinator.""" config_entry: TibberConfigEntry + def __init__( + self, + hass: HomeAssistant, + config_entry: TibberConfigEntry, + *, + name: str, + update_interval: timedelta, + ) -> None: + """Initialize the coordinator.""" + super().__init__( + hass, + _LOGGER, + config_entry=config_entry, + name=name, + update_interval=update_interval, + ) + self._runtime_data = config_entry.runtime_data + + async def _async_get_client(self) -> tibber.Tibber: + """Get the Tibber client with error handling.""" + try: + return await self._runtime_data.async_get_client(self.hass) + except (ClientError, TimeoutError, tibber.exceptions.HttpExceptionError) as err: + raise UpdateFailed(f"Unable to create Tibber client: {err}") from err + + +class TibberDataCoordinator(TibberCoordinator[None]): + """Handle Tibber data and insert statistics.""" + def __init__( self, hass: HomeAssistant, @@ -106,17 +134,14 @@ class TibberDataCoordinator(DataUpdateCoordinator[None]): """Initialize the data handler.""" super().__init__( hass, - _LOGGER, - config_entry=config_entry, + config_entry, name=f"Tibber {tibber_connection.name}", update_interval=timedelta(minutes=20), ) async def _async_update_data(self) -> None: """Update data via API.""" - tibber_connection = await self.config_entry.runtime_data.async_get_client( - self.hass - ) + tibber_connection = await self._async_get_client() try: await tibber_connection.fetch_consumption_data_active_homes() @@ -132,9 +157,7 @@ class TibberDataCoordinator(DataUpdateCoordinator[None]): async def _insert_statistics(self) -> None: """Insert Tibber statistics.""" - tibber_connection = await self.config_entry.runtime_data.async_get_client( - self.hass - ) + tibber_connection = await self._async_get_client() for home in tibber_connection.get_homes(): sensors: list[tuple[str, bool, str | None, str]] = [] if home.hourly_consumption_data: @@ -254,11 +277,9 @@ class TibberDataCoordinator(DataUpdateCoordinator[None]): async_add_external_statistics(self.hass, metadata, statistics) -class TibberPriceCoordinator(DataUpdateCoordinator[dict[str, TibberHomeData]]): +class TibberPriceCoordinator(TibberCoordinator[dict[str, TibberHomeData]]): """Handle Tibber price data and insert statistics.""" - config_entry: TibberConfigEntry - def __init__( self, hass: HomeAssistant, @@ -267,8 +288,7 @@ class TibberPriceCoordinator(DataUpdateCoordinator[dict[str, TibberHomeData]]): """Initialize the price coordinator.""" super().__init__( hass, - _LOGGER, - config_entry=config_entry, + config_entry, name=f"{DOMAIN} price", update_interval=timedelta(minutes=1), ) @@ -290,9 +310,7 @@ class TibberPriceCoordinator(DataUpdateCoordinator[dict[str, TibberHomeData]]): async def _async_update_data(self) -> dict[str, TibberHomeData]: """Update data via API and return per-home data for sensors.""" - tibber_connection = await self.config_entry.runtime_data.async_get_client( - self.hass - ) + tibber_connection = await self._async_get_client() active_homes = tibber_connection.get_homes(only_active=True) now = dt_util.now() @@ -347,11 +365,9 @@ class TibberPriceCoordinator(DataUpdateCoordinator[dict[str, TibberHomeData]]): return result -class TibberDataAPICoordinator(DataUpdateCoordinator[dict[str, TibberDevice]]): +class TibberDataAPICoordinator(TibberCoordinator[dict[str, TibberDevice]]): """Fetch and cache Tibber Data API device capabilities.""" - config_entry: TibberConfigEntry - def __init__( self, hass: HomeAssistant, @@ -360,12 +376,10 @@ class TibberDataAPICoordinator(DataUpdateCoordinator[dict[str, TibberDevice]]): """Initialize the coordinator.""" super().__init__( hass, - _LOGGER, + entry, name=f"{DOMAIN} Data API", update_interval=timedelta(minutes=1), - config_entry=entry, ) - self._runtime_data = entry.runtime_data self.sensors_by_device: dict[str, dict[str, tibber.data_api.Sensor]] = {} def _build_sensor_lookup(self, devices: dict[str, TibberDevice]) -> None: @@ -383,15 +397,6 @@ class TibberDataAPICoordinator(DataUpdateCoordinator[dict[str, TibberDevice]]): return device_sensors.get(sensor_id) return None - async def _async_get_client(self) -> tibber.Tibber: - """Get the Tibber client with error handling.""" - try: - return await self._runtime_data.async_get_client(self.hass) - except ConfigEntryAuthFailed: - raise - except (ClientError, TimeoutError, tibber.UserAgentMissingError) as err: - raise UpdateFailed(f"Unable to create Tibber client: {err}") from err - async def _async_setup(self) -> None: """Initial load of Tibber Data API devices.""" client = await self._async_get_client()