Signed-off-by: Daniel Hjelseth Høyer <github@dahoiv.net>
This commit is contained in:
Daniel Hjelseth Høyer
2026-02-15 12:37:13 +01:00
parent ca2dc20709
commit cfc85cfd29
5 changed files with 39 additions and 51 deletions

View File

@@ -23,7 +23,7 @@ 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
from .coordinator import TibberCoordinator, TibberDataAPICoordinator
from .services import async_setup_services
PLATFORMS = [Platform.BINARY_SENSOR, Platform.NOTIFY, Platform.SENSOR]
@@ -39,7 +39,7 @@ class TibberRuntimeData:
session: OAuth2Session
data_api_coordinator: TibberDataAPICoordinator | None = field(default=None)
data_coordinator: TibberDataCoordinator | None = field(default=None)
data_coordinator: TibberCoordinator | None = field(default=None)
_client: tibber.Tibber | None = None
async def async_get_client(self, hass: HomeAssistant) -> tibber.Tibber:
@@ -129,7 +129,7 @@ async def async_setup_entry(hass: HomeAssistant, entry: TibberConfigEntry) -> bo
await data_api_coordinator.async_config_entry_first_refresh()
entry.runtime_data.data_api_coordinator = data_api_coordinator
data_coordinator = TibberDataCoordinator(hass, entry, entry.runtime_data)
data_coordinator = TibberCoordinator(hass, entry, entry.runtime_data)
await data_coordinator.async_config_entry_first_refresh()
entry.runtime_data.data_coordinator = data_coordinator

View File

@@ -88,7 +88,7 @@ def _build_home_data(home: TibberHome) -> TibberHomeData:
)
class TibberDataCoordinator(DataUpdateCoordinator[dict[str, TibberHomeData]]):
class TibberCoordinator(DataUpdateCoordinator[dict[str, TibberHomeData]]):
"""Handle Tibber data, insert statistics, and expose per-home data for sensors."""
config_entry: TibberConfigEntry

View File

@@ -12,20 +12,20 @@ from homeassistant.helpers.update_coordinator import CoordinatorEntity
from homeassistant.util import dt as dt_util
from .const import DOMAIN
from .coordinator import TibberDataCoordinator, TibberHomeData, TibberRtDataCoordinator
from .coordinator import TibberCoordinator, TibberHomeData, TibberRtDataCoordinator
if TYPE_CHECKING:
from tibber import TibberHome
class TibberDataCoordinatorEntity(CoordinatorEntity[TibberDataCoordinator]):
"""Base entity for Tibber sensors using TibberDataCoordinator."""
class TibberCoordinatorEntity(CoordinatorEntity[TibberCoordinator]):
"""Base entity for Tibber sensors using TibberCoordinator."""
_attr_has_entity_name = True
def __init__(
self,
coordinator: TibberDataCoordinator,
coordinator: TibberCoordinator,
tibber_home: TibberHome,
) -> None:
"""Initialize the entity."""
@@ -52,34 +52,7 @@ class TibberDataCoordinatorEntity(CoordinatorEntity[TibberDataCoordinator]):
)
class TibberSensor:
"""Mixin for Tibber sensors that have a Tibber home and device info.
Used as the first base for real-time sensors (TibberSensorRT with
CoordinatorEntity["TibberRtDataCoordinator"]). Provides _tibber_home,
_home_name, _model, _device_name and device_info; does not inherit
CoordinatorEntity so the second base can be the coordinator entity.
"""
def __init__(self, coordinator: object, tibber_home: TibberHome) -> None:
"""Initialize the mixin."""
super().__init__(coordinator) # type: ignore[call-arg]
self._tibber_home = tibber_home
self._home_name: str = tibber_home.name or tibber_home.home_id
self._model: str | None = None
self._device_name: str = self._home_name
@property
def device_info(self) -> DeviceInfo:
"""Return device information."""
return DeviceInfo(
identifiers={(DOMAIN, self._tibber_home.home_id)},
name=self._device_name,
model=self._model,
)
class TibberSensorRT(TibberSensor, CoordinatorEntity[TibberRtDataCoordinator]):
class TibberRTCoordinatorEntity(CoordinatorEntity[TibberRtDataCoordinator]):
"""Representation of a Tibber sensor for real time consumption."""
def __init__(
@@ -90,10 +63,12 @@ class TibberSensorRT(TibberSensor, CoordinatorEntity[TibberRtDataCoordinator]):
coordinator: TibberRtDataCoordinator,
) -> None:
"""Initialize the sensor."""
super().__init__(coordinator=coordinator, tibber_home=tibber_home)
super().__init__(coordinator)
self._tibber_home = tibber_home
self._home_name: str = tibber_home.name or tibber_home.home_id
self._model: str = "Tibber Pulse"
self._device_name: str = f"{self._model} {self._home_name}"
self.entity_description = description
self._model = "Tibber Pulse"
self._device_name = f"{self._model} {self._home_name}"
self._attr_native_value = initial_state
self._attr_last_reset: datetime | None = None
@@ -102,6 +77,15 @@ class TibberSensorRT(TibberSensor, CoordinatorEntity[TibberRtDataCoordinator]):
if description.key in ("accumulatedCost", "accumulatedReward"):
self._attr_native_unit_of_measurement = tibber_home.currency
@property
def device_info(self) -> DeviceInfo:
"""Return device information."""
return DeviceInfo(
identifiers={(DOMAIN, self._tibber_home.home_id)},
name=self._device_name,
model=self._model,
)
@property
def available(self) -> bool:
"""Return True if entity is available."""

View File

@@ -36,11 +36,11 @@ from homeassistant.helpers.update_coordinator import CoordinatorEntity
from .const import DOMAIN, TibberConfigEntry
from .coordinator import (
TibberCoordinator,
TibberDataAPICoordinator,
TibberDataCoordinator,
TibberRtDataCoordinator,
)
from .entity import TibberDataCoordinatorEntity, TibberSensorRT
from .entity import TibberCoordinatorEntity, TibberRTCoordinatorEntity
_LOGGER = logging.getLogger(__name__)
@@ -674,17 +674,15 @@ async def _async_setup_graphql_sensors(
).async_set_updated_data
)
entities: list[TibberDataSensor] = []
entities: list[TibberSensor] = []
coordinator = entry.runtime_data.data_coordinator
if coordinator is not None and active_homes:
for home in active_homes:
entities.extend(
TibberDataSensor(home, coordinator, desc, model="Price Sensor")
TibberSensor(home, coordinator, desc, model="Price Sensor")
for desc in PRICE_SENSORS
)
entities.extend(
TibberDataSensor(home, coordinator, desc) for desc in SENSORS
)
entities.extend(TibberSensor(home, coordinator, desc) for desc in SENSORS)
async_add_entities(entities)
@@ -746,13 +744,13 @@ class TibberDataAPISensor(CoordinatorEntity[TibberDataAPICoordinator], SensorEnt
return sensor.value if sensor else None
class TibberDataSensor(TibberDataCoordinatorEntity):
class TibberSensor(TibberCoordinatorEntity):
"""Representation of a Tibber sensor reading from coordinator data."""
def __init__(
self,
tibber_home: TibberHome,
coordinator: TibberDataCoordinator,
coordinator: TibberCoordinator,
entity_description: SensorEntityDescription,
*,
model: str | None = None,
@@ -777,7 +775,13 @@ class TibberDataSensor(TibberDataCoordinatorEntity):
"""Return the value of the sensor from coordinator data."""
home_data = self._get_home_data()
if home_data is None:
_LOGGER.error("Home data not found for home %s", self._tibber_home.home_id)
return None
_LOGGER.error(
"Home data found for home %s: %s",
self._tibber_home.home_id,
getattr(home_data, self.entity_description.key, None),
)
return cast(
StateType,
getattr(home_data, self.entity_description.key, None),
@@ -884,7 +888,7 @@ class TibberRtEntityCreator:
continue
self._migrate_unique_id(sensor_description)
entity = TibberSensorRT(
entity = TibberRTCoordinatorEntity(
self._tibber_home,
sensor_description,
state,

View File

@@ -4,7 +4,7 @@ from unittest.mock import AsyncMock
from homeassistant.components.recorder import Recorder
from homeassistant.components.recorder.statistics import statistics_during_period
from homeassistant.components.tibber.coordinator import TibberDataCoordinator
from homeassistant.components.tibber.coordinator import TibberCoordinator
from homeassistant.core import HomeAssistant
from homeassistant.util import dt as dt_util
@@ -24,7 +24,7 @@ async def test_async_setup_entry(
tibber_connection.fetch_production_data_active_homes.return_value = None
tibber_connection.get_homes = mock_get_homes
coordinator = TibberDataCoordinator(hass, config_entry, tibber_connection)
coordinator = TibberCoordinator(hass, config_entry, tibber_connection)
await coordinator._async_update_data()
await async_wait_recording_done(hass)