"""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, SendResult from gateway.platforms.event import MessageEvent, MessageType except ImportError: # Allow importing and testing utility functions without gateway in environment class BasePlatformAdapter: # type: ignore[no-redef] def __init__(self, config: Any): self.config = config 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"(? 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) 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}" rx_snr = packet.get("rxSnr", 0.0) hop_limit = packet.get("hopLimit", 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 MessageEvent event = MessageEvent( platform="meshtastic", message_type=MessageType.TEXT, text=text, sender_id=from_id, sender_name=from_id, chat_id=chat_id, message_id=str(packet.get("id", int(time.time()))), raw_event=packet, ) 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_message( self, chat_id: str, message: str, reply_to_id: Optional[str] = None, **kwargs: Any, ) -> SendResult: """Send a message out over LoRa.""" if not self._connected or not self._iface: return SendResult(success=False, error="Meshtastic radio not connected") clean_text = strip_markdown(message) 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 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")