From 72a01500060c49248d00f1c4a54d4675bf55f12e Mon Sep 17 00:00:00 2001 From: l5y <220195275+l5yth@users.noreply.github.com> Date: Sun, 14 Sep 2025 11:37:56 +0200 Subject: [PATCH] data: enable serial collection of messages on channel 0 (#25) * data: enable serial collection of messages on channel 0 * remove tests --- data/mesh.py | 187 +++++++++++++++++++++++++++++++ data/{nodes.sh => mesh.sh} | 2 +- data/messages.sql | 20 ++++ data/nodes.py | 103 ----------------- test/test.sh | 5 - test/test_nodes_serialization.py | 65 ----------- test/test_web_app.py | 127 --------------------- web/app.rb | 4 +- web/public/index.html | 10 +- 9 files changed, 218 insertions(+), 305 deletions(-) create mode 100644 data/mesh.py rename data/{nodes.sh => mesh.sh} (85%) create mode 100644 data/messages.sql delete mode 100644 data/nodes.py delete mode 100755 test/test.sh delete mode 100644 test/test_nodes_serialization.py delete mode 100644 test/test_web_app.py diff --git a/data/mesh.py b/data/mesh.py new file mode 100644 index 0000000..1d9a79d --- /dev/null +++ b/data/mesh.py @@ -0,0 +1,187 @@ +#!/usr/bin/env python3 +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 + +# --- 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 + +# --- DB setup ----------------------------------------------------------------- +nodeSchema = Path(__file__).with_name("nodes.sql").read_text() +conn = sqlite3.connect(DB, check_same_thread=False) +conn.executescript(nodeSchema) +msgSchema = Path(__file__).with_name("messages.sql").read_text() +conn.executescript(msgSchema) +conn.commit() + +DB_LOCK = threading.Lock() + +def _get(obj, key, default=None): + """Return value for key/attribute from dicts or objects.""" + if isinstance(obj, dict): + return obj.get(key, default) + return getattr(obj, key, default) + +# --- Node upsert -------------------------------------------------------------- +def upsert_node(node_id, n): + user = _get(n, "user") or {} + met = _get(n, "deviceMetrics") or {} + pos = _get(n, "position") or {} + lh = _get(n, "lastHeard") + row = ( + node_id, + _get(n, "num"), + _get(user, "shortName"), + _get(user, "longName"), + _get(user, "macaddr"), + _get(user, "hwModel") or _get(n, "hwModel"), + _get(user, "role"), + _get(user, "publicKey"), + _get(user, "isUnmessagable"), + _get(n, "isFavorite"), + _get(n, "hopsAway"), + _get(n, "snr"), + lh, + lh, + _get(met, "batteryLevel"), + _get(met, "voltage"), + _get(met, "channelUtilization"), + _get(met, "airUtilTx"), + _get(met, "uptimeSeconds"), + _get(pos, "time"), + _get(pos, "locationSource"), + _get(pos, "latitude"), + _get(pos, "longitude"), + _get(pos, "altitude"), + ) + with DB_LOCK: + conn.execute( + """ + INSERT INTO nodes(node_id,num,short_name,long_name,macaddr,hw_model,role,public_key,is_unmessagable,is_favorite, + hops_away,snr,last_heard,first_heard,battery_level,voltage,channel_utilization,air_util_tx,uptime_seconds, + position_time,location_source,latitude,longitude,altitude) + VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?) + ON CONFLICT(node_id) DO UPDATE SET + num=excluded.num, short_name=excluded.short_name, long_name=excluded.long_name, macaddr=excluded.macaddr, + hw_model=excluded.hw_model, role=excluded.role, public_key=excluded.public_key, is_unmessagable=excluded.is_unmessagable, + is_favorite=excluded.is_favorite, hops_away=excluded.hops_away, snr=excluded.snr, last_heard=excluded.last_heard, + battery_level=excluded.battery_level, voltage=excluded.voltage, channel_utilization=excluded.channel_utilization, + air_util_tx=excluded.air_util_tx, uptime_seconds=excluded.uptime_seconds, position_time=excluded.position_time, + location_source=excluded.location_source, latitude=excluded.latitude, longitude=excluded.longitude, + altitude=excluded.altitude + """, + 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 ----------------------------- +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")) + if ch is None: + ch = 0 # default to main if radio didn't annotate + try: + ch = int(ch) + except Exception: + ch = 0 + + # if ch != CHANNEL_INDEX: + # return # only log main channel (override via env if needed) + + 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") + + row = ( + rx_time, + _iso(rx_time), + from_id, + to_id, + ch, + str(portnum) if portnum is not None else None, + text, + 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( + """INSERT INTO messages + (rx_time, rx_iso, from_id, to_id, channel, portnum, text, snr, rssi, hop_limit) + VALUES (?,?,?,?,?,?,?,?,?,?)""", + row, + ) + +# --- Main --------------------------------------------------------------------- +def main(): + if SerialInterface is None: + raise RuntimeError("meshtastic library not installed") + + 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() + signal.signal(signal.SIGINT, handle_sig) + signal.signal(signal.SIGTERM, handle_sig) + + print(f"Mesh daemon: nodes+messages → {DB} | port={PORT} | channel={CHANNEL_INDEX}") + while not stop.is_set(): + try: + nodes = getattr(iface, "nodes", {}) or {} + for node_id, n in nodes.items(): + upsert_node(node_id, n) + with DB_LOCK: + conn.commit() + except Exception as e: + print("node snapshot error:", e) + stop.wait(SNAPSHOT_SECS) + + try: + iface.close() + except Exception: + pass + with DB_LOCK: + conn.commit() + conn.close() + +if __name__ == "__main__": + main() diff --git a/data/nodes.sh b/data/mesh.sh similarity index 85% rename from data/nodes.sh rename to data/mesh.sh index 621afda..cd2e7ff 100755 --- a/data/nodes.sh +++ b/data/mesh.sh @@ -2,4 +2,4 @@ python -m venv .venv && source .venv/bin/activate pip install meshtastic -python nodes.py +python mesh.py diff --git a/data/messages.sql b/data/messages.sql new file mode 100644 index 0000000..89916a8 --- /dev/null +++ b/data/messages.sql @@ -0,0 +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 +); +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); diff --git a/data/nodes.py b/data/nodes.py deleted file mode 100644 index 3425894..0000000 --- a/data/nodes.py +++ /dev/null @@ -1,103 +0,0 @@ -import json, os, sqlite3, time, threading -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 - -DB = os.environ.get("MESH_DB", "nodes.db") - -schema = Path(__file__).with_name("nodes.sql").read_text() -conn = sqlite3.connect(DB, check_same_thread=False) -conn.executescript(schema) -conn.commit() - - -def _get(obj, key, default=None): - """Return value for key/attribute from dicts or objects.""" - if isinstance(obj, dict): - return obj.get(key, default) - return getattr(obj, key, default) - - -def upsert_node(node_id, n): - user = _get(n, "user") or {} - met = _get(n, "deviceMetrics") or {} - pos = _get(n, "position") or {} - lh = _get(n, "lastHeard") - row = ( - node_id, - _get(n, "num"), - _get(user, "shortName"), - _get(user, "longName"), - _get(user, "macaddr"), - _get(user, "hwModel") or _get(n, "hwModel"), - _get(user, "role"), - _get(user, "publicKey"), - _get(user, "isUnmessagable"), - _get(n, "isFavorite"), - _get(n, "hopsAway"), - _get(n, "snr"), - lh, - lh, - _get(met, "batteryLevel"), - _get(met, "voltage"), - _get(met, "channelUtilization"), - _get(met, "airUtilTx"), - _get(met, "uptimeSeconds"), - _get(pos, "time"), - _get(pos, "locationSource"), - _get(pos, "latitude"), - _get(pos, "longitude"), - _get(pos, "altitude"), - ) - conn.execute( - """ - INSERT INTO nodes(node_id,num,short_name,long_name,macaddr,hw_model,role,public_key,is_unmessagable,is_favorite, - hops_away,snr,last_heard,first_heard,battery_level,voltage,channel_utilization,air_util_tx,uptime_seconds, - position_time,location_source,latitude,longitude,altitude) - VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?) - ON CONFLICT(node_id) DO UPDATE SET - num=excluded.num, short_name=excluded.short_name, long_name=excluded.long_name, macaddr=excluded.macaddr, - hw_model=excluded.hw_model, role=excluded.role, public_key=excluded.public_key, is_unmessagable=excluded.is_unmessagable, - is_favorite=excluded.is_favorite, hops_away=excluded.hops_away, snr=excluded.snr, last_heard=excluded.last_heard, - battery_level=excluded.battery_level, voltage=excluded.voltage, channel_utilization=excluded.channel_utilization, - air_util_tx=excluded.air_util_tx, uptime_seconds=excluded.uptime_seconds, position_time=excluded.position_time, - location_source=excluded.location_source, latitude=excluded.latitude, longitude=excluded.longitude, - altitude=excluded.altitude - """, - row, - ) - - -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) - conn.commit() - - -def main(): - if SerialInterface is None: - raise RuntimeError("meshtastic library not installed") - iface = SerialInterface( - # or whatever serial interface it is - devPath="/dev/ttyACM0" - ) - print("Nodes ingestor running. Ctrl+C to stop.") - while True: - try: - for node_id, n in (getattr(iface, "nodes", {}) or {}).items(): - upsert_node(node_id, n) - conn.commit() - except Exception as e: - print("node snapshot error:", e) - time.sleep(30) - - -if __name__ == "__main__": - main() diff --git a/test/test.sh b/test/test.sh deleted file mode 100755 index 90a7bad..0000000 --- a/test/test.sh +++ /dev/null @@ -1,5 +0,0 @@ -#!/usr/bin/env bash - -python -m venv .venv && source .venv/bin/activate -pip install pytest -pytest diff --git a/test/test_nodes_serialization.py b/test/test_nodes_serialization.py deleted file mode 100644 index 785defa..0000000 --- a/test/test_nodes_serialization.py +++ /dev/null @@ -1,65 +0,0 @@ -import os -import sqlite3 -import sys -import types -from dataclasses import dataclass -from pathlib import Path - - -def test_upsert_node_handles_position(tmp_path): - data_dir = Path(__file__).resolve().parent.parent / "data" - cwd = os.getcwd() - os.chdir(data_dir) - try: - # Provide minimal stubs for the meshtastic modules imported by nodes.py - meshtastic = types.ModuleType("meshtastic") - serial_module = types.ModuleType("serial_interface") - serial_module.SerialInterface = object - mesh_module = types.ModuleType("mesh_interface") - mesh_module.MeshInterface = object - meshtastic.serial_interface = serial_module - meshtastic.mesh_interface = mesh_module - sys.modules.setdefault("meshtastic", meshtastic) - sys.modules.setdefault("meshtastic.serial_interface", serial_module) - sys.modules.setdefault("meshtastic.mesh_interface", mesh_module) - - sys.path.insert(0, str(data_dir)) - import nodes - # Close original on-disk connection and use in-memory DB for testing - nodes.conn.close() - dbfile = Path("nodes.db") - if dbfile.exists(): - dbfile.unlink() - nodes.conn = sqlite3.connect(":memory:", check_same_thread=False) - nodes.conn.executescript(nodes.schema) - nodes.conn.commit() - - @dataclass - class Position: - time: int = 123 - locationSource: str = "GPS" - latitude: float = 52.5 - longitude: float = 13.4 - altitude: float = 34.0 - - n = {"num": 7, "position": Position(), "lastHeard": 100} - nodes.upsert_node("node1", n) - nodes.conn.commit() - row = nodes.conn.execute( - "SELECT latitude, first_heard, last_heard FROM nodes WHERE node_id=?", - ("node1",), - ).fetchone() - assert row is not None - assert row[0] == 52.5 - assert row[1] == 100 - assert row[2] == 100 - - n["lastHeard"] = 200 - nodes.upsert_node("node1", n) - nodes.conn.commit() - row2 = nodes.conn.execute( - "SELECT first_heard, last_heard FROM nodes WHERE node_id=?", ("node1",) - ).fetchone() - assert row2 == (100, 200) - finally: - os.chdir(cwd) diff --git a/test/test_web_app.py b/test/test_web_app.py deleted file mode 100644 index 6560c37..0000000 --- a/test/test_web_app.py +++ /dev/null @@ -1,127 +0,0 @@ -import json -import os -import subprocess -import sys -from pathlib import Path -import time -import sqlite3 - -import pytest - - -def test_query_nodes_from_web_app(tmp_path): - db_path = tmp_path / "nodes.db" - os.environ["MESH_DB"] = str(db_path) - - # import nodes module after setting env var - sys.path.insert(0, str(Path(__file__).resolve().parents[1])) - from data import nodes # type: ignore - - nodes.load_nodes_from_file(Path(__file__).with_name("nodes.json")) - - web_dir = Path(__file__).resolve().parents[1] / "web" - try: - subprocess.run(["bundle", "install"], cwd=web_dir, check=True, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) - env = os.environ.copy() - env["MESH_DB"] = str(db_path) - out = subprocess.check_output( - [ - "bundle", - "exec", - "ruby", - "-e", - 'require_relative "app"; require "json"; puts query_nodes(1000).to_json' - ], - cwd=web_dir, - env=env, - stderr=subprocess.DEVNULL, - ) - except subprocess.CalledProcessError: - pytest.skip("ruby dependencies not installed") - - data = json.loads(out) - conn = sqlite3.connect(db_path) - threshold = int(time.time()) - 7 * 24 * 60 * 60 - expected = conn.execute("SELECT COUNT(*) FROM nodes WHERE last_heard >= ?", (threshold,)).fetchone()[0] - old_count = conn.execute("SELECT COUNT(*) FROM nodes WHERE last_heard < ?", (threshold,)).fetchone()[0] - conn.close() - assert old_count > 0 - assert len(data) == expected - last_heards = [item["last_heard"] for item in data] - assert last_heards == sorted(last_heards, reverse=True) - assert all(lh is None or lh >= threshold for lh in last_heards) - - -def test_post_nodes_to_web_app(tmp_path): - db_path = tmp_path / "nodes.db" - os.environ["MESH_DB"] = str(db_path) - os.environ["API_TOKEN"] = "secrettoken" - - web_dir = Path(__file__).resolve().parents[1] / "web" - nodes_json = Path(__file__).with_name("nodes.json") - try: - subprocess.run(["bundle", "install"], cwd=web_dir, check=True, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) - env = os.environ.copy() - env["MESH_DB"] = str(db_path) - env["API_TOKEN"] = "secrettoken" - ruby = ( - "require_relative 'app'; require 'json'; require 'rack/mock'; require 'sqlite3';"\ - f"nodes = File.read({json.dumps(str(nodes_json))});"\ - "req = Rack::MockRequest.new(Sinatra::Application);"\ - "res = req.post('/api/nodes', 'CONTENT_TYPE' => 'application/json', 'HTTP_AUTHORIZATION' => 'Bearer secrettoken', :input => nodes);"\ - "puts res.status;"\ - "db = SQLite3::Database.new(ENV['MESH_DB']);"\ - "puts db.get_first_value('SELECT COUNT(*) FROM nodes');" - ) - out = subprocess.check_output( - ["bundle", "exec", "ruby", "-e", ruby], - cwd=web_dir, - env=env, - stderr=subprocess.DEVNULL, - ) - except subprocess.CalledProcessError: - pytest.skip("ruby dependencies not installed") - - lines = out.decode().strip().splitlines() - assert lines[0] == "200" - expected = len(json.load(open(nodes_json))) - assert int(lines[1]) == expected - - -def test_null_role_defaults_to_client(tmp_path): - db_path = tmp_path / "nodes.db" - os.environ["MESH_DB"] = str(db_path) - os.environ["API_TOKEN"] = "tok" - - web_dir = Path(__file__).resolve().parents[1] / "web" - node = { - "nodeA": { - "num": 1, - "lastHeard": int(time.time()), - "user": {"shortName": "Foo"}, - } - } - nodes_json = json.dumps(node) - try: - subprocess.run(["bundle", "install"], cwd=web_dir, check=True, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) - env = os.environ.copy() - env["MESH_DB"] = str(db_path) - env["API_TOKEN"] = "tok" - ruby = ( - "require_relative 'app'; require 'json'; require 'rack/mock';" - f"nodes = {json.dumps(nodes_json)};" - "req = Rack::MockRequest.new(Sinatra::Application);" - "req.post('/api/nodes', 'CONTENT_TYPE' => 'application/json', 'HTTP_AUTHORIZATION' => 'Bearer tok', :input => nodes);" - "puts query_nodes(1000).to_json;" - ) - out = subprocess.check_output(["bundle", "exec", "ruby", "-e", ruby], cwd=web_dir, env=env, stderr=subprocess.DEVNULL) - except subprocess.CalledProcessError: - pytest.skip("ruby dependencies not installed") - - data = json.loads(out.decode().splitlines()[-1]) - assert any(n["role"] == "CLIENT" for n in data if n["node_id"] == "nodeA") - - conn = sqlite3.connect(db_path) - role = conn.execute("SELECT role FROM nodes WHERE node_id=?", ("nodeA",)).fetchone()[0] - conn.close() - assert role == "CLIENT" diff --git a/web/app.rb b/web/app.rb index 7077f07..f4b5679 100644 --- a/web/app.rb +++ b/web/app.rb @@ -3,8 +3,8 @@ require "sinatra" require "json" require "sqlite3" -# run ../data/nodes.sh to nodespopulate nodes database -DB_PATH = ENV.fetch("MESH_DB", File.join(__dir__, "../data/nodes.db")) +# run ../data/mesh.sh to populate nodes and messages database +DB_PATH = ENV.fetch("MESH_DB", File.join(__dir__, "../data/mesh.db")) set :public_folder, File.join(__dir__, "public") diff --git a/web/public/index.html b/web/public/index.html index 2fa397c..b2ce9ee 100644 --- a/web/public/index.html +++ b/web/public/index.html @@ -161,6 +161,12 @@ pad(d.getSeconds()); } + function fmtHw(v) { + if (v == null) return ""; + if (v == "UNSET") return ""; + return String(v); + } + function fmtCoords(v, d = 5) { if (v == null) return ""; const n = Number(v); @@ -218,7 +224,7 @@ ${n.long_name || ""} ${timeAgo(n.last_heard)} ${n.role || "CLIENT"} - ${n.hw_model || ""} + ${fmtHw(n.hw_model)} ${fmtAlt(n.battery_level, "%")} ${fmtAlt(n.voltage, "V")} ${timeHum(n.uptime_seconds)} @@ -252,7 +258,7 @@ const lines = [ `${n.long_name || ''}`, `${n.short_name || ''} ${n.node_id || ''}`, - n.hw_model ? `Model: ${n.hw_model}` : null, + n.hw_model ? `Model: ${fmtHw(n.hw_model)}` : null, `Role: ${n.role || 'CLIENT'}`, (n.battery_level != null ? `Battery: ${fmtAlt(n.battery_level, "%")}, ${fmtAlt(n.voltage, "V")}` : null), (n.last_heard ? `Last seen: ${timeAgo(n.last_heard)}` : null),