deepseek

__init__.py

@@ -1,14 +0,0 @@
-from homeassistant.config_entries import ConfigEntry
-from homeassistant.core import HomeAssistant
-from homeassistant.helpers.typing import ConfigType
-from .const import DOMAIN
-
-async def async_setup(hass: HomeAssistant, config: ConfigType):
-    return True
-
-async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry):
-    await hass.config_entries.async_forward_entry_setups(entry, ["sensor"])
-    return True
-
-async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry):
-    return await hass.config_entries.async_unload_platforms(entry, ["sensor"])

config_flow.py

@@ -1,21 +1,93 @@
-from homeassistant import config_entries
+"""Config flow for Linky Tariff integration."""
+from __future__ import annotations
+
+import logging
 import voluptuous as vol
 
-from .const import DOMAIN, CONF_IEEE, CONF_INTERVAL, DEFAULT_INTERVAL
+from homeassistant import config_entries
+from homeassistant.core import callback
+from homeassistant.data_entry_flow import FlowResult
+from homeassistant.helpers import device_registry as dr
+
+from .const import (
+    DOMAIN,
+    CONF_IEEE,
+    CONF_POLL_INTERVAL,
+    DEFAULT_POLL_INTERVAL,
+)
+
+_LOGGER = logging.getLogger(__name__)
+
+async def async_get_devices(hass):
+    """Get ZHA devices."""
+    device_registry = dr.async_get(hass)
+    return [
+        entry
+        for entry in device_registry.devices.values()
+        if entry.via_device_id is not None and "zha" in entry.via_device_id
+    ]
+
+class LinkyTariffConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
+    """Handle a config flow for Linky Tariff."""
+
+    VERSION = 1
 
-class LinkyTarifConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
     async def async_step_user(self, user_input=None):
+        """Handle the initial step."""
         errors = {}
+        devices = await async_get_devices(self.hass)
+
+        if not devices:
+            return self.async_abort(reason="no_zha_devices")
 
         if user_input is not None:
-            return self.async_create_entry(title="Linky Tarif", data={
+            await self.async_set_unique_id(user_input[CONF_IEEE])
-                CONF_IEEE: user_input[CONF_IEEE],
+            self._abort_if_unique_id_configured()
-            }, options={
+            return self.async_create_entry(title="Linky Tariff", data=user_input)
-                CONF_INTERVAL: user_input.get(CONF_INTERVAL, DEFAULT_INTERVAL)
-            })
 
-        schema = vol.Schema({
+        device_options = {
-            vol.Required(CONF_IEEE): str,
+            device.id: f"{device.name_by_user or device.name} ({device.id})"
-            vol.Optional(CONF_INTERVAL, default=DEFAULT_INTERVAL): int,
+            for device in devices
-        })
+        }
-        return self.async_show_form(step_id="user", data_schema=schema, errors=errors)
+
+        return self.async_show_form(
+            step_id="user",
+            data_schema=vol.Schema({
+                vol.Required(CONF_IEEE): vol.In(device_options),
+                vol.Optional(
+                    CONF_POLL_INTERVAL,
+                    default=DEFAULT_POLL_INTERVAL
+                ): int,
+            }),
+            errors=errors,
+        )
+
+    @staticmethod
+    @callback
+    def async_get_options_flow(config_entry):
+        """Get the options flow for this handler."""
+        return LinkyTariffOptionsFlow(config_entry)
+
+class LinkyTariffOptionsFlow(config_entries.OptionsFlow):
+    """Handle options flow for Linky Tariff."""
+
+    def __init__(self, config_entry):
+        """Initialize options flow."""
+        self.config_entry = config_entry
+
+    async def async_step_init(self, user_input=None):
+        """Manage the options."""
+        if user_input is not None:
+            return self.async_create_entry(title="", data=user_input)
+
+        return self.async_show_form(
+            step_id="init",
+            data_schema=vol.Schema({
+                vol.Optional(
+                    CONF_POLL_INTERVAL,
+                    default=self.config_entry.options.get(
+                        CONF_POLL_INTERVAL, DEFAULT_POLL_INTERVAL
+                    ),
+                ): int,
+            }),
+        )

const.py

@@ -1,4 +1,13 @@
+"""Constants for the Linky Tariff integration."""
+from homeassistant.const import Platform
+
 DOMAIN = "linky_tarif"
+PLATFORMS = [Platform.SENSOR]
+
 CONF_IEEE = "ieee"
-CONF_INTERVAL = "interval"
+CONF_POLL_INTERVAL = "poll_interval"
-DEFAULT_INTERVAL = 300
+DEFAULT_POLL_INTERVAL = 60
+
+CLUSTER_ID = 0xFF66
+ATTRIBUTE_ID = 0x0010
+ENDPOINT_ID = 1

coordinator.py

@@ -1,49 +0,0 @@
-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):
-        try:
-            zha_gateway = self.hass.data["zha"]
-            app_ctrl = zha_gateway.application_controller
-        except (AttributeError, KeyError) as e:
-            raise UpdateFailed(f"ZHA Gateway not ready: {e}")
-
-        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}")

init.py

@@ -0,0 +1,94 @@
+"""The Linky Tariff integration."""
+from __future__ import annotations
+
+import logging
+from typing import Any
+
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.const import CONF_IEEE, CONF_SCAN_INTERVAL
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
+
+from .const import (
+    DOMAIN,
+    CONF_POLL_INTERVAL,
+    CLUSTER_ID,
+    ATTRIBUTE_ID,
+    ENDPOINT_ID,
+    PLATFORMS,
+)
+
+_LOGGER = logging.getLogger(__name__)
+
+async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
+    """Set up Linky Tariff from a config entry."""
+    hass.data.setdefault(DOMAIN, {})
+
+    coordinator = LinkyTariffCoordinator(
+        hass,
+        entry.data[CONF_IEEE],
+        entry.options.get(CONF_POLL_INTERVAL, entry.data.get(CONF_POLL_INTERVAL, 60)),
+    )
+
+    await coordinator.async_config_entry_first_refresh()
+
+    hass.data[DOMAIN][entry.entry_id] = coordinator
+
+    await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
+
+    entry.async_on_unload(entry.add_update_listener(async_update_options))
+
+    return True
+
+async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
+    """Unload a config entry."""
+    if unload_ok := await hass.config_entries.async_unload_platforms(entry, PLATFORMS):
+        hass.data[DOMAIN].pop(entry.entry_id)
+
+    return unload_ok
+
+async def async_update_options(hass: HomeAssistant, entry: ConfigEntry) -> None:
+    """Update options."""
+    await hass.config_entries.async_reload(entry.entry_id)
+
+class LinkyTariffCoordinator(DataUpdateCoordinator):
+    """Class to manage fetching Linky Tariff data."""
+
+    def __init__(self, hass: HomeAssistant, ieee: str, poll_interval: int) -> None:
+        """Initialize."""
+        super().__init__(
+            hass,
+            _LOGGER,
+            name=DOMAIN,
+            update_interval=poll_interval,
+        )
+        self.ieee = ieee
+        self._attr = None
+
+    async def _async_update_data(self) -> dict[str, Any]:
+        """Fetch data from ZHA device."""
+        try:
+            result = await self.hass.services.async_call(
+                "zha_toolkit",
+                "attr_read",
+                {
+                    "ieee": self.ieee,
+                    "endpoint_id": ENDPOINT_ID,
+                    "cluster_id": CLUSTER_ID,
+                    "attribute_id": ATTRIBUTE_ID,
+                    "manufacturer": None,
+                },
+                blocking=True,
+                return_response=True,
+            )
+            
+            if result and "value" in result:
+                return {
+                    "value": result["value"],
+                    "last_update": self.hass.config.time_utc(),
+                }
+            return {"value": "unknown", "last_update": self.hass.config.time_utc()}
+
+        except Exception as ex:
+            _LOGGER.error("Error reading Linky tariff: %s", ex)
+            return {"value": "error", "last_update": self.hass.config.time_utc()}

manifest.json

@@ -1,9 +1,12 @@
 {
   "domain": "linky_tarif",
-  "name": "Linky Tarif Sensor",
+  "name": "Linky Tariff Period",
   "version": "1.0.0",
   "config_flow": true,
   "documentation": "https://git.flavien.ovh/flavien/linky_tarif",
+  "codeowners": ["@flavien"],
   "requirements": [],
-  "dependencies": ["zha"]
+  "iot_class": "local_polling",
+  "loggers": ["zha_toolkit"],
+  "integration_type": "device"
 }

sensor.py

@@ -1,44 +1,46 @@
+"""Sensor platform for Linky Tariff integration."""
+from __future__ import annotations
+
 from homeassistant.components.sensor import SensorEntity
 from homeassistant.config_entries import ConfigEntry
 from homeassistant.core import HomeAssistant
 from homeassistant.helpers.entity_platform import AddEntitiesCallback
-from homeassistant.helpers.entity import DeviceInfo
+from homeassistant.helpers.update_coordinator import CoordinatorEntity
-from homeassistant.exceptions import ConfigEntryNotReady
 
-from .coordinator import LinkyTarifCoordinator
+from .const import DOMAIN
-from .const import DOMAIN, CONF_IEEE, CONF_INTERVAL
 
-async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback):
+async def async_setup_entry(
-    coordinator = LinkyTarifCoordinator(
+    hass: HomeAssistant,
-        hass,
+    entry: ConfigEntry,
-        ieee=entry.data[CONF_IEEE],
+    async_add_entities: AddEntitiesCallback,
-        interval=entry.options.get(CONF_INTERVAL, 300),
+) -> None:
-    )
+    """Set up the Linky Tariff sensor."""
+    coordinator = hass.data[DOMAIN][entry.entry_id]
+    async_add_entities([LinkyTariffSensor(coordinator)])
 
-    try:
+class LinkyTariffSensor(CoordinatorEntity, SensorEntity):
-        await coordinator.async_refresh()
+    """Representation of a Linky Tariff sensor."""
-        if coordinator.last_update_success is False:
-            raise ConfigEntryNotReady
-    except Exception:
-        raise ConfigEntryNotReady
 
-    async_add_entities([LinkyTarifSensor(coordinator)])
+    _attr_icon = "mdi:lightning-bolt"
+    _attr_has_entity_name = True
+    _attr_name = "Linky Tariff Period"
 
-class LinkyTarifSensor(SensorEntity):
+    def __init__(self, coordinator) -> None:
-    def __init__(self, coordinator: LinkyTarifCoordinator):
+        """Initialize the sensor."""
-        self.coordinator = coordinator
+        super().__init__(coordinator)
-        self._attr_unique_id = f"linky_tarif_{coordinator._ieee}"
+        self._attr_unique_id = f"{coordinator.ieee}_tariff_period"
-        self._attr_name = "Linky Tarif Period"
+        self._attr_device_info = {
-        self._attr_icon = "mdi:flash"
+            "identifiers": {(DOMAIN, coordinator.ieee)},
-        self._attr_device_info = DeviceInfo(identifiers={(DOMAIN, coordinator._ieee)})
+        }
 
     @property
-    def native_value(self):
+    def native_value(self) -> str:
-        return self.coordinator.data.get("tariff") if self.coordinator.data else None
+        """Return the state of the sensor."""
+        return self.coordinator.data["value"]
 
     @property
-    def extra_state_attributes(self):
+    def extra_state_attributes(self) -> dict[str, Any]:
-        return {"last_update": self.coordinator.data.get("last_update")} if self.coordinator.data else {}
+        """Return the state attributes."""
-
+        return {
-    async def async_update(self):
+            "last_update": self.coordinator.data["last_update"],
-        await self.coordinator.async_request_refresh()
+        }

strings.json

@@ -1,13 +0,0 @@
-{
-  "domain": "linky_tarif",
-  "name": "Linky Tarif",
-  "config_flow": {
-    "title": "Linky Tarif",
-    "description": "Monitors Linky energy meter tariff periods using ZHA.",
-    "step": {
-      "user": {
-        "title": "Configure Linky Tarif"
-      }
-    }
-  }
-}

translations/en.json

@@ -2,13 +2,15 @@
   "config": {
     "step": {
       "user": {
-        "title": "Configure Linky Tarif",
+        "title": "Configure Linky Tariff",
-        "description": "Enter the IEEE address of your ZLinky_TIC device.",
         "data": {
-          "ieee": "Device IEEE address",
+          "ieee": "ZHA Device",
-          "interval": "Polling interval (seconds)"
+          "poll_interval": "Polling Interval (seconds)"
         }
       }
+    },
+    "abort": {
+      "no_zha_devices": "No ZHA devices found. Please set up ZHA first."
     }
   }
-}
+}