274 lines
8.9 KiB
Python
274 lines
8.9 KiB
Python
"""Health diagnostic sensors for SWS-12500."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from collections.abc import Callable
|
|
from dataclasses import dataclass
|
|
from functools import cached_property
|
|
from typing import Any, cast
|
|
|
|
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
|
|
from homeassistant.helpers.entity import DeviceInfo
|
|
from homeassistant.helpers.entity_platform import AddEntitiesCallback
|
|
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
|
|
|
|
|
|
@dataclass(frozen=True, kw_only=True)
|
|
class HealthSensorEntityDescription(SensorEntityDescription):
|
|
"""Description for health diagnostic sensors."""
|
|
|
|
data_path: tuple[str, ...]
|
|
value_fn: Callable[[Any], Any] | None = None
|
|
|
|
|
|
def _resolve_path(data: dict[str, Any], path: tuple[str, ...]) -> Any:
|
|
"""Resolve a nested path from a dictionary."""
|
|
current: Any = data
|
|
for key in path:
|
|
if checked(current, dict[str, Any]) is None:
|
|
return None
|
|
current = current.get(key)
|
|
return current
|
|
|
|
|
|
def _on_off(value: Any) -> str:
|
|
"""Render a boolean-ish value as `on` / `off`."""
|
|
return "on" if bool(value) else "off"
|
|
|
|
|
|
def _accepted_state(value: Any) -> str:
|
|
"""Render ingress acceptance state."""
|
|
return "accepted" if bool(value) else "rejected"
|
|
|
|
|
|
def _authorized_state(value: Any) -> str:
|
|
"""Render ingress authorization state."""
|
|
if value is None:
|
|
return "unknown"
|
|
return "authorized" if bool(value) else "unauthorized"
|
|
|
|
|
|
def _timestamp_or_none(value: Any) -> Any:
|
|
"""Convert ISO timestamp string to datetime for HA rendering."""
|
|
if not isinstance(value, str):
|
|
return None
|
|
return dt_util.parse_datetime(value)
|
|
|
|
|
|
HEALTH_SENSOR_DESCRIPTIONS: tuple[HealthSensorEntityDescription, ...] = (
|
|
HealthSensorEntityDescription(
|
|
key="integration_health",
|
|
translation_key="integration_health",
|
|
icon="mdi:heart-pulse",
|
|
data_path=("integration_status",),
|
|
),
|
|
HealthSensorEntityDescription(
|
|
key="active_protocol",
|
|
translation_key="active_protocol",
|
|
icon="mdi:swap-horizontal",
|
|
data_path=("active_protocol",),
|
|
),
|
|
HealthSensorEntityDescription(
|
|
key="wslink_addon_status",
|
|
translation_key="wslink_addon_status",
|
|
icon="mdi:server-network",
|
|
data_path=("addon", "online"),
|
|
value_fn=lambda value: "online" if value else "offline",
|
|
),
|
|
HealthSensorEntityDescription(
|
|
key="wslink_addon_name",
|
|
translation_key="wslink_addon_name",
|
|
icon="mdi:package-variant-closed",
|
|
data_path=("addon", "name"),
|
|
),
|
|
HealthSensorEntityDescription(
|
|
key="wslink_addon_version",
|
|
translation_key="wslink_addon_version",
|
|
icon="mdi:label-outline",
|
|
data_path=("addon", "version"),
|
|
),
|
|
HealthSensorEntityDescription(
|
|
key="wslink_addon_listen_port",
|
|
translation_key="wslink_addon_listen_port",
|
|
icon="mdi:lan-connect",
|
|
data_path=("addon", "listen_port"),
|
|
),
|
|
HealthSensorEntityDescription(
|
|
key="wslink_upstream_ha_port",
|
|
translation_key="wslink_upstream_ha_port",
|
|
icon="mdi:transit-connection-variant",
|
|
data_path=("addon", "upstream_ha_port"),
|
|
),
|
|
HealthSensorEntityDescription(
|
|
key="route_wu_enabled",
|
|
translation_key="route_wu_enabled",
|
|
icon="mdi:transit-connection-horizontal",
|
|
data_path=("routes", "wu_enabled"),
|
|
value_fn=_on_off,
|
|
),
|
|
HealthSensorEntityDescription(
|
|
key="route_wslink_enabled",
|
|
translation_key="route_wslink_enabled",
|
|
icon="mdi:transit-connection-horizontal",
|
|
data_path=("routes", "wslink_enabled"),
|
|
value_fn=_on_off,
|
|
),
|
|
HealthSensorEntityDescription(
|
|
key="last_ingress_time",
|
|
translation_key="last_ingress_time",
|
|
icon="mdi:clock-outline",
|
|
device_class=SensorDeviceClass.TIMESTAMP,
|
|
data_path=("last_ingress", "time"),
|
|
value_fn=_timestamp_or_none,
|
|
),
|
|
HealthSensorEntityDescription(
|
|
key="last_ingress_protocol",
|
|
translation_key="last_ingress_protocol",
|
|
icon="mdi:download-network",
|
|
data_path=("last_ingress", "protocol"),
|
|
),
|
|
HealthSensorEntityDescription(
|
|
key="last_ingress_route_enabled",
|
|
translation_key="last_ingress_route_enabled",
|
|
icon="mdi:check-network",
|
|
data_path=("last_ingress", "route_enabled"),
|
|
value_fn=_on_off,
|
|
),
|
|
HealthSensorEntityDescription(
|
|
key="last_ingress_accepted",
|
|
translation_key="last_ingress_accepted",
|
|
icon="mdi:check-decagram",
|
|
data_path=("last_ingress", "accepted"),
|
|
value_fn=_accepted_state,
|
|
),
|
|
HealthSensorEntityDescription(
|
|
key="last_ingress_authorized",
|
|
translation_key="last_ingress_authorized",
|
|
icon="mdi:key",
|
|
data_path=("last_ingress", "authorized"),
|
|
value_fn=_authorized_state,
|
|
),
|
|
HealthSensorEntityDescription(
|
|
key="last_ingress_reason",
|
|
translation_key="last_ingress_reason",
|
|
icon="mdi:message-alert-outline",
|
|
data_path=("last_ingress", "reason"),
|
|
),
|
|
HealthSensorEntityDescription(
|
|
key="forward_windy_enabled",
|
|
translation_key="forward_windy_enabled",
|
|
icon="mdi:weather-windy",
|
|
data_path=("forwarding", "windy", "enabled"),
|
|
value_fn=_on_off,
|
|
),
|
|
HealthSensorEntityDescription(
|
|
key="forward_windy_status",
|
|
translation_key="forward_windy_status",
|
|
icon="mdi:weather-windy",
|
|
data_path=("forwarding", "windy", "last_status"),
|
|
),
|
|
HealthSensorEntityDescription(
|
|
key="forward_pocasi_enabled",
|
|
translation_key="forward_pocasi_enabled",
|
|
icon="mdi:cloud-upload-outline",
|
|
data_path=("forwarding", "pocasi", "enabled"),
|
|
value_fn=_on_off,
|
|
),
|
|
HealthSensorEntityDescription(
|
|
key="forward_pocasi_status",
|
|
translation_key="forward_pocasi_status",
|
|
icon="mdi:cloud-upload-outline",
|
|
data_path=("forwarding", "pocasi", "last_status"),
|
|
),
|
|
)
|
|
|
|
|
|
async def async_setup_entry(
|
|
hass: HomeAssistant,
|
|
entry: ConfigEntry,
|
|
async_add_entities: AddEntitiesCallback,
|
|
) -> None:
|
|
"""Set up health diagnostic sensors."""
|
|
|
|
if (data := checked(hass.data.get(DOMAIN), dict[str, Any])) is None:
|
|
return
|
|
|
|
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
|
|
|
|
entities = [
|
|
HealthDiagnosticSensor(coordinator=coordinator, description=description)
|
|
for description in HEALTH_SENSOR_DESCRIPTIONS
|
|
]
|
|
async_add_entities(entities)
|
|
|
|
|
|
class HealthDiagnosticSensor( # pyright: ignore[reportIncompatibleVariableOverride]
|
|
CoordinatorEntity, SensorEntity
|
|
):
|
|
"""Health diagnostic sensor for SWS-12500."""
|
|
|
|
_attr_has_entity_name = True
|
|
_attr_should_poll = False
|
|
|
|
def __init__(
|
|
self,
|
|
coordinator: Any,
|
|
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"
|
|
|
|
@property
|
|
def native_value(self) -> Any: # pyright: ignore[reportIncompatibleVariableOverride]
|
|
"""Return the current diagnostic value."""
|
|
|
|
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)
|
|
return value
|
|
|
|
@property
|
|
def extra_state_attributes(self) -> dict[str, Any] | None: # pyright: ignore[reportIncompatibleVariableOverride]
|
|
"""Expose the full health JSON on the main health sensor for debugging."""
|
|
if self.entity_description.key != "integration_health":
|
|
return None
|
|
|
|
return checked_or(self.coordinator.data, dict[str, Any], None)
|
|
|
|
@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",
|
|
)
|