meshtastic/plugin/hermes_meshtastic/adapter.py

"""Meshtastic LoRa platform adapter for Hermes Agent.

Connects to a Meshtastic node via TCP (e.g. meshtastic.local:4403), listening for
LoRa packets, stripping formatting, and dispatching to Hermes agent sessions. Outbound
replies are chunked into concise LoRa-friendly packets.
"""

from __future__ import annotations

import asyncio
import logging
import re
import time
from typing import Any, Dict, List, Optional

try:
    from gateway.platforms.base import (
        BasePlatformAdapter,
        MessageEvent,
        MessageType,
        SendResult,
    )
except ImportError:
    # Allow importing and testing utility functions without gateway in environment
    class BasePlatformAdapter:  # type: ignore[no-redef]
        def __init__(self, config: Any, platform: Any = "meshtastic"):
            self.config = config
            self.platform = platform

        def build_source(self, **kwargs):
            return kwargs

        async def handle_message(self, event: Any) -> None:
            pass

    class SendResult:  # type: ignore[no-redef]
        def __init__(self, success: bool, message_id: Optional[str] = None, error: Optional[str] = None):
            self.success = success
            self.message_id = message_id
            self.error = error

    class MessageType:  # type: ignore[no-redef]
        TEXT = "text"

    class MessageEvent:  # type: ignore[no-redef]
        def __init__(self, **kwargs):
            for k, v in kwargs.items():
                setattr(self, k, v)

logger = logging.getLogger(__name__)

MAX_LORA_MESSAGE_LENGTH = 180  # Characters per packet to prevent LoRa airtime congestion

_MARKDOWN_RULES = (
    (r"\*\*(.+?)\*\*", r"\1"),  # bold
    (r"__(.+?)__", r"\1"),
    (r"\*(.+?)\*", r"\1"),      # italic
    (r"(?<!\w)_(.+?)_(?!\w)", r"\1"),
    (r"`(.+?)`", r"\1"),        # inline code
    (r"```[\w]*\n?", ""),       # code fences
    (r"!\[([^\]]*)\]\(([^)]+)\)", r"\1"),
    (r"\[([^\]]+)\]\(([^)]+)\)", r"\1 (\2)"),
    (r"^#+\s*", ""),            # headings
    (r"^\s*[-*+]\s+", "- "),    # normalize lists
)


def strip_markdown(text: str) -> str:
    """Convert markdown formatting into plain readable text suitable for radio."""
    if not text:
        return ""
    result = text
    for pattern, replacement in _MARKDOWN_RULES:
        result = re.sub(pattern, replacement, result, flags=re.MULTILINE)
    # Collapse multiple blank lines
    result = re.sub(r"\n{3,}", "\n\n", result)
    return result.strip()


def chunk_text(text: str, limit: int = MAX_LORA_MESSAGE_LENGTH) -> List[str]:
    """Split text into smaller chunks for LoRa transmission, breaking at whitespace/newlines."""
    if not text:
        return []
    if len(text) <= limit:
        return [text]

    chunks: List[str] = []
    remaining = text
    while remaining:
        if len(remaining) <= limit:
            chunks.append(remaining.strip())
            break
        split_at = remaining.rfind(" ", 0, limit)
        if split_at <= 0 or split_at < (limit // 3):
            newline_at = remaining.rfind("\n", 0, limit)
            if newline_at > 0:
                split_at = newline_at
            else:
                split_at = limit
        chunk = remaining[:split_at].strip()
        if chunk:
            chunks.append(chunk)
        remaining = remaining[split_at:].lstrip()
    return chunks


def check_requirements() -> bool:
    """Check if meshtastic is installed."""
    try:
        import meshtastic
        import meshtastic.tcp_interface
        return True
    except ImportError:
        return False


class MeshtasticAdapter(BasePlatformAdapter):
    """Hermes gateway adapter for Meshtastic LoRa radios over TCP."""

    def __init__(self, config: Any):
        super().__init__(config, platform="meshtastic")
        self.extra = getattr(config, "extra", {}) or {}
        self.host = self.extra.get("host", "meshtastic.local")
        self.port = int(self.extra.get("port", 4403))
        self.channel_index = int(self.extra.get("channel_index", 1))
        self.allowed_nodes: List[str] = [
            n.strip() for n in self.extra.get("allowed_nodes", []) if n.strip()
        ]
        self._iface = None
        self._loop: Optional[asyncio.AbstractEventLoop] = None
        self._connected = False
        self._running = False
        self._task: Optional[asyncio.Task] = None

    async def connect(self) -> None:
        """Connect to Meshtastic radio via TCP and start subscriber."""
        self._loop = asyncio.get_running_loop()
        self._running = True
        self._task = asyncio.create_task(self._connection_supervisor())

    async def _connection_supervisor(self) -> None:
        backoff = 2.0
        while self._running:
            try:
                import meshtastic.tcp_interface
                from pubsub import pub

                logger.info("[Meshtastic] Connecting to radio at %s:%s...", self.host, self.port)
                # TCPInterface handles initial protobuf handshake synchronously
                self._iface = await asyncio.to_thread(
                    meshtastic.tcp_interface.TCPInterface,
                    hostname=self.host,
                    portNumber=self.port,
                    noProto=False,
                )
                self._connected = True
                backoff = 2.0
                my_id = getattr(self._iface.myInfo, "my_node_num", "unknown")
                logger.info("[Meshtastic] Connected to radio (my_node_num=%s, listening on ch=%d)", my_id, self.channel_index)

                pub.subscribe(self._on_meshtastic_receive, "meshtastic.receive.text")

                # Keep-alive loop
                while self._running and self._connected:
                    await asyncio.sleep(5)
                    if not self._iface or not self._iface.isConnected:
                        break

            except asyncio.CancelledError:
                break
            except Exception as exc:
                logger.warning("[Meshtastic] Radio connection failed (%s). Retrying in %.1fs...", exc, backoff)
                self._connected = False
                await asyncio.sleep(backoff)
                backoff = min(backoff * 1.5, 60.0)

    def _on_meshtastic_receive(self, packet: Dict[str, Any], interface: Any) -> None:
        """Handle incoming text packet from Meshtastic pubsub (runs in worker thread)."""
        try:
            ch = packet.get("channel", 0)
            if self.channel_index is not None and ch != self.channel_index:
                # Ignore packets from different channels (e.g. public channel 0 when listening on 1)
                return

            from_id = packet.get("fromId") or f"!{packet.get('from', 0):08x}"
            my_info = getattr(interface, "myInfo", None)
            my_node_num = getattr(my_info, "my_node_num", 0)
            if packet.get("from") == my_node_num:
                # Prevent self-message loops
                return

            if self.allowed_nodes and from_id not in self.allowed_nodes:
                logger.debug("[Meshtastic] Ignoring message from unauthorized node: %s", from_id)
                return

            decoded = packet.get("decoded", {})
            text = decoded.get("text", "").strip()
            if not text:
                return

            # Determine sender and target
            to_id = packet.get("toId")
            is_dm = to_id != "^all" and packet.get("to") == my_node_num
            chat_id = from_id if is_dm else f"channel_{ch}"
            chat_type = "dm" if is_dm else "channel"

            rx_snr = packet.get("rxSnr", 0.0)
            logger.info(
                "[Meshtastic] Inbound message from %s on ch=%s (is_dm=%s, SNR=%.1fdB): %s",
                from_id, ch, is_dm, rx_snr, text
            )

            # Build Hermes SessionSource and MessageEvent
            source = self.build_source(
                chat_id=chat_id,
                chat_name=f"Meshtastic {chat_id}",
                chat_type=chat_type,
                user_id=from_id,
                user_name=from_id,
                message_id=str(packet.get("id", int(time.time()))),
            )

            event = MessageEvent(
                text=text,
                message_type=MessageType.TEXT,
                user_id=from_id,
                user_name=from_id,
                source=source,
                raw_message=packet,
                message_id=str(packet.get("id", int(time.time()))),
            )

            if self._loop and self._loop.is_running():
                asyncio.run_coroutine_threadsafe(self.handle_message(event), self._loop)

        except Exception as exc:
            logger.exception("[Meshtastic] Error processing received packet: %s", exc)

    async def send(
        self,
        chat_id: str,
        content: str,
        reply_to: Optional[str] = None,
        metadata: Optional[Dict[str, Any]] = None,
    ) -> SendResult:
        """Send a message out over LoRa (implements BasePlatformAdapter.send)."""
        if not self._connected or not self._iface:
            return SendResult(success=False, error="Meshtastic radio not connected")

        clean_text = strip_markdown(content)
        chunks = chunk_text(clean_text, MAX_LORA_MESSAGE_LENGTH)

        # Destination: if chat_id starts with '!', it is a direct node ID
        destination = chat_id if chat_id.startswith("!") else "^all"
        channel_index = self.channel_index if destination == "^all" else 0

        last_id = None
        try:
            for idx, chunk in enumerate(chunks):
                if len(chunks) > 1:
                    formatted_chunk = f"({idx+1}/{len(chunks)}) {chunk}"
                else:
                    formatted_chunk = chunk

                logger.info(
                    "[Meshtastic] Sending chunk to %s on ch=%s: %s",
                    destination, channel_index, formatted_chunk
                )

                # Send via mesh interface thread
                await asyncio.to_thread(
                    self._iface.sendText,
                    text=formatted_chunk,
                    destinationId=destination,
                    channelIndex=channel_index,
                )
                last_id = str(int(time.time() * 1000))
                # Inter-packet delay to comply with LoRa duty cycle
                if len(chunks) > 1:
                    await asyncio.sleep(1.2)

            return SendResult(success=True, message_id=last_id or str(int(time.time())))
        except Exception as exc:
            logger.exception("[Meshtastic] Failed to send message: %s", exc)
            return SendResult(success=False, error=str(exc))

    async def get_chat_info(self, chat_id: str) -> Dict[str, Any]:
        """Get chat/channel info (implements BasePlatformAdapter.get_chat_info)."""
        is_dm = chat_id.startswith("!")
        return {
            "name": f"Node {chat_id}" if is_dm else f"Channel {chat_id}",
            "type": "dm" if is_dm else "channel",
        }

    async def disconnect(self) -> None:
        """Disconnect and clean up resources."""
        self._running = False
        if self._task:
            self._task.cancel()
        if self._iface:
            try:
                from pubsub import pub
                pub.unsubscribe(self._on_meshtastic_receive, "meshtastic.receive.text")
            except Exception:
                pass
            await asyncio.to_thread(self._iface.close)
            self._connected = False
            logger.info("[Meshtastic] Radio interface closed")