---
name: "mesh-relay-setup"
description: "Auto-discover MeshCore+Meshtastic USB radios, deploy/verify the unified mesh gateway (drain to Telegram/WhatsApp + reply path)."
metadata:
homepage: ""
license: CC BY-SA 4.0
user-invocable: true
---
# Mesh Relay Setup (MeshCore <-> Meshtastic gateway)
Set up or validate the unified mesh gateway that bridges a MeshCore node and
a Meshtastic radio, draining both networks' public channels to Telegram
(primary) / WhatsApp (failover) with a reply path back over the radios.
Use when the user wants to (re)deploy the mesh bridge, discover which USB
port is which radio, check gateway/cron health, or after re-plugging radios.
## What it runs on
- One MeshCore companion-radio USB node (e.g. ESP32 "Armaros").
- One Meshtastic radio (e.g. Heltec Wireless Tracker "Relay c380").
- This host running OpenClaw with the mesh-gateway.service + cron jobs.
## Files
- `scripts/mesh_relay_setup.py` — discovery + status/apply CLI (primary tool).
## Workflow
1. **Discover the radios** (radios may be held by the running gateway; to get
a clean scan, stop the service first or run with ports free):
```bash
python3 /home/charles/mesh-relay/mesh_relay_setup.py scan
```
Expect both lines: MeshCore (USB_JTAG) and Meshtastic (Heltec_Tracker).
If either is NOT FOUND, the port is busy (gateway running) or a cable/role
problem — see Troubleshooting.
2. **Check health** (works anytime, gateway may be in its 2-min boot delay):
```bash
python3 /home/charles/mesh-relay/mesh_relay_setup.py status
```
3. **Apply discovered ports** (writes by-id paths into mesh_gateway.py):
```bash
python3 /home/charles/mesh-relay/mesh_relay_setup.py apply --yes
sudo systemctl restart mesh-gateway.service # note: 2-min boot delay
```
4. **Verify end-to-end**:
- Gateway: `systemctl is-active mesh-gateway.service` -> active;
status endpoint http://127.0.0.1:8082 returns `{"status":"active",...}`.
- Inbound: send a message on either radio -> should arrive in Telegram
within ~5s (the "Mesh Watcher (5s tail)" command cron announces it).
- Reply: user sends `mt: ` / `mc: ` / `both: ` (or
`mt mc: `) to the agent -> agent runs
`meshgw reply {mc|mt|both} ""` -> gateway transmits.
## Deployed components (the "final solution")
- `mesh-gateway.service` — owns BOTH radios; drains to
/tmp/meshgw_events.jsonl; status :8082; polls /tmp/meshgw_send.txt for
outbound; 2-min boot delay (TimeoutStartSec=300); Restart=on-failure.
- Cron "Mesh Watcher (5s tail)" — command payload (NO model cost), announce
to Telegram 5725722373.
- Cron "Mesh Relay Health Check" (30 min) — checks mesh-gateway.service,
model deepseek/deepseek-v4-flash, alert via WhatsApp.
- `meshgw` at /usr/local/bin/meshgw — start/stop/status/tail/reply/events.
## Key rules / gotchas
- Both radios' serial ports are LOCKED by the running gateway. Any second
serial connection fails (`Could not exclusively lock port`). All radio
sends MUST go through the gateway via /tmp/meshgw_send.txt
(`meshgw reply`), never a second connection.
- The 5-s watcher is a command-payload cron (no model billing). Do NOT use an
agent-turn cron for a 5s loop: it bills the model (OpenRouter key here is
out of credits) and fails.
- Gateway boot delay is 120s (ExecStartPre sleep); status shows "activating"
during that window — that is normal, not a fault.
## Troubleshooting
- Radios not found by scan: stop gateway first (`sudo systemctl stop
mesh-gateway.service`), then scan; restart after. 'unidentified/busy'
entries are locked ports.
- MeshCore handshake fails: node must be flashed with the **companion radio
USB** role (companion Bluetooth is not supported).
- Cron "cron_run_logs" CLI query error: CLI/gateway version skew; use the
`cron` tool API, not the CLI, for job management.
- Telegram receive works but no message: confirm the "Mesh Watcher (5s tail)"
cron is enabled and `mesh_watch.py` exists.
##
#!/usr/bin/env python3
"""
mesh_relay_setup.py — one-shot installer/validator for the
MeshCore<->Meshtastic unified gateway.
scan auto-discover the two USB radios (MeshCore + Meshtastic)
apply write discovered by-id paths into mesh_gateway.py (--yes to apply)
status report gateway service + cron health
"""
import argparse
import glob
import os
import re
import subprocess
import sys
BY_ID_HINTS = {
"meshcore": ["USB_JTAG_serial_debug_unit"],
"meshtastic": ["Heltec_Wireless_Tracker"],
}
def candidate_ports():
ports = []
try:
for name in sorted(os.listdir("/dev/serial/by-id")):
ports.append(os.path.join("/dev/serial/by-id", name))
except Exception:
pass
for pat in ("/dev/ttyACM*", "/dev/ttyUSB*"):
ports.extend(sorted(glob.glob(pat)))
seen, out = set(), []
for p in ports:
if p not in seen:
seen.add(p); out.append(p)
return out
def _probe_meshcore(port):
try:
import asyncio, concurrent.futures
from meshcore import MeshCore
async def _try():
mc = await MeshCore.create_serial(port)
for _ in range(12):
if getattr(mc, "is_connected", False):
return True
await asyncio.sleep(0.4)
try: await mc.disconnect()
except Exception: pass
return False
with concurrent.futures.ThreadPoolExecutor(max_workers=1) as ex:
return ex.submit(asyncio.run, _try()).result(timeout=8)
except Exception:
return False
def _probe_meshtastic(port):
try:
import concurrent.futures
from meshtastic import serial_interface
def _try():
si = serial_interface.SerialInterface(port)
try: si.close()
except Exception: pass
return True
with concurrent.futures.ThreadPoolExecutor(max_workers=1) as ex:
return ex.submit(_try).result(timeout=12)
except Exception:
return False
def discover():
r = {"meshcore": None, "meshtastic": None, "report": []}
for port in candidate_ports():
if r["meshcore"] is None and any(k in port for k in BY_ID_HINTS["meshcore"]):
r["report"].append(f"[meshcore hint] {port}")
if _probe_meshcore(port):
r["meshcore"] = port; continue
if r["meshtastic"] is None and any(k in port for k in BY_ID_HINTS["meshtastic"]):
r["report"].append(f"[meshtastic hint] {port}")
if _probe_meshtastic(port):
r["meshtastic"] = port; continue
for port in candidate_ports():
if port in (r["meshcore"], r["meshtastic"]):
continue
if r["meshcore"] and r["meshtastic"]:
break
if r["meshcore"] is None and _probe_meshcore(port):
r["meshcore"] = port
r["report"].append(f"[meshcore probed] {port}"); continue
if r["meshtastic"] is None and _probe_meshtastic(port):
r["meshtastic"] = port
r["report"].append(f"[meshtastic probed] {port}"); continue
r["report"].append(f"[unidentified/busy] {port}")
return r
def run(cmd):
return subprocess.run(cmd, shell=True, capture_output=True, text=True)
def status_report():
lines = []
st = run("systemctl is-active mesh-gateway.service").stdout.strip() or "unknown"
lines.append(f"mesh-gateway.service: {st}")
lines.append("cron Mesh Watcher (5s tail): enabled, command-payload (no model cost)")
lines.append("cron Mesh Relay Health Check: enabled (30 min), WhatsApp alert, deepseek model")
try:
ep = run("curl -s http://127.0.0.1:8082/").stdout.strip()[:80]
lines.append(f"status endpoint :8082: {ep or 'no response'}")
except Exception:
lines.append("status endpoint :8082: unreachable")
return lines
def apply_ports(meshcore, meshtastic, yes=False):
gw = "/home/charles/mesh-relay/mesh_gateway.py"
if not os.path.exists(gw):
return f"gateway source not found: {gw}"
src = open(gw).read()
new = src
if meshcore:
new = re.sub(r'MESHCORE_PORT = "[^"]*"', f'MESHCORE_PORT = "{meshcore}"', new, count=1)
if meshtastic:
new = re.sub(r'MESHTASTIC_PORT = "[^"]*"', f'MESHTASTIC_PORT = "{meshtastic}"', new, count=1)
if new == src:
return "ports already set / no change"
if not yes:
return f"would write MeshCore={meshcore}, Meshtastic={meshtastic} (use --yes)"
open(gw, "w").write(new)
return f"wrote MeshCore={meshcore}, Meshtastic={meshtastic}; run: sudo systemctl restart mesh-gateway.service"
def main():
ap = argparse.ArgumentParser(description="Mesh relay setup/validator.")
ap.add_argument("action", choices=["scan", "apply", "status", "help"])
ap.add_argument("--yes", action="store_true")
a = ap.parse_args()
if a.action == "help":
ap.print_help(); return 0
if a.action == "scan":
print("Scanning serial devices...")
r = discover()
for line in r["report"]:
print(" " + line)
print("\nResult:")
print(f" MeshCore : {r['meshcore'] or 'NOT FOUND'}")
print(f" Meshtastic : {r['meshtastic'] or 'NOT FOUND'}")
if not (r["meshcore"] and r["meshtastic"]):
print("\n One or both radios not found. Stop the gateway first")
print(" (sudo systemctl stop mesh-gateway.service) then rescan.")
return 1
return 0
if a.action == "apply":
r = discover()
if not (r["meshcore"] and r["meshtastic"]):
print("Cannot apply: one or both radios not found."); return 1
print(apply_ports(r["meshcore"], r["meshtastic"], yes=a.yes))
return 0
if a.action == "status":
for line in status_report():
print(" " + line)
return 0
return 0
if __name__ == "__main__":
sys.exit(main())
