c3b98043f7
The greeting option returns, but it is now sent as an automatic reply to incoming messages only when "Route incoming messages to Assist" is disabled, so senders get an acknowledgement instead of silence. When Assist handles messages the greeting is not used. Blank disables it. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
418 lines
14 KiB
Python
418 lines
14 KiB
Python
"""The Reticulum integration."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
from datetime import timedelta
|
|
from typing import Any
|
|
|
|
import voluptuous as vol
|
|
|
|
from homeassistant.config_entries import ConfigEntry
|
|
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_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,
|
|
)
|
|
from .reticulum_client import ReticulumError, ReticulumManager
|
|
|
|
_LOGGER = logging.getLogger(__name__)
|
|
|
|
PLATFORMS: list[Platform] = [
|
|
Platform.BINARY_SENSOR,
|
|
Platform.BUTTON,
|
|
Platform.NOTIFY,
|
|
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."""
|
|
|
|
def __init__(self, manager: ReticulumManager) -> None:
|
|
self.manager = manager
|
|
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.
|
|
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
|
|
storage_dir = hass.config.path(STORAGE_SUBDIR)
|
|
|
|
manager = ReticulumManager(
|
|
hass,
|
|
storage_dir,
|
|
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)),
|
|
)
|
|
|
|
try:
|
|
await manager.async_start()
|
|
except Exception as err: # noqa: BLE001
|
|
raise ConfigEntryNotReady(f"Could not start Reticulum: {err}") from err
|
|
|
|
runtime = RuntimeData(manager)
|
|
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)
|
|
|
|
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)
|
|
|
|
# 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).
|
|
runtime.unsubs.append(
|
|
async_track_time_interval(
|
|
hass, manager.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()
|
|
|
|
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))
|
|
_async_register_services(hass)
|
|
|
|
return True
|
|
|
|
|
|
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
|
|
"""Unload a config 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()
|
|
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_options_updated(hass: HomeAssistant, entry: ConfigEntry) -> None:
|
|
"""Reload the entry so option changes take effect."""
|
|
await hass.config_entries.async_reload(entry.entry_id)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Periodic tasks
|
|
# ---------------------------------------------------------------------------
|
|
def _schedule_periodic(
|
|
hass: HomeAssistant, entry: ConfigEntry, runtime: RuntimeData
|
|
) -> None:
|
|
manager = runtime.manager
|
|
announce_interval = entry.options.get(
|
|
CONF_ANNOUNCE_INTERVAL, DEFAULT_ANNOUNCE_INTERVAL
|
|
)
|
|
if announce_interval and announce_interval > 0:
|
|
|
|
async def _do_announce(_now) -> None:
|
|
await _safe_announce(manager)
|
|
|
|
runtime.unsubs.append(
|
|
async_track_time_interval(
|
|
hass, _do_announce, timedelta(seconds=announce_interval)
|
|
)
|
|
)
|
|
|
|
sync_interval = entry.options.get(CONF_SYNC_INTERVAL, DEFAULT_SYNC_INTERVAL)
|
|
propagation_node = entry.options.get(CONF_PROPAGATION_NODE)
|
|
if propagation_node:
|
|
hass.async_create_task(_set_propagation(manager, propagation_node))
|
|
if sync_interval and sync_interval > 0:
|
|
|
|
async def _do_sync(_now) -> None:
|
|
try:
|
|
await manager.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: ReticulumManager, 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:
|
|
try:
|
|
await manager.async_announce()
|
|
except ReticulumError as err:
|
|
_LOGGER.debug("Announce skipped: %s", err)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Incoming message pipeline
|
|
# ---------------------------------------------------------------------------
|
|
def _make_incoming_handler(
|
|
hass: HomeAssistant, entry: ConfigEntry, runtime: RuntimeData
|
|
):
|
|
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)
|
|
allowed = {
|
|
a.strip().lower()
|
|
for a in options.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()
|
|
if greeting:
|
|
await _reply(runtime.manager, source, greeting, "Home Assistant")
|
|
return
|
|
|
|
text = (payload.get("content") or "").strip()
|
|
if not text:
|
|
return
|
|
|
|
await _run_assist(hass, entry, runtime, source, text)
|
|
|
|
return _handle
|
|
|
|
|
|
async def _run_assist(
|
|
hass: HomeAssistant,
|
|
entry: ConfigEntry,
|
|
runtime: RuntimeData,
|
|
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
|
|
|
|
try:
|
|
result = await conversation.async_converse(
|
|
hass=hass,
|
|
text=text,
|
|
conversation_id=runtime.conversation_ids.get(source),
|
|
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")
|
|
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
|
|
|
|
reply = _extract_speech(result)
|
|
if reply:
|
|
await _reply(runtime.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: ReticulumManager, 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)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 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,
|
|
}
|
|
)
|
|
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))}
|
|
)
|
|
|
|
|
|
def _first_manager(hass: HomeAssistant) -> ReticulumManager:
|
|
data: dict[str, RuntimeData] = hass.data.get(DOMAIN, {})
|
|
if not data:
|
|
raise HomeAssistantError("Reticulum is not configured")
|
|
return next(iter(data.values())).manager
|
|
|
|
|
|
def _async_register_services(hass: HomeAssistant) -> None:
|
|
if hass.services.has_service(DOMAIN, SERVICE_SEND_MESSAGE):
|
|
return
|
|
|
|
async def _send(call: ServiceCall) -> ServiceResponse:
|
|
manager = _first_manager(hass)
|
|
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}
|
|
|
|
async def _announce(call: ServiceCall) -> None:
|
|
try:
|
|
await _first_manager(hass).async_announce()
|
|
except ReticulumError as err:
|
|
raise HomeAssistantError(str(err)) from err
|
|
|
|
async def _request_path(call: ServiceCall) -> None:
|
|
try:
|
|
await _first_manager(hass).async_request_path(call.data[ATTR_DESTINATION])
|
|
except ReticulumError as err:
|
|
raise HomeAssistantError(str(err)) from err
|
|
|
|
async def _set_propagation(call: ServiceCall) -> None:
|
|
try:
|
|
await _first_manager(hass).async_set_propagation_node(
|
|
call.data[ATTR_DESTINATION]
|
|
)
|
|
except ReticulumError as err:
|
|
raise HomeAssistantError(str(err)) from err
|
|
|
|
async def _sync(call: ServiceCall) -> None:
|
|
try:
|
|
await _first_manager(hass).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)
|
|
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)
|