ha-reticulum/custom_components/reticulum/sensor.py

"""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)