custom_components/sws12500/__init__.py

@@ -28,7 +28,7 @@ period where no entities are subscribed, causing stale states until another full
 
 from asyncio import timeout
 import logging
-from typing import Any
+from typing import Any, cast
 
 from aiohttp import ClientConnectionError
 import aiohttp.web
@@ -242,16 +242,10 @@ class WeatherDataUpdateCoordinator(DataUpdateCoordinator):
 
         # Optional forwarding to external services. This is kept here (in the webhook handler)
         # to avoid additional background polling tasks.
-
+        if self.config.options.get(WINDY_ENABLED, False):
-        _windy_enabled = checked_or(self.config.options.get(WINDY_ENABLED), bool, False)
-        _pocasi_enabled = checked_or(
-            self.config.options.get(POCASI_CZ_ENABLED), bool, False
-        )
-
-        if _windy_enabled:
             await self.windy.push_data_to_windy(data, _wslink)
 
-        if _pocasi_enabled:
+        if self.config.options.get(POCASI_CZ_ENABLED, False):
             await self.pocasi.push_data_to_server(data, "WSLINK" if _wslink else "WU")
 
         # Convert raw payload keys to our internal sensor keys (stable identifiers).
@@ -408,17 +402,18 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
 
     """
 
-    hass_data = hass.data.setdefault(DOMAIN, {})
+    hass_data_any = hass.data.setdefault(DOMAIN, {})
-    # hass_data = cast("dict[str, Any]", hass_data_any)
+    hass_data = cast("dict[str, Any]", hass_data_any)
 
     # Per-entry runtime storage:
     # hass.data[DOMAIN][entry_id] is always a dict (never the coordinator itself).
     # Mixing types here (sometimes dict, sometimes coordinator) is a common source of hard-to-debug
     # issues where entities stop receiving updates.
-
+    entry_data_any = hass_data.get(entry.entry_id)
-    if (entry_data := checked(hass_data.get(entry.entry_id), dict[str, Any])) is None:
+    if not isinstance(entry_data_any, dict):
-        entry_data = {}
+        entry_data_any = {}
-    hass_data[entry.entry_id] = entry_data
+        hass_data[entry.entry_id] = entry_data_any
+    entry_data = cast("dict[str, Any]", entry_data_any)
 
     # Reuse the existing coordinator across reloads so webhook handlers and entities
     # remain connected to the same coordinator instance.
@@ -426,13 +421,14 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
     # Note: Routes store a bound method (`coordinator.received_data`). If we replaced the coordinator
     # instance on reload, the dispatcher could keep calling the old instance while entities listen
     # to the new one, causing updates to "disappear".
-    coordinator = entry_data.get(ENTRY_COORDINATOR)
+    coordinator_any = entry_data.get(ENTRY_COORDINATOR)
-    if isinstance(coordinator, WeatherDataUpdateCoordinator):
+    if isinstance(coordinator_any, WeatherDataUpdateCoordinator):
-        coordinator.config = entry
+        coordinator_any.config = entry
 
         # Recreate helper instances so they pick up updated options safely.
-        coordinator.windy = WindyPush(hass, entry)
+        coordinator_any.windy = WindyPush(hass, entry)
-        coordinator.pocasi = PocasiPush(hass, entry)
+        coordinator_any.pocasi = PocasiPush(hass, entry)
+        coordinator = coordinator_any
     else:
         coordinator = WeatherDataUpdateCoordinator(hass, entry)
         entry_data[ENTRY_COORDINATOR] = coordinator
@@ -486,16 +482,16 @@ async def update_listener(hass: HomeAssistant, entry: ConfigEntry):
     - Reloading a push-based integration temporarily unloads platforms and removes
       coordinator listeners, which can make the UI appear "stuck" until restart.
     """
+    hass_data_any = hass.data.get(DOMAIN)
+    if isinstance(hass_data_any, dict):
+        hass_data = cast("dict[str, Any]", hass_data_any)
+        entry_data_any = hass_data.get(entry.entry_id)
+        if isinstance(entry_data_any, dict):
+            entry_data = cast("dict[str, Any]", entry_data_any)
 
-    if (hass_data := checked(hass.data.get(DOMAIN), dict[str, Any])) is not None:
+            old_options_any = entry_data.get(ENTRY_LAST_OPTIONS)
-        if (
+            if isinstance(old_options_any, dict):
-            entry_data := checked(hass_data.get(entry.entry_id), dict[str, Any])
+                old_options = cast("dict[str, Any]", old_options_any)
-        ) is not None:
-            if (
-                old_options := checked(
-                    entry_data.get(ENTRY_LAST_OPTIONS), dict[str, Any]
-                )
-            ) is not None:
                 new_options = dict(entry.options)
 
                 changed_keys = {

custom_components/sws12500/health_sensor.py

@@ -6,15 +6,12 @@ This file is a helper module and must be wired from `sensor.py`.
 
 from __future__ import annotations
 
-from functools import cached_property
 from typing import Any, cast
 
-from homeassistant.components.sensor import SensorEntity, SensorEntityDescription
+from homeassistant.components.sensor import SensorEntity
 from homeassistant.config_entries import ConfigEntry
 from homeassistant.const import EntityCategory
 from homeassistant.core import HomeAssistant
-from homeassistant.helpers.device_registry import DeviceEntryType
-from homeassistant.helpers.entity import DeviceInfo
 from homeassistant.helpers.entity_platform import AddEntitiesCallback
 from homeassistant.helpers.update_coordinator import CoordinatorEntity
 
@@ -22,34 +19,6 @@ from .const import DOMAIN
 from .data import ENTRY_HEALTH_COORD
 
 
-class HealthSensorEntityDescription(SensorEntityDescription):
-    """Description for health diagnostic sensors."""
-
-
-HEALTH_SENSOR_DESCRIPTIONS: tuple[HealthSensorEntityDescription, ...] = (
-    HealthSensorEntityDescription(
-        key="Integration status",
-        name="Integration status",
-        icon="mdi:heart-pulse",
-    ),
-    HealthSensorEntityDescription(
-        key="HomeAssistant source_ip",
-        name="Home Assistant source IP",
-        icon="mdi:ip",
-    ),
-    HealthSensorEntityDescription(
-        key="HomeAssistant base_url",
-        name="Home Assistant base URL",
-        icon="mdi:link-variant",
-    ),
-    HealthSensorEntityDescription(
-        key="WSLink Addon response",
-        name="WSLink Addon response",
-        icon="mdi:server-network",
-    ),
-)
-
-
 async def async_setup_entry(
     hass: HomeAssistant,
     entry: ConfigEntry,
@@ -71,14 +40,7 @@ async def async_setup_entry(
     if coordinator_any is None:
         return
 
-    entities = [
+    async_add_entities([HealthDiagnosticSensor(coordinator_any, entry)])
-        HealthDiagnosticSensor(
-            coordinator=coordinator_any, entry=entry, description=description
-        )
-        for description in HEALTH_SENSOR_DESCRIPTIONS
-    ]
-    async_add_entities(entities)
-    # async_add_entities([HealthDiagnosticSensor(coordinator_any, entry)])
 
 
 class HealthDiagnosticSensor(  # pyright: ignore[reportIncompatibleVariableOverride]
@@ -89,19 +51,14 @@ class HealthDiagnosticSensor(  # pyright: ignore[reportIncompatibleVariableOverr
     _attr_has_entity_name = True
     _attr_should_poll = False
 
-    def __init__(
+    def __init__(self, coordinator: Any, entry: ConfigEntry) -> None:
-        self,
-        coordinator: Any,
-        entry: ConfigEntry,
-        description: HealthSensorEntityDescription,
-    ) -> None:
         """Initialize the sensor."""
         super().__init__(coordinator)
 
         self._attr_entity_category = EntityCategory.DIAGNOSTIC
-        self._attr_unique_id = f"{description.key}_health"
+        self._attr_unique_id = f"{entry.entry_id}_health"
-        # self._attr_name = description.name
+        self._attr_name = "Health"
-        # self._attr_icon = "mdi:heart-pulse"
+        self._attr_icon = "mdi:heart-pulse"
 
     @property
     def native_value(self) -> str | None:  # pyright: ignore[reportIncompatibleVariableOverride]
@@ -119,15 +76,3 @@ class HealthDiagnosticSensor(  # pyright: ignore[reportIncompatibleVariableOverr
         if not isinstance(data_any, dict):
             return None
         return cast("dict[str, Any]", data_any)
-
-    @cached_property
-    def device_info(self) -> DeviceInfo:
-        """Device info."""
-        return DeviceInfo(
-            connections=set(),
-            name="Weather Station SWS 12500",
-            entry_type=DeviceEntryType.SERVICE,
-            identifiers={(DOMAIN,)},  # type: ignore[arg-type]
-            manufacturer="Schizza",
-            model="Weather Station SWS 12500",
-        )

custom_components/sws12500/sensor.py

@@ -20,7 +20,7 @@ from functools import cached_property
 import logging
 from typing import Any, cast
 
-from py_typecheck import checked, checked_or
+from py_typecheck import checked_or
 
 from homeassistant.components.sensor import SensorEntity
 from homeassistant.config_entries import ConfigEntry
@@ -92,17 +92,15 @@ async def async_setup_entry(
     so the webhook handler can add newly discovered entities dynamically without
     reloading the config entry.
     """
+    hass_data_any = hass.data.setdefault(DOMAIN, {})
+    hass_data = cast("dict[str, Any]", hass_data_any)
 
-    if (hass_data := checked(hass.data.setdefault(DOMAIN, {}), dict[str, Any])) is None:
+    entry_data_any = hass_data.get(config_entry.entry_id)
-        return
+    if not isinstance(entry_data_any, dict):
-
+        # Created by the integration setup, but keep this defensive for safety.
-    # we have to check if entry_data are present
+        entry_data_any = {}
-    # It is created by integration setup, so it should be presnet
+        hass_data[config_entry.entry_id] = entry_data_any
-    if (
+    entry_data = cast("dict[str, Any]", entry_data_any)
-        entry_data := checked(hass_data.get(config_entry.entry_id), dict[str, Any])
-    ) is None:
-        # This should not happen in normal operation.
-        return
 
     coordinator = entry_data.get(ENTRY_COORDINATOR)
     if coordinator is None:
@@ -153,31 +151,34 @@ def add_new_sensors(
       finished setting up yet (e.g. callback/description map missing).
     - Unknown payload keys are ignored (only keys with an entity description are added).
     """
+    hass_data_any = hass.data.get(DOMAIN)
+    if not isinstance(hass_data_any, dict):
+        return
+    hass_data = cast("dict[str, Any]", hass_data_any)
 
-    if (hass_data := checked(hass.data.get(DOMAIN), dict[str, Any])) is None:
+    entry_data_any = hass_data.get(config_entry.entry_id)
+    if not isinstance(entry_data_any, dict):
+        return
+    entry_data = cast("dict[str, Any]", entry_data_any)
+
+    add_entities_any = entry_data.get(ENTRY_ADD_ENTITIES)
+    descriptions_any = entry_data.get(ENTRY_DESCRIPTIONS)
+    coordinator_any = entry_data.get(ENTRY_COORDINATOR)
+
+    if add_entities_any is None or descriptions_any is None or coordinator_any is None:
         return
 
-    if (
+    add_entities_fn = cast("_AddEntitiesFn", add_entities_any)
-        entry_data := checked(hass_data.get(config_entry.entry_id), dict[str, Any])
+    descriptions_map = cast(
-    ) is None:
+        "dict[str, WeatherSensorEntityDescription]", descriptions_any
-        return
+    )
-
-    add_entities = entry_data.get(ENTRY_ADD_ENTITIES)
-    descriptions = entry_data.get(ENTRY_DESCRIPTIONS)
-    coordinator = entry_data.get(ENTRY_COORDINATOR)
-
-    if add_entities is None or descriptions is None or coordinator is None:
-        return
-
-    add_entities_fn = cast("_AddEntitiesFn", add_entities)
-    descriptions_map = cast("dict[str, WeatherSensorEntityDescription]", descriptions)
 
     new_entities: list[SensorEntity] = []
     for key in keys:
         desc = descriptions_map.get(key)
         if desc is None:
             continue
-        new_entities.append(WeatherSensor(desc, coordinator))
+        new_entities.append(WeatherSensor(desc, coordinator_any))
 
     if new_entities:
         add_entities_fn(new_entities)