"""Sensor platform for Reticulum.""" from __future__ import annotations from collections.abc import Callable from dataclasses import dataclass from datetime import datetime, timezone from typing import Any from homeassistant.components.sensor import ( SensorDeviceClass, SensorEntity, SensorEntityDescription, SensorStateClass, ) from homeassistant.config_entries import ConfigEntry from homeassistant.const import ( PERCENTAGE, SIGNAL_STRENGTH_DECIBELS_MILLIWATT, EntityCategory, UnitOfDataRate, UnitOfInformation, ) from homeassistant.core import HomeAssistant from homeassistant.helpers.entity_platform import AddEntitiesCallback from .const import DOMAIN from .entity import ReticulumEntity from .reticulum_client import ManagerState, ReticulumManager @dataclass(frozen=True, kw_only=True) class ReticulumSensorDescription(SensorEntityDescription): """Describes a Reticulum sensor.""" value_fn: Callable[[ManagerState], Any] attr_fn: Callable[[ManagerState], dict[str, Any]] | None = None def _last_message_time(state: ManagerState) -> datetime | None: if state.last_message_time is None: return None return datetime.fromtimestamp(state.last_message_time, tz=timezone.utc) SENSORS: tuple[ReticulumSensorDescription, ...] = ( ReticulumSensorDescription( key="lxmf_address", translation_key="lxmf_address", icon="mdi:identifier", entity_category=None, value_fn=lambda s: s.lxmf_address, ), ReticulumSensorDescription( key="messages_received", translation_key="messages_received", icon="mdi:inbox-arrow-down", state_class=SensorStateClass.TOTAL_INCREASING, value_fn=lambda s: s.messages_received, ), ReticulumSensorDescription( key="messages_sent", translation_key="messages_sent", icon="mdi:inbox-arrow-up", state_class=SensorStateClass.TOTAL_INCREASING, value_fn=lambda s: s.messages_sent, ), ReticulumSensorDescription( key="messages_failed", translation_key="messages_failed", icon="mdi:message-alert", state_class=SensorStateClass.TOTAL_INCREASING, value_fn=lambda s: s.messages_failed, ), ReticulumSensorDescription( key="peers", translation_key="peers", icon="mdi:account-group", state_class=SensorStateClass.MEASUREMENT, value_fn=lambda s: len(s.peers), attr_fn=lambda s: { "peers": [ { "address": p.destination_hash, "name": p.display_name, "hops": p.hops, "stamp_cost": p.stamp_cost, } for p in sorted( s.peers.values(), key=lambda x: x.last_heard, reverse=True ) ] }, ), ReticulumSensorDescription( key="last_message", translation_key="last_message", icon="mdi:message-text", value_fn=lambda s: (s.last_message or "")[:255] or None, attr_fn=lambda s: { "source": s.last_message_source, "title": s.last_message_title, "full_content": s.last_message, }, ), ReticulumSensorDescription( key="last_message_time", translation_key="last_message_time", icon="mdi:clock-outline", device_class=SensorDeviceClass.TIMESTAMP, value_fn=_last_message_time, ), # --- Interface telemetry ------------------------------------------------- ReticulumSensorDescription( key="rx_bytes", translation_key="rx_bytes", icon="mdi:download-network", device_class=SensorDeviceClass.DATA_SIZE, native_unit_of_measurement=UnitOfInformation.BYTES, state_class=SensorStateClass.TOTAL_INCREASING, suggested_unit_of_measurement=UnitOfInformation.KIBIBYTES, suggested_display_precision=1, entity_category=EntityCategory.DIAGNOSTIC, value_fn=lambda s: s.tel_rxb, ), ReticulumSensorDescription( key="tx_bytes", translation_key="tx_bytes", icon="mdi:upload-network", device_class=SensorDeviceClass.DATA_SIZE, native_unit_of_measurement=UnitOfInformation.BYTES, state_class=SensorStateClass.TOTAL_INCREASING, suggested_unit_of_measurement=UnitOfInformation.KIBIBYTES, suggested_display_precision=1, entity_category=EntityCategory.DIAGNOSTIC, value_fn=lambda s: s.tel_txb, ), ReticulumSensorDescription( key="bitrate", translation_key="bitrate", icon="mdi:speedometer", device_class=SensorDeviceClass.DATA_RATE, native_unit_of_measurement=UnitOfDataRate.BITS_PER_SECOND, state_class=SensorStateClass.MEASUREMENT, entity_category=EntityCategory.DIAGNOSTIC, value_fn=lambda s: s.tel_bitrate, ), # RSSI / SNR / quality are only meaningful on physical interfaces # (RNode/LoRa). Disabled by default; enable them if a physical link is used. ReticulumSensorDescription( key="rssi", translation_key="rssi", icon="mdi:signal", device_class=SensorDeviceClass.SIGNAL_STRENGTH, native_unit_of_measurement=SIGNAL_STRENGTH_DECIBELS_MILLIWATT, state_class=SensorStateClass.MEASUREMENT, entity_category=EntityCategory.DIAGNOSTIC, entity_registry_enabled_default=False, value_fn=lambda s: s.tel_rssi, ), ReticulumSensorDescription( key="snr", translation_key="snr", icon="mdi:signal-variant", native_unit_of_measurement="dB", state_class=SensorStateClass.MEASUREMENT, entity_category=EntityCategory.DIAGNOSTIC, entity_registry_enabled_default=False, value_fn=lambda s: s.tel_snr, ), ReticulumSensorDescription( key="link_quality", translation_key="link_quality", icon="mdi:gauge", native_unit_of_measurement=PERCENTAGE, state_class=SensorStateClass.MEASUREMENT, entity_category=EntityCategory.DIAGNOSTIC, entity_registry_enabled_default=False, value_fn=lambda s: s.tel_quality, ), ) async def async_setup_entry( hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback, ) -> None: """Set up Reticulum sensors.""" manager: ReticulumManager = hass.data[DOMAIN][entry.entry_id].manager async_add_entities( ReticulumSensor(manager, entry.entry_id, desc) for desc in SENSORS ) class ReticulumSensor(ReticulumEntity, SensorEntity): """A Reticulum sensor.""" entity_description: ReticulumSensorDescription def __init__( self, manager: ReticulumManager, entry_id: str, description: ReticulumSensorDescription, ) -> None: super().__init__(manager, entry_id) self.entity_description = description self._attr_unique_id = f"{entry_id}_{description.key}" @property def native_value(self) -> Any: return self.entity_description.value_fn(self._manager.state) @property def extra_state_attributes(self) -> dict[str, Any] | None: if self.entity_description.attr_fn is None: return None return self.entity_description.attr_fn(self._manager.state)