Add Reticulum/LXMF integration for Home Assistant
Custom integration connecting Home Assistant to a Reticulum network over a TCPClientInterface and exchanging LXMF messages, with an Assist conversation bridge, notify entity, sensors (incl. interface telemetry), buttons, services, bus events and device triggers. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
claude
219
custom_components/reticulum/sensor.py
Normal file
219
custom_components/reticulum/sensor.py
Normal file
@@ -0,0 +1,219 @@
|
||||
"""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)
|
||||
Reference in New Issue
Block a user