StalkerSVR/ha-reticulum
Public Access
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Multiple identities via config subentries + regenerate + editable name

Browse Source 
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>
This commit is contained in:
claude
2026-07-22 20:29:05 +03:00
parent c3b98043f7
commit 027ff685b7
16 changed files with 1572 additions and 1069 deletions
Download Patch File Download Diff File Expand all files Collapse all files

244
custom_components/reticulum/__init__.py
View File

@@ -1,4 +1,8 @@
"""The Reticulum integration."""
"""The Reticulum integration.
One config entry = the shared Reticulum stack (RNS + TCP interface).
Each config subentry (type ``identity``) = an independent LXMF identity.
"""
from __future__ import annotations
@@ -8,7 +12,7 @@ from typing import Any
import voluptuous as vol
from homeassistant.config_entries import ConfigEntry
from homeassistant.config_entries import ConfigEntry, ConfigSubentry
from homeassistant.const import EVENT_HOMEASSISTANT_STOP, Platform
from homeassistant.core import (
HomeAssistant,
@@ -24,6 +28,7 @@ from .const import (
ATTR_CONTENT,
ATTR_DESTINATION,
ATTR_FIELDS,
ATTR_IDENTITY,
ATTR_MAX_MESSAGES,
ATTR_METHOD,
ATTR_TITLE,
@@ -55,8 +60,13 @@ from .const import (
SERVICE_SET_PROPAGATION_NODE,
SERVICE_SYNC_PROPAGATION,
STORAGE_SUBDIR,
SUBENTRY_TYPE_IDENTITY,
)
from .reticulum_client import (
IdentityManager,
ReticulumError,
ReticulumStack,
)
from .reticulum_client import ReticulumError, ReticulumManager
_LOGGER = logging.getLogger(__name__)
@@ -67,86 +77,99 @@ PLATFORMS: list[Platform] = [
Platform.SENSOR,
]
# Interface-status poll: RNS has no push for link up/down, so poll cheaply.
INTERFACE_POLL = timedelta(seconds=15)
class RuntimeData:
"""Container stored on the config entry."""
"""Runtime objects for the hub entry."""
def __init__(self, manager: ReticulumManager) -> None:
self.manager = manager
def __init__(self, stack: ReticulumStack) -> None:
self.stack = stack
self.identities: dict[str, IdentityManager] = {}
self.unsubs: list[Any] = []
# Maps a peer address -> the HA conversation_id we opened for it, so
# each remote user keeps its own Assist conversation context.
# f"{subentry_id}:{source}" -> HA conversation_id
self.conversation_ids: dict[str, str] = {}
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Set up Reticulum from a config entry."""
options = entry.options
"""Set up the Reticulum stack and its identities."""
storage_dir = hass.config.path(STORAGE_SUBDIR)
manager = ReticulumManager(
stack = ReticulumStack(
hass,
storage_dir,
entry.entry_id,
target_host=entry.data[CONF_TARGET_HOST],
target_port=entry.data[CONF_TARGET_PORT],
interface_name=entry.data.get(CONF_INTERFACE_NAME, DEFAULT_INTERFACE_NAME),
display_name=entry.data.get(CONF_DISPLAY_NAME, DEFAULT_DISPLAY_NAME),
loglevel=int(options.get(CONF_LOGLEVEL, DEFAULT_LOGLEVEL)),
loglevel=int(entry.options.get(CONF_LOGLEVEL, DEFAULT_LOGLEVEL)),
)
try:
await manager.async_start()
await stack.async_start()
except Exception as err: # noqa: BLE001
raise ConfigEntryNotReady(f"Could not start Reticulum: {err}") from err
runtime = RuntimeData(manager)
runtime = RuntimeData(stack)
hass.data.setdefault(DOMAIN, {})[entry.entry_id] = runtime
# Wire the incoming-message pipeline (assist bridge + event).
manager.incoming_handler = _make_incoming_handler(hass, entry, runtime)
# Start each identity subentry.
for subentry in entry.subentries.values():
if subentry.subentry_type != SUBENTRY_TYPE_IDENTITY:
continue
display_name = subentry.data.get(CONF_DISPLAY_NAME) or subentry.title
manager = IdentityManager(
hass, stack, subentry.subentry_id, display_name=display_name
)
try:
await manager.async_start()
except Exception as err: # noqa: BLE001
raise ConfigEntryNotReady(
f"Could not start identity '{display_name}': {err}"
) from err
manager.incoming_handler = _make_incoming_handler(
hass, entry, subentry, runtime, manager
)
runtime.identities[subentry.subentry_id] = manager
await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
# Announce on startup so peers learn our address, then periodically.
await _safe_announce(manager)
_schedule_periodic(hass, entry, runtime)
# Per-identity announce + periodic schedules.
for subentry in entry.subentries.values():
if subentry.subentry_type != SUBENTRY_TYPE_IDENTITY:
continue
manager = runtime.identities.get(subentry.subentry_id)
if manager is None:
continue
await _safe_announce(manager)
_schedule_identity(hass, runtime, manager, subentry)
# Poll interface status. refresh_interface_status is a @callback, so pass it
# directly (a lambda would be a plain sync job and get run in an executor
# thread, where its async_dispatcher_send call is not thread-safe).
# Hub interface-status poll.
runtime.unsubs.append(
async_track_time_interval(
hass, manager.refresh_interface_status, INTERFACE_POLL
hass, stack.refresh_interface_status, INTERFACE_POLL
)
)
# Persist RNS/LXMF state off-loop when Home Assistant stops (we unregister
# RNS's own atexit handlers, which would otherwise do this with blocking I/O
# on the event loop and detach HA's stdout).
async def _on_ha_stop(_event) -> None:
await manager.async_persist()
await stack.async_persist()
runtime.unsubs.append(
hass.bus.async_listen_once(EVENT_HOMEASSISTANT_STOP, _on_ha_stop)
)
entry.async_on_unload(entry.add_update_listener(_async_options_updated))
entry.async_on_unload(entry.add_update_listener(_async_entry_updated))
_async_register_services(hass)
return True
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Unload a config entry."""
"""Unload the hub entry."""
unload_ok = await hass.config_entries.async_unload_platforms(entry, PLATFORMS)
runtime: RuntimeData | None = hass.data.get(DOMAIN, {}).get(entry.entry_id)
if runtime is not None:
for unsub in runtime.unsubs:
unsub()
await runtime.manager.async_stop()
for manager in runtime.identities.values():
await manager.async_stop()
await runtime.stack.async_stop()
if unload_ok:
hass.data[DOMAIN].pop(entry.entry_id, None)
if not hass.data[DOMAIN]:
@@ -154,25 +177,26 @@ async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
return unload_ok
async def _async_options_updated(hass: HomeAssistant, entry: ConfigEntry) -> None:
"""Reload the entry so option changes take effect."""
async def _async_entry_updated(hass: HomeAssistant, entry: ConfigEntry) -> None:
"""Reload when options or subentries change."""
await hass.config_entries.async_reload(entry.entry_id)
# ---------------------------------------------------------------------------
# Periodic tasks
# Periodic scheduling
# ---------------------------------------------------------------------------
def _schedule_periodic(
hass: HomeAssistant, entry: ConfigEntry, runtime: RuntimeData
def _schedule_identity(
hass: HomeAssistant,
runtime: RuntimeData,
manager: IdentityManager,
subentry: ConfigSubentry,
) -> None:
manager = runtime.manager
announce_interval = entry.options.get(
CONF_ANNOUNCE_INTERVAL, DEFAULT_ANNOUNCE_INTERVAL
)
data = subentry.data
announce_interval = data.get(CONF_ANNOUNCE_INTERVAL, DEFAULT_ANNOUNCE_INTERVAL)
if announce_interval and announce_interval > 0:
async def _do_announce(_now) -> None:
await _safe_announce(manager)
async def _do_announce(_now, _mgr=manager) -> None:
await _safe_announce(_mgr)
runtime.unsubs.append(
async_track_time_interval(
@@ -180,15 +204,15 @@ def _schedule_periodic(
)
)
sync_interval = entry.options.get(CONF_SYNC_INTERVAL, DEFAULT_SYNC_INTERVAL)
propagation_node = entry.options.get(CONF_PROPAGATION_NODE)
propagation_node = data.get(CONF_PROPAGATION_NODE)
if propagation_node:
hass.async_create_task(_set_propagation(manager, propagation_node))
sync_interval = data.get(CONF_SYNC_INTERVAL, DEFAULT_SYNC_INTERVAL)
if sync_interval and sync_interval > 0:
async def _do_sync(_now) -> None:
async def _do_sync(_now, _mgr=manager) -> None:
try:
await manager.async_sync_propagation()
await _mgr.async_sync_propagation()
except ReticulumError as err:
_LOGGER.warning("Propagation sync failed: %s", err)
@@ -199,14 +223,14 @@ def _schedule_periodic(
)
async def _set_propagation(manager: ReticulumManager, node: str) -> None:
async def _set_propagation(manager: IdentityManager, node: str) -> None:
try:
await manager.async_set_propagation_node(node)
except ReticulumError as err:
_LOGGER.warning("Could not set propagation node: %s", err)
async def _safe_announce(manager: ReticulumManager) -> None:
async def _safe_announce(manager: IdentityManager) -> None:
try:
await manager.async_announce()
except ReticulumError as err:
@@ -214,82 +238,82 @@ async def _safe_announce(manager: ReticulumManager) -> None:
# ---------------------------------------------------------------------------
# Incoming message pipeline
# Incoming message pipeline (per identity)
# ---------------------------------------------------------------------------
def _make_incoming_handler(
hass: HomeAssistant, entry: ConfigEntry, runtime: RuntimeData
hass: HomeAssistant,
entry: ConfigEntry,
subentry: ConfigSubentry,
runtime: RuntimeData,
manager: IdentityManager,
):
async def _handle(payload: dict[str, Any]) -> None:
source = payload["source"]
# Always surface the message as an event for automations.
hass.bus.async_fire(EVENT_MESSAGE_RECEIVED, payload)
options = entry.options
allow_all = options.get(CONF_ALLOW_ALL, True)
data = subentry.data
allow_all = data.get(CONF_ALLOW_ALL, True)
allowed = {
a.strip().lower()
for a in options.get(CONF_ALLOWED_IDENTITIES, [])
for a in data.get(CONF_ALLOWED_IDENTITIES, [])
if a and a.strip()
}
if not allow_all and source.lower() not in allowed:
_LOGGER.debug("Ignoring message from non-allowed sender %s", source)
return
# When messages are NOT routed to Assist, optionally auto-reply with the
# greeting so senders get an acknowledgement instead of silence.
if not options.get(CONF_ENABLE_ASSIST, True):
greeting = (options.get(CONF_GREETING) or "").strip()
# When not routing to Assist, optionally auto-reply with the greeting.
if not data.get(CONF_ENABLE_ASSIST, True):
greeting = (data.get(CONF_GREETING) or "").strip()
if greeting:
await _reply(runtime.manager, source, greeting, "Home Assistant")
await _reply(manager, source, greeting, "Home Assistant")
return
text = (payload.get("content") or "").strip()
if not text:
return
await _run_assist(hass, entry, runtime, source, text)
await _run_assist(hass, subentry, runtime, manager, source, text)
return _handle
async def _run_assist(
hass: HomeAssistant,
entry: ConfigEntry,
subentry: ConfigSubentry,
runtime: RuntimeData,
manager: IdentityManager,
source: str,
text: str,
) -> None:
"""Route a message through the Assist conversation pipeline and reply."""
from homeassistant.components import conversation # noqa: PLC0415
from homeassistant.core import Context # noqa: PLC0415
options = entry.options
agent_id = options.get(CONF_ASSIST_AGENT) or None
language = options.get(CONF_ASSIST_LANGUAGE) or hass.config.language
data = subentry.data
agent_id = data.get(CONF_ASSIST_AGENT) or None
language = data.get(CONF_ASSIST_LANGUAGE) or hass.config.language
conv_key = f"{subentry.subentry_id}:{source}"
try:
result = await conversation.async_converse(
hass=hass,
text=text,
conversation_id=runtime.conversation_ids.get(source),
conversation_id=runtime.conversation_ids.get(conv_key),
context=Context(),
language=language,
agent_id=agent_id,
)
except Exception as err: # noqa: BLE001
_LOGGER.exception("Assist conversation failed")
await _reply(runtime.manager, source, f"⚠️ Assist error: {err}", "Error")
await _reply(manager, source, f"⚠️ Assist error: {err}", "Error")
return
# Remember the conversation so this peer keeps context.
conv_id = getattr(result, "conversation_id", None)
if conv_id:
runtime.conversation_ids[source] = conv_id
runtime.conversation_ids[conv_key] = conv_id
reply = _extract_speech(result)
if reply:
await _reply(runtime.manager, source, reply, "Assist")
await _reply(manager, source, reply, "Assist")
def _extract_speech(result: Any) -> str | None:
@@ -303,12 +327,12 @@ def _extract_speech(result: Any) -> str | None:
async def _reply(
manager: ReticulumManager, destination: str, content: str, title: str
manager: IdentityManager, destination: str, content: str, title: str
) -> None:
try:
await manager.async_send_message(destination, content, title=title)
except ReticulumError as err:
_LOGGER.warning("Could not send Assist reply to %s: %s", destination, err)
_LOGGER.warning("Could not send reply to %s: %s", destination, err)
# ---------------------------------------------------------------------------
@@ -321,20 +345,42 @@ SEND_MESSAGE_SCHEMA = vol.Schema(
vol.Optional(ATTR_TITLE, default=""): cv.string,
vol.Optional(ATTR_METHOD, default="direct"): vol.In(DELIVERY_METHODS),
vol.Optional(ATTR_FIELDS): dict,
vol.Optional(ATTR_IDENTITY): cv.string,
}
)
REQUEST_PATH_SCHEMA = vol.Schema({vol.Required(ATTR_DESTINATION): cv.string})
SET_PROPAGATION_SCHEMA = vol.Schema({vol.Required(ATTR_DESTINATION): cv.string})
SYNC_SCHEMA = vol.Schema(
{vol.Optional(ATTR_MAX_MESSAGES): vol.All(vol.Coerce(int), vol.Range(min=1))}
REQUEST_PATH_SCHEMA = vol.Schema(
{vol.Required(ATTR_DESTINATION): cv.string, vol.Optional(ATTR_IDENTITY): cv.string}
)
SET_PROPAGATION_SCHEMA = vol.Schema(
{vol.Required(ATTR_DESTINATION): cv.string, vol.Optional(ATTR_IDENTITY): cv.string}
)
SYNC_SCHEMA = vol.Schema(
{
vol.Optional(ATTR_MAX_MESSAGES): vol.All(vol.Coerce(int), vol.Range(min=1)),
vol.Optional(ATTR_IDENTITY): cv.string,
}
)
ANNOUNCE_SCHEMA = vol.Schema({vol.Optional(ATTR_IDENTITY): cv.string})
def _first_manager(hass: HomeAssistant) -> ReticulumManager:
def _resolve_manager(hass: HomeAssistant, identity: str | None) -> IdentityManager:
"""Pick an identity manager by name/address, or the first one."""
data: dict[str, RuntimeData] = hass.data.get(DOMAIN, {})
if not data:
raise HomeAssistantError("Reticulum is not configured")
return next(iter(data.values())).manager
managers: list[IdentityManager] = []
for runtime in data.values():
managers.extend(runtime.identities.values())
if not managers:
raise HomeAssistantError("No Reticulum identity is configured")
if identity:
key = identity.strip().lower()
for manager in managers:
if (
manager.display_name.lower() == key
or (manager.state.lxmf_address or "").lower() == key
):
return manager
raise HomeAssistantError(f"No Reticulum identity matches '{identity}'")
return managers[0]
def _async_register_services(hass: HomeAssistant) -> None:
@@ -342,7 +388,7 @@ def _async_register_services(hass: HomeAssistant) -> None:
return
async def _send(call: ServiceCall) -> ServiceResponse:
manager = _first_manager(hass)
manager = _resolve_manager(hass, call.data.get(ATTR_IDENTITY))
try:
message_hash = await manager.async_send_message(
call.data[ATTR_DESTINATION],
@@ -353,33 +399,33 @@ def _async_register_services(hass: HomeAssistant) -> None:
)
except ReticulumError as err:
raise HomeAssistantError(str(err)) from err
return {"message_hash": message_hash}
return {"message_hash": message_hash, "identity": manager.display_name}
async def _announce(call: ServiceCall) -> None:
manager = _resolve_manager(hass, call.data.get(ATTR_IDENTITY))
try:
await _first_manager(hass).async_announce()
await manager.async_announce()
except ReticulumError as err:
raise HomeAssistantError(str(err)) from err
async def _request_path(call: ServiceCall) -> None:
manager = _resolve_manager(hass, call.data.get(ATTR_IDENTITY))
try:
await _first_manager(hass).async_request_path(call.data[ATTR_DESTINATION])
await manager.async_request_path(call.data[ATTR_DESTINATION])
except ReticulumError as err:
raise HomeAssistantError(str(err)) from err
async def _set_propagation(call: ServiceCall) -> None:
manager = _resolve_manager(hass, call.data.get(ATTR_IDENTITY))
try:
await _first_manager(hass).async_set_propagation_node(
call.data[ATTR_DESTINATION]
)
await manager.async_set_propagation_node(call.data[ATTR_DESTINATION])
except ReticulumError as err:
raise HomeAssistantError(str(err)) from err
async def _sync(call: ServiceCall) -> None:
manager = _resolve_manager(hass, call.data.get(ATTR_IDENTITY))
try:
await _first_manager(hass).async_sync_propagation(
call.data.get(ATTR_MAX_MESSAGES)
)
await manager.async_sync_propagation(call.data.get(ATTR_MAX_MESSAGES))
except ReticulumError as err:
raise HomeAssistantError(str(err)) from err
@@ -390,7 +436,9 @@ def _async_register_services(hass: HomeAssistant) -> None:
schema=SEND_MESSAGE_SCHEMA,
supports_response=SupportsResponse.OPTIONAL,
)
hass.services.async_register(DOMAIN, SERVICE_ANNOUNCE, _announce)
hass.services.async_register(
DOMAIN, SERVICE_ANNOUNCE, _announce, schema=ANNOUNCE_SCHEMA
)
hass.services.async_register(
DOMAIN, SERVICE_REQUEST_PATH, _request_path, schema=REQUEST_PATH_SCHEMA
)