"""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 import logging from datetime import timedelta from typing import Any import voluptuous as vol from homeassistant.config_entries import ConfigEntry, ConfigSubentry from homeassistant.const import EVENT_HOMEASSISTANT_STOP, Platform from homeassistant.core import ( HomeAssistant, ServiceCall, ServiceResponse, SupportsResponse, ) from homeassistant.exceptions import ConfigEntryNotReady, HomeAssistantError from homeassistant.helpers import config_validation as cv from homeassistant.helpers.event import async_track_time_interval from .const import ( ATTR_CONTENT, ATTR_DESTINATION, ATTR_FIELDS, ATTR_IDENTITY, ATTR_MAX_MESSAGES, ATTR_METHOD, ATTR_TITLE, CONF_ALLOW_ALL, CONF_ALLOWED_IDENTITIES, CONF_ANNOUNCE_INTERVAL, CONF_ASSIST_AGENT, CONF_ASSIST_LANGUAGE, CONF_DISPLAY_NAME, CONF_ENABLE_ASSIST, CONF_GREETING, CONF_INTERFACE_NAME, CONF_LOGLEVEL, CONF_PROPAGATION_NODE, CONF_SYNC_INTERVAL, CONF_TARGET_HOST, CONF_TARGET_PORT, DEFAULT_ANNOUNCE_INTERVAL, DEFAULT_DISPLAY_NAME, DEFAULT_INTERFACE_NAME, DEFAULT_LOGLEVEL, DEFAULT_SYNC_INTERVAL, DELIVERY_METHODS, DOMAIN, EVENT_MESSAGE_RECEIVED, SERVICE_ANNOUNCE, SERVICE_REQUEST_PATH, SERVICE_SEND_MESSAGE, SERVICE_SET_PROPAGATION_NODE, SERVICE_SYNC_PROPAGATION, STORAGE_SUBDIR, SUBENTRY_TYPE_IDENTITY, ) from .reticulum_client import ( IdentityManager, ReticulumError, ReticulumStack, ) _LOGGER = logging.getLogger(__name__) PLATFORMS: list[Platform] = [ Platform.BINARY_SENSOR, Platform.BUTTON, Platform.NOTIFY, Platform.SENSOR, ] INTERFACE_POLL = timedelta(seconds=15) class RuntimeData: """Runtime objects for the hub entry.""" def __init__(self, stack: ReticulumStack) -> None: self.stack = stack self.identities: dict[str, IdentityManager] = {} self.unsubs: list[Any] = [] # 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 the Reticulum stack and its identities.""" storage_dir = hass.config.path(STORAGE_SUBDIR) 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), loglevel=int(entry.options.get(CONF_LOGLEVEL, DEFAULT_LOGLEVEL)), ) try: await stack.async_start() except Exception as err: # noqa: BLE001 raise ConfigEntryNotReady(f"Could not start Reticulum: {err}") from err runtime = RuntimeData(stack) hass.data.setdefault(DOMAIN, {})[entry.entry_id] = 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) # 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) # Hub interface-status poll. runtime.unsubs.append( async_track_time_interval( hass, stack.refresh_interface_status, INTERFACE_POLL ) ) async def _on_ha_stop(_event) -> None: 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_entry_updated)) _async_register_services(hass) return True async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool: """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() 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]: _async_unregister_services(hass) return unload_ok 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 scheduling # --------------------------------------------------------------------------- def _schedule_identity( hass: HomeAssistant, runtime: RuntimeData, manager: IdentityManager, subentry: ConfigSubentry, ) -> None: 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, _mgr=manager) -> None: await _safe_announce(_mgr) runtime.unsubs.append( async_track_time_interval( hass, _do_announce, timedelta(seconds=announce_interval) ) ) 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, _mgr=manager) -> None: try: await _mgr.async_sync_propagation() except ReticulumError as err: _LOGGER.warning("Propagation sync failed: %s", err) runtime.unsubs.append( async_track_time_interval( hass, _do_sync, timedelta(seconds=sync_interval) ) ) 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: IdentityManager) -> None: try: await manager.async_announce() except ReticulumError as err: _LOGGER.debug("Announce skipped: %s", err) # --------------------------------------------------------------------------- # Incoming message pipeline (per identity) # --------------------------------------------------------------------------- def _make_incoming_handler( hass: HomeAssistant, entry: ConfigEntry, subentry: ConfigSubentry, runtime: RuntimeData, manager: IdentityManager, ): async def _handle(payload: dict[str, Any]) -> None: source = payload["source"] hass.bus.async_fire(EVENT_MESSAGE_RECEIVED, payload) data = subentry.data allow_all = data.get(CONF_ALLOW_ALL, True) allowed = { a.strip().lower() 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 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(manager, source, greeting, "Home Assistant") return text = (payload.get("content") or "").strip() if not text: return await _run_assist(hass, subentry, runtime, manager, source, text) return _handle async def _run_assist( hass: HomeAssistant, subentry: ConfigSubentry, runtime: RuntimeData, manager: IdentityManager, source: str, text: str, ) -> None: from homeassistant.components import conversation # noqa: PLC0415 from homeassistant.core import Context # noqa: PLC0415 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(conv_key), context=Context(), language=language, agent_id=agent_id, ) except Exception as err: # noqa: BLE001 _LOGGER.exception("Assist conversation failed") await _reply(manager, source, f"⚠️ Assist error: {err}", "Error") return conv_id = getattr(result, "conversation_id", None) if conv_id: runtime.conversation_ids[conv_key] = conv_id reply = _extract_speech(result) if reply: await _reply(manager, source, reply, "Assist") def _extract_speech(result: Any) -> str | None: try: speech = result.response.speech if isinstance(speech, dict): return speech.get("plain", {}).get("speech") except AttributeError: return None return None async def _reply( 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 reply to %s: %s", destination, err) # --------------------------------------------------------------------------- # Services # --------------------------------------------------------------------------- SEND_MESSAGE_SCHEMA = vol.Schema( { vol.Required(ATTR_DESTINATION): cv.string, vol.Required(ATTR_CONTENT): cv.string, 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, 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 _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, {}) 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: if hass.services.has_service(DOMAIN, SERVICE_SEND_MESSAGE): return async def _send(call: ServiceCall) -> ServiceResponse: manager = _resolve_manager(hass, call.data.get(ATTR_IDENTITY)) try: message_hash = await manager.async_send_message( call.data[ATTR_DESTINATION], call.data[ATTR_CONTENT], title=call.data.get(ATTR_TITLE, ""), method=call.data.get(ATTR_METHOD, "direct"), fields=call.data.get(ATTR_FIELDS), ) except ReticulumError as err: raise HomeAssistantError(str(err)) from err 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 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 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 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 manager.async_sync_propagation(call.data.get(ATTR_MAX_MESSAGES)) except ReticulumError as err: raise HomeAssistantError(str(err)) from err hass.services.async_register( DOMAIN, SERVICE_SEND_MESSAGE, _send, schema=SEND_MESSAGE_SCHEMA, supports_response=SupportsResponse.OPTIONAL, ) 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 ) hass.services.async_register( DOMAIN, SERVICE_SET_PROPAGATION_NODE, _set_propagation, schema=SET_PROPAGATION_SCHEMA, ) hass.services.async_register( DOMAIN, SERVICE_SYNC_PROPAGATION, _sync, schema=SYNC_SCHEMA ) def _async_unregister_services(hass: HomeAssistant) -> None: for service in ( SERVICE_SEND_MESSAGE, SERVICE_ANNOUNCE, SERVICE_REQUEST_PATH, SERVICE_SET_PROPAGATION_NODE, SERVICE_SYNC_PROPAGATION, ): if hass.services.has_service(DOMAIN, service): hass.services.async_remove(DOMAIN, service)