47 lines
1.5 KiB
Python
47 lines
1.5 KiB
Python
from datetime import timedelta
|
|
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
|
|
from homeassistant.core import HomeAssistant
|
|
from homeassistant.helpers.entity import DeviceInfo
|
|
from zigpy import types
|
|
import logging
|
|
|
|
from .const import DOMAIN
|
|
|
|
_LOGGER = logging.getLogger(__name__)
|
|
|
|
class LinkyTarifCoordinator(DataUpdateCoordinator):
|
|
def __init__(self, hass: HomeAssistant, ieee: str, interval: int):
|
|
super().__init__(
|
|
hass,
|
|
_LOGGER,
|
|
name="Linky Tarif Coordinator",
|
|
update_interval=timedelta(seconds=interval),
|
|
)
|
|
self._ieee = ieee
|
|
|
|
async def _async_update_data(self):
|
|
zha_storage = self.hass.data["zha"]
|
|
app_ctrl = zha_storage.gateway.application
|
|
|
|
device = app_ctrl.get_device(types.EUI64.convert(self._ieee))
|
|
if not device:
|
|
raise UpdateFailed(f"Device with IEEE {self._ieee} not found")
|
|
|
|
endpoint = device.endpoints.get(1)
|
|
if not endpoint:
|
|
raise UpdateFailed("Endpoint 1 not found")
|
|
|
|
cluster = endpoint[0xFF66]
|
|
if not cluster:
|
|
raise UpdateFailed("Cluster 0xFF66 not found")
|
|
|
|
try:
|
|
res = await cluster.read_attributes([0x0010])
|
|
value = res.get(0x0010)
|
|
return {
|
|
"tariff": value,
|
|
"last_update": self.hass.helpers.event.dt_util.utcnow().isoformat(),
|
|
}
|
|
except Exception as e:
|
|
raise UpdateFailed(f"Error reading tariff: {e}")
|