eric/meshtastic
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 5 Wiki Activity

fix(plugin): bind adapter to real Hermes gateway API; add contract-test gate
release / contract-test (push) Successful in 54s
Details
release / test-and-release (push) Successful in 11s
Details

Browse Source 
- Resolve the Hermes gateway API lazily via get_adapter_class() so the adapter
  always inherits the real gateway.platforms.base.BasePlatformAdapter instead
  of the ImportError fallback stub (root cause of the AttributeError:
  'set_message_handler' crash at gateway startup).
- Pass Platform('meshtastic') enum to super().__init__ and align
  connect(*, is_reconnect=False) -> bool with the gateway contract.
- Split pure formatting utils into formatting.py; remove the stub fallback.
- Add tests/contract (L0-L3): real-base binding, ABC instantiation, gateway
  run.py wiring replay, signature and event-shape conformance, run against
  hermes-agent pinned to the deployed image source (v2026.8.31).
- Add scripts/contract-test.sh and a contract-test CI job; release publish
  now requires contract tests to pass.
- Untrack stray __pycache__ artifacts.
This commit is contained in:
Eric X. Liu
2026-09-07 20:50:21 -07:00
parent 2d9106b864
commit 310c0244df
12 changed files with 712 additions and 256 deletions
Download Patch File Download Diff File Expand all files Collapse all files
plugin/hermes_meshtastic/adapter.py
+289 -250
View File
@@ -3,306 +3,345 @@
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 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)
from .formatting import MAX_LORA_MESSAGE_LENGTH, chunk_text, strip_markdown
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
)
__all__ = [
"MAX_LORA_MESSAGE_LENGTH",
"check_requirements",
"chunk_text",
"get_adapter_class",
"strip_markdown",
]
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()
# ---------------------------------------------------------------------------
# Hermes gateway API — resolved lazily and cached (see module docstring).
# ---------------------------------------------------------------------------
_gateway_api: Optional[Dict[str, Any]] = None
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]
def _load_gateway_api() -> Dict[str, Any]:
"""Import the real Hermes gateway types and bind them as module globals.
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
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
import meshtastic.tcp_interface
import meshtastic # noqa: F401
import meshtastic.tcp_interface # noqa: F401
return True
except ImportError:
return False
class MeshtasticAdapter(BasePlatformAdapter):
"""Hermes gateway adapter for Meshtastic LoRa radios over TCP."""
# ---------------------------------------------------------------------------
# Adapter — class body defined once the real gateway API is available.
# ---------------------------------------------------------------------------
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
_adapter_class = 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
def get_adapter_class():
"""Return the ``MeshtasticAdapter`` class bound to the real gateway base.
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)
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()
pub.subscribe(self._on_meshtastic_receive, "meshtastic.receive.text")
class MeshtasticAdapter(BasePlatformAdapter):
"""Hermes gateway adapter for Meshtastic LoRa radios over TCP."""
# 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)
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
await asyncio.sleep(backoff)
backoff = min(backoff * 1.5, 60.0)
self._running = False
self._task: Optional[asyncio.Task] = None
self._connecting = False
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
async def connect(self, *, is_reconnect: bool = False) -> bool:
"""Establish the Meshtastic TCP session; return success.
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
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
if self.allowed_nodes and from_id not in self.allowed_nodes:
logger.debug("[Meshtastic] Ignoring message from unauthorized node: %s", from_id)
return
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
decoded = packet.get("decoded", {})
text = decoded.get("text", "").strip()
if not text:
return
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
# 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"
def _ensure_watchdog(self) -> None:
if self._task is None or self._task.done():
self._task = asyncio.create_task(self._connection_watchdog())
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
)
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)
# 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()))),
)
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)
if self.channel_index is not None and ch != self.channel_index:
# Ignore packets from different channels (e.g. public channel 0).
return
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()))),
)
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._loop and self._loop.is_running():
asyncio.run_coroutine_threadsafe(self.handle_message(event), self._loop)
if self.allowed_nodes and from_id not in self.allowed_nodes:
logger.debug(
"[Meshtastic] Ignoring message from unauthorized node: %s", from_id
)
return
except Exception as exc:
logger.exception("[Meshtastic] Error processing received packet: %s", exc)
decoded = packet.get("decoded", {})
text = decoded.get("text", "").strip()
if not text:
return
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")
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"
clean_text = strip_markdown(content)
chunks = chunk_text(clean_text, MAX_LORA_MESSAGE_LENGTH)
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,
)
# 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
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()))),
)
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
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()))),
)
logger.info(
"[Meshtastic] Sending chunk to %s on ch=%s: %s",
destination, channel_index, formatted_chunk
)
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)
# 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)
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")
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))
clean_text = strip_markdown(content)
chunks = chunk_text(clean_text, MAX_LORA_MESSAGE_LENGTH)
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",
}
# 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
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")
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