"""Reticulum / LXMF client manager for Home Assistant. This module isolates all interaction with the (blocking, thread-based) Reticulum Network Stack (``RNS``) and the LXMF messaging layer, and bridges their callbacks (which fire on RNS-owned threads) back onto the Home Assistant event loop. The neighbouring machine runs a full Reticulum instance exposing a ``TCPServerInterface``. We connect out to it with a ``TCPClientInterface`` and run a standalone (non-shared) RNS instance inside the Home Assistant process. """ from __future__ import annotations import asyncio import logging import os import time from collections import deque from dataclasses import dataclass, field from typing import TYPE_CHECKING, Any, Callable from homeassistant.core import HomeAssistant, callback from homeassistant.helpers.dispatcher import async_dispatcher_send from .const import ( ATTACHMENTS_SUBDIR, CONFIG_FILENAME, DEST_HASH_LEN, IDENTITY_FILENAME, SIGNAL_STATE_UPDATED, ) if TYPE_CHECKING: import LXMF import RNS _LOGGER = logging.getLogger(__name__) # A process may only ever hold a single RNS.Reticulum instance, and RNS/LXMF # cannot be cleanly torn down inside a running process. Home Assistant is a # single process, so we cache the running stack objects module-side and reuse # them across config-entry reloads instead of rebuilding (which would raise # "Attempt to reinitialise Reticulum" and/or leak duplicate LXMF routers). _RNS_INSTANCE: Any = None _LXM_ROUTER: Any = None _LOCAL_DESTINATION: Any = None _IDENTITY: Any = None _ANNOUNCE_HANDLER: Any = None # How long (seconds) to wait for a path/identity to resolve before giving up on # an outbound message. PATH_RESOLVE_TIMEOUT = 15.0 # Ring buffer size for recently seen peers / messages surfaced as attributes. MAX_TRACKED_PEERS = 200 MAX_RECENT_MESSAGES = 25 @dataclass(slots=True) class Peer: """A Reticulum LXMF peer we have heard announce.""" destination_hash: str display_name: str | None stamp_cost: int | None last_heard: float hops: int | None = None @dataclass(slots=True) class ManagerState: """Mutable state surfaced to entities and diagnostics.""" started: bool = False interface_online: bool = False lxmf_address: str | None = None display_name: str | None = None messages_received: int = 0 messages_sent: int = 0 messages_failed: int = 0 # Interface telemetry (populated by refresh_interface_status). tel_rxb: int | None = None tel_txb: int | None = None tel_bitrate: float | None = None tel_rssi: float | None = None tel_snr: float | None = None tel_quality: float | None = None last_message: str | None = None last_message_source: str | None = None last_message_title: str | None = None last_message_time: float | None = None peers: dict[str, Peer] = field(default_factory=dict) recent_messages: deque = field(default_factory=lambda: deque(maxlen=MAX_RECENT_MESSAGES)) class ReticulumError(Exception): """Raised for recoverable Reticulum operations (surfaced to the UI).""" class ReticulumManager: """Owns the RNS instance, the LXMF router and all bridging logic.""" def __init__( self, hass: HomeAssistant, storage_dir: str, *, target_host: str, target_port: int, interface_name: str, display_name: str, loglevel: int, ) -> None: self.hass = hass self.storage_dir = storage_dir self.target_host = target_host self.target_port = target_port self.interface_name = interface_name self.display_name = display_name self.loglevel = loglevel self.state = ManagerState(display_name=display_name) # Populated in _start (executor). self._rns: Any = None self._router: Any = None self._identity: Any = None self._local_destination: Any = None self._announce_handler: Any = None # Incoming-message handler installed by __init__.py once options are # known (assist bridge, allow-list, …). self.incoming_handler: Callable[[dict[str, Any]], Any] | None = None self._lock = asyncio.Lock() # ------------------------------------------------------------------ # Lifecycle # ------------------------------------------------------------------ async def async_start(self) -> None: """Initialise RNS + LXMF (runs the blocking parts in an executor).""" async with self._lock: await self.hass.async_add_executor_job(self._start) self._push_state() def _start(self) -> None: """Blocking init. Runs in the executor thread.""" global _RNS_INSTANCE, _LXM_ROUTER, _LOCAL_DESTINATION # noqa: PLW0603 global _IDENTITY, _ANNOUNCE_HANDLER # noqa: PLW0603 import LXMF # noqa: PLC0415 import RNS # noqa: PLC0415 os.makedirs(self.storage_dir, exist_ok=True) os.makedirs(os.path.join(self.storage_dir, ATTACHMENTS_SUBDIR), exist_ok=True) self._write_config_file() # RNS is a hard process-wide singleton: calling RNS.Reticulum() twice # raises "Attempt to reinitialise Reticulum, when it was already # running". Our module-level cache can get out of sync with RNS's own # internal singleton (e.g. the integration module is re-imported after # an update, or a partial-setup retry), so treat RNS itself as the # source of truth via get_instance(). existing = None try: existing = RNS.Reticulum.get_instance() except Exception: # noqa: BLE001 - older RNS may lack get_instance existing = _RNS_INSTANCE if existing is not None: _LOGGER.debug("Reusing already-running Reticulum instance") _RNS_INSTANCE = existing else: _LOGGER.debug("Initialising Reticulum instance at %s", self.storage_dir) try: _RNS_INSTANCE = RNS.Reticulum( configdir=self.storage_dir, loglevel=self.loglevel ) except OSError as err: # Lost a race (or divergent state): adopt whatever is running. adopted = None try: adopted = RNS.Reticulum.get_instance() except Exception: # noqa: BLE001 adopted = None if adopted is None: raise _LOGGER.debug("Adopted existing Reticulum instance after %s", err) _RNS_INSTANCE = adopted self._rns = _RNS_INSTANCE # Stable identity so our LXMF address survives restarts. if _IDENTITY is None: identity_path = os.path.join(self.storage_dir, IDENTITY_FILENAME) identity = None if os.path.isfile(identity_path): identity = RNS.Identity.from_file(identity_path) if identity is None: identity = RNS.Identity() identity.to_file(identity_path) _IDENTITY = identity self._identity = _IDENTITY # Reuse the LXMF router + delivery destination across reloads; only the # delivery callback (which is bound to this manager) is (re)registered. if _LXM_ROUTER is None: _LXM_ROUTER = LXMF.LXMRouter( storagepath=os.path.join(self.storage_dir, "lxmf"), ) _LOCAL_DESTINATION = _LXM_ROUTER.register_delivery_identity( self._identity, display_name=self.display_name ) self._router = _LXM_ROUTER self._local_destination = _LOCAL_DESTINATION self._router.register_delivery_callback(self._delivery_callback) # Discover peers via their LXMF delivery announces. Deregister any # previous handler (bound to a stale manager) before installing ours. if _ANNOUNCE_HANDLER is not None: try: RNS.Transport.deregister_announce_handler(_ANNOUNCE_HANDLER) except Exception: # noqa: BLE001 - best effort / older RNS _LOGGER.debug("Could not deregister old announce handler", exc_info=True) self._announce_handler = _AnnounceHandler(self) _ANNOUNCE_HANDLER = self._announce_handler RNS.Transport.register_announce_handler(self._announce_handler) self.state.lxmf_address = RNS.hexrep( self._local_destination.hash, delimit=False ) self.state.started = True _LOGGER.info( "Reticulum ready. LXMF address: %s (%s)", self.state.lxmf_address, self.display_name, ) def _write_config_file(self) -> None: """Write a minimal RNS config with only our TCP client interface.""" config_path = os.path.join(self.storage_dir, CONFIG_FILENAME) contents = ( "[reticulum]\n" " enable_transport = False\n" " share_instance = No\n" " panic_on_interface_error = No\n\n" "[logging]\n" f" loglevel = {self.loglevel}\n\n" "[interfaces]\n" f" [[{self.interface_name}]]\n" " type = TCPClientInterface\n" " enabled = yes\n" f" target_host = {self.target_host}\n" f" target_port = {self.target_port}\n" ) with open(config_path, "w", encoding="utf-8") as handle: handle.write(contents) async def async_stop(self) -> None: """Detach our callbacks. The RNS instance itself lives for the process. RNS/LXMF do not support a clean per-entry teardown inside a running process, so a full stack restart requires restarting Home Assistant. We at least stop delivering into a torn-down entry. """ self.incoming_handler = None if self._router is not None: try: self._router.register_delivery_callback(lambda _msg: None) except Exception: # noqa: BLE001 - best effort _LOGGER.debug("Could not reset delivery callback", exc_info=True) if self._announce_handler is not None: try: import RNS # noqa: PLC0415 RNS.Transport.deregister_announce_handler(self._announce_handler) except Exception: # noqa: BLE001 - best effort / older RNS _LOGGER.debug("Could not deregister announce handler", exc_info=True) self.state.started = False # ------------------------------------------------------------------ # Outbound # ------------------------------------------------------------------ async def async_send_message( self, destination: str, content: str, title: str = "", method: str = "direct", fields: dict | None = None, ) -> str: """Send an LXMF message. Returns the message hash (hex).""" dest_hash = self._normalise_hash(destination) return await self.hass.async_add_executor_job( self._send, dest_hash, content, title, method, fields ) def _send( self, dest_hash: bytes, content: str, title: str, method: str, fields: dict | None, ) -> str: import LXMF # noqa: PLC0415 import RNS # noqa: PLC0415 recipient_identity = self._resolve_identity(dest_hash) if recipient_identity is None: raise ReticulumError( f"Could not resolve a path/identity for " f"{RNS.prettyhexrep(dest_hash)} within {PATH_RESOLVE_TIMEOUT:.0f}s" ) dest = RNS.Destination( recipient_identity, RNS.Destination.OUT, RNS.Destination.SINGLE, "lxmf", "delivery", ) desired = { "direct": LXMF.LXMessage.DIRECT, "opportunistic": LXMF.LXMessage.OPPORTUNISTIC, "propagated": LXMF.LXMessage.PROPAGATED, }.get(method, LXMF.LXMessage.DIRECT) lxm = LXMF.LXMessage( dest, self._local_destination, content, title, desired_method=desired, ) if fields: lxm.fields = fields lxm.register_delivery_callback(self._outbound_delivered) lxm.register_failed_callback(self._outbound_failed) self._router.handle_outbound(lxm) _LOGGER.debug( "Queued LXMF message to %s via %s", RNS.prettyhexrep(dest_hash), method ) return RNS.hexrep(lxm.hash, delimit=False) def _resolve_identity(self, dest_hash: bytes) -> Any: """Recall (and if needed, request a path for) a destination identity.""" import RNS # noqa: PLC0415 identity = RNS.Identity.recall(dest_hash) if identity is not None: return identity if not RNS.Transport.has_path(dest_hash): RNS.Transport.request_path(dest_hash) deadline = time.monotonic() + PATH_RESOLVE_TIMEOUT while time.monotonic() < deadline: identity = RNS.Identity.recall(dest_hash) if identity is not None: return identity time.sleep(0.25) return RNS.Identity.recall(dest_hash) async def async_announce(self) -> None: """Announce our LXMF delivery destination on the network.""" if self._router is None or self._local_destination is None: raise ReticulumError("Reticulum is not started") await self.hass.async_add_executor_job( self._router.announce, self._local_destination.hash ) _LOGGER.debug("Announced LXMF destination %s", self.state.lxmf_address) async def async_request_path(self, destination: str) -> None: """Request a network path to a destination.""" dest_hash = self._normalise_hash(destination) def _request() -> None: import RNS # noqa: PLC0415 RNS.Transport.request_path(dest_hash) await self.hass.async_add_executor_job(_request) async def async_set_propagation_node(self, destination: str) -> None: """Set the outbound LXMF propagation (store-and-forward) node.""" dest_hash = self._normalise_hash(destination) await self.hass.async_add_executor_job( self._router.set_outbound_propagation_node, dest_hash ) async def async_sync_propagation(self, max_messages: int | None = None) -> None: """Pull any queued messages from the configured propagation node.""" def _sync() -> None: if max_messages is None: self._router.request_messages_from_propagation_node(self._identity) else: self._router.request_messages_from_propagation_node( self._identity, max_messages ) await self.hass.async_add_executor_job(_sync) # ------------------------------------------------------------------ # Inbound (RNS thread -> HA loop bridging) # ------------------------------------------------------------------ def _delivery_callback(self, message: Any) -> None: """LXMF delivery callback. Runs on an RNS-owned thread.""" import RNS # noqa: PLC0415 try: source_hash = RNS.hexrep(message.source_hash, delimit=False) payload: dict[str, Any] = { "content": message.content_as_string() or "", "title": message.title_as_string() or "", "source": source_hash, "destination": RNS.hexrep(message.destination_hash, delimit=False), "timestamp": getattr(message, "timestamp", None), "signature_validated": getattr(message, "signature_validated", None), "fields": dict(getattr(message, "fields", {}) or {}), } except Exception: # noqa: BLE001 - never let a bad message kill the thread _LOGGER.exception("Failed to parse inbound LXMF message") return # Hop back onto the event loop thread. self.hass.loop.call_soon_threadsafe(self._dispatch_incoming, payload) @callback def _dispatch_incoming(self, payload: dict[str, Any]) -> None: """Runs on the event loop: update state, hand off to handler.""" self.state.messages_received += 1 self.state.last_message = payload["content"] self.state.last_message_source = payload["source"] self.state.last_message_title = payload["title"] self.state.last_message_time = payload.get("timestamp") or time.time() self.state.recent_messages.appendleft( { "source": payload["source"], "title": payload["title"], "content": payload["content"], "time": self.state.last_message_time, } ) self._push_state() if self.incoming_handler is not None: self.hass.async_create_task( _maybe_await(self.incoming_handler, payload), name="reticulum_incoming", ) # ------------------------------------------------------------------ # Outbound delivery status (RNS thread) # ------------------------------------------------------------------ def _outbound_delivered(self, message: Any) -> None: self.hass.loop.call_soon_threadsafe(self._on_outbound_delivered, message) def _outbound_failed(self, message: Any) -> None: self.hass.loop.call_soon_threadsafe(self._on_outbound_failed, message) @callback def _on_outbound_delivered(self, message: Any) -> None: import RNS # noqa: PLC0415 self.state.messages_sent += 1 self._push_state() _fire_event( self.hass, "delivered", { "message_hash": RNS.hexrep(message.hash, delimit=False), "destination": RNS.hexrep(message.destination_hash, delimit=False), }, ) @callback def _on_outbound_failed(self, message: Any) -> None: import RNS # noqa: PLC0415 self.state.messages_failed += 1 self._push_state() _fire_event( self.hass, "failed", { "message_hash": RNS.hexrep(message.hash, delimit=False), "destination": RNS.hexrep(message.destination_hash, delimit=False), }, ) # ------------------------------------------------------------------ # Peer tracking (called from announce handler on RNS thread) # ------------------------------------------------------------------ def _register_peer( self, destination_hash: str, display_name: str | None, stamp_cost: int | None, hops: int | None, ) -> None: self.hass.loop.call_soon_threadsafe( self._store_peer, destination_hash, display_name, stamp_cost, hops ) @callback def _store_peer( self, destination_hash: str, display_name: str | None, stamp_cost: int | None, hops: int | None, ) -> None: peer = Peer( destination_hash=destination_hash, display_name=display_name, stamp_cost=stamp_cost, last_heard=time.time(), hops=hops, ) self.state.peers[destination_hash] = peer # Keep the peer table bounded. if len(self.state.peers) > MAX_TRACKED_PEERS: oldest = min(self.state.peers.values(), key=lambda p: p.last_heard) self.state.peers.pop(oldest.destination_hash, None) self._push_state() _fire_event( self.hass, "announce", { "destination": destination_hash, "display_name": display_name, "stamp_cost": stamp_cost, "hops": hops, }, ) # ------------------------------------------------------------------ # Helpers # ------------------------------------------------------------------ def refresh_interface_status(self) -> None: """Recompute interface online status and collect telemetry. Called on the event loop from a periodic timer. RX/TX byte counters and bitrate are available for the TCP client interface; RSSI/SNR/quality are only populated when the underlying interface is a physical one (e.g. an RNode/LoRa interface) and stay ``None`` for a plain TCP link. """ if self._rns is None: return online = False rxb = txb = bitrate = rssi = snr = quality = None try: import RNS # noqa: PLC0415 target = None interfaces = list(RNS.Transport.interfaces) for iface in interfaces: name = getattr(iface, "name", "") or str(iface) if self.interface_name in name: target = iface break if target is None and interfaces: target = interfaces[0] if target is not None: online = bool(getattr(target, "online", False)) rxb = getattr(target, "rxb", None) txb = getattr(target, "txb", None) bitrate = getattr(target, "bitrate", None) rssi = getattr(target, "rssi", None) snr = getattr(target, "snr", None) quality = getattr(target, "q", None) except Exception: # noqa: BLE001 online = False snapshot = (online, rxb, txb, bitrate, rssi, snr, quality) current = ( self.state.interface_online, self.state.tel_rxb, self.state.tel_txb, self.state.tel_bitrate, self.state.tel_rssi, self.state.tel_snr, self.state.tel_quality, ) if snapshot == current: return ( self.state.interface_online, self.state.tel_rxb, self.state.tel_txb, self.state.tel_bitrate, self.state.tel_rssi, self.state.tel_snr, self.state.tel_quality, ) = snapshot self._push_state() @callback def _push_state(self) -> None: async_dispatcher_send(self.hass, SIGNAL_STATE_UPDATED) @staticmethod def _normalise_hash(value: str) -> bytes: cleaned = value.strip().lower().replace(":", "").replace(" ", "") if cleaned.startswith("0x"): cleaned = cleaned[2:] if len(cleaned) != DEST_HASH_LEN: raise ReticulumError( f"Invalid Reticulum address '{value}': expected {DEST_HASH_LEN} " f"hex characters (a 16-byte destination hash)" ) try: return bytes.fromhex(cleaned) except ValueError as err: raise ReticulumError(f"Invalid Reticulum address '{value}'") from err class _AnnounceHandler: """RNS announce handler filtered to the LXMF delivery aspect.""" aspect_filter = "lxmf.delivery" def __init__(self, manager: ReticulumManager) -> None: self._manager = manager # LXMF requests receipt of the full announce data path. self.receive_path_responses = False def received_announce( self, destination_hash: bytes, announced_identity: Any, app_data: Any, ) -> None: # NOTE: RNS dispatches this callback by *exact* parameter count and # passes arguments by keyword. The 3-parameter form (destination_hash, # announced_identity, app_data) is supported by every RNS version; the # optional announce_packet_hash / is_path_response are unused here. import RNS # noqa: PLC0415 display_name: str | None = None stamp_cost: int | None = None try: import LXMF # noqa: PLC0415 if app_data is not None: display_name = LXMF.display_name_from_app_data(app_data) stamp_cost = LXMF.stamp_cost_from_app_data(app_data) except Exception: # noqa: BLE001 - app_data helpers vary by version if isinstance(app_data, (bytes, bytearray)): try: display_name = app_data.decode("utf-8", errors="replace") except Exception: # noqa: BLE001 display_name = None hops = None try: hops = RNS.Transport.hops_to(destination_hash) except Exception: # noqa: BLE001 hops = None self._manager._register_peer( RNS.hexrep(destination_hash, delimit=False), display_name, stamp_cost, hops, ) def _fire_event(hass: HomeAssistant, kind: str, data: dict[str, Any]) -> None: """Fire one of the reticulum_* bus events (imported lazily to avoid cycle).""" from .const import ( # noqa: PLC0415 EVENT_ANNOUNCE_RECEIVED, EVENT_MESSAGE_DELIVERED, EVENT_MESSAGE_FAILED, ) mapping = { "delivered": EVENT_MESSAGE_DELIVERED, "failed": EVENT_MESSAGE_FAILED, "announce": EVENT_ANNOUNCE_RECEIVED, } hass.bus.async_fire(mapping[kind], data) async def _maybe_await(func: Callable[[dict[str, Any]], Any], payload: dict[str, Any]) -> None: """Call ``func`` supporting both coroutine and plain callables.""" result = func(payload) if asyncio.iscoroutine(result): await result