027ff685b7
Refactor from one-entry-one-identity to a hub config entry (shared RNS stack) plus config subentries of type "identity", each an independent LXMF identity with its own LXMRouter, address, device and entities. This is required because LXMRouter allows only one delivery identity per instance. - ReticulumStack: owns the single process-wide RNS instance, peer discovery and interface telemetry (hub device). - IdentityManager: one per subentry, its own LXMRouter/identity/destination and delivery callback (identity device). Signal handlers suppressed and atexit handlers tamed for each router too. - Config subentry flow (add / reconfigure identity); a first identity is seeded on stack creation. Editable announce display name = subentry title. - Per-identity "Regenerate identity" button: new address + re-announce (clears the router's single delivery destination first). - Services and events gain an "identity" field to target/distinguish identities. - Per-identity assist bridge, greeting-when-assist-off, allow-list, notify, announce/sync buttons, message counters; hub-level connectivity/telemetry/peers. Verified against current HA config_entries/entity_platform/selector and RNS/LXMF sources. Docs and en/ru translations updated. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
803 lines
27 KiB
Python
803 lines
27 KiB
Python
"""Reticulum / LXMF client for Home Assistant.
|
|
|
|
Architecture
|
|
------------
|
|
Reticulum (``RNS``) is a process-wide singleton, so there is exactly one
|
|
:class:`ReticulumStack` (the config entry / hub) that owns the RNS instance, the
|
|
outbound TCP interface, peer discovery and interface telemetry.
|
|
|
|
LXMF's ``LXMRouter`` supports only a single delivery identity per instance, so
|
|
each LXMF identity (a config subentry) gets its own :class:`IdentityManager`
|
|
with its own ``LXMRouter``, delivery destination and delivery callback. All of
|
|
them share the single RNS instance.
|
|
|
|
Both ``RNS.Reticulum()`` and ``LXMF.LXMRouter()`` call ``signal.signal()`` in
|
|
their constructors (main-thread only) and register blocking ``atexit`` handlers,
|
|
so construction happens in an executor thread with signals neutralised and the
|
|
atexit handlers untamed; state is persisted off-loop on ``homeassistant_stop``.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import atexit
|
|
import logging
|
|
import os
|
|
import signal
|
|
import threading
|
|
import time
|
|
from collections import deque
|
|
from contextlib import contextmanager
|
|
from dataclasses import dataclass, field
|
|
from typing import TYPE_CHECKING, Any, Callable, Iterator
|
|
|
|
from homeassistant.core import HomeAssistant, callback
|
|
from homeassistant.helpers.dispatcher import async_dispatcher_send
|
|
|
|
from .const import (
|
|
ATTACHMENTS_SUBDIR,
|
|
CONFIG_FILENAME,
|
|
DEST_HASH_LEN,
|
|
IDENTITIES_SUBDIR,
|
|
IDENTITY_FILENAME,
|
|
identity_signal,
|
|
stack_signal,
|
|
)
|
|
|
|
if TYPE_CHECKING:
|
|
import LXMF
|
|
import RNS
|
|
|
|
_LOGGER = logging.getLogger(__name__)
|
|
|
|
# Process-wide singletons / caches (survive config-entry reloads).
|
|
_RNS_INSTANCE: Any = None
|
|
_ANNOUNCE_HANDLER: Any = None
|
|
# Per-identity caches keyed by subentry_id, so a reload reuses the running
|
|
# LXMRouter/identity instead of spawning duplicates.
|
|
_ROUTERS: dict[str, Any] = {}
|
|
_IDENTITIES: dict[str, Any] = {}
|
|
_DESTINATIONS: dict[str, Any] = {}
|
|
|
|
PATH_RESOLVE_TIMEOUT = 15.0
|
|
MAX_TRACKED_PEERS = 200
|
|
MAX_RECENT_MESSAGES = 25
|
|
|
|
|
|
def _noop_signal(*_args: Any, **_kwargs: Any) -> None:
|
|
"""Drop-in for ``signal.signal`` used while initialising off the main thread."""
|
|
return None
|
|
|
|
|
|
@contextmanager
|
|
def _suppressed_signal_handlers() -> Iterator[None]:
|
|
"""Neutralise ``signal.signal`` when not on the main thread.
|
|
|
|
RNS and LXMF register SIGINT/SIGTERM handlers in their constructors, which
|
|
only works on the main thread (we init in an executor). Suppressing this
|
|
also keeps Home Assistant in control of its own signals.
|
|
"""
|
|
patched = threading.current_thread() is not threading.main_thread()
|
|
saved = signal.signal
|
|
if patched:
|
|
signal.signal = _noop_signal # type: ignore[assignment]
|
|
try:
|
|
yield
|
|
finally:
|
|
if patched:
|
|
signal.signal = saved # type: ignore[assignment]
|
|
|
|
|
|
def _tame_exit_handler(obj: Any) -> None:
|
|
"""Unregister an RNS/LXMF ``atexit`` handler (blocking I/O / stdout detach)."""
|
|
handler = getattr(obj, "exit_handler", None)
|
|
if handler is None:
|
|
return
|
|
try:
|
|
atexit.unregister(handler)
|
|
except Exception: # noqa: BLE001 - best effort
|
|
_LOGGER.debug("Could not unregister exit handler for %s", obj)
|
|
|
|
|
|
class ReticulumError(Exception):
|
|
"""Recoverable Reticulum error surfaced to the UI."""
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# State containers
|
|
# ---------------------------------------------------------------------------
|
|
@dataclass(slots=True)
|
|
class Peer:
|
|
"""An LXMF peer heard announcing."""
|
|
|
|
destination_hash: str
|
|
display_name: str | None
|
|
stamp_cost: int | None
|
|
last_heard: float
|
|
hops: int | None = None
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class StackState:
|
|
"""Hub-level state (shared RNS instance)."""
|
|
|
|
started: bool = False
|
|
interface_online: bool = False
|
|
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
|
|
peers: dict[str, Peer] = field(default_factory=dict)
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class IdentityState:
|
|
"""Per-identity state."""
|
|
|
|
started: bool = False
|
|
lxmf_address: str | None = None
|
|
display_name: str | None = None
|
|
messages_received: int = 0
|
|
messages_sent: int = 0
|
|
messages_failed: int = 0
|
|
last_message: str | None = None
|
|
last_message_source: str | None = None
|
|
last_message_title: str | None = None
|
|
last_message_time: float | None = None
|
|
recent_messages: deque = field(default_factory=lambda: deque(maxlen=MAX_RECENT_MESSAGES))
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Stack (hub) — one shared RNS instance
|
|
# ---------------------------------------------------------------------------
|
|
class ReticulumStack:
|
|
"""Owns the process-wide RNS instance, peer discovery and telemetry."""
|
|
|
|
def __init__(
|
|
self,
|
|
hass: HomeAssistant,
|
|
storage_dir: str,
|
|
entry_id: str,
|
|
*,
|
|
target_host: str,
|
|
target_port: int,
|
|
interface_name: str,
|
|
loglevel: int,
|
|
) -> None:
|
|
self.hass = hass
|
|
self.storage_dir = storage_dir
|
|
self.entry_id = entry_id
|
|
self.target_host = target_host
|
|
self.target_port = target_port
|
|
self.interface_name = interface_name
|
|
self.loglevel = loglevel
|
|
self.state = StackState()
|
|
self._rns: Any = None
|
|
|
|
async def async_start(self) -> None:
|
|
await self.hass.async_add_executor_job(self._start)
|
|
self._push_state()
|
|
|
|
def _start(self) -> None:
|
|
global _RNS_INSTANCE, _ANNOUNCE_HANDLER # noqa: PLW0603
|
|
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()
|
|
|
|
existing = None
|
|
try:
|
|
existing = RNS.Reticulum.get_instance()
|
|
except Exception: # noqa: BLE001
|
|
existing = _RNS_INSTANCE
|
|
|
|
if existing is not None:
|
|
_RNS_INSTANCE = existing
|
|
else:
|
|
with _suppressed_signal_handlers():
|
|
try:
|
|
_RNS_INSTANCE = RNS.Reticulum(
|
|
configdir=self.storage_dir, loglevel=self.loglevel
|
|
)
|
|
except OSError as err:
|
|
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: %s", err)
|
|
_RNS_INSTANCE = adopted
|
|
self._rns = _RNS_INSTANCE
|
|
_tame_exit_handler(RNS.Reticulum)
|
|
|
|
# (Re)install the peer-discovery announce handler bound to this stack.
|
|
if _ANNOUNCE_HANDLER is not None:
|
|
try:
|
|
RNS.Transport.deregister_announce_handler(_ANNOUNCE_HANDLER)
|
|
except Exception: # noqa: BLE001
|
|
_LOGGER.debug("Could not deregister old announce handler", exc_info=True)
|
|
_ANNOUNCE_HANDLER = _AnnounceHandler(self)
|
|
RNS.Transport.register_announce_handler(_ANNOUNCE_HANDLER)
|
|
|
|
self.state.started = True
|
|
_LOGGER.info(
|
|
"Reticulum stack ready (%s:%s via %s)",
|
|
self.target_host,
|
|
self.target_port,
|
|
self.interface_name,
|
|
)
|
|
|
|
def _write_config_file(self) -> None:
|
|
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:
|
|
self.state.started = False
|
|
|
|
async def async_persist(self) -> None:
|
|
if self._rns is None:
|
|
return
|
|
await self.hass.async_add_executor_job(self._persist)
|
|
|
|
def _persist(self) -> None:
|
|
import RNS # noqa: PLC0415
|
|
|
|
for label, fn in (
|
|
("transport", getattr(RNS.Transport, "persist_data", None)),
|
|
("identity", getattr(RNS.Identity, "persist_data", None)),
|
|
):
|
|
if fn is None:
|
|
continue
|
|
try:
|
|
fn()
|
|
except Exception: # noqa: BLE001
|
|
_LOGGER.debug("Failed to persist RNS %s data", label, exc_info=True)
|
|
|
|
@callback
|
|
def refresh_interface_status(self, now: Any = None) -> None:
|
|
"""Recompute interface online status + telemetry (on the event loop)."""
|
|
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()
|
|
|
|
# Peer tracking (announce handler runs on an 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:
|
|
self.state.peers[destination_hash] = Peer(
|
|
destination_hash=destination_hash,
|
|
display_name=display_name,
|
|
stamp_cost=stamp_cost,
|
|
last_heard=time.time(),
|
|
hops=hops,
|
|
)
|
|
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_announce_event(
|
|
self.hass, destination_hash, display_name, stamp_cost, hops
|
|
)
|
|
|
|
@callback
|
|
def _push_state(self) -> None:
|
|
async_dispatcher_send(self.hass, stack_signal(self.entry_id))
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Identity — one LXMRouter per identity
|
|
# ---------------------------------------------------------------------------
|
|
class IdentityManager:
|
|
"""Owns one LXMF identity, its router and delivery destination."""
|
|
|
|
def __init__(
|
|
self,
|
|
hass: HomeAssistant,
|
|
stack: ReticulumStack,
|
|
subentry_id: str,
|
|
*,
|
|
display_name: str,
|
|
) -> None:
|
|
self.hass = hass
|
|
self.stack = stack
|
|
self.subentry_id = subentry_id
|
|
self.display_name = display_name
|
|
self.state = IdentityState(display_name=display_name)
|
|
self._router: Any = None
|
|
self._identity: Any = None
|
|
self._local_destination: Any = None
|
|
self.incoming_handler: Callable[[dict[str, Any]], Any] | None = None
|
|
|
|
@property
|
|
def _dir(self) -> str:
|
|
return os.path.join(
|
|
self.stack.storage_dir, IDENTITIES_SUBDIR, self.subentry_id
|
|
)
|
|
|
|
async def async_start(self) -> None:
|
|
await self.hass.async_add_executor_job(self._start)
|
|
self._push_state()
|
|
|
|
def _start(self) -> None:
|
|
import LXMF # noqa: PLC0415
|
|
import RNS # noqa: PLC0415
|
|
|
|
os.makedirs(self._dir, exist_ok=True)
|
|
sid = self.subentry_id
|
|
|
|
if sid not in _ROUTERS:
|
|
identity_path = os.path.join(self._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)
|
|
with _suppressed_signal_handlers():
|
|
router = LXMF.LXMRouter(storagepath=os.path.join(self._dir, "lxmf"))
|
|
_tame_exit_handler(router)
|
|
destination = router.register_delivery_identity(
|
|
identity, display_name=self.display_name
|
|
)
|
|
_ROUTERS[sid] = router
|
|
_IDENTITIES[sid] = identity
|
|
_DESTINATIONS[sid] = destination
|
|
else:
|
|
router = _ROUTERS[sid]
|
|
identity = _IDENTITIES[sid]
|
|
destination = _DESTINATIONS[sid]
|
|
# Apply a possibly-changed announce display name.
|
|
try:
|
|
destination.display_name = self.display_name
|
|
except Exception: # noqa: BLE001
|
|
pass
|
|
|
|
router.register_delivery_callback(self._delivery_callback)
|
|
self._router = router
|
|
self._identity = identity
|
|
self._local_destination = destination
|
|
self.state.lxmf_address = RNS.hexrep(destination.hash, delimit=False)
|
|
self.state.display_name = self.display_name
|
|
self.state.started = True
|
|
_LOGGER.info(
|
|
"Reticulum identity '%s' ready: %s",
|
|
self.display_name,
|
|
self.state.lxmf_address,
|
|
)
|
|
|
|
async def async_stop(self) -> None:
|
|
self.incoming_handler = None
|
|
if self._router is not None:
|
|
try:
|
|
self._router.register_delivery_callback(lambda _msg: None)
|
|
except Exception: # noqa: BLE001
|
|
_LOGGER.debug("Could not reset delivery callback", 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:
|
|
dest_hash = _normalise_hash(destination)
|
|
message_hash = await self.hass.async_add_executor_job(
|
|
self._send, dest_hash, content, title, method, fields
|
|
)
|
|
self.state.messages_sent += 1
|
|
self._push_state()
|
|
return message_hash
|
|
|
|
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)
|
|
return RNS.hexrep(lxm.hash, delimit=False)
|
|
|
|
def _resolve_identity(self, dest_hash: bytes) -> Any:
|
|
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:
|
|
if self._router is None or self._local_destination is None:
|
|
raise ReticulumError("Identity is not started")
|
|
await self.hass.async_add_executor_job(
|
|
self._router.announce, self._local_destination.hash
|
|
)
|
|
|
|
async def async_request_path(self, destination: str) -> None:
|
|
dest_hash = _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:
|
|
dest_hash = _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:
|
|
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)
|
|
|
|
async def async_regenerate_identity(self) -> str:
|
|
"""Create a brand-new identity/address for this manager."""
|
|
address = await self.hass.async_add_executor_job(self._regenerate)
|
|
self._push_state()
|
|
return address
|
|
|
|
def _regenerate(self) -> str:
|
|
import RNS # noqa: PLC0415
|
|
|
|
os.makedirs(self._dir, exist_ok=True)
|
|
new_identity = RNS.Identity()
|
|
new_identity.to_file(os.path.join(self._dir, IDENTITY_FILENAME))
|
|
|
|
router = self._router
|
|
# LXMRouter allows only one delivery identity; clear the old one first.
|
|
try:
|
|
router.delivery_destinations.clear()
|
|
except Exception: # noqa: BLE001
|
|
_LOGGER.debug("Could not clear old delivery destination", exc_info=True)
|
|
destination = router.register_delivery_identity(
|
|
new_identity, display_name=self.display_name
|
|
)
|
|
_IDENTITIES[self.subentry_id] = new_identity
|
|
_DESTINATIONS[self.subentry_id] = destination
|
|
self._identity = new_identity
|
|
self._local_destination = destination
|
|
self.state.lxmf_address = RNS.hexrep(destination.hash, delimit=False)
|
|
try:
|
|
router.announce(destination.hash)
|
|
except Exception: # noqa: BLE001
|
|
_LOGGER.debug("Announce after regenerate failed", exc_info=True)
|
|
_LOGGER.info(
|
|
"Regenerated identity '%s': new address %s",
|
|
self.display_name,
|
|
self.state.lxmf_address,
|
|
)
|
|
return self.state.lxmf_address
|
|
|
|
# Inbound -----------------------------------------------------------
|
|
def _delivery_callback(self, message: Any) -> None:
|
|
"""LXMF delivery callback (runs on an RNS thread)."""
|
|
import RNS # noqa: PLC0415
|
|
|
|
try:
|
|
payload: dict[str, Any] = {
|
|
"content": message.content_as_string() or "",
|
|
"title": message.title_as_string() or "",
|
|
"source": RNS.hexrep(message.source_hash, delimit=False),
|
|
"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 {}),
|
|
"identity": self.subentry_id,
|
|
"identity_name": self.display_name,
|
|
"local_address": self.state.lxmf_address,
|
|
}
|
|
except Exception: # noqa: BLE001
|
|
_LOGGER.exception("Failed to parse inbound LXMF message")
|
|
return
|
|
self.hass.loop.call_soon_threadsafe(self._dispatch_incoming, payload)
|
|
|
|
@callback
|
|
def _dispatch_incoming(self, payload: dict[str, Any]) -> None:
|
|
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 -----------------------------------------
|
|
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
|
|
|
|
_fire_message_event(
|
|
self.hass,
|
|
"delivered",
|
|
self,
|
|
{
|
|
"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_message_event(
|
|
self.hass,
|
|
"failed",
|
|
self,
|
|
{
|
|
"message_hash": RNS.hexrep(message.hash, delimit=False),
|
|
"destination": RNS.hexrep(message.destination_hash, delimit=False),
|
|
},
|
|
)
|
|
|
|
@callback
|
|
def _push_state(self) -> None:
|
|
async_dispatcher_send(self.hass, identity_signal(self.subentry_id))
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Announce handler (peer discovery)
|
|
# ---------------------------------------------------------------------------
|
|
class _AnnounceHandler:
|
|
"""RNS announce handler filtered to the LXMF delivery aspect."""
|
|
|
|
aspect_filter = "lxmf.delivery"
|
|
|
|
def __init__(self, stack: ReticulumStack) -> None:
|
|
self._stack = stack
|
|
self.receive_path_responses = False
|
|
|
|
def received_announce(
|
|
self,
|
|
destination_hash: bytes,
|
|
announced_identity: Any,
|
|
app_data: Any,
|
|
) -> None:
|
|
# RNS dispatches by exact parameter count with keyword args; the
|
|
# 3-parameter form is supported by every RNS version.
|
|
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
|
|
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._stack.register_peer(
|
|
RNS.hexrep(destination_hash, delimit=False),
|
|
display_name,
|
|
stamp_cost,
|
|
hops,
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helpers
|
|
# ---------------------------------------------------------------------------
|
|
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} hex "
|
|
f"characters (a 16-byte destination hash)"
|
|
)
|
|
try:
|
|
return bytes.fromhex(cleaned)
|
|
except ValueError as err:
|
|
raise ReticulumError(f"Invalid Reticulum address '{value}'") from err
|
|
|
|
|
|
def _fire_message_event(
|
|
hass: HomeAssistant, kind: str, manager: IdentityManager, data: dict[str, Any]
|
|
) -> None:
|
|
from .const import EVENT_MESSAGE_DELIVERED, EVENT_MESSAGE_FAILED # noqa: PLC0415
|
|
|
|
event = {"delivered": EVENT_MESSAGE_DELIVERED, "failed": EVENT_MESSAGE_FAILED}[kind]
|
|
hass.bus.async_fire(
|
|
event,
|
|
{
|
|
**data,
|
|
"identity": manager.subentry_id,
|
|
"identity_name": manager.display_name,
|
|
"local_address": manager.state.lxmf_address,
|
|
},
|
|
)
|
|
|
|
|
|
def _fire_announce_event(
|
|
hass: HomeAssistant,
|
|
destination_hash: str,
|
|
display_name: str | None,
|
|
stamp_cost: int | None,
|
|
hops: int | None,
|
|
) -> None:
|
|
from .const import EVENT_ANNOUNCE_RECEIVED # noqa: PLC0415
|
|
|
|
hass.bus.async_fire(
|
|
EVENT_ANNOUNCE_RECEIVED,
|
|
{
|
|
"destination": destination_hash,
|
|
"display_name": display_name,
|
|
"stamp_cost": stamp_cost,
|
|
"hops": hops,
|
|
},
|
|
)
|
|
|
|
|
|
async def _maybe_await(
|
|
func: Callable[[dict[str, Any]], Any], payload: dict[str, Any]
|
|
) -> None:
|
|
import asyncio # noqa: PLC0415
|
|
|
|
result = func(payload)
|
|
if asyncio.iscoroutine(result):
|
|
await result
|