Michal
/
SWS-12500-custom-component
mirror of https://github.com/schizza/SWS-12500-custom-component
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

fix: coordinator reuse, route dispatcher, binary sensor translations & test suite alignment 

- Reuse existing `WeatherDataUpdateCoordinator` instance across reloads instead of
  creating a new one; routes keep pointing to the same coordinator, so entities
  stay subscribed and updates never silently drop
- `update_listener` now skips full reload when only `SENSORS_TO_LOAD` changes
  (auto-discovery); avoids the "frozen UI" window caused by temporary platform unload
- Replaced direct `route._handler` mutation with a proper `Routes` dispatcher;
  all aiohttp routes are registered once and an internal dispatcher decides which
  handler is active — safe to switch at runtime without touching the router
- Fixed `self.config_entry` / `self.config` inconsistency in coordinator

- `BatteryBinarySensor` now carries correct `device_info` (same identifiers as
  `WeatherSensor`) → single device card in UI instead of two
- Translations for binary sensors must live under `entity.binary_sensor.*`,
  not `entity.sensor.*`

- Fixed all tests broken by new `register_path(hass, coordinator, coordinator_h, entry)` signature
- Updated `_RouterStub` to accept `name=` kwargs and return objects with `.method`
- Added `async post()` to `_RequestStub` stubs
- Aligned `Routes` tests with new `method:path` dispatch key format and `show_enabled()` output
- Replaced `WindyApiKeyError` with `WindyPasswordMissing` to match actual exceptions
- Updated `_FakeResponse` to use HTTP status codes instead of response text strings
- Patched `persistent_notification.create` in Windy push tests to avoid `SimpleNamespace` errors
ecowitt_support
SchiZzA 2026-06-01 17:01:40 +02:00
parent e3d00389de
commit 6363351d9c
No known key found for this signature in database
23 changed files with 953 additions and 917 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

490
custom_components/sws12500/__init__.py
View File

@ -26,395 +26,43 @@ With a high-frequency push source (webhook), a reload at the wrong moment can le
period where no entities are subscribed, causing stale states until another full reload/restart.
"""
import hmac
from __future__ import annotations
import logging
from typing import Any
import aiohttp.web
from aiohttp.web_exceptions import HTTPUnauthorized
from py_typecheck import checked, checked_or
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import Platform
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryNotReady, InvalidStateError, PlatformNotReady
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
from homeassistant.exceptions import ConfigEntryNotReady, PlatformNotReady
from .const import (
API_ID,
API_KEY,
DEFAULT_URL,
DEV_DBG,
DOMAIN,
ECOWITT_ENABLED,
ECOWITT_URL_PREFIX,
HEALTH_URL,
LEGACY_ENABLED,
POCASI_CZ_ENABLED,
SENSORS_TO_LOAD,
WINDY_ENABLED,
WSLINK,
WSLINK_URL,
)
from .data import ENTRY_COORDINATOR, ENTRY_HEALTH_COORD, ENTRY_LAST_OPTIONS
from .ecowitt import EcowittBridge # noqa: PLC0415
from .coordinator import WeatherDataUpdateCoordinator
from .data import SWSConfigEntry, SWSRuntimeData
from .health_coordinator import HealthCoordinator
from .pocasti_cz import PocasiPush
from .legacy import update_legacy_battery_issue
from .routes import Routes
from .utils import (
anonymize,
check_disabled,
loaded_sensors,
remap_items,
remap_wslink_items,
translated_notification,
translations,
update_options,
)
from .windy_func import WindyPush
_LOGGER = logging.getLogger(__name__)
PLATFORMS: list[Platform] = [Platform.SENSOR, Platform.BINARY_SENSOR]
class IncorrectDataError(InvalidStateError):
"""Invalid exception."""
# NOTE:
# We intentionally avoid importing the sensor platform module at import-time here.
# Home Assistant can import modules in different orders; keeping imports acyclic
# prevents "partially initialized module" failures (circular imports / partially initialized modules).
#
# When we need to dynamically add sensors, we do a local import inside the webhook handler.
class WeatherDataUpdateCoordinator(DataUpdateCoordinator):
"""Coordinator for push updates.
Even though Home Assistant's `DataUpdateCoordinator` is often used for polling,
it also works well as a "fan-out" mechanism for push integrations:
- webhook handler updates `self.data` via `async_set_updated_data`
- all `CoordinatorEntity` instances subscribed to this coordinator update themselves
"""
def __init__(self, hass: HomeAssistant, config: ConfigEntry) -> None:
"""Initialize the coordinator.
`config` is the config entry for this integration instance. We store it because
the webhook handler needs access to options (auth data, enabled features, etc.).
"""
self.hass: HomeAssistant = hass
self.config: ConfigEntry = config
self.windy: WindyPush = WindyPush(hass, config)
self.pocasi: PocasiPush = PocasiPush(hass, config)
# Ecowitt bridge - aioecowitt parser without HTTP server
self.ecowitt_bridge: EcowittBridge = EcowittBridge(hass, config)
super().__init__(hass, _LOGGER, name=DOMAIN)
def _health_coordinator(self) -> HealthCoordinator | None:
"""Return the health coordinator for this config entry."""
if (data := checked(self.hass.data.get(DOMAIN), dict[str, Any])) is None:
return None
if (entry := checked(data.get(self.config.entry_id), dict[str, Any])) is None:
return None
coordinator = entry.get(ENTRY_HEALTH_COORD)
return coordinator if isinstance(coordinator, HealthCoordinator) else None
async def recieved_ecowitt_data(self, webdata: aiohttp.web.Request) -> aiohttp.web.Response:
"""Handle incoming Ecowitt webhook payload.
We are using aioecowitt for parsing payload. Sensors with internal
mapping will use SWS pipline. Sensors withou mapping will create
native Ecowitt entity trough bridge callback.
"""
from .const import ECOWITT_ENABLED, ECOWITT_WEBHOOK_ID # noqa: PLC0415
health = self._health_coordinator()
# Do we have Ecowitt enabled?
if not checked_or(self.config.options.get(ECOWITT_ENABLED), bool, False):
if health:
health.update_ingress_result(
webdata,
accepted=False,
authorized=None,
reason="ecowitt_disabled",
)
return aiohttp.web.Response(text="Ecowitt disabled", status=403)
# Check webhook ID from URL
expected_webhook = self.config.options.get(ECOWITT_WEBHOOK_ID, "")
actual_webhook = webdata.match_info.get("webhook_id", "")
if not expected_webhook or actual_webhook != expected_webhook:
_LOGGER.error("Ecowitt: invalid webhook ID")
if health:
health.update_ingress_result(
webdata,
accepted=False,
authorized=False,
reason="ecowitt_invalid_webhook_id",
)
raise HTTPUnauthorized
# Parse POST body
post_data = await webdata.post()
data: dict[str, Any] = dict(post_data)
# Bridge: aioecowitt parsing + internal remap
mapped_data = await self.ecowitt_bridge.process_payload(data)
# Mapped sensors to SWS pipline (auto-discovery + fan-out)
if mapped_data:
if sensors := check_disabled(mapped_data, self.config):
newly_discovered = list(sensors)
if _loaded_senosrs := loaded_sensors(self.config):
sensors.extend(_loaded_senosrs)
await update_options(self.hass, self.config, SENSORS_TO_LOAD, sensors)
from .binary_sensor import add_new_binary_sensors # noqa: PLC0415
from .sensor import add_new_sensors # noqa: PLC0415
add_new_binary_sensors(self.hass, self.config, newly_discovered)
add_new_sensors(self.hass, self.config, newly_discovered)
self.async_set_updated_data(mapped_data)
if health:
health.update_ingress_result(
webdata,
accepted=True,
authorized=True,
reason="accepted",
)
# Forwarding (mapped data in WU units)
_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, False)
# Will push just WU payload to POCASI
# TODO: create ecowitt protocol to send full payload to Pocasi CZ
if _pocasi_enabled:
await self.pocasi.push_data_to_server(data, "WU")
if health:
health.update_forwarding(self.windy, self.pocasi)
if checked_or(self.config.options.get(DEV_DBG), bool, False):
_LOGGER.info("Dev log (ecowitt): %s", anonymize(data))
return aiohttp.web.Response(body="OK", status=200)
async def received_data(self, webdata: aiohttp.web.Request) -> aiohttp.web.Response:
"""Handle incoming webhook payload from the station.
This method:
- validates authentication (different keys for WU vs WSLink)
- optionally forwards data to third-party services (Windy / Pocasi)
- remaps payload keys to internal sensor keys
- auto-discovers new sensor fields and adds entities dynamically
- updates coordinator data so existing entities refresh immediately
"""
# WSLink uses different auth and payload field naming than the legacy endpoint.
_wslink: bool = checked_or(self.config.options.get(WSLINK), bool, False)
# Incoming station payload is delivered as query params.
# Some stations posts data in body, so we need to contracts those data.
#
# We copy it to a plain dict so it can be passed around safely.
get_data = webdata.query
post_data = await webdata.post()
# normalize incoming data to dict[str, Any]
data: dict[str, Any] = {**dict(get_data), **dict(post_data)}
# Get health data coordinator
health = self._health_coordinator()
# Validate auth keys (different parameter names depending on endpoint mode).
if not _wslink and ("ID" not in data or "PASSWORD" not in data):
_LOGGER.error("Invalid request. No security data provided!")
if health:
health.update_ingress_result(
webdata,
accepted=False,
authorized=False,
reason="missing_credentials",
)
raise HTTPUnauthorized
if _wslink and ("wsid" not in data or "wspw" not in data):
_LOGGER.error("Invalid request. No security data provided!")
if health:
health.update_ingress_result(
webdata,
accepted=False,
authorized=False,
reason="missing_credentials",
)
raise HTTPUnauthorized
id_data: str = ""
key_data: str = ""
if _wslink:
id_data = data.get("wsid", "")
key_data = data.get("wspw", "")
else:
id_data = data.get("ID", "")
key_data = data.get("PASSWORD", "")
# Validate credentials against the integration's configured options.
# If auth doesn't match, we reject the request (prevents random pushes from the LAN/Internet).
if (_id := checked(self.config.options.get(API_ID), str)) is None:
_LOGGER.error("We don't have API ID set! Update your config!")
if health:
health.update_ingress_result(
webdata,
accepted=False,
authorized=None,
reason="config_missing_api_id",
)
raise IncorrectDataError
if (_key := checked(self.config.options.get(API_KEY), str)) is None:
_LOGGER.error("We don't have API KEY set! Update your config!")
if health:
health.update_ingress_result(
webdata,
accepted=False,
authorized=None,
reason="config_missing_api_key",
)
raise IncorrectDataError
# Constant-time comaprision to avoid lekaing credential length/content via timig.
# Both operands are compared even if the first fails, so the branch order doesn't
# short-circut. Encode to bytes so non-ASCII credentials are handeled safely.
id_ok = hmac.compare_digest(id_data.encode("utf-8"), _id.encode("utf-8"))
key_ok = hmac.compare_digest(key_data.encode("utf-8"), _key.encode("utf-8"))
if not (id_ok & key_ok):
_LOGGER.error("Unauthorised access!")
if health:
health.update_ingress_result(
webdata,
accepted=False,
authorized=False,
reason="unauthorized",
)
raise HTTPUnauthorized
# Convert raw payload keys to our internal sensor keys (stable identifiers).
remaped_items: dict[str, str] = remap_wslink_items(data) if _wslink else remap_items(data)
# Auto-discovery: if payload contains keys that are not enabled/loaded yet,
# add them to the option list and create entities dynamically.
if sensors := check_disabled(remaped_items, self.config):
if (
translate_sensors := checked(
[
await translations(
self.hass,
DOMAIN,
f"sensor.{t_key}",
key="name",
category="entity",
)
for t_key in sensors
if await translations(
self.hass,
DOMAIN,
f"sensor.{t_key}",
key="name",
category="entity",
)
is not None
],
list[str],
)
) is not None:
human_readable: str = "\n".join(translate_sensors)
else:
human_readable = ""
await translated_notification(
self.hass,
DOMAIN,
"added",
{"added_sensors": f"{human_readable}\n"},
)
# Persist newly discovered sensor keys to options (so they remain enabled after restart).
newly_discovered = list(sensors)
if _loaded_sensors := loaded_sensors(self.config):
sensors.extend(_loaded_sensors)
await update_options(self.hass, self.config, SENSORS_TO_LOAD, sensors)
# Dynamically add newly discovered sensors *without* reloading the entry.
#
# Why: Reloading a config entry unloads platforms temporarily. That removes coordinator
# listeners; with frequent webhook pushes the UI can appear "frozen" until the listeners
# are re-established. Dynamic adds avoid this window completely.
#
# We do a local import to avoid circular imports at module import time.
#
# NOTE: Some linters prefer top-level imports. In this case the local import is
# intentional and prevents "partially initialized module" errors.
from .binary_sensor import add_new_binary_sensors # noqa: PLC0415 (local import is intentional)
from .sensor import add_new_sensors # noqa: PLC0415 (local import is intentional)
add_new_sensors(self.hass, self.config, newly_discovered)
add_new_binary_sensors(self.hass, self.config, newly_discovered)
# Fan-out update: notify all subscribed entities.
self.async_set_updated_data(remaped_items)
if health:
health.update_ingress_result(
webdata,
accepted=True,
authorized=True,
reason="accepted",
)
# Optional forwarding to external services. This is kept here (in the webhook handler)
# to avoid additional background polling tasks.
_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:
await self.pocasi.push_data_to_server(data, "WSLINK" if _wslink else "WU")
if health:
health.update_forwarding(self.windy, self.pocasi)
# Optional dev logging (keep it lightweight to avoid log spam under high-frequency updates).
if checked_or(self.config.options.get(DEV_DBG), bool, False):
_LOGGER.info("Dev log: %s", anonymize(data))
return aiohttp.web.Response(body="OK", status=200)
def register_path(
hass: HomeAssistant,
coordinator: WeatherDataUpdateCoordinator,
coordinator_h: HealthCoordinator,
config: ConfigEntry,
config: SWSConfigEntry,
) -> bool:
"""Register webhook paths.
@ -474,7 +122,7 @@ def register_path(
routes.add_route(
_ecowitt_path,
_ecowitt_route,
coordinator.recieved_ecowitt_data,
coordinator.received_ecowitt_data,
enabled=_ecowitt_enabled,
sticky=True,
)
@ -484,61 +132,25 @@ def register_path(
return True
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
async def async_setup_entry(hass: HomeAssistant, entry: SWSConfigEntry) -> bool:
"""Set up a config entry.
Important:
- We store per-entry runtime state under `hass.data[DOMAIN][entry_id]` as a dict.
- We reuse the same coordinator instance across reloads so that:
- the webhook handler keeps updating the same coordinator
- already-created entities remain subscribed
Per-entry state is held on `entry.runtime_data`. Only the shared aiohttp route dispatcher
lives in `hass.data[DOMAIN]` because it must outlive a single entry reload. This separation is critical
to avoid issues where entities stop receiving updates after a reload because the coordinator instance they
are subscribed to is replaced in `hass.data[DOMAIN]` but the aiohttp route dispatcher still calls the old instance.
"""
hass_data = hass.data.setdefault(DOMAIN, {})
# hass_data = cast("dict[str, Any]", hass_data_any)
hass.data.setdefault(DOMAIN, {})
# 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.
if (entry_data := checked(hass_data.get(entry.entry_id), dict[str, Any])) is None:
entry_data = {}
hass_data[entry.entry_id] = entry_data
# Reuse the existing coordinator across reloads so webhook handlers and entities
# remain connected to the same coordinator instance.
#
# 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)
if isinstance(coordinator, WeatherDataUpdateCoordinator):
coordinator.config = entry
# Recreate helper instances so they pick up updated options safely.
coordinator.windy = WindyPush(hass, entry)
coordinator.pocasi = PocasiPush(hass, entry)
else:
coordinator = WeatherDataUpdateCoordinator(hass, entry)
entry_data[ENTRY_COORDINATOR] = coordinator
# Similar to the coordinator, we want to reuse the same health coordinator instance across
# reloads so that the health endpoint remains responsive and doesn't lose its listeners.
coordinator_health = entry_data.get(ENTRY_HEALTH_COORD)
if isinstance(coordinator_health, HealthCoordinator):
coordinator_health.config = entry
else:
coordinator_health = HealthCoordinator(hass, entry)
entry_data[ENTRY_HEALTH_COORD] = coordinator_health
routes: Routes | None = hass_data.get("routes", None)
# Keep an options snapshot so update_listener can skip reloads when only `SENSORS_TO_LOAD` changes.
# Auto-discovery updates this option frequently and we do not want to reload for that case.
entry_data[ENTRY_LAST_OPTIONS] = dict(entry.options)
coordinator = WeatherDataUpdateCoordinator(hass, entry)
coordinator_health = HealthCoordinator(hass, entry)
entry.runtime_data = SWSRuntimeData(
coordinator=coordinator,
health_coordinator=coordinator_health,
last_options=dict(entry.options),
)
_wslink = checked_or(entry.options.get(WSLINK), bool, False)
_legacy = checked_or(entry.options.get(LEGACY_ENABLED), bool, True)
_ecowitt_enabled = checked_or(entry.options.get(ECOWITT_ENABLED), bool, False)
@ -546,21 +158,22 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
_LOGGER.debug("WS Link is %s", "enabled" if _wslink else "disabled")
if routes:
routes: Routes | None = hass.data[DOMAIN].get("routes")
if routes is not None:
_LOGGER.debug("We have routes registered, will try to switch dispatcher.")
routes.switch_route(coordinator.received_data, DEFAULT_URL if not _wslink else WSLINK_URL, enabled=_legacy)
routes.set_ecowitt_enabled(_ecowitt_path, coordinator.recieved_ecowitt_data, _ecowitt_enabled)
routes.set_ecowitt_enabled(_ecowitt_path, coordinator.received_ecowitt_data, _ecowitt_enabled)
routes.set_ingress_observer(coordinator_health.record_dispatch)
coordinator_health.update_routing(routes)
_LOGGER.debug("%s", routes.show_enabled())
else:
routes_enabled = register_path(hass, coordinator, coordinator_health, entry)
if not routes_enabled:
if not register_path(hass, coordinator, coordinator_health, entry):
_LOGGER.error("Fatal: path not registered!")
raise PlatformNotReady
routes = hass_data.get("routes", None)
if isinstance(routes, Routes):
routes = hass.data[DOMAIN].get("routes")
if routes is not None:
coordinator_health.update_routing(routes)
await coordinator_health.async_config_entry_first_refresh()
@ -569,11 +182,12 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
entry.async_on_unload(entry.add_update_listener(update_listener))
update_legacy_battery_issue(hass, entry)
return True
async def update_listener(hass: HomeAssistant, entry: ConfigEntry):
async def update_listener(hass: HomeAssistant, entry: SWSConfigEntry):
"""Handle config entry option updates.
We skip reloading when only `SENSORS_TO_LOAD` changes.
@ -584,36 +198,30 @@ async def update_listener(hass: HomeAssistant, entry: ConfigEntry):
coordinator listeners, which can make the UI appear "stuck" until restart.
"""
if (hass_data := checked(hass.data.get(DOMAIN), dict[str, Any])) is not None:
if (entry_data := checked(hass_data.get(entry.entry_id), dict[str, 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)
runtime = getattr(entry, "runtime_data", None)
if isinstance(runtime, SWSRuntimeData):
old_options = runtime.last_options
new_options = dict(entry.options)
changed_keys = {
k
for k in set(old_options.keys()) | set(new_options.keys())
if old_options.get(k) != new_options.get(k)
}
changed_keys = {k for k in set(old_options) | set(new_options) if old_options.get(k) != new_options.get(k)}
# Update snapshot early for the next comparison.
entry_data[ENTRY_LAST_OPTIONS] = new_options
runtime.last_options = new_options
if changed_keys == {SENSORS_TO_LOAD}:
_LOGGER.debug("Options updated (%s); skipping reload.", SENSORS_TO_LOAD)
return
else:
# No/invalid snapshot: store current options for next comparison.
entry_data[ENTRY_LAST_OPTIONS] = dict(entry.options)
if changed_keys == {SENSORS_TO_LOAD}:
_LOGGER.debug("Options updated (%s); skipping reload.", SENSORS_TO_LOAD)
return
_ = await hass.config_entries.async_reload(entry.entry_id)
update_legacy_battery_issue(hass, entry)
await hass.config_entries.async_reload(entry.entry_id)
_LOGGER.info("Settings updated")
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Unload a config entry."""
async def async_unload_entry(hass: HomeAssistant, entry: SWSConfigEntry) -> bool:
"""Unload a config entry.
_ok = await hass.config_entries.async_unload_platforms(entry, PLATFORMS)
if _ok:
hass.data[DOMAIN].pop(entry.entry_id)
`entry.runtime_data` becomes irrelevant once entry is unloaded.
On next `async_setup_entry` we overwrite it. The shared `hass.data[DOMAIN]["routes"]` survives by design.
aiohttp routes stay registered and the dispatcher is re-wired on the next setup.
"""
return _ok
return await hass.config_entries.async_unload_platforms(entry, PLATFORMS)

59
custom_components/sws12500/battery_sensors_def.py
View File

@ -1,51 +1,20 @@
"""Battery sensors."""
"""Battery sensors templates.
We create a sensor tempate here.
Actualy loaded senors are gated in coordinator.
"""
from __future__ import annotations
from homeassistant.components.binary_sensor import BinarySensorDeviceClass, BinarySensorEntityDescription
BATTERY_BINARY_SENSORS: tuple[BinarySensorEntityDescription, ...] = (
from .const import BATTERY_LIST
BATTERY_BINARY_SENSORS: tuple[BinarySensorEntityDescription, ...] = tuple(
BinarySensorEntityDescription(
key="outside_battery",
translation_key="outside_battery",
key=key,
translation_key=key,
device_class=BinarySensorDeviceClass.BATTERY,
),
BinarySensorEntityDescription(
key="indoor_battery",
translation_key="indoor_battery",
device_class=BinarySensorDeviceClass.BATTERY,
),
BinarySensorEntityDescription(
key="ch2_battery",
translation_key="ch2_battery",
device_class=BinarySensorDeviceClass.BATTERY,
),
BinarySensorEntityDescription(
key="ch3_battery",
translation_key="ch3_battery",
device_class=BinarySensorDeviceClass.BATTERY,
),
BinarySensorEntityDescription(
key="ch4_battery",
translation_key="ch4_battery",
device_class=BinarySensorDeviceClass.BATTERY,
),
BinarySensorEntityDescription(
key="ch5_battery",
translation_key="ch5_battery",
device_class=BinarySensorDeviceClass.BATTERY,
),
BinarySensorEntityDescription(
key="ch6_battery",
translation_key="ch6_battery",
device_class=BinarySensorDeviceClass.BATTERY,
),
BinarySensorEntityDescription(
key="ch7_battery",
translation_key="ch7_battery",
device_class=BinarySensorDeviceClass.BATTERY,
),
BinarySensorEntityDescription(
key="ch8_battery",
translation_key="ch8_battery",
device_class=BinarySensorDeviceClass.BATTERY,
),
)
for key in BATTERY_LIST
)

85
custom_components/sws12500/binary_sensor.py
View File

@ -1,104 +1,81 @@
"""Binary sensor platform for SWS12500.
Exposes low-battery warnings as binary sensors.
Auto-discovery adds entities without reloading the entry, using callbacks stored on `runtime_data`.
"""
from __future__ import annotations
import logging
from typing import Any, cast
from py_typecheck import checked
from homeassistant.components.binary_sensor import BinarySensorEntity
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from .battery_sensors import BatteryBinarySensor
from .battery_sensors_def import BATTERY_BINARY_SENSORS
from .const import DOMAIN, SENSORS_TO_LOAD
from .data import ENTRY_ADD_BINARY_ENTITIES, ENTRY_ADDED_BINARY_KEYS, ENTRY_BINARY_DESCRIPTION, ENTRY_COORDINATOR
from .const import SENSORS_TO_LOAD
from .data import SWSConfigEntry
_LOGGER = logging.getLogger(__name__)
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback) -> None:
async def async_setup_entry(
hass: HomeAssistant, entry: SWSConfigEntry, async_add_entities: AddEntitiesCallback
) -> None:
"""Set up battery binary sensors."""
if (hass_data := checked(hass.data.get(DOMAIN), dict[str, Any])) is None:
return
del hass
if (entry_data := checked(hass_data.get(entry.entry_id), dict[str, Any])) is None:
return
runtime = entry.runtime_data
coordinator = runtime.coordinator
coordinator = entry_data.get(ENTRY_COORDINATOR)
if coordinator is None:
return
# Save callback and descriptions for later auto-discovery.
# webhook needs this to dynamicaly add entities withou reload
description = {desc.key: desc for desc in BATTERY_BINARY_SENSORS}
entry_data[ENTRY_ADD_BINARY_ENTITIES] = async_add_entities
entry_data[ENTRY_BINARY_DESCRIPTION] = description
added_keys: set[str] = set()
entry_data[ENTRY_ADDED_BINARY_KEYS] = added_keys
# Create binary sensors for battery key that station send.
# SENSORS_TO_LOAD contains all discovered keys.
# Persist platform callback + description map for dynamic entity creation.
runtime.add_binary_entities = async_add_entities
runtime.binary_descriptions = {desc.key: desc for desc in BATTERY_BINARY_SENSORS}
runtime.added_binary_keys = set()
# Initial entities for battery keys that station already reports.
# `SENSORS_TO_LOAD` accumulates all discovered keys across runs.
loaded = set(entry.options.get(SENSORS_TO_LOAD, []))
entities: list[BatteryBinarySensor] = []
for desc in BATTERY_BINARY_SENSORS:
if desc.key in loaded:
entities.append(BatteryBinarySensor(coordinator, desc))
added_keys.add(desc.key)
runtime.added_binary_keys.add(desc.key)
if entities:
async_add_entities(entities)
def add_new_binary_sensors(hass: HomeAssistant, entry: ConfigEntry, keys: list[str]) -> None:
"""Dynamic add newly discovered enetities.
def add_new_binary_sensors(hass: HomeAssistant, entry: SWSConfigEntry, keys: list[str]) -> None:
"""Dynamic add newly discovered binary sensors without reloading the entry.
Called from webhook handler in __init__.py.
Safe no-op if the platform hasn't finished setting up yet (e.g. callback/description map missing).
Duplicate keys are ignored (only keys with an entity description that haven't been added yet are added).
"""
if (hass_data := checked(hass.data.get(DOMAIN), dict[str, Any])) is None:
del hass # kept for backwards-compatible call signature; not used after runtime_data migration
runtime = entry.runtime_data
add_entities = runtime.add_binary_entities
if add_entities is None:
return
if (entry_data := checked(hass_data.get(entry.entry_id), dict[str, Any])) is None:
return
add_entities = entry_data.get(ENTRY_ADD_BINARY_ENTITIES)
description = entry_data.get(ENTRY_BINARY_DESCRIPTION)
coordinator = entry_data.get(ENTRY_COORDINATOR)
added_keys: set[str] | None = entry_data.get(ENTRY_ADDED_BINARY_KEYS)
if add_entities is None or description is None or coordinator is None:
return
if added_keys is None:
added_keys = set()
entry_data[ENTRY_ADDED_BINARY_KEYS] = added_keys
description_map = cast("dict[str, Any]", description)
added = added_keys
descriptions = runtime.binary_descriptions
coordinator = runtime.coordinator
added = runtime.added_binary_keys
new_entities: list[BinarySensorEntity] = []
for key in keys:
if key in added:
continue
desc = description_map.get(key)
if desc is None:
if (desc := descriptions.get(key)) is None:
continue
new_entities.append(BatteryBinarySensor(coordinator, desc))
added.add(key)
if new_entities:
add_fn = cast("AddEntitiesCallback", add_entities)
add_fn(new_entities)
add_entities(new_entities)

2
custom_components/sws12500/config_flow.py
View File

@ -1,5 +1,7 @@
"""Config flow for Sencor SWS 12500 Weather Station integration."""
from __future__ import annotations
import secrets
from typing import Any

174
custom_components/sws12500/const.py
View File

@ -1,12 +1,12 @@
"""Constants."""
from __future__ import annotations
from enum import StrEnum
from typing import Final
# Integration specific constants.
DOMAIN = "sws12500"
DATABASE_PATH = "/config/home-assistant_v2.db"
ICON = "mdi:weather"
DEV_DBG: Final = "dev_debug_checkbox"
@ -15,7 +15,6 @@ API_KEY = "API_KEY"
API_ID = "API_ID"
SENSORS_TO_LOAD: Final = "sensors_to_load"
SENSOR_TO_MIGRATE: Final = "sensor_to_migrate"
INVALID_CREDENTIALS: Final = [
"API",
@ -109,6 +108,38 @@ PURGE_DATA: Final = [
"dailyrainin",
]
"""NOTE: These are sensors that should be available with PWS protocol acording to https://support.weather.com/s/article/PWS-Upload-Protocol?language=en_US:
I have no option to test, if it will work correctly. So their implementatnion will be in future releases.
leafwetness - [%]
+ for sensor 2 use leafwetness2
visibility - [nm visibility]
pweather - [text] -- metar style (+RA)
clouds - [text] -- SKC, FEW, SCT, BKN, OVC
Pollution Fields:
AqNO - [ NO (nitric oxide) ppb ]
AqNO2T - (nitrogen dioxide), true measure ppb
AqNO2 - NO2 computed, NOx-NO ppb
AqNO2Y - NO2 computed, NOy-NO ppb
AqNOX - NOx (nitrogen oxides) - ppb
AqNOY - NOy (total reactive nitrogen) - ppb
AqNO3 - NO3 ion (nitrate, not adjusted for ammonium ion) UG/M3
AqSO4 - SO4 ion (sulfate, not adjusted for ammonium ion) UG/M3
AqSO2 - (sulfur dioxide), conventional ppb
AqSO2T - trace levels ppb
AqCO - CO (carbon monoxide), conventional ppm
AqCOT -CO trace levels ppb
AqEC - EC (elemental carbon) – PM2.5 UG/M3
AqOC - OC (organic carbon, not adjusted for oxygen and hydrogen) – PM2.5 UG/M3
AqBC - BC (black carbon at 880 nm) UG/M3
AqUV-AETH - UV-AETH (second channel of Aethalometer at 370 nm) UG/M3
AqPM2.5 - PM2.5 mass - UG/M3
AqPM10 - PM10 mass - PM10 mass
AqOZONE - Ozone - ppb
"""
REMAP_ITEMS: dict[str, str] = {
"baromin": BARO_PRESSURE,
"tempf": OUTSIDE_TEMP,
@ -150,92 +181,6 @@ LEGACY_ENABLED: Final = "legacy_enabled"
WINDY_MAX_RETRIES: Final = 3
WSLINK_ADDON_PORT: Final = "WSLINK_ADDON_PORT"
__all__ = [
"DOMAIN",
"DEFAULT_URL",
"WSLINK_URL",
"HEALTH_URL",
"WINDY_URL",
"DATABASE_PATH",
"POCASI_CZ_URL",
"POCASI_CZ_SEND_MINIMUM",
"ICON",
"API_KEY",
"API_ID",
"SENSORS_TO_LOAD",
"SENSOR_TO_MIGRATE",
"DEV_DBG",
"WSLINK",
"LEGACY_ENABLED",
"ECOWITT",
"ECOWITT_WEBHOOK_ID",
"ECOWITT_ENABLED",
"POCASI_CZ_API_KEY",
"POCASI_CZ_API_ID",
"POCASI_CZ_SEND_INTERVAL",
"POCASI_CZ_ENABLED",
"POCASI_CZ_LOGGER_ENABLED",
"POCASI_INVALID_KEY",
"POCASI_CZ_SUCCESS",
"POCASI_CZ_UNEXPECTED",
"WINDY_STATION_ID",
"WINDY_STATION_PW",
"WINDY_ENABLED",
"WINDY_LOGGER_ENABLED",
"WINDY_NOT_INSERTED",
"WINDY_INVALID_KEY",
"WINDY_SUCCESS",
"WINDY_UNEXPECTED",
"INVALID_CREDENTIALS",
"PURGE_DATA",
"PURGE_DATA_POCAS",
"BARO_PRESSURE",
"OUTSIDE_TEMP",
"DEW_POINT",
"OUTSIDE_HUMIDITY",
"OUTSIDE_CONNECTION",
"OUTSIDE_BATTERY",
"WIND_SPEED",
"WIND_GUST",
"WIND_DIR",
"WIND_AZIMUT",
"RAIN",
"HOURLY_RAIN",
"WEEKLY_RAIN",
"MONTHLY_RAIN",
"YEARLY_RAIN",
"DAILY_RAIN",
"SOLAR_RADIATION",
"INDOOR_TEMP",
"INDOOR_HUMIDITY",
"INDOOR_BATTERY",
"UV",
"CH2_TEMP",
"CH2_HUMIDITY",
"CH2_CONNECTION",
"CH2_BATTERY",
"CH3_TEMP",
"CH3_HUMIDITY",
"CH3_CONNECTION",
"CH4_TEMP",
"CH4_HUMIDITY",
"CH4_CONNECTION",
"HEAT_INDEX",
"CHILL_INDEX",
"WBGT_TEMP",
"REMAP_ITEMS",
"REMAP_WSLINK_ITEMS",
"DISABLED_BY_DEFAULT",
"BATTERY_LIST",
"UnitOfDir",
"AZIMUT",
"UnitOfBat",
"BATTERY_LEVEL",
"ECOWITT_URL",
"ECOWITT_META_KEYS",
"REMAP_ECOWITT_COMPAT",
]
ECOWITT: Final = "ecowitt"
ECOWITT_WEBHOOK_ID: Final = "ecowitt_webhook_id"
ECOWITT_ENABLED: Final = "ecowitt_enabled"
@ -301,39 +246,6 @@ PURGE_DATA_POCAS: Final = [
]
"""NOTE: These are sensors that should be available with PWS protocol acording to https://support.weather.com/s/article/PWS-Upload-Protocol?language=en_US:
I have no option to test, if it will work correctly. So their implementatnion will be in future releases.
leafwetness - [%]
+ for sensor 2 use leafwetness2
visibility - [nm visibility]
pweather - [text] -- metar style (+RA)
clouds - [text] -- SKC, FEW, SCT, BKN, OVC
Pollution Fields:
AqNO - [ NO (nitric oxide) ppb ]
AqNO2T - (nitrogen dioxide), true measure ppb
AqNO2 - NO2 computed, NOx-NO ppb
AqNO2Y - NO2 computed, NOy-NO ppb
AqNOX - NOx (nitrogen oxides) - ppb
AqNOY - NOy (total reactive nitrogen) - ppb
AqNO3 - NO3 ion (nitrate, not adjusted for ammonium ion) UG/M3
AqSO4 - SO4 ion (sulfate, not adjusted for ammonium ion) UG/M3
AqSO2 - (sulfur dioxide), conventional ppb
AqSO2T - trace levels ppb
AqCO - CO (carbon monoxide), conventional ppm
AqCOT -CO trace levels ppb
AqEC - EC (elemental carbon) – PM2.5 UG/M3
AqOC - OC (organic carbon, not adjusted for oxygen and hydrogen) – PM2.5 UG/M3
AqBC - BC (black carbon at 880 nm) UG/M3
AqUV-AETH - UV-AETH (second channel of Aethalometer at 370 nm) UG/M3
AqPM2.5 - PM2.5 mass - UG/M3
AqPM10 - PM10 mass - PM10 mass
AqOZONE - Ozone - ppb
"""
REMAP_WSLINK_ITEMS: dict[str, str] = {
"intem": INDOOR_TEMP,
"inhum": INDOOR_HUMIDITY,
@ -477,22 +389,32 @@ DISABLED_BY_DEFAULT: Final = [
WBGT_TEMP,
]
BATTERY_LIST = [
# Station reports batteries as 0/1 (low/normal) for most of sensors.
# Batteries reported as 0-5 level are stored in `BATTERY_NON_BINARY` tuple
BATTERY_LIST: Final[tuple[str, ...]] = (
OUTSIDE_BATTERY,
INDOOR_BATTERY,
CH2_BATTERY,
CH2_BATTERY,
CH3_BATTERY,
CH4_BATTERY,
CH5_BATTERY,
CH6_BATTERY,
CH7_BATTERY,
CH8_BATTERY,
]
)
BATTERY_NON_BINARY: list[str] = [T9_BATTERY]
BATTERY_NON_BINARY: Final[tuple[str, ...]] = (T9_BATTERY,)
CONNECTION_GATED_SENSORS: Final[dict[str, list[str]]] = {
# Multi-channel temp/humidity probes (CH2 - CH8)
"t234c1cn": [CH2_TEMP, CH2_HUMIDITY, CH2_BATTERY],
"t234c2cn": [CH3_TEMP, CH3_HUMIDITY, CH3_BATTERY],
"t234c3cn": [CH4_TEMP, CH4_HUMIDITY, CH4_BATTERY],
"t234c4cn": [CH5_TEMP, CH5_HUMIDITY, CH5_BATTERY],
"t234c5cn": [CH6_TEMP, CH6_HUMIDITY, CH6_BATTERY],
"t234c6cn": [CH7_TEMP, CH7_HUMIDITY, CH7_BATTERY],
"t234c7cn": [CH8_TEMP, CH8_HUMIDITY, CH8_BATTERY],
# T9 HCHO/VOC probe
"t9cn": [HCHO, VOC, T9_BATTERY],
}

325
custom_components/sws12500/coordinator.py Normal file
View File

@ -0,0 +1,325 @@
"""Push coordinator for the SWS-12500 weather station integration.
This module is the runtime heart of the integration:
- `WeatherDataUpdateCoordinator` is a fan-out hub for push payloads.
Webhook handlers call `async_set_updated_data(...)` and every CoordinatorEntity
subscribed to the coordinator updates its state.
- `received_data` handles the legacy PWS / WSLink endpoints.
- `received_ecowitt_data` handles the Ecowitt endpoint via the aioecowitt parser.
- `IncorrectDataError` is raised when the integration's auth options are missing.
Kept separate from `__init__.py` so the platforms (sensor / binary_sensor) can
import `WeatherDataUpdateCoordinator` without re-entering the package's
initialization machinery — no more local-import `# noqa: PLC0415` shims.
"""
from __future__ import annotations
import hmac
import logging
from typing import Any
import aiohttp.web
from aiohttp.web_exceptions import HTTPUnauthorized
from py_typecheck import checked, checked_or
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import InvalidStateError
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
from .binary_sensor import add_new_binary_sensors
from .const import (
API_ID,
API_KEY,
DEV_DBG,
DOMAIN,
ECOWITT_ENABLED,
ECOWITT_WEBHOOK_ID,
POCASI_CZ_ENABLED,
SENSORS_TO_LOAD,
WINDY_ENABLED,
WSLINK,
)
from .data import SWSConfigEntry
from .ecowitt import EcowittBridge
from .health_coordinator import HealthCoordinator
from .pocasti_cz import PocasiPush
from .sensor import add_new_sensors
from .utils import (
anonymize,
check_disabled,
loaded_sensors,
remap_items,
remap_wslink_items,
translated_notification,
translations,
update_options,
)
from .windy_func import WindyPush
_LOGGER = logging.getLogger(__name__)
class IncorrectDataError(InvalidStateError):
"""Raised when the integration's auth options are missing or invalid."""
class WeatherDataUpdateCoordinator(DataUpdateCoordinator):
"""Coordinator for push updates.
Even though Home Assistant's `DataUpdateCoordinator` is often used for polling,
it also works well as a fan-out mechanism for push integrations:
- webhook handler updates `self.data` via `async_set_updated_data`
- all `CoordinatorEntity` instances subscribed update themselves
"""
def __init__(self, hass: HomeAssistant, config: SWSConfigEntry) -> None:
"""Initialize the coordinator."""
self.hass: HomeAssistant = hass
self.config: SWSConfigEntry = config
self.windy: WindyPush = WindyPush(hass, config)
self.pocasi: PocasiPush = PocasiPush(hass, config)
self.ecowitt_bridge: EcowittBridge = EcowittBridge(hass, config)
super().__init__(hass, _LOGGER, name=DOMAIN)
def _health_coordinator(self) -> HealthCoordinator | None:
"""Return the health coordinator for this config entry."""
try:
return self.config.runtime_data.health_coordinator
except AttributeError:
return None
async def received_ecowitt_data(self, webdata: aiohttp.web.Request) -> aiohttp.web.Response:
"""Handle incoming Ecowitt webhook payload.
Uses aioecowitt for payload parsing. Sensors with internal mapping
join the SWS pipeline; sensors without mapping create native Ecowitt
entities through the bridge callback.
"""
health = self._health_coordinator()
if not checked_or(self.config.options.get(ECOWITT_ENABLED), bool, False):
if health:
health.update_ingress_result(
webdata,
accepted=False,
authorized=None,
reason="ecowitt_disabled",
)
return aiohttp.web.Response(text="Ecowitt disabled", status=403)
expected_webhook = self.config.options.get(ECOWITT_WEBHOOK_ID, "")
actual_webhook = webdata.match_info.get("webhook_id", "")
if not expected_webhook or actual_webhook != expected_webhook:
_LOGGER.error("Ecowitt: invalid webhook ID")
if health:
health.update_ingress_result(
webdata,
accepted=False,
authorized=False,
reason="ecowitt_invalid_webhook_id",
)
raise HTTPUnauthorized
post_data = await webdata.post()
data: dict[str, Any] = dict(post_data)
mapped_data = await self.ecowitt_bridge.process_payload(data)
if mapped_data:
if sensors := check_disabled(mapped_data, self.config):
newly_discovered = list(sensors)
if _loaded_sensors := loaded_sensors(self.config):
sensors.extend(_loaded_sensors)
await update_options(self.hass, self.config, SENSORS_TO_LOAD, sensors)
add_new_binary_sensors(self.hass, self.config, newly_discovered)
add_new_sensors(self.hass, self.config, newly_discovered)
self.async_set_updated_data(mapped_data)
if health:
health.update_ingress_result(
webdata,
accepted=True,
authorized=True,
reason="accepted",
)
_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, False)
# TODO: create ecowitt protocol to send full payload to Pocasi CZ
if _pocasi_enabled:
await self.pocasi.push_data_to_server(data, "WU")
if health:
health.update_forwarding(self.windy, self.pocasi)
if checked_or(self.config.options.get(DEV_DBG), bool, False):
_LOGGER.info("Dev log (ecowitt): %s", anonymize(data))
return aiohttp.web.Response(body="OK", status=200)
async def received_data(self, webdata: aiohttp.web.Request) -> aiohttp.web.Response:
"""Handle incoming webhook payload from the station.
- validates authentication (different keys for WU vs WSLink)
- optionally forwards data to third-party services (Windy / Pocasi)
- remaps payload keys to internal sensor keys
- auto-discovers new sensor fields and adds entities dynamically
- updates coordinator data so existing entities refresh immediately
"""
# WSLink uses different auth and payload field naming than the legacy endpoint.
_wslink: bool = checked_or(self.config.options.get(WSLINK), bool, False)
get_data = webdata.query
post_data = await webdata.post()
data: dict[str, Any] = {**dict(get_data), **dict(post_data)}
health = self._health_coordinator()
if not _wslink and ("ID" not in data or "PASSWORD" not in data):
_LOGGER.error("Invalid request. No security data provided!")
if health:
health.update_ingress_result(
webdata,
accepted=False,
authorized=False,
reason="missing_credentials",
)
raise HTTPUnauthorized
if _wslink and ("wsid" not in data or "wspw" not in data):
_LOGGER.error("Invalid request. No security data provided!")
if health:
health.update_ingress_result(
webdata,
accepted=False,
authorized=False,
reason="missing_credentials",
)
raise HTTPUnauthorized
if _wslink:
id_data = data.get("wsid", "")
key_data = data.get("wspw", "")
else:
id_data = data.get("ID", "")
key_data = data.get("PASSWORD", "")
if (_id := checked(self.config.options.get(API_ID), str)) is None:
_LOGGER.error("We don't have API ID set! Update your config!")
if health:
health.update_ingress_result(
webdata,
accepted=False,
authorized=None,
reason="config_missing_api_id",
)
raise IncorrectDataError
if (_key := checked(self.config.options.get(API_KEY), str)) is None:
_LOGGER.error("We don't have API KEY set! Update your config!")
if health:
health.update_ingress_result(
webdata,
accepted=False,
authorized=None,
reason="config_missing_api_key",
)
raise IncorrectDataError
# Constant-time comparison to avoid leaking credential length/content via timing.
# Both operands are compared even if the first fails, so the branch order doesn't
# short-circuit. Encode to bytes so non-ASCII credentials are handled safely.
id_ok = hmac.compare_digest(id_data.encode("utf-8"), _id.encode("utf-8"))
key_ok = hmac.compare_digest(key_data.encode("utf-8"), _key.encode("utf-8"))
if not (id_ok & key_ok):
_LOGGER.error("Unauthorised access!")
if health:
health.update_ingress_result(
webdata,
accepted=False,
authorized=False,
reason="unauthorized",
)
raise HTTPUnauthorized
remaped_items: dict[str, str] = remap_wslink_items(data) if _wslink else remap_items(data)
if sensors := check_disabled(remaped_items, self.config):
if (
translate_sensors := checked(
[
await translations(
self.hass,
DOMAIN,
f"sensor.{t_key}",
key="name",
category="entity",
)
for t_key in sensors
if await translations(
self.hass,
DOMAIN,
f"sensor.{t_key}",
key="name",
category="entity",
)
is not None
],
list[str],
)
) is not None:
human_readable: str = "\n".join(translate_sensors)
else:
human_readable = ""
await translated_notification(
self.hass,
DOMAIN,
"added",
{"added_sensors": f"{human_readable}\n"},
)
newly_discovered = list(sensors)
if _loaded_sensors := loaded_sensors(self.config):
sensors.extend(_loaded_sensors)
await update_options(self.hass, self.config, SENSORS_TO_LOAD, sensors)
# Dynamic adds avoid the listener-drop window of a full reload.
add_new_sensors(self.hass, self.config, newly_discovered)
add_new_binary_sensors(self.hass, self.config, newly_discovered)
self.async_set_updated_data(remaped_items)
if health:
health.update_ingress_result(
webdata,
accepted=True,
authorized=True,
reason="accepted",
)
_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:
await self.pocasi.push_data_to_server(data, "WSLINK" if _wslink else "WU")
if health:
health.update_forwarding(self.windy, self.pocasi)
if checked_or(self.config.options.get(DEV_DBG), bool, False):
_LOGGER.info("Dev log: %s", anonymize(data))
return aiohttp.web.Response(body="OK", status=200)

43
custom_components/sws12500/data.py
View File

@ -1,22 +1,22 @@
"""Shared keys for storing integration runtime data.
HA 2025+ pattern: structured runtime state stored in entry.runtime_data
instead of loosely-typed hass.data[][] dicts.
HA 2025+ pattern: typed `ConfigEntry[SWSRuntimeData]` replaces ad-hoc `hass.data[DOMAIN][entry_id]` dicts.
All per-entry state lives here. Cross-reload shared state (aiohttp route registrations) stays under
hass.data[DOMAIN]["routes"] because it must outlive a single entry reload.
"""
from __future__ import annotations
from dataclasses import dataclass, field
from typing import TYPE_CHECKING, Any, Final
from typing import TYPE_CHECKING, Any
from homeassistant.config_entries import ConfigEntry
from homeassistant.core_config import Config
from homeassistant.exceptions import NoEntitySpecifiedError
from homeassistant.helpers.entity_platform import AddEntitiesCallback
if TYPE_CHECKING:
from . import WeatherDataUpdateCoordinator
from .ecowitt import EcowittBridge
from homeassistant.components.binary_sensor import BinarySensorEntityDescription
from .coordinator import WeatherDataUpdateCoordinator
from .health_coordinator import HealthCoordinator
from .sensors_common import WeatherSensorEntityDescription
@ -25,41 +25,26 @@ if TYPE_CHECKING:
class SWSRuntimeData:
"""Per-entry runtime state for SWS12500 integration.
Stored in entry.runtime_data. Type-safe, no string key lookups,
no checked() boilerplate
Stored in entry.runtime_data. Type-safe.
"""
# Core coordinators
# Core coordinators - required - create during `async_setup_entry`.
coordinator: WeatherDataUpdateCoordinator
health_coordinator: HealthCoordinator
last_options: dict[str, Any]
# Sensor platform callbacks (set by sensor.async_setup_entry)
add_sensor_entities: AddEntitiesCallback | None = None
sensor_descriptions: dict[str, WeatherSensorEntityDescription] | None = None
sensor_descriptions: dict[str, WeatherSensorEntityDescription] = field(default_factory=dict)
# Binary sensor platform callbacks
add_binary_entities: AddEntitiesCallback | None = None
binary_description: dict[str, Any] | None = None
added_binary_keys: set[str] = field(default_factory=dict)
binary_descriptions: dict[str, BinarySensorEntityDescription] = field(default_factory=dict)
added_binary_keys: set[str] = field(default_factory=set)
# Health data cache for diagnostics
# Health data cache for diagnostics - refreshed by `HealthCoordinator` on each tick.
health_data: dict[str, Any] | None = None
# Type alias for typed ConfigEntry
type SWSConfigEntry = ConfigEntry[SWSRuntimeData]
# Per-entry dict keys stored under hass.data[DOMAIN][entry_id]
ENTRY_COORDINATOR: Final[str] = "coordinator"
ENTRY_ADD_ENTITIES: Final[str] = "async_add_entities"
ENTRY_DESCRIPTIONS: Final[str] = "sensor_descriptions"
# Binary sensor dynamic support
ENTRY_ADD_BINARY_ENTITIES: Final[str] = "async_add_binary_entities"
ENTRY_BINARY_DESCRIPTION: Final[str] = "binary_sensor_description"
ENTRY_ADDED_BINARY_KEYS: Final[str] = "added_binary_keys"
ENTRY_LAST_OPTIONS: Final[str] = "last_options"
ENTRY_HEALTH_COORD: Final[str] = "coord_h"
ENTRY_HEALTH_DATA: Final[str] = "health_data"

34
custom_components/sws12500/diagnostics.py
View File

@ -5,24 +5,11 @@ from __future__ import annotations
from copy import deepcopy
from typing import Any
from py_typecheck import checked, checked_or
from homeassistant.components.diagnostics import (
async_redact_data, # pyright: ignore[reportUnknownVariableType]
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.components.diagnostics import async_redact_data # pyright: ignore[reportUnknownVariableType]
from homeassistant.core import HomeAssistant
from .const import (
API_ID,
API_KEY,
DOMAIN,
POCASI_CZ_API_ID,
POCASI_CZ_API_KEY,
WINDY_STATION_ID,
WINDY_STATION_PW,
)
from .data import ENTRY_HEALTH_COORD, ENTRY_HEALTH_DATA
from .const import API_ID, API_KEY, POCASI_CZ_API_ID, POCASI_CZ_API_KEY, WINDY_STATION_ID, WINDY_STATION_PW
from .data import SWSConfigEntry
TO_REDACT = {
API_ID,
@ -38,20 +25,17 @@ TO_REDACT = {
}
async def async_get_config_entry_diagnostics(
hass: HomeAssistant, entry: ConfigEntry
) -> dict[str, Any]:
async def async_get_config_entry_diagnostics(hass: HomeAssistant, entry: SWSConfigEntry) -> dict[str, Any]:
"""Return diagnostics for a config entry."""
data = checked_or(hass.data.get(DOMAIN), dict[str, Any], {})
del hass # Unused, but required by the interface
if (entry_data := checked(data.get(entry.entry_id), dict[str, Any])) is None:
entry_data = {}
runtime = entry.runtime_data
health_data = runtime.health_data
health_data = checked(entry_data.get(ENTRY_HEALTH_DATA), dict[str, Any])
if health_data is None:
coordinator = entry_data.get(ENTRY_HEALTH_COORD)
health_data = getattr(coordinator, "data", None)
# Fallback to the live coordinator snapshot if no payload has been persisted yet.
health_data = runtime.health_coordinator.data
return {
"entry_data": async_redact_data(dict(entry.data), TO_REDACT),

22
custom_components/sws12500/ecowitt.py
View File

@ -5,12 +5,13 @@ but does NOT start a separate HTTP server. Instead, the HA webhook handler
feeds raw POST data into EcoWittListener.process_data().
Sensors that have an internal mapping (REMAP_ECOWITT_COMPACT) are unified
with the existing SWS sensor pipline. Unmapped sensors are exposed as
native Ecowitt entites for forward compatibility.
with the existing SWS sensor pipeline. Unmapped sensors are exposed as
native Ecowitt entities for forward compatibility.
"""
from __future__ import annotations
from datetime import datetime
import logging
from typing import Any
@ -22,17 +23,18 @@ from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers.device_registry import DeviceEntryType
from homeassistant.helpers.entity import DeviceInfo
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.typing import StateType
from .const import DOMAIN, REMAP_ECOWITT_COMPAT
_LOGGER = logging.getLogger(__name__)
# Reverse mapping: internal key to ecowitt field name
# we need to know which key is internaly covered.
# we need to know which key is internally covered.
_MAPPED_ECOWITT_KEYS: set[str] = set(REMAP_ECOWITT_COMPAT.keys())
# aioecowitt sensor type to HA device class + unit
# We cover most common types, addidional will be covered later.
# We cover most common types, additional will be covered later.
STYPE_TO_HA: dict[EcoWittSensorTypes, tuple[SensorDeviceClass | None, str | None, SensorStateClass | None]] = {
EcoWittSensorTypes.TEMPERATURE_C: (
SensorDeviceClass.TEMPERATURE,
@ -157,7 +159,7 @@ class EcowittBridge:
# Callback for new sensors
self._listener.new_sensor_cb.append(self._on_new_sensor)
# We need to know which native ecowitt senosrs have an entity
# We need to know which native ecowitt sensors have an entity
self._know_native_keys: set[str] = set()
# Callback for new entities
@ -181,7 +183,7 @@ class EcowittBridge:
# then call new_sensors_cb for new sensors
self._listener.process_data(data)
# Get values for internaly mapped sensors
# Get values for internally mapped sensors
mapped_result: dict[str, str] = {}
for ecowitt_key, internal_key in REMAP_ECOWITT_COMPAT.items():
if ecowitt_key in data:
@ -233,7 +235,7 @@ class EcowittBridge:
class EcoWittNativeSensor(SensorEntity):
"""Sensor entity for Ecowitt sensors without internal mapping.
These entities are "pass-trough" - theri values are directly from EcoWittSensor
These entities are "pass-through" - their values are directly from EcoWittSensor
and maps `stype` to HA device class. They do not have coordinator, because
EcoWittSensor have his own update_cb callback mechanism.
"""
@ -268,7 +270,7 @@ class EcoWittNativeSensor(SensorEntity):
)
@property
def native_value(self) -> str | int | float | None:
def native_value(self) -> StateType | datetime: # pyright: ignore[reportIncompatibleVariableOverride]
"""Current value from Ecowitt sensor."""
value = self._ecowitt_sensor.value
if value is None or value == "":
@ -280,11 +282,11 @@ class EcoWittNativeSensor(SensorEntity):
self._ecowitt_sensor.update_cb.append(self._handle_update)
async def async_will_remove_from_hass(self) -> None:
"""REmove update callback when entity is removed."""
"""Remove update callback when entity is removed."""
if self._handle_update in self._ecowitt_sensor.update_cb:
self._ecowitt_sensor.update_cb.remove(self._handle_update)
@callback
def _handle_update(self) -> None:
"""Handle sensro values update from aioecowitt."""
"""Handle sensor values update from aioecowitt."""
self.async_write_ha_state()

21
custom_components/sws12500/health_coordinator.py
View File

@ -26,7 +26,6 @@ import aiohttp.web
from py_typecheck import checked, checked_or
from homeassistant.components.network import async_get_source_ip
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from homeassistant.helpers.network import get_url
@ -44,7 +43,7 @@ from .const import (
WSLINK_ADDON_PORT,
WSLINK_URL,
)
from .data import ENTRY_HEALTH_DATA
from .data import SWSConfigEntry
from .pocasti_cz import PocasiPush
from .routes import Routes
from .windy_func import WindyPush
@ -80,7 +79,7 @@ def _empty_forwarding_state(enabled: bool) -> dict[str, Any]:
}
def _default_health_data(config: ConfigEntry) -> dict[str, Any]:
def _default_health_data(config: SWSConfigEntry) -> dict[str, Any]:
"""Build the default health/debug payload for this config entry."""
configured_protocol = _protocol_name(checked_or(config.options.get(WSLINK), bool, False))
return {
@ -137,10 +136,10 @@ class HealthCoordinator(DataUpdateCoordinator):
All of that is stored as one structured JSON-like dict in `self.data`.
"""
def __init__(self, hass: HomeAssistant, config: ConfigEntry) -> None:
def __init__(self, hass: HomeAssistant, config: SWSConfigEntry) -> None:
"""Initialize the health coordinator."""
self.hass: HomeAssistant = hass
self.config: ConfigEntry = config
self.config: SWSConfigEntry = config
super().__init__(
hass,
@ -154,14 +153,12 @@ class HealthCoordinator(DataUpdateCoordinator):
def _store_runtime_health(self, data: dict[str, Any]) -> None:
"""Persist the latest health payload into entry runtime storage."""
if (domain := checked(self.hass.data.get(DOMAIN), dict[str, Any])) is None:
try:
self.config.runtime_data.health_data = deepcopy(data)
except AttributeError:
# runtime_data may not be set up yet during early initialization; that's fine, we'll populate it on the next tick.
return
if (entry := checked(domain.get(self.config.entry_id), dict[str, Any])) is None:
return
entry[ENTRY_HEALTH_DATA] = deepcopy(data)
def _commit(self, data: dict[str, Any]) -> dict[str, Any]:
"""Publish a new health snapshot."""
self.async_set_updated_data(data)
@ -197,7 +194,7 @@ class HealthCoordinator(DataUpdateCoordinator):
url = get_url(self.hass)
ip = await async_get_source_ip(self.hass)
port = checked_or(self.config_entry.options.get(WSLINK_ADDON_PORT), int, 443)
port = checked_or(self.config.options.get(WSLINK_ADDON_PORT), int, 443)
health_url = f"https://{ip}:{port}/healthz"
info_url = f"https://{ip}:{port}/status/internal"

34
custom_components/sws12500/health_sensor.py
View File

@ -5,12 +5,11 @@ from __future__ import annotations
from collections.abc import Callable
from dataclasses import dataclass
from functools import cached_property
from typing import Any, cast
from typing import TYPE_CHECKING, Any
from py_typecheck import checked, checked_or
from homeassistant.components.sensor import SensorDeviceClass, SensorEntity, SensorEntityDescription
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import EntityCategory
from homeassistant.core import HomeAssistant
from homeassistant.helpers.device_registry import DeviceEntryType
@ -20,7 +19,11 @@ from homeassistant.helpers.update_coordinator import CoordinatorEntity
from homeassistant.util import dt as dt_util
from .const import DOMAIN
from .data import ENTRY_HEALTH_COORD
from .data import SWSConfigEntry
from .health_coordinator import HealthCoordinator
if TYPE_CHECKING:
from .health_coordinator import HealthCoordinator
@dataclass(frozen=True, kw_only=True)
@ -195,20 +198,14 @@ HEALTH_SENSOR_DESCRIPTIONS: tuple[HealthSensorEntityDescription, ...] = (
async def async_setup_entry(
hass: HomeAssistant,
entry: ConfigEntry,
entry: SWSConfigEntry,
async_add_entities: AddEntitiesCallback,
) -> None:
"""Set up health diagnostic sensors."""
if (data := checked(hass.data.get(DOMAIN), dict[str, Any])) is None:
return
del hass # kept for backwards-compatible call signature; not used after runtime_data migration
if (entry_data := checked(data.get(entry.entry_id), dict[str, Any])) is None:
return
coordinator = entry_data.get(ENTRY_HEALTH_COORD)
if coordinator is None:
return
coordinator = entry.runtime_data.health_coordinator
entities = [
HealthDiagnosticSensor(coordinator=coordinator, description=description)
@ -222,19 +219,21 @@ class HealthDiagnosticSensor( # pyright: ignore[reportIncompatibleVariableOverr
):
"""Health diagnostic sensor for SWS-12500."""
entity_description: HealthSensorEntityDescription # pyright: ignore[reportIncompatibleVariableOverride] type: ignore[assignment]
_attr_has_entity_name = True
_attr_should_poll = False
def __init__(
self,
coordinator: Any,
coordinator: HealthCoordinator,
description: HealthSensorEntityDescription,
) -> None:
"""Initialize the sensor."""
super().__init__(coordinator)
self.entity_description = description
self._attr_entity_category = EntityCategory.DIAGNOSTIC
self._attr_unique_id = f"{description.key}_health"
self.entity_description = description # pyright: ignore[reportIncompatibleVariableOverride] type: ignore[assignment]
@property
def native_value(self) -> Any: # pyright: ignore[reportIncompatibleVariableOverride]
@ -242,10 +241,9 @@ class HealthDiagnosticSensor( # pyright: ignore[reportIncompatibleVariableOverr
data = checked_or(self.coordinator.data, dict[str, Any], {})
description = cast("HealthSensorEntityDescription", self.entity_description)
value = _resolve_path(data, description.data_path)
if description.value_fn is not None:
return description.value_fn(value)
value = _resolve_path(data, self.entity_description.data_path)
if self.entity_description.value_fn is not None:
return self.entity_description.value_fn(value)
return value
@property

78
custom_components/sws12500/legacy.py
View File

@ -1,47 +1,66 @@
"""Legacy definitions."""
"""Legacy battery sensor deprecation.
The integration used to expose battery state as regular SensorEntity instance
(unique_id == bettery key), they have been migrated to BinarySensorEntity (uniqui_id == `key`_binary). Old entity-registry entries from
pre-migration installs orphan. This module raises a Repairs issue so user can celan them up.
"""
from __future__ import annotations
from typing import Final
from py_typecheck import checked_or
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from homeassistant.helpers import issue_registry as ir
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers import entity_registry as er, issue_registry as ir
from homeassistant.helpers.issue_registry import IssueSeverity
from .const import DOMAIN, SENSORS_TO_LOAD
from .const import DOMAIN
from .data import SWSConfigEntry
LEGACY_REMOVE_VERSION: Final = "2.1.0"
LEGACY_BATTERY_KEYS: Final[set[str]] = {
"outside_battery",
"indoor_battery",
"ch2_battery",
"ch3_battery",
"ch4_battery",
"ch5_battery",
"ch6_battery",
"ch7_battery",
"ch8_battery",
}
LEGACY_BATTERY_KEYS: Final[frozenset[str]] = frozenset(
{
"outside_battery",
"indoor_battery",
"ch2_battery",
"ch3_battery",
"ch4_battery",
"ch5_battery",
"ch6_battery",
"ch7_battery",
"ch8_battery",
}
)
def _legacy_battery_issue_id(entry: ConfigEntry) -> str:
"""Issued id."""
def _legacy_battery_issue_id(entry: SWSConfigEntry) -> str:
"""Return Repairs issue id fpr this config entry."""
return f"legacy_battery_sensor_deprecation_{entry.entry_id}"
def _has_legacy_battery_loaded(entry: ConfigEntry) -> bool:
loaded = set(checked_or(entry.options.get(SENSORS_TO_LOAD), list[str], []))
return bool(loaded & LEGACY_BATTERY_KEYS)
@callback
def _orphan_legacy_battery_etries(hass: HomeAssistant, entry: SWSConfigEntry) -> list[str]:
"""Return entity_ids of legacy battery sensors still present in entity registry.
Old non-binary battery entities have:
- domian == "sensor"
- unique_id matches a LEGACY_BATTERY_KESY entry (without `_binary` suffix)
"""
ent_reg = er.async_get(hass)
return [
ent.entity_id
for ent in er.async_entries_for_config_entry(ent_reg, entry.entry_id)
if ent.domain == "sensor" and ent.unique_id in LEGACY_BATTERY_KEYS
]
def update_legacy_battery_issue(hass: HomeAssistant, entry: ConfigEntry) -> None:
"""Update legacy battery issue."""
@callback
def update_legacy_battery_issue(hass: HomeAssistant, entry: SWSConfigEntry) -> None:
"""Create or clear a Repairs issue for orphan legacy battery snesors."""
issue_id = _legacy_battery_issue_id(entry=entry)
orphans = _orphan_legacy_battery_etries(hass, entry)
if _has_legacy_battery_loaded(entry=entry):
if orphans:
ir.async_create_issue(
hass,
DOMAIN,
@ -50,7 +69,10 @@ def update_legacy_battery_issue(hass: HomeAssistant, entry: ConfigEntry) -> None
is_fixable=False,
severity=IssueSeverity.WARNING,
translation_key="legacy_battery_sensor_deprecated",
translation_placeholders={"remove_version": LEGACY_REMOVE_VERSION},
translation_placeholders={
"remove_version": LEGACY_REMOVE_VERSION,
"entities": ", ".join(orphans),
},
)
else:
ir.async_delete_issue(hass, DOMAIN, issue_id=issue_id)

22
custom_components/sws12500/pocasti_cz.py
View File

@ -1,5 +1,7 @@
"""Pocasi CZ resend functions."""
from __future__ import annotations
from datetime import datetime, timedelta
import logging
from typing import Any, Literal
@ -74,9 +76,7 @@ class PocasiPush:
return None
async def push_data_to_server(
self, data: dict[str, Any], mode: Literal["WU", "WSLINK"]
):
async def push_data_to_server(self, data: dict[str, Any], mode: Literal["WU", "WSLINK"]):
"""Pushes weather data to server."""
_data = data.copy()
@ -85,19 +85,13 @@ class PocasiPush:
self.last_error = None
if (_api_id := checked(self.config.options.get(POCASI_CZ_API_ID), str)) is None:
_LOGGER.error(
"No API ID is provided for Pocasi Meteo. Check your configuration."
)
_LOGGER.error("No API ID is provided for Pocasi Meteo. Check your configuration.")
self.last_status = "config_error"
self.last_error = "Missing API ID."
return
if (
_api_key := checked(self.config.options.get(POCASI_CZ_API_KEY), str)
) is None:
_LOGGER.error(
"No API Key is provided for Pocasi Meteo. Check your configuration."
)
if (_api_key := checked(self.config.options.get(POCASI_CZ_API_KEY), str)) is None:
_LOGGER.error("No API Key is provided for Pocasi Meteo. Check your configuration.")
self.last_status = "config_error"
self.last_error = "Missing API key."
return
@ -148,9 +142,7 @@ class PocasiPush:
self.last_error = POCASI_INVALID_KEY
self.enabled = False
_LOGGER.critical(POCASI_INVALID_KEY)
await update_options(
self.hass, self.config, POCASI_CZ_ENABLED, False
)
await update_options(self.hass, self.config, POCASI_CZ_ENABLED, False)
except PocasiSuccess:
self.last_status = "ok"
self.last_error = None

2
custom_components/sws12500/routes.py
View File

@ -15,6 +15,8 @@ Important note:
an old coordinator while entities listen to a new one (result: UI appears "frozen").
"""
from __future__ import annotations
from collections.abc import Awaitable, Callable
from dataclasses import dataclass, field
import logging

123
custom_components/sws12500/sensor.py
View File

@ -4,26 +4,25 @@ This module creates sensor entities based on the config entry options.
The integration is push-based (webhook), so we avoid reloading the entry for
auto-discovered sensors. Instead, we dynamically add new entities at runtime
using the `async_add_entities` callback stored in `hass.data`.
using the `async_add_entities` callback stored in `entry.runtime_data`.
Why not reload on auto-discovery?
Reloading a config entry unloads platforms temporarily, which removes coordinator
listeners. With frequent webhook pushes, this can create a window where nothing is
subscribed and the frontend appears "frozen" until another full reload/restart.
Runtime state is stored under:
hass.data[DOMAIN][entry_id] -> dict with known keys (see `data.py`)
Per-entry runtime state lives on `entry.runtime_data` (see data.SWSRuntimeData)
"""
from collections.abc import Callable
from __future__ import annotations
from functools import cached_property
import logging
from typing import Any, cast
from typing import TYPE_CHECKING, Any
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
from homeassistant.core import HomeAssistant
from homeassistant.helpers.device_registry import DeviceEntryType
from homeassistant.helpers.entity import DeviceInfo, generate_entity_id
@ -33,6 +32,7 @@ from homeassistant.helpers.update_coordinator import CoordinatorEntity
from . import health_sensor
from .const import (
CHILL_INDEX,
DEV_DBG,
DOMAIN,
HEAT_INDEX,
OUTSIDE_HUMIDITY,
@ -43,17 +43,15 @@ from .const import (
WIND_SPEED,
WSLINK,
)
from .data import ENTRY_ADD_ENTITIES, ENTRY_COORDINATOR, ENTRY_DESCRIPTIONS
from .data import SWSConfigEntry
from .sensors_common import WeatherSensorEntityDescription
from .sensors_weather import SENSOR_TYPES_WEATHER_API
from .sensors_wslink import SENSOR_TYPES_WSLINK
_LOGGER = logging.getLogger(__name__)
if TYPE_CHECKING:
from .coordinator import WeatherDataUpdateCoordinator
# The `async_add_entities` callback accepts a list of Entity-like objects.
# We keep the type loose here to avoid propagating HA generics (`DataUpdateCoordinator[T]`)
# that often end up as "partially unknown" under type-checkers.
_AddEntitiesFn = Callable[[list[SensorEntity]], None]
_LOGGER = logging.getLogger(__name__)
def _auto_enable_derived_sensors(requested: set[str]) -> set[str]:
@ -83,45 +81,27 @@ def _auto_enable_derived_sensors(requested: set[str]) -> set[str]:
async def async_setup_entry(
hass: HomeAssistant,
config_entry: ConfigEntry,
config_entry: SWSConfigEntry,
async_add_entities: AddEntitiesCallback,
) -> None:
"""Set up Weather Station sensors.
We also store `async_add_entities` and a map of sensor descriptions in `hass.data`
so the webhook handler can add newly discovered entities dynamically without
reloading the config entry.
Stores `async_add_entities` and the sensor-description map on `entry.runtime_data`
so the webhook handler can add newly discovered entities dynamically without reloading the config entry.
"""
if (hass_data := checked(hass.data.setdefault(DOMAIN, {}), dict[str, Any])) is None:
return
runtime = config_entry.runtime_data
coordinator = runtime.coordinator
# we have to check if entry_data are present
# It is created by integration setup, so it should be presnet
if (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:
# Coordinator is created by the integration (`__init__.py`). Without it, we cannot set up entities.
# This should not happen in normal operation; treat it as a no-op setup.
return
# Store the platform callback so we can add entities later (auto-discovery) without reload.
entry_data[ENTRY_ADD_ENTITIES] = async_add_entities
# Wire up the integration health diagnostic sensor.
# This is kept in a dedicated module (`health_sensor.py`) for readability.
# Wire up integration health diagnostic sensor.
await health_sensor.async_setup_entry(hass, config_entry, async_add_entities)
wslink_enabled = checked_or(config_entry.options.get(WSLINK), bool, False)
sensor_types = SENSOR_TYPES_WSLINK if wslink_enabled else SENSOR_TYPES_WEATHER_API
# Keep a descriptions map for dynamic entity creation by key.
# When the station starts sending a new payload field, the webhook handler can
# look up its description here and instantiate the matching entity.
entry_data[ENTRY_DESCRIPTIONS] = {desc.key: desc for desc in sensor_types}
# Persist platform callback + description map for dynamic entity creation.
runtime.add_sensor_entities = async_add_entities
runtime.sensor_descriptions = {desc.key: desc for desc in sensor_types}
sensors_to_load = checked_or(config_entry.options.get(SENSORS_TO_LOAD), list[str], [])
if not sensors_to_load:
@ -134,13 +114,12 @@ async def async_setup_entry(
]
async_add_entities(entities)
# Connect Ecowitt bridge to sensor platform,
# so it can dynamically add native Ecowitt entities
if hasattr(coordinator, "ecowitt_bridge"):
coordinator.ecowitt_bridge.set_add_entities(async_add_entities)
# Connect Ecowitt bridge to sensor platform so it can dynamically add
# native Ecowitt entities (sensors without internal SWS mapping).
coordinator.ecowitt_bridge.set_add_entities(async_add_entities)
def add_new_sensors(hass: HomeAssistant, config_entry: ConfigEntry, keys: list[str]) -> None:
def add_new_sensors(hass: HomeAssistant, config_entry: SWSConfigEntry, keys: list[str]) -> None:
"""Dynamically add newly discovered sensors without reloading the entry.
Called by the webhook handler when the station starts sending new fields.
@ -151,31 +130,22 @@ def add_new_sensors(hass: HomeAssistant, config_entry: ConfigEntry, keys: list[s
- Unknown payload keys are ignored (only keys with an entity description are added).
"""
if (hass_data := checked(hass.data.get(DOMAIN), dict[str, Any])) is None:
del hass # kept for backwards-compatible call signature; not used after runtime_data migration
runtime = config_entry.runtime_data
add_entities = runtime.add_sensor_entities
if add_entities is None:
return
if (entry_data := checked(hass_data.get(config_entry.entry_id), dict[str, Any])) is None:
return
descriptions = runtime.sensor_descriptions
coordinator = runtime.coordinator
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: list[SensorEntity] = [
WeatherSensor(desc, coordinator) for key in keys if (desc := descriptions.get(key)) is not None
]
if new_entities:
add_entities_fn(new_entities)
add_entities(new_entities)
class WeatherSensor( # pyright: ignore[reportIncompatibleVariableOverride]
@ -187,26 +157,21 @@ class WeatherSensor( # pyright: ignore[reportIncompatibleVariableOverride]
propagating HA's generic `DataUpdateCoordinator[T]` typing into this module.
"""
entity_description: WeatherSensorEntityDescription # pyright: ignore[reportIncompatibleVariableOverride]
_attr_has_entity_name = True
_attr_should_poll = False
def __init__(
self,
description: WeatherSensorEntityDescription,
coordinator: Any,
coordinator: WeatherDataUpdateCoordinator,
) -> None:
"""Initialize sensor."""
super().__init__(coordinator)
self.entity_description = description
self._attr_unique_id = description.key
config_entry = getattr(self.coordinator, "config", None)
self._dev_log = checked_or(
config_entry.options.get("dev_debug_checkbox") if config_entry is not None else False,
bool,
False,
)
self.entity_description = description # pyright: ignore[reportIncompatibleVariableOverride] type: ignore[assignment]
self._dev_log = checked_or(coordinator.config.options.get(DEV_DBG), bool, False)
@property
def native_value(self): # pyright: ignore[reportIncompatibleVariableOverride]
@ -223,11 +188,9 @@ class WeatherSensor( # pyright: ignore[reportIncompatibleVariableOverride]
data: dict[str, Any] = checked_or(self.coordinator.data, dict[str, Any], {})
key = self.entity_description.key
description = cast("WeatherSensorEntityDescription", self.entity_description)
if description.value_from_data_fn is not None:
if self.entity_description.value_from_data_fn is not None:
try:
value = description.value_from_data_fn(data)
value = self.entity_description.value_from_data_fn(data)
except Exception: # noqa: BLE001
_LOGGER.exception("native_value compute failed via value_from_data_fn for key=%s", key)
return None
@ -240,13 +203,13 @@ class WeatherSensor( # pyright: ignore[reportIncompatibleVariableOverride]
_LOGGER.debug("native_value missing raw: key=%s raw=%s", key, raw)
return None
if description.value_fn is None:
if self.entity_description.value_fn is None:
if self._dev_log:
_LOGGER.debug("native_value has no value_fn: key=%s raw=%s", key, raw)
return None
try:
value = description.value_fn(raw)
value = self.entity_description.value_fn(raw)
except Exception: # noqa: BLE001
_LOGGER.exception("native_value compute failed via value_fn for key=%s raw=%s", key, raw)
return None

5
custom_components/sws12500/sensors_common.py
View File

@ -1,12 +1,15 @@
"""Common classes for sensors."""
from __future__ import annotations
from collections.abc import Callable
from dataclasses import dataclass
from typing import Any
from dev.custom_components.sws12500.const import VOCLevel
from homeassistant.components.sensor import SensorEntityDescription
from .const import VOCLevel
@dataclass(frozen=True, kw_only=True)
class WeatherSensorEntityDescription(SensorEntityDescription):

2
custom_components/sws12500/sensors_weather.py
View File

@ -1,5 +1,7 @@
"""Sensor entities for the SWS12500 integration for old endpoint."""
from __future__ import annotations
from homeassistant.components.sensor import SensorDeviceClass, SensorStateClass
from homeassistant.const import (
DEGREE,

2
custom_components/sws12500/sensors_wslink.py
View File

@ -1,5 +1,7 @@
"""Sensor entities for the SWS12500 integration for old endpoint."""
from __future__ import annotations
from homeassistant.components.sensor import SensorDeviceClass, SensorStateClass
from homeassistant.const import (
CONCENTRATION_PARTS_PER_BILLION,

237
custom_components/sws12500/strings.json
View File

@ -49,7 +49,8 @@
"valid_credentials_api": "Provide valid API ID.",
"valid_credentials_key": "Provide valid API KEY.",
"valid_credentials_match": "API ID and API KEY should not be the same.",
"windy_key_required": "Windy API key is required if you want to enable this function."
"windy_id_required": "Windy API ID is required if you want to enable this function.",
"windy_pw_required": "Windy API password is required if you want to enable this function."
},
"step": {
"init": {
@ -61,31 +62,33 @@
}
},
"basic": {
"description": "Configure the PWS/WSLink endpoint. Turn off 'Enable PWS/WSLink' for an Ecowitt-only setup - API ID/KEY are not required."""title": "Configure PWS/WSLink","data": {
"description": "Provide API ID and API KEY so the Weather Station can access HomeAssistant",
"title": "Configure credentials",
"data": {
"API_ID": "API ID / Station ID",
"API_KEY": "API KEY / Password",
"wslink": "WSLink protocol",
"legacy_enbaled": "Enable PWS/WSLink endpoint (disable for Ecowitt-only setup)",
"WSLINK": "WSLink API",
"dev_debug_checkbox": "Developer log"
},
"data_description": {
"dev_debug_checkbox": " Enable only if you want to send debuging data to the developer.",
"API_ID": "API ID is the Station ID you set in the Weather Station.",
"API_KEY": "API KEY is the password you set in the Weather Station.",
"wslink": "Enable WSLink API if the station is set to send data via WSLink. (If you are unsure, use https://test-station.schizza.cz/)",
"legacy_enbaled": "Turn off if your station uses Ecowitt only."
"WSLINK": "Enable WSLink API if the station is set to send data via WSLink."
}
},
"windy": {
"description": "Resend weather data to your Windy stations.",
"title": "Configure Windy",
"data": {
"WINDY_API_KEY": "API KEY provided by Windy",
"WINDY_STATION_ID": "Station ID obtained form Windy",
"WINDY_STATION_PWD": "Station password obtained from Windy",
"windy_enabled_checkbox": "Enable resending data to Windy",
"windy_logger_checkbox": "Log Windy data and responses"
},
"data_description": {
"WINDY_API_KEY": "Windy API KEY obtained from https://https://api.windy.com/keys",
"WINDY_STATION_ID": "Windy station ID obtained from https://stations.windy.com/stations",
"WINDY_STATION_PWD": "Windy station password obtained from https://stations.windy.com/stations",
"windy_logger_checkbox": "Enable only if you want to send debuging data to the developer."
}
},
@ -134,7 +137,131 @@
}
},
"entity": {
"binary_sensor": {
"outside_battery": {
"name": "Outside battery"
},
"indoor_battery": {
"name": "Console battery"
},
"ch2_battery": {
"name": "Channel 2 battery"
},
"ch3_battery": {
"name": "Channel 3 battery"
},
"ch4_battery": {
"name": "Channel 4 battery"
},
"ch5_battery": {
"name": "Channel 5 battery"
},
"ch6_battery": {
"name": "Channel 6 battery"
},
"ch7_battery": {
"name": "Channel 7 battery"
},
"ch8_battery": {
"name": "Channel 8 battery"
}
},
"sensor": {
"integration_health": {
"name": "Integration status",
"state": {
"online_wu": "Online PWS/WU",
"online_wslink": "Online WSLink",
"online_idle": "Waiting for data",
"degraded": "Degraded",
"error": "Error"
}
},
"active_protocol": {
"name": "Active protocol",
"state": {
"wu": "PWS/WU",
"wslink": "WSLink API"
}
},
"wslink_addon_status": {
"name": "WSLink Addon Status",
"state": {
"online": "Running",
"offline": "Offline"
}
},
"wslink_addon_name": {
"name": "WSLink Addon Name"
},
"wslink_addon_version": {
"name": "WSLink Addon Version"
},
"wslink_addon_listen_port": {
"name": "WSLink Addon Listen Port"
},
"wslink_upstream_ha_port": {
"name": "WSLink Addon Upstream HA Port"
},
"route_wu_enabled": {
"name": "PWS/WU Protocol"
},
"route_wslink_enabled": {
"name": "WSLink Protocol"
},
"last_ingress_time": {
"name": "Last access time"
},
"last_ingress_protocol": {
"name": "Last access protocol",
"state": {
"wu": "PWS/WU",
"wslink": "WSLink API"
}
},
"last_ingress_route_enabled": {
"name": "Last ingress route enabled"
},
"last_ingress_accepted": {
"name": "Last access",
"state": {
"accepted": "Accepted",
"rejected": "Rejected"
}
},
"last_ingress_authorized": {
"name": "Last access authorization",
"state": {
"authorized": "Authorized",
"unauthorized": "Unauthorized",
"unknown": "Unknown"
}
},
"last_ingress_reason": {
"name": "Last access reason"
},
"forward_windy_enabled": {
"name": "Forwarding to Windy"
},
"forward_windy_status": {
"name": "Forwarding status to Windy",
"state": {
"disabled": "Disabled",
"idle": "Waiting to send",
"ok": "Ok"
}
},
"forward_pocasi_enabled": {
"name": "Forwarding to Počasí Meteo"
},
"forward_pocasi_status": {
"name": "Forwarding status to Počasí Meteo",
"state": {
"disabled": "Disabled",
"idle": "Waiting to send",
"ok": "Ok"
}
},
"indoor_temp": {
"name": "Indoor temperature"
},
@ -192,6 +319,30 @@
"ch4_humidity": {
"name": "Channel 4 humidity"
},
"ch5_temp": {
"name": "Channel 5 temperature"
},
"ch5_humidity": {
"name": "Channel 5 humidity"
},
"ch6_temp": {
"name": "Channel 6 temperature"
},
"ch6_humidity": {
"name": "Channel 6 humidity"
},
"ch7_temp": {
"name": "Channel 7 temperature"
},
"ch7_humidity": {
"name": "Channel 7 humidity"
},
"ch8_temp": {
"name": "Channel 8 temperature"
},
"ch8_humidity": {
"name": "Channel 8 humidity"
},
"heat_index": {
"name": "Apparent temperature"
},
@ -213,6 +364,22 @@
"wbgt_index": {
"name": "WBGT index"
},
"hcho": {
"name": "Formaldehyde (HCHO)"
},
"voc": {
"name": "VOC level",
"state": {
"unhealthy": "Unhealthy",
"poor": "Poor",
"moderate": "Moderate",
"good": "Good",
"excellent": "Excellent"
}
},
"t9_battery": {
"name": "HCHO/VOC sensor battery"
},
"wind_azimut": {
"name": "Bearing",
"state": {
@ -250,6 +417,54 @@
"unknown": "Unknown / drained out"
}
},
"ch3_battery": {
"name": "Channel 3 battery level",
"state": {
"normal": "OK",
"low": "Low",
"unknown": "Unknown / drained out"
}
},
"ch4_battery": {
"name": "Channel 4 battery level",
"state": {
"normal": "OK",
"low": "Low",
"unknown": "Unknown / drained out"
}
},
"ch5_battery": {
"name": "Channel 5 battery level",
"state": {
"normal": "OK",
"low": "Low",
"unknown": "Unknown / drained out"
}
},
"ch6_battery": {
"name": "Channel 6 battery level",
"state": {
"normal": "OK",
"low": "Low",
"unknown": "Unknown / drained out"
}
},
"ch7_battery": {
"name": "Channel 7 battery level",
"state": {
"normal": "OK",
"low": "Low",
"unknown": "Unknown / drained out"
}
},
"ch8_battery": {
"name": "Channel 8 battery level",
"state": {
"normal": "OK",
"low": "Low",
"unknown": "Unknown / drained out"
}
},
"indoor_battery": {
"name": "Console battery level",
"state": {
@ -260,6 +475,12 @@
}
}
},
"issues": {
"legacy_battery_sensor_deprecated": {
"title": "Legacy battery sensor detected",
"description": "Old battery sensor entities ({entities}) were found in the entity registry. They have been replaced by binary battery sensors and will be removed in version {remove_version}. Please delete the legacy entities manually via Settings Old battery sensor entities ({entities}) were found in the entity registry. They have been replaced by binary battery sensors and will be removed in version {remove_version}. Please delete the legacy entities manually via Settings -> Devices & Services -> SWS 12500."
}
},
"notify": {
"added": {
"title": "New sensors for SWS 12500 found.",

35
custom_components/sws12500/translations/cs.json
View File

@ -156,6 +156,35 @@
}
},
"entity": {
"binary_sensor": {
"outside_battery": {
"name": "Venkovní baterie"
},
"indoor_battery": {
"name": "Baterie kozole"
},
"ch2_battery": {
"name": "Baterie senzoru 2"
},
"ch3_battery": {
"name": "Baterie senzoru 3"
},
"ch4_battery": {
"name": "Baterie senzoru 4"
},
"ch5_battery": {
"name": "Baterie senzoru 5"
},
"ch6_battery": {
"name": "Baterie senzoru 6"
},
"ch7_battery": {
"name": "Baterie senzoru 7"
},
"ch8_battery": {
"name": "Baterie senzoru 8"
}
},
"sensor": {
"integration_health": {
"name": "Stav integrace",
@ -393,6 +422,12 @@
}
}
},
"issues": {
"legacy_battery_sensor_deprecated": {
"title": "Detekovány zastaralé senzory baterií.",
"description": "V registru entit byly nalezeny staré senzory baterií ({entities}). Byly nahrazeny binárními senzory baterií a budou odstraněny ve verzi {remove_version}. Smažte prosím staré entity ručně přes Nastavení -> Zařízení a služby -> SWS-12500."
}
},
"notify": {
"added": {
"title": "Nalezeny nové senzory pro SWS 12500.",

35
custom_components/sws12500/translations/en.json
View File

@ -137,6 +137,35 @@
}
},
"entity": {
"binary_sensor": {
"outside_battery": {
"name": "Outside battery"
},
"indoor_battery": {
"name": "Console battery"
},
"ch2_battery": {
"name": "Channel 2 battery"
},
"ch3_battery": {
"name": "Channel 3 battery"
},
"ch4_battery": {
"name": "Channel 4 battery"
},
"ch5_battery": {
"name": "Channel 5 battery"
},
"ch6_battery": {
"name": "Channel 6 battery"
},
"ch7_battery": {
"name": "Channel 7 battery"
},
"ch8_battery": {
"name": "Channel 8 battery"
}
},
"sensor": {
"integration_health": {
"name": "Integration status",
@ -446,6 +475,12 @@
}
}
},
"issues": {
"legacy_battery_sensor_deprecated": {
"title": "Legacy battery sensor detected",
"description": "Old battery sensor entities ({entities}) were found in the entity registry. They have been replaced by binary battery sensors and will be removed in version {remove_version}. Please delete the legacy entities manually via Settings Old battery sensor entities ({entities}) were found in the entity registry. They have been replaced by binary battery sensors and will be removed in version {remove_version}. Please delete the legacy entities manually via Settings -> Devices & Services -> SWS 12500."
}
},
"notify": {
"added": {
"title": "New sensors for SWS 12500 found.",

10
custom_components/sws12500/utils.py
View File

@ -11,9 +11,11 @@ Notable responsibilities:
Keeping these concerns in one place avoids duplicating logic in the webhook handler and entity code.
"""
from __future__ import annotations
import logging
import math
from typing import Any, cast
from typing import Any
from py_typecheck.core import checked_or
@ -78,7 +80,7 @@ async def translated_notification(
localize_title = f"component.{translation_domain}.{category}.{translation_key}.title"
language: str = cast("str", hass.config.language)
language: str = hass.config.language
_translations = await async_get_translations(hass, language, category, [translation_domain])
if localize_key in _translations:
@ -291,7 +293,7 @@ def heat_index(data: dict[str, int | float | str], convert: bool = False) -> flo
"""Calculate heat index from temperature.
data: dict with temperature and humidity
convert: bool, convert recieved data from Celsius to Fahrenheit
convert: bool, convert received data from Celsius to Fahrenheit
"""
if (temp := to_float(data.get(OUTSIDE_TEMP))) is None:
_LOGGER.error(
@ -340,7 +342,7 @@ def chill_index(data: dict[str, str | float | int], convert: bool = False) -> fl
"""Calculate wind chill index from temperature and wind speed.
data: dict with temperature and wind speed
convert: bool, convert recieved data from Celsius to Fahrenheit
convert: bool, convert received data from Celsius to Fahrenheit
"""
temp = to_float(data.get(OUTSIDE_TEMP))
wind = to_float(data.get(WIND_SPEED))

30
custom_components/sws12500/windy_func.py
View File

@ -1,5 +1,7 @@
"""Windy functions."""
from __future__ import annotations
from datetime import datetime, timedelta
import logging
@ -155,9 +157,7 @@ class WindyPush:
persistent_notification.create(self.hass, reason, "Windy resending disabled.")
async def push_data_to_windy(
self, data: dict[str, str], wslink: bool = False
) -> bool:
async def push_data_to_windy(self, data: dict[str, str], wslink: bool = False) -> bool:
"""Pushes weather data do Windy stations.
Interval is 5 minutes, otherwise Windy would not accepts data.
@ -171,9 +171,7 @@ class WindyPush:
self.last_attempt_at = datetime.now().isoformat()
self.last_error = None
if (
windy_station_id := checked(self.config.options.get(WINDY_STATION_ID), str)
) is None:
if (windy_station_id := checked(self.config.options.get(WINDY_STATION_ID), str)) is None:
_LOGGER.error("Windy API key is not provided! Check your configuration.")
self.last_status = "config_error"
await self._disable_windy(
@ -181,12 +179,8 @@ class WindyPush:
)
return False
if (
windy_station_pw := checked(self.config.options.get(WINDY_STATION_PW), str)
) is None:
_LOGGER.error(
"Windy station password is missing! Check your configuration."
)
if (windy_station_pw := checked(self.config.options.get(WINDY_STATION_PW), str)) is None:
_LOGGER.error("Windy station password is missing! Check your configuration.")
self.last_status = "config_error"
await self._disable_windy(
"Windy password is not provided. Resending is disabled for now. Reconfigure your integration."
@ -226,9 +220,7 @@ class WindyPush:
_LOGGER.info("Dataset for windy: %s", purged_data)
session = async_get_clientsession(self.hass)
try:
async with session.get(
request_url, params=purged_data, headers=headers
) as resp:
async with session.get(request_url, params=purged_data, headers=headers) as resp:
try:
self.verify_windy_response(response=resp)
except WindyNotInserted:
@ -286,9 +278,7 @@ class WindyPush:
)
finally:
if self.invalid_response_count >= 3:
_LOGGER.critical(
"Invalid response from Windy 3 times. Disabling resend option."
)
_LOGGER.critical("Invalid response from Windy 3 times. Disabling resend option.")
await self._disable_windy(
reason="Unable to send data to Windy (3 times). Disabling resend option for now. Please check your Windy configuration and enable this feature afterwards."
)
@ -304,9 +294,7 @@ class WindyPush:
self.invalid_response_count += 1
if self.invalid_response_count >= WINDY_MAX_RETRIES:
_LOGGER.critical(WINDY_UNEXPECTED)
await self._disable_windy(
reason="Invalid response from Windy 3 times. Disabling resending option."
)
await self._disable_windy(reason="Invalid response from Windy 3 times. Disabling resending option.")
self.last_update = datetime.now()
self.next_update = self.last_update + timed(minutes=5)