adam/potato-mesh
1
0
Fork 0
mirror of https://github.com/l5yth/potato-mesh.git synced 2026-07-06 09:52:02 +02:00
Code Issues Packages Projects Releases Wiki Activity

switch to pubsub for message parsing

Browse Source
This commit is contained in:
l5y
2025-09-14 13:35:58 +02:00
parent 72a0150006
commit c52b917f84
3 changed files with 195 additions and 64 deletions
Download Patch File Download Diff File Expand all files Collapse all files
data/mesh.py
+75 -46
View File
@@ -2,18 +2,17 @@
import json, os, sqlite3, time, threading, signal
from pathlib import Path
try: # meshtastic is optional for tests
from meshtastic.serial_interface import SerialInterface
from meshtastic.mesh_interface import MeshInterface
except ModuleNotFoundError: # pragma: no cover - imported lazily for hardware usage
SerialInterface = None # type: ignore
MeshInterface = None # type: ignore
from meshtastic.serial_interface import SerialInterface
from meshtastic.mesh_interface import MeshInterface
from pubsub import pub
from google.protobuf.json_format import MessageToDict
from google.protobuf.message import Message as ProtoMessage
# --- Config (env overrides) ---------------------------------------------------
DB = os.environ.get("MESH_DB", "mesh.db")
PORT = os.environ.get("MESH_SERIAL", "/dev/ttyACM0")
SNAPSHOT_SECS = int(os.environ.get("MESH_SNAPSHOT_SECS", "30"))
CHANNEL_INDEX = int(os.environ.get("MESH_CHANNEL_INDEX", "0")) # main #MediumFast
CHANNEL_INDEX = int(os.environ.get("MESH_CHANNEL_INDEX", "0"))
# --- DB setup -----------------------------------------------------------------
nodeSchema = Path(__file__).with_name("nodes.sql").read_text()
@@ -82,42 +81,71 @@ def upsert_node(node_id, n):
row,
)
# --- Nodes.json loader (unchanged) -------------------------------------------
def load_nodes_from_file(path: str | Path):
"""Populate the database from a nodes.json file."""
nodes = json.loads(Path(path).read_text())
for node_id, node in nodes.items():
upsert_node(node_id, node)
with DB_LOCK:
conn.commit()
# --- Message logging via the same SerialInterface -----------------------------
# --- Message logging via PubSub -----------------------------------------------
def _iso(ts: int | float) -> str:
import datetime
return datetime.datetime.utcfromtimestamp(int(ts)).isoformat() + "Z"
def store_packet(packet: dict):
"""Store a received packet into messages table (filtered by channel)."""
dec = packet.get("decoded") or {}
ch = dec.get("channel", packet.get("channel"))
def _first(d: dict, *names, default=None):
"""Return first present key from names (supports nested 'a.b' lookups)."""
for name in names:
cur = d
parts = name.split(".")
ok = True
for p in parts:
if isinstance(cur, dict) and p in cur:
cur = cur[p]
else:
ok = False
break
if ok:
return cur
return default
def _pkt_to_dict(packet) -> dict:
"""Convert protobuf MeshPacket or already-dict into a JSON-friendly dict."""
if isinstance(packet, dict):
return packet
if isinstance(packet, ProtoMessage):
return MessageToDict(packet, preserving_proto_field_name=True, use_integers_for_enums=False)
# Last resort: try to read attributes
try:
return json.loads(json.dumps(packet, default=lambda o: str(o)))
except Exception:
return {"_unparsed": str(packet)}
def store_packet_dict(p: dict):
"""
Store only TEXT messages (decoded.payload.text) to the DB.
Safe against snake/camel case differences.
"""
dec = p.get("decoded") or {}
text = _first(dec, "payload.text", "text", default=None)
if not text:
return # ignore non-text packets
# port filter (optional): ensure it's TEXT_MESSAGE_APP if present
portnum = _first(dec, "portnum", default=None)
# If you want to enforce: if portnum and portnum != "TEXT_MESSAGE_APP": return
# channel (prefer decoded.channel if present; else top-level)
ch = _first(dec, "channel", default=None)
if ch is None:
ch = 0 # default to main if radio didn't annotate
ch = _first(p, "channel", default=0)
try:
ch = int(ch)
except Exception:
ch = 0
# if ch != CHANNEL_INDEX:
# return # only log main channel (override via env if needed)
# timestamps & ids
rx_time = int(_first(p, "rxTime", "rx_time", default=time.time()))
from_id = _first(p, "fromId", "from_id", "from", default=None)
to_id = _first(p, "toId", "to_id", "to", default=None)
rx_time = int(packet.get("rxTime") or time.time())
from_id = packet.get("fromId")
to_id = packet.get("toId")
portnum = dec.get("portnum") # can be enum name or numeric
text = dec.get("text")
snr = packet.get("snr")
rssi = packet.get("rssi")
hop = packet.get("hopLimit") or packet.get("hop_limit")
# link metrics
snr = _first(p, "snr", "rx_snr", "rxSnr", default=None)
rssi = _first(p, "rssi", "rx_rssi", "rxRssi", default=None)
hop = _first(p, "hopLimit", "hop_limit", default=None)
row = (
rx_time,
@@ -130,7 +158,6 @@ def store_packet(packet: dict):
float(snr) if snr is not None else None,
int(rssi) if rssi is not None else None,
int(hop) if hop is not None else None,
# json.dumps(packet, ensure_ascii=False),
)
with DB_LOCK:
conn.execute(
@@ -139,24 +166,26 @@ def store_packet(packet: dict):
VALUES (?,?,?,?,?,?,?,?,?,?)""",
row,
)
conn.commit()
# PubSub receive handler
def on_receive(packet, interface):
try:
p = _pkt_to_dict(packet)
store_packet_dict(p)
except Exception as e:
try:
print("[warn] failed to store packet:", e, "| keys:", list(p.keys()) if isinstance(p, dict) else type(p))
except Exception:
print("[warn] failed to store packet:", e)
# --- Main ---------------------------------------------------------------------
def main():
if SerialInterface is None:
raise RuntimeError("meshtastic library not installed")
# Subscribe to PubSub topics (reliable in current meshtastic)
pub.subscribe(on_receive, "meshtastic.receive")
iface = SerialInterface(devPath=PORT)
# Packet callback runs in iface reader thread
def on_receive(packet, _interface):
try:
store_packet(packet)
except Exception as e:
# Keep daemon resilient
print(f"[warn] failed to store packet: {e}")
iface.onReceive = on_receive
stop = threading.Event()
def handle_sig(*_):
stop.set()
@@ -172,7 +201,7 @@ def main():
with DB_LOCK:
conn.commit()
except Exception as e:
print("node snapshot error:", e)
print(f"[warn] failed to update node snapshot: {e}")
stop.wait(SNAPSHOT_SECS)
try:
data/messages.sql
+18 -18
View File
@@ -1,20 +1,20 @@
PRAGMA journal_mode=WAL;
PRAGMA synchronous=NORMAL;
CREATE TABLE IF NOT EXISTS messages (
id INTEGER PRIMARY KEY AUTOINCREMENT,
rx_time INTEGER, -- unix seconds
rx_iso TEXT,
from_id TEXT,
to_id TEXT,
channel INTEGER,
portnum TEXT,
text TEXT,
snr REAL,
rssi INTEGER,
hop_limit INTEGER,
packet_json TEXT
id INTEGER PRIMARY KEY AUTOINCREMENT,
rx_time INTEGER NOT NULL, -- unix seconds when received
rx_iso TEXT NOT NULL, -- ISO8601 UTC timestamp
from_id TEXT, -- sender node id (string form)
to_id TEXT, -- recipient node id
channel INTEGER, -- channel index
portnum TEXT, -- application portnum (e.g. TEXT_MESSAGE_APP)
text TEXT, -- decoded text payload if present
snr REAL, -- signal-to-noise ratio
rssi INTEGER, -- received signal strength
hop_limit INTEGER, -- hops left when received
raw_json TEXT -- entire packet JSON dump
);
CREATE INDEX IF NOT EXISTS idx_messages_rx_time ON messages(rx_time);
CREATE INDEX IF NOT EXISTS idx_messages_channel ON messages(channel);
CREATE INDEX IF NOT EXISTS idx_messages_portnum ON messages(portnum);
CREATE INDEX IF NOT EXISTS idx_messages_rx_time ON messages(rx_time);
CREATE INDEX IF NOT EXISTS idx_messages_from_id ON messages(from_id);
CREATE INDEX IF NOT EXISTS idx_messages_to_id ON messages(to_id);
CREATE INDEX IF NOT EXISTS idx_messages_channel ON messages(channel);
CREATE INDEX IF NOT EXISTS idx_messages_portnum ON messages(portnum);
test/debug.py
+102
View File
@@ -0,0 +1,102 @@
#!/usr/bin/env python3
import time, json, base64, threading
from pubsub import pub # comes with meshtastic
from meshtastic.serial_interface import SerialInterface
from google.protobuf.json_format import MessageToDict
from google.protobuf.message import Message as ProtoMessage
PORT = "/dev/ttyACM0"
packet_count = 0
last_rx_ts = None
stop = threading.Event()
def to_jsonable(obj):
"""Recursively convert protobuf/bytes/etc. into JSON-serializable structures."""
if obj is None:
return None
if isinstance(obj, ProtoMessage):
# Convert protobuf to dict; bytes become base64 by default
return MessageToDict(obj, preserving_proto_field_name=True, use_integers_for_enums=False)
if isinstance(obj, bytes):
return {"__bytes_b64__": base64.b64encode(obj).decode("ascii")}
if isinstance(obj, (str, int, float, bool)):
return obj
if isinstance(obj, dict):
return {str(k): to_jsonable(v) for k, v in obj.items()}
if isinstance(obj, (list, tuple)):
return [to_jsonable(v) for v in obj]
# fallback
return str(obj)
def extract_text(d):
"""Best-effort pull of decoded text from a dict produced by to_jsonable()."""
dec = d.get("decoded") or {}
# Text packets usually at decoded.payload.text
payload = dec.get("payload") or {}
if isinstance(payload, dict) and "text" in payload:
return payload.get("text")
# Some versions flatten 'text' at decoded.text
if "text" in dec:
return dec.get("text")
return None
def on_receive(packet, interface):
global packet_count, last_rx_ts
packet_count += 1
last_rx_ts = time.time()
d = to_jsonable(packet)
text = extract_text(d)
frm = d.get("from") or d.get("from_id") or d.get("fromId")
to = d.get("to") or d.get("to_id") or d.get("toId")
portnum = (d.get("decoded") or {}).get("portnum")
print(f"\n=== PACKET #{packet_count} RECEIVED ===")
if text:
print(f"[summary] {frm} → {to} port={portnum} text={text!r}")
else:
print(f"[summary] {frm} → {to} port={portnum} (no text)")
try:
print(json.dumps(d, indent=2, ensure_ascii=False))
except Exception as e:
# Shouldn't happen after to_jsonable, but keep a guard
print("[warn] JSON dump failed even after conversion:", e)
def on_connected(interface, *args, **kwargs):
print("[info] connection established")
def on_disconnected(interface, *args, **kwargs):
print("[info] disconnected")
def main():
print(f"Opening Meshtastic on {PORT} …")
# Use PubSub topics (reliable in current meshtastic)
pub.subscribe(on_receive, "meshtastic.receive")
pub.subscribe(on_connected, "meshtastic.connection.established")
pub.subscribe(on_disconnected, "meshtastic.connection.lost")
iface = SerialInterface(devPath=PORT)
try:
last_heartbeat = time.time()
while not stop.is_set():
time.sleep(0.5)
now = time.time()
if now - last_heartbeat >= 5:
since = "never" if last_rx_ts is None else f"{int(now - last_rx_ts)}s ago"
print(f"[heartbeat] alive; packets={packet_count} (last packet {since})")
last_heartbeat = now
except KeyboardInterrupt:
pass
finally:
try:
iface.close()
except Exception:
pass
print("\nExiting.")
if __name__ == "__main__":
main()