ha-reticulum/custom_components/reticulum/__init__.py

"""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] = {}
        # Senders we've already greeted this session (first-contact greeting).
        self.greeted: set[str] = set()


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

        # First-contact greeting: send the configured greeting once per sender.
        greeting = (options.get(CONF_GREETING) or "").strip()
        if greeting and source not in runtime.greeted:
            runtime.greeted.add(source)
            await _reply(runtime.manager, source, greeting, "Home Assistant")

        if not options.get(CONF_ENABLE_ASSIST, True):
            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)