"""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. Contract note ------------- The Hermes gateway API (``gateway.platforms.base.BasePlatformAdapter`` and friends) is resolved LAZILY, inside ``get_adapter_class()``, never at module import time. Hermes' own guidance (``hermes_cli/plugins.py``) puts gateway imports inside factory bodies because third-party plugin modules are imported during plugin discovery, which runs before/independently of the gateway runtime being fully importable. Importing ``gateway.platforms.base`` at module scope silently failed in the deployed runtime (an ``ImportError`` swallowed by an earlier fallback bound the adapter to a local stub class lacking ``set_message_handler`` — see git history). Deferred resolution keeps this module importable in isolation (unit tests / CI without Hermes) while always binding the adapter to the REAL gateway base at registration/creation time. The contract tests in ``tests/contract/`` assert this invariant against the real ``hermes-agent`` package (pinned to the deployed image tag). """ from __future__ import annotations import asyncio import logging import time from typing import Any, Dict, List, Optional from .formatting import MAX_LORA_MESSAGE_LENGTH, chunk_text, strip_markdown logger = logging.getLogger(__name__) __all__ = [ "MAX_LORA_MESSAGE_LENGTH", "check_requirements", "chunk_text", "get_adapter_class", "strip_markdown", ] # --------------------------------------------------------------------------- # Hermes gateway API — resolved lazily and cached (see module docstring). # --------------------------------------------------------------------------- _gateway_api: Optional[Dict[str, Any]] = None def _load_gateway_api() -> Dict[str, Any]: """Import the real Hermes gateway types and bind them as module globals. Called from ``get_adapter_class()`` at adapter-creation time (the gateway runtime is fully loaded by then). Method bodies reference these names as module globals, so they are injected into this module's namespace. """ global _gateway_api if _gateway_api is None: from gateway.config import Platform from gateway.platforms.base import ( # type: ignore[import-not-found] BasePlatformAdapter, MessageEvent, MessageType, SendResult, ) _gateway_api = { "Platform": Platform, "BasePlatformAdapter": BasePlatformAdapter, "MessageEvent": MessageEvent, "MessageType": MessageType, "SendResult": SendResult, } globals().update(_gateway_api) return _gateway_api # --------------------------------------------------------------------------- # Dependency probe (passive — never installs anything). # --------------------------------------------------------------------------- def check_requirements() -> bool: """Check if meshtastic is installed.""" try: import meshtastic # noqa: F401 import meshtastic.tcp_interface # noqa: F401 return True except ImportError: return False # --------------------------------------------------------------------------- # Adapter — class body defined once the real gateway API is available. # --------------------------------------------------------------------------- _adapter_class = None def get_adapter_class(): """Return the ``MeshtasticAdapter`` class bound to the real gateway base. The class is defined (and cached) on first call so that ``BasePlatformAdapter`` is the REAL Hermes class — a fallback stub at this point would reproduce the ``set_message_handler`` AttributeError this design exists to prevent. """ global _adapter_class if _adapter_class is None: _load_gateway_api() class MeshtasticAdapter(BasePlatformAdapter): """Hermes gateway adapter for Meshtastic LoRa radios over TCP.""" def __init__(self, config: Any): super().__init__(config=config, platform=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 self._connecting = False async def connect(self, *, is_reconnect: bool = False) -> bool: """Establish the Meshtastic TCP session; return success. Mirrors the gateway contract ``async connect(*, is_reconnect=False) -> bool`` (see tests/contract/). On initial failure the watchdog keeps retrying with exponential backoff in the background. """ self._loop = asyncio.get_running_loop() self._running = True try: ok = await self._try_connect() except Exception as exc: logger.warning( "[Meshtastic] Initial connect failed (%s); watchdog will retry.", exc ) ok = False self._connected = ok self._ensure_watchdog() return ok async def _try_connect(self) -> bool: if self._connecting: return False # single-flight: gateway + watchdog must not double-connect self._connecting = True try: import meshtastic.tcp_interface from pubsub import pub logger.info( "[Meshtastic] Connecting to radio at %s:%s (is_reconnect=%s)...", self.host, self.port, getattr(self, "_last_reconnect", False), ) # TCPInterface performs the protobuf handshake synchronously. self._iface = await asyncio.to_thread( meshtastic.tcp_interface.TCPInterface, hostname=self.host, portNumber=self.port, noProto=False, ) 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") return True finally: self._connecting = False def _ensure_watchdog(self) -> None: if self._task is None or self._task.done(): self._task = asyncio.create_task(self._connection_watchdog()) async def _connection_watchdog(self) -> None: """Keep the radio session alive; reconnect with backoff on drop.""" backoff = 2.0 while self._running: await asyncio.sleep(5) try: iface = self._iface if iface is not None and not iface.isConnected: self._connected = False except Exception: self._connected = False if self._connected: backoff = 2.0 continue # Radio went away (or never came up) — reconnect. self._last_reconnect = True try: ok = await self._try_connect() self._connected = ok backoff = 2.0 except Exception as exc: logger.warning( "[Meshtastic] Radio reconnect 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 (worker thread).""" try: ch = packet.get("channel", 0) my_info = getattr(interface, "myInfo", None) my_node_num = getattr(my_info, "my_node_num", 0) is_direct_to_me = packet.get("to") == my_node_num is_pki_dm = is_direct_to_me and packet.get("pkiEncrypted", False) if ( self.channel_index is not None and ch != self.channel_index and not is_pki_dm ): # Ignore packets from different channels (e.g. public channel 0). # PKI direct messages omit the channel field and decode as # channel 0 even when sent from a secondary channel. return from_id = packet.get("fromId") or f"!{packet.get('from', 0):08x}" 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 to_id = packet.get("toId") is_dm = to_id != "^all" and is_direct_to_me 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, ) 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, ) 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") _adapter_class = MeshtasticAdapter return _adapter_class