diff --git a/src/meshchat/database/schema.py b/src/meshchat/database/schema.py index 383a4a3..99920e8 100644 --- a/src/meshchat/database/schema.py +++ b/src/meshchat/database/schema.py @@ -64,7 +64,21 @@ first_seen TEXT, last_heard TEXT, packet_count INTEGER DEFAULT 0, - text_count INTEGER DEFAULT 0 + text_count INTEGER DEFAULT 0, + -- Signal/hop/transport state of the most recently heard packet, plus + -- session-lifetime RF/MQTT/position/telemetry counters — previously + -- tracked only on the in-memory NodeSnapshot, so a node the app had + -- tracked for weeks started every restart with these reset to + -- zero/unknown until it was heard again live. See MonitorStore._write_node. + last_snr REAL, + last_rssi INTEGER, + last_hops_used INTEGER, + last_hop_start INTEGER, + last_via_mqtt INTEGER, + rf_count INTEGER DEFAULT 0, + via_mqtt_count INTEGER DEFAULT 0, + position_count INTEGER DEFAULT 0, + telemetry_count INTEGER DEFAULT 0 ); CREATE TABLE IF NOT EXISTS positions ( @@ -155,3 +169,18 @@ def _migrate(conn) -> None: conn.execute("ALTER TABLE messages ADD COLUMN text TEXT") if "destination_num" not in cols: conn.execute("ALTER TABLE messages ADD COLUMN destination_num INTEGER") + + node_cols = {row[1] for row in conn.execute("PRAGMA table_info(nodes)").fetchall()} + for col, decl in ( + ("last_snr", "REAL"), + ("last_rssi", "INTEGER"), + ("last_hops_used", "INTEGER"), + ("last_hop_start", "INTEGER"), + ("last_via_mqtt", "INTEGER"), + ("rf_count", "INTEGER DEFAULT 0"), + ("via_mqtt_count", "INTEGER DEFAULT 0"), + ("position_count", "INTEGER DEFAULT 0"), + ("telemetry_count", "INTEGER DEFAULT 0"), + ): + if col not in node_cols: + conn.execute(f"ALTER TABLE nodes ADD COLUMN {col} {decl}") diff --git a/src/meshchat/services/export_service.py b/src/meshchat/services/export_service.py index 6c98c7a..ea0e996 100644 --- a/src/meshchat/services/export_service.py +++ b/src/meshchat/services/export_service.py @@ -63,7 +63,13 @@ def export_packets_csv( "hop_start": pkt.hop_start if pkt.hop_start is not None else "", "hop_limit": pkt.hop_limit if pkt.hop_limit is not None else "", "hops_used": pkt.hops_used if pkt.hops_used is not None else "", - "via_mqtt": int(bool(pkt.via_mqtt)), + # is-not-None here too: via_mqtt is bool | None on older + # firmware that never reports the field at all, and + # `int(bool(None))` == 0 was indistinguishable from a + # packet confirmed to be direct RF — silently collapsing + # "unknown" into "confirmed not MQTT" for anyone doing + # offline RF-vs-MQTT analysis of the export. + "via_mqtt": int(pkt.via_mqtt) if pkt.via_mqtt is not None else "", "transport_mechanism": pkt.transport_mechanism or "", } if include_text: diff --git a/src/meshchat/services/monitor_store.py b/src/meshchat/services/monitor_store.py index bd1686c..f58d8a6 100644 --- a/src/meshchat/services/monitor_store.py +++ b/src/meshchat/services/monitor_store.py @@ -38,6 +38,15 @@ def _dt_str(dt: datetime | None) -> str | None: return dt.isoformat() +def _parse_dt(iso_str: str | None) -> datetime | None: + if not iso_str: + return None + try: + return datetime.fromisoformat(iso_str) + except ValueError: + return None + + class MonitorStore: """ SQLite persistence for the Network Monitor. @@ -128,6 +137,50 @@ def read_packets( log.error("MonitorStore.read_packets failed: %s", exc) return [] + def read_packets_as_objects(self, session_id: str, limit: int = 200_000) -> list[NetworkPacket]: + """Like read_packets(), reconstructed as NetworkPacket objects. + + Used for exporting a session's full packet history — beyond what + PacketIngestor's bounded in-memory ring buffer retains. `text` and + `raw_metadata_json` are always None here: message text isn't stored + in the packets table at all (it lives in `messages`, keyed by + chat-relevant fields the packets table doesn't carry), so callers + that need it must merge it back in themselves from whatever + still-in-memory packets happen to overlap. + """ + rows = self.read_packets(session_id, limit=limit) + packets = [] + for r in rows: + via_mqtt = r.get("via_mqtt") + pki_encrypted = r.get("pki_encrypted") + want_ack = r.get("want_ack") + packets.append(NetworkPacket( + session_id=r["session_id"], + observed_at=_parse_dt(r.get("observed_at")) or datetime.now(timezone.utc), + rx_time=_parse_dt(r.get("rx_time")), + sender_num=r.get("sender_num"), + sender_id=r.get("sender_id"), + destination_num=r.get("destination_num"), + packet_id=r.get("packet_id"), + channel_index=r.get("channel_index"), + portnum=r.get("portnum"), + portnum_name=r.get("portnum_name") or "", + text=None, + payload_size=r.get("payload_size"), + rx_snr=r.get("rx_snr"), + rx_rssi=r.get("rx_rssi"), + hop_start=r.get("hop_start"), + hop_limit=r.get("hop_limit"), + hops_used=r.get("hops_used"), + via_mqtt=bool(via_mqtt) if via_mqtt is not None else None, + transport_mechanism=r.get("transport_mechanism"), + pki_encrypted=bool(pki_encrypted) if pki_encrypted is not None else None, + want_ack=bool(want_ack) if want_ack is not None else None, + priority=r.get("priority"), + raw_metadata_json=None, + )) + return packets + def packet_count(self, session_id: str) -> int: try: conn = self._read_conn() @@ -414,8 +467,10 @@ def _write_node(self, conn: sqlite3.Connection, node: NodeSnapshot) -> None: conn.execute( """INSERT INTO nodes (node_num, node_id, long_name, short_name, role, hw_model, - first_seen, last_heard, packet_count, text_count) - VALUES (?,?,?,?,?,?,?,?,?,?) + first_seen, last_heard, packet_count, text_count, + last_snr, last_rssi, last_hops_used, last_hop_start, last_via_mqtt, + rf_count, via_mqtt_count, position_count, telemetry_count) + VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?) ON CONFLICT(node_num) DO UPDATE SET node_id=excluded.node_id, long_name=COALESCE(excluded.long_name, long_name), @@ -436,12 +491,41 @@ def _write_node(self, conn: sqlite3.Connection, node: NodeSnapshot) -> None: -- fields default to 0, never None), so a bare MAX is fine. last_heard=COALESCE(MAX(excluded.last_heard, last_heard), excluded.last_heard, last_heard), packet_count=MAX(excluded.packet_count, packet_count), - text_count=MAX(excluded.text_count, text_count)""", + text_count=MAX(excluded.text_count, text_count), + -- The last_* fields describe the single most recent packet, + -- not a running total — MAX() would be wrong here (e.g. a + -- fresher 0-hop packet must not lose to a staler 3-hop one + -- just because 3 > 0). Take them together only when this + -- write's last_heard is at least as new as what's stored, + -- so they always describe the same observation last_heard + -- itself was just updated to. + last_snr = CASE WHEN excluded.last_heard IS NOT NULL + AND (last_heard IS NULL OR excluded.last_heard >= last_heard) + THEN excluded.last_snr ELSE last_snr END, + last_rssi = CASE WHEN excluded.last_heard IS NOT NULL + AND (last_heard IS NULL OR excluded.last_heard >= last_heard) + THEN excluded.last_rssi ELSE last_rssi END, + last_hops_used = CASE WHEN excluded.last_heard IS NOT NULL + AND (last_heard IS NULL OR excluded.last_heard >= last_heard) + THEN excluded.last_hops_used ELSE last_hops_used END, + last_hop_start = CASE WHEN excluded.last_heard IS NOT NULL + AND (last_heard IS NULL OR excluded.last_heard >= last_heard) + THEN excluded.last_hop_start ELSE last_hop_start END, + last_via_mqtt = CASE WHEN excluded.last_heard IS NOT NULL + AND (last_heard IS NULL OR excluded.last_heard >= last_heard) + THEN excluded.last_via_mqtt ELSE last_via_mqtt END, + rf_count=MAX(excluded.rf_count, rf_count), + via_mqtt_count=MAX(excluded.via_mqtt_count, via_mqtt_count), + position_count=MAX(excluded.position_count, position_count), + telemetry_count=MAX(excluded.telemetry_count, telemetry_count)""", ( node.node_num, node.node_id, node.long_name, node.short_name, node.role, node.hw_model, _dt_str(node.first_seen), _dt_str(node.last_heard), node.packet_count, node.text_count, + node.last_snr, node.last_rssi, node.last_hops_used, node.last_hop_start, + int(node.last_via_mqtt) if node.last_via_mqtt is not None else None, + node.rf_count, node.via_mqtt_count, node.position_count, node.telemetry_count, ), ) diff --git a/src/meshchat/services/packet_ingestor.py b/src/meshchat/services/packet_ingestor.py index bc6133a..6f2c772 100644 --- a/src/meshchat/services/packet_ingestor.py +++ b/src/meshchat/services/packet_ingestor.py @@ -349,9 +349,15 @@ def _update_node(self, pkt: NetworkPacket, raw: dict) -> None: node.last_hop_start = pkt.hop_start node.last_via_mqtt = pkt.via_mqtt - if pkt.is_via_mqtt: + # Tri-state, not pkt.is_via_mqtt (bool(via_mqtt), which coerces + # None to False): older firmware that never reports viaMqtt at + # all must not be counted as confirmed RF — these counters now + # persist across restarts (see MonitorStore._write_node / + # PacketIngestor.seed_from_store), so misclassifying "unknown" + # as "confirmed RF" here would keep compounding every session. + if pkt.via_mqtt is True: node.via_mqtt_count += 1 - else: + elif pkt.via_mqtt is False: node.rf_count += 1 if pkt.portnum == PORTNUM_TEXT: @@ -635,6 +641,28 @@ def seed_from_store(self, node_rows: list[dict], positions_by_num: dict[int, dic node.last_heard = node.last_heard or _parse_dt(row.get("last_heard")) node.packet_count = max(node.packet_count, row.get("packet_count") or 0) node.text_count = max(node.text_count, row.get("text_count") or 0) + node.rf_count = max(node.rf_count, row.get("rf_count") or 0) + node.via_mqtt_count = max(node.via_mqtt_count, row.get("via_mqtt_count") or 0) + node.position_count = max(node.position_count, row.get("position_count") or 0) + node.telemetry_count = max(node.telemetry_count, row.get("telemetry_count") or 0) + # Only restore the "last observation" fields if this node + # hasn't already been heard live this session — a fresh live + # packet is always newer than whatever was persisted before + # this app run started, and seed_from_store only fills in + # nodes not yet touched this session anyway (see the + # `node is None` branch above), but a defensive guard here + # keeps that true even if seeding order ever changes. + if node.last_snr is None: + node.last_snr = row.get("last_snr") + if node.last_rssi is None: + node.last_rssi = row.get("last_rssi") + if node.last_hops_used is None: + node.last_hops_used = row.get("last_hops_used") + if node.last_hop_start is None: + node.last_hop_start = row.get("last_hop_start") + if node.last_via_mqtt is None: + raw_via_mqtt = row.get("last_via_mqtt") + node.last_via_mqtt = bool(raw_via_mqtt) if raw_via_mqtt is not None else None snap = _copy_node_snapshot(node) self.node_updated.emit(snap) diff --git a/src/meshchat/ui/main_window.py b/src/meshchat/ui/main_window.py index b515e2f..83bde78 100644 --- a/src/meshchat/ui/main_window.py +++ b/src/meshchat/ui/main_window.py @@ -7,7 +7,7 @@ from pathlib import Path import platformdirs -from PySide6.QtCore import Qt, QSettings, QTimer +from PySide6.QtCore import QObject, Qt, QSettings, QThread, QTimer, Signal from PySide6.QtGui import QAction from PySide6.QtWidgets import ( QApplication, @@ -42,6 +42,42 @@ _SETTINGS_KEY = "MeshChat/MainWindow" +# ── Packet export worker ────────────────────────────────────────────────── +# A full-session export can be up to read_packets_as_objects()'s 200,000-row +# cap — CSV-formatting and writing that many rows is real, unbounded disk +# I/O that would otherwise block the GUI thread (input, repaints) for as +# long as it takes. Runs on its own QThread; the row list itself was +# already gathered on the GUI thread (a single bounded SQLite read, fast +# enough not to need its own worker) before this is started. + +class _PacketExportWorker(QObject): + finished = Signal(int) # rows written + failed = Signal(str) # error message + + def __init__(self, rows, path: Path, include_text: bool, parent=None): + super().__init__(parent) + self._rows = rows + self._path = path + self._include_text = include_text + + def run(self) -> None: + from meshchat.services.export_service import ExportService + try: + count = ExportService.export_packets_csv( + self._rows, self._path, include_text=self._include_text, + ) + self.finished.emit(count) + except Exception as exc: + # Not narrowed to OSError: any uncaught exception here would + # otherwise leave this slot without ever emitting finished/ + # failed, so the caller's thread.quit() (only wired to those + # two signals) never runs — the worker thread's event loop + # stays up forever and the Export menu action stays disabled + # for the rest of the session. + log.exception("Packet export worker failed") + self.failed.emit(str(exc)) + + # ── Nav rail button ──────────────────────────────────────────────────────── class _NavButton(QPushButton): @@ -66,6 +102,8 @@ def __init__(self): self._session = NetworkSession.new() self._store = MonitorStore() self._ingestor = PacketIngestor(self._session, self._store) + self._export_thread: QThread | None = None + self._export_worker: _PacketExportWorker | None = None # ── Central layout ──────────────────────────────────────────── central = QWidget() @@ -238,9 +276,9 @@ def _build_menu(self) -> None: file_menu = menu_bar.addMenu("File") - export_pkts_act = QAction("Export Packet Log to CSV…", self) - export_pkts_act.triggered.connect(self._export_packets) - file_menu.addAction(export_pkts_act) + self._export_pkts_act = QAction("Export Packet Log to CSV…", self) + self._export_pkts_act.triggered.connect(self._export_packets) + file_menu.addAction(self._export_pkts_act) export_nodes_act = QAction("Export Nodes to CSV…", self) export_nodes_act.triggered.connect(self._export_nodes) @@ -284,13 +322,51 @@ def _prune_old_data(self) -> None: self._store.prune_async() def _export_packets(self) -> None: - from meshchat.services.export_service import ExportService + if self._export_thread is not None: + self._status_bar.showMessage("A packet export is already in progress", 5000) + return + + # PacketIngestor.get_recent_packets() is a bounded 10,000-packet + # in-memory ring buffer — everything still in it is authoritative + # (it's the exact same data ChatView/rankings use, complete and + # with text) and always included as-is. MonitorStore only fills in + # whatever fell out of that buffer on a long/busy session: every + # ingested packet is durably written to the packets table, but that + # write is asynchronous and the table has no text column at all + # (message content lives in `messages`), so store rows are only + # used for packets NOT already covered by the in-memory set — + # never to replace or race against it. + recent = self._ingestor.get_recent_packets() + recent_keys = {(p.sender_num, p.packet_id, p.observed_at) for p in recent} + store_rows = self._store.read_packets_as_objects(self._session.id) + older = [p for p in store_rows if (p.sender_num, p.packet_id, p.observed_at) not in recent_keys] + # read_packets_as_objects() returns newest-first, get_recent_packets() + # returns insertion order — without this the merged CSV would have a + # reverse-chronological historical section followed by a + # chronological recent one. + rows = sorted(older + recent, key=lambda pkt: pkt.observed_at) - rows = self._ingestor.get_recent_packets() if not rows: self._status_bar.showMessage("Nothing to export — no packets captured yet", 5000) return + # read_packets_as_objects() caps at 200,000 rows — warn rather than + # silently truncate on the rare session that exceeds it, instead of + # exporting a file that looks complete but isn't. + total = self._store.packet_count(self._session.id) + if total > len(rows): + proceed = QMessageBox.question( + self, + "Partial Export", + f"This session has captured {total} packets, more than this export " + f"can include at once ({len(rows)} available) — the earliest packets " + "will be missing.\n\nExport the available packets anyway?", + QMessageBox.StandardButton.Yes | QMessageBox.StandardButton.Cancel, + QMessageBox.StandardButton.Cancel, + ) + if proceed != QMessageBox.StandardButton.Yes: + return + path, _ = QFileDialog.getSaveFileName( self, "Export Packet Log", "meshchat-packets.csv", "CSV files (*.csv)" ) @@ -303,17 +379,57 @@ def _export_packets(self) -> None: self, "Include message text?", "Include the text of received messages in the export?\n\n" - "Message content is personal — leave this out if you plan to share the file.", + "Message content is personal — leave this out if you plan to share the file.\n\n" + "Text is only filled in for packets still held in this session's " + "in-memory buffer (the most recent ~10,000) — older rows in this " + "export will have it blank.", QMessageBox.StandardButton.Yes | QMessageBox.StandardButton.No, QMessageBox.StandardButton.No, ) == QMessageBox.StandardButton.Yes - try: - count = ExportService.export_packets_csv(rows, Path(path), include_text=include_text) - self._status_bar.showMessage(f"Exported {count} packet(s) to {path}", 8000) - except OSError as exc: - log.exception("Packet export failed") - QMessageBox.warning(self, "Export Failed", f"Could not write the file:\n{exc}") + # CSV-formatting and writing up to 200,000 rows is real, unbounded + # disk I/O — done on a background QThread so it can't freeze the + # GUI (input, repaints) for as long as it takes on a large session. + # Not parented to `self`/MainWindow: run() is a blocking write, so + # if the window closes before it finishes, closeEvent must be able + # to wait it out without this thread being a child object Qt might + # try to tear down mid-run. + self._status_bar.showMessage(f"Exporting {len(rows)} packet(s)…") + self._export_pkts_act.setEnabled(False) + self._export_thread = QThread() + self._export_worker = _PacketExportWorker(rows, Path(path), include_text) + self._export_worker.moveToThread(self._export_thread) + self._export_thread.started.connect(self._export_worker.run) + self._export_worker.finished.connect(self._on_packet_export_finished) + self._export_worker.failed.connect(self._on_packet_export_failed) + self._export_worker.finished.connect(self._export_thread.quit) + self._export_worker.failed.connect(self._export_thread.quit) + # deleteLater from the object's OWN thread while its event loop is + # still running it (worker: still on _export_thread when finished/ + # failed fires; thread: on the GUI thread when its `finished` fires) + # — not from _on_packet_export_thread_finished after the fact, which + # runs after the worker's thread affinity/event loop is already gone. + self._export_worker.finished.connect(self._export_worker.deleteLater) + self._export_worker.failed.connect(self._export_worker.deleteLater) + self._export_thread.finished.connect(self._export_thread.deleteLater) + self._export_thread.finished.connect(self._on_packet_export_thread_finished) + self._export_thread.start() + + def _on_packet_export_finished(self, count: int) -> None: + self._status_bar.showMessage(f"Exported {count} packet(s)", 8000) + + def _on_packet_export_failed(self, message: str) -> None: + log.error("Packet export failed: %s", message) + QMessageBox.warning(self, "Export Failed", f"Could not write the file:\n{message}") + + def _on_packet_export_thread_finished(self) -> None: + # Runs after both finished/failed have already updated the status + # bar — this only clears the tracking references and re-enables the + # menu action, regardless of which outcome occurred. deleteLater + # for both objects is already wired above; don't call it again here. + self._export_worker = None + self._export_thread = None + self._export_pkts_act.setEnabled(True) def _export_nodes(self) -> None: from meshchat.services.export_service import ExportService @@ -614,4 +730,24 @@ def closeEvent(self, event) -> None: self._spectrum_page.shutdown() self._controller.shutdown() self._store.shutdown() + if self._export_thread is not None: + # Closing mid-export: block until the write actually finishes + # rather than destroying a still-running QThread out from under + # it, which Qt warns about and can crash on some platforms. + # + # The explicit quit() here is required, not redundant with the + # worker's finished/failed -> thread.quit connections: those are + # QUEUED (the QThread object lives on this GUI thread, the + # worker emits from _export_thread), so they only get delivered + # once THIS thread's event loop is pumping — which it isn't + # while blocked in wait() below. Without this direct call, a + # write that finishes during that wait() would leave the + # worker thread parked in exec() forever with nothing left to + # ever tell it to quit, hanging app shutdown indefinitely. + # quit() itself does not interrupt the blocking write in + # progress — only ensures the thread doesn't idle in its event + # loop once that write actually returns — so this still waits + # as long as the write itself takes. + self._export_thread.quit() + self._export_thread.wait() super().closeEvent(event) diff --git a/src/meshchat/ui/nodes/nodes_page.py b/src/meshchat/ui/nodes/nodes_page.py index 2256302..cee7ec7 100644 --- a/src/meshchat/ui/nodes/nodes_page.py +++ b/src/meshchat/ui/nodes/nodes_page.py @@ -252,12 +252,21 @@ def _on_context_menu(self, pos) -> None: menu.exec(self._table.viewport().mapToGlobal(pos)) def _confirm_remove(self, node_num: int, name: str) -> None: - # Removing writes to the radio's own NodeDB — confirm before doing it. + # Removing writes to the radio's own NodeDB — confirm before doing + # it. This does NOT touch MeshChat's own local node history at all: + # MeshChat's node table is intentionally exempt from history + # pruning, so the entry keeps showing here (and in Rankings/the + # Map) indefinitely regardless of whether the node ever transmits + # again — say so explicitly rather than implying it only sticks + # around until its next transmission, which overstates how much + # this action actually does on the MeshChat side. reply = QMessageBox.question( self, "Remove Node", - f"Remove “{name}” from the connected radio's node database?\n\n" - "The node will reappear if it transmits again.", + f"Remove “{name}” from the connected radio's own node database?\n\n" + "This only affects the physical radio. MeshChat's own node " + "history here is untouched — the entry stays in the Nodes " + "table, Rankings, and Map regardless.", QMessageBox.StandardButton.Yes | QMessageBox.StandardButton.Cancel, QMessageBox.StandardButton.Cancel, ) diff --git a/src/meshchat/utils/node_names.py b/src/meshchat/utils/node_names.py deleted file mode 100644 index 0af8ee9..0000000 --- a/src/meshchat/utils/node_names.py +++ /dev/null @@ -1,60 +0,0 @@ -"""MeshChat – utility: node name resolution.""" -from __future__ import annotations - - -def resolve_node_name( - node_num: int | None, - node_id: str | None = None, - nodes_by_num: dict | None = None, -) -> str: - """ - Resolve a node display name using the priority order from spec: - 1. NodeDB longName - 2. NodeDB shortName - 3. packet fromId - 4. formatted node number - 5. "Unknown node" - """ - # node_num is not None, not `and node_num`: node_num == 0 is a - # legitimate (if unusual) node number and would otherwise be treated - # as falsy, silently skipping the NodeDB lookup for it. - if nodes_by_num and node_num is not None and node_num in nodes_by_num: - user = nodes_by_num[node_num].get("user", {}) - long_name = user.get("longName") or user.get("long_name") - if long_name: - return long_name - short_name = user.get("shortName") or user.get("short_name") - if short_name: - return short_name - - if node_id: - return node_id - - if node_num is not None: - return f"!{node_num:08x}" - - return "Unknown node" - - -def resolve_short_name( - node_num: int | None, - node_id: str | None = None, - nodes_by_num: dict | None = None, -) -> str: - """Return the shortest useful name for a node.""" - # node_num is not None, not `and node_num`: node_num == 0 is a - # legitimate (if unusual) node number and would otherwise be treated - # as falsy, silently skipping the NodeDB lookup for it. - if nodes_by_num and node_num is not None and node_num in nodes_by_num: - user = nodes_by_num[node_num].get("user", {}) - short = user.get("shortName") or user.get("short_name") - if short: - return short - - if node_id: - return node_id[-4:] if len(node_id) > 4 else node_id - - if node_num is not None: - return f"{node_num:04x}" - - return "????" diff --git a/tests/test_export_service.py b/tests/test_export_service.py index 7e378be..c93792f 100644 --- a/tests/test_export_service.py +++ b/tests/test_export_service.py @@ -85,3 +85,20 @@ def test_none_values_still_render_as_empty(self, tmp_path): assert row["packet_id"] == "" assert row["portnum"] == "" assert row["payload_size"] == "" + + +class TestViaMqttTriState: + def test_via_mqtt_false_exports_as_zero(self, tmp_path): + row = _export_and_read_row(tmp_path, _pkt(via_mqtt=False)) + assert row["via_mqtt"] == "0" + + def test_via_mqtt_true_exports_as_one(self, tmp_path): + row = _export_and_read_row(tmp_path, _pkt(via_mqtt=True)) + assert row["via_mqtt"] == "1" + + def test_via_mqtt_unknown_exports_as_empty_not_zero(self, tmp_path): + # via_mqtt is bool | None — older firmware never reports it at all. + # int(bool(None)) == 0 used to collapse that into "confirmed not + # MQTT", indistinguishable from a real False. + row = _export_and_read_row(tmp_path, _pkt(via_mqtt=None)) + assert row["via_mqtt"] == "" diff --git a/tests/test_monitor_store_packets.py b/tests/test_monitor_store_packets.py new file mode 100644 index 0000000..f8062b0 --- /dev/null +++ b/tests/test_monitor_store_packets.py @@ -0,0 +1,97 @@ +"""Tests for MonitorStore.read_packets_as_objects(). + +Backs "Export Packet Log" for a full session's history instead of just +PacketIngestor's bounded 10,000-packet in-memory ring buffer — the packets +table has no text column (message content lives in `messages`), so text +always comes back None here; callers that need it merge it back in from +whatever packets are still in memory. +""" +from __future__ import annotations + +from datetime import datetime, timezone + +import pytest + +from meshchat.models.network_packet import NetworkPacket +from meshchat.services.monitor_store import MonitorStore + + +@pytest.fixture +def store(tmp_path): + s = MonitorStore(db_path=tmp_path / "test.db") + yield s + s.shutdown() + + +def _pkt(**overrides) -> NetworkPacket: + defaults = dict( + session_id="sess-1", + observed_at=datetime.now(timezone.utc), + rx_time=None, + sender_num=111111, + sender_id="!0001b1c7", + destination_num=None, + packet_id=42, + channel_index=0, + portnum=1, + portnum_name="TEXT_MESSAGE_APP", + text="hello", + payload_size=5, + rx_snr=8.25, + rx_rssi=-72, + hop_start=3, + hop_limit=3, + hops_used=0, + via_mqtt=False, + transport_mechanism=None, + pki_encrypted=False, + want_ack=True, + priority="DEFAULT", + raw_metadata_json="{}", + ) + defaults.update(overrides) + return NetworkPacket(**defaults) + + +def _flush(store: MonitorStore) -> None: + # Force the async writer to drain its queue without tearing the store + # down — shutdown() only stops the writer thread; read connections are + # opened fresh per call and remain usable afterward. + store.shutdown() + + +class TestReadPacketsAsObjects: + def test_round_trips_scalar_fields(self, store): + store.save_packet(_pkt()) + _flush(store) + [pkt] = store.read_packets_as_objects("sess-1") + assert pkt.sender_num == 111111 + assert pkt.packet_id == 42 + assert pkt.portnum == 1 + assert pkt.portnum_name == "TEXT_MESSAGE_APP" + assert pkt.rx_snr == 8.25 + assert pkt.rx_rssi == -72 + assert pkt.hops_used == 0 + assert pkt.via_mqtt is False + assert pkt.pki_encrypted is False + assert pkt.want_ack is True + + def test_text_is_always_none(self, store): + # The packets table has no text column at all. + store.save_packet(_pkt(text="this should not survive")) + _flush(store) + [pkt] = store.read_packets_as_objects("sess-1") + assert pkt.text is None + + def test_unknown_via_mqtt_round_trips_as_none_not_false(self, store): + store.save_packet(_pkt(via_mqtt=None)) + _flush(store) + [pkt] = store.read_packets_as_objects("sess-1") + assert pkt.via_mqtt is None + + def test_scoped_to_the_requested_session(self, store): + store.save_packet(_pkt(session_id="sess-1", packet_id=1)) + store.save_packet(_pkt(session_id="sess-2", packet_id=2)) + _flush(store) + pkts = store.read_packets_as_objects("sess-1") + assert {p.packet_id for p in pkts} == {1} diff --git a/tests/test_monitor_store_upsert.py b/tests/test_monitor_store_upsert.py index 1a369b8..558da92 100644 --- a/tests/test_monitor_store_upsert.py +++ b/tests/test_monitor_store_upsert.py @@ -75,3 +75,63 @@ def test_second_upsert_with_null_last_heard_keeps_existing_timestamp(self, tmp_p row = _row(s, 4) assert row["last_heard"] is not None assert row["packet_count"] == 2 + + +class TestUpsertNodeSignalHopTransportPersistence: + """last_snr/last_rssi/last_hops_used/last_hop_start/last_via_mqtt and the + rf_count/via_mqtt_count/position_count/telemetry_count counters used to + have no DB column at all — a node the app had tracked for weeks reset to + zero/unknown on every restart until it was heard again live.""" + + def test_signal_and_hop_fields_round_trip(self, store): + now = datetime.now(timezone.utc) + store.upsert_node(NodeSnapshot( + node_num=5, last_heard=now, + last_snr=8.25, last_rssi=-72, last_hops_used=0, last_hop_start=3, + last_via_mqtt=False, rf_count=4, via_mqtt_count=1, + position_count=2, telemetry_count=1, + )) + store.shutdown() + row = _row(store, 5) + assert row["last_snr"] == 8.25 + assert row["last_rssi"] == -72 + assert row["last_hops_used"] == 0 + assert row["last_hop_start"] == 3 + assert row["last_via_mqtt"] == 0 + assert row["rf_count"] == 4 + assert row["via_mqtt_count"] == 1 + assert row["position_count"] == 2 + assert row["telemetry_count"] == 1 + + def test_older_upsert_does_not_overwrite_newer_last_hops_used(self, store): + # A fresher 0-hop packet must not lose to a staler 3-hop one just + # because a naive MAX() would prefer the bigger number. + now = datetime.now(timezone.utc) + store.upsert_node(NodeSnapshot( + node_num=6, last_heard=now, last_hops_used=0, last_hop_start=3, + )) + store.upsert_node(NodeSnapshot( + node_num=6, last_heard=now - timedelta(hours=1), + last_hops_used=3, last_hop_start=3, + )) + store.shutdown() + row = _row(store, 6) + assert row["last_hops_used"] == 0 + + def test_newer_upsert_advances_last_snr(self, store): + now = datetime.now(timezone.utc) + store.upsert_node(NodeSnapshot(node_num=7, last_heard=now, last_snr=4.0)) + store.upsert_node(NodeSnapshot( + node_num=7, last_heard=now + timedelta(minutes=5), last_snr=9.5, + )) + store.shutdown() + row = _row(store, 7) + assert row["last_snr"] == 9.5 + + def test_counters_use_max_not_last_observation(self, store): + store.upsert_node(NodeSnapshot(node_num=8, rf_count=5, via_mqtt_count=2)) + store.upsert_node(NodeSnapshot(node_num=8, rf_count=3, via_mqtt_count=1)) + store.shutdown() + row = _row(store, 8) + assert row["rf_count"] == 5 + assert row["via_mqtt_count"] == 2 diff --git a/tests/test_node_names.py b/tests/test_node_names.py deleted file mode 100644 index e19b0c7..0000000 --- a/tests/test_node_names.py +++ /dev/null @@ -1,111 +0,0 @@ -"""Tests for utils.node_names.""" -from meshchat.utils.node_names import resolve_node_name, resolve_short_name - - -_NODES = { - 111111: { - "user": { - "id": "!0001b1c7", - "longName": "Alice Node", - "shortName": "ALIC", - } - }, - 222222: { - "user": { - "id": "!000364ce", - "shortName": "BOB", - # No longName — tests short name fallback - } - }, - 333333: { - "user": { - "id": "!00051615", - # Neither longName nor shortName - } - }, -} - - -# ── resolve_node_name ───────────────────────────────────────────────────────── - -class TestResolveNodeName: - def test_long_name_takes_priority(self): - name = resolve_node_name(111111, nodes_by_num=_NODES) - assert name == "Alice Node" - - def test_short_name_fallback_when_no_long_name(self): - name = resolve_node_name(222222, nodes_by_num=_NODES) - assert name == "BOB" - - def test_node_id_fallback_in_db_but_no_names(self): - """When NodeDB has only an id, fall through to node_id arg.""" - name = resolve_node_name(333333, node_id="!00051615", nodes_by_num=_NODES) - # NodeDB user has id but no longName/shortName → fall through to node_id param - assert name == "!00051615" - - def test_node_id_arg_fallback_when_not_in_db(self): - name = resolve_node_name(999999, node_id="!000f423f", nodes_by_num=_NODES) - assert name == "!000f423f" - - def test_formatted_num_fallback(self): - name = resolve_node_name(12345678, nodes_by_num=_NODES) - assert name == "!00bc614e" - - def test_unknown_node_when_no_info(self): - name = resolve_node_name(None, node_id=None, nodes_by_num=None) - assert name == "Unknown node" - - def test_empty_nodes_by_num_uses_node_id(self): - name = resolve_node_name(111111, node_id="!0001b1c7", nodes_by_num={}) - assert name == "!0001b1c7" - - def test_none_nodes_by_num_uses_node_id(self): - name = resolve_node_name(111111, node_id="!0001b1c7", nodes_by_num=None) - assert name == "!0001b1c7" - - def test_snake_case_long_name(self): - """Accepts long_name (snake_case) as well as longName.""" - nodes = { - 500: {"user": {"long_name": "Snake Node", "short_name": "SNK"}} - } - assert resolve_node_name(500, nodes_by_num=nodes) == "Snake Node" - - def test_node_num_zero_is_looked_up_not_treated_as_falsy(self): - # node_num == 0 is a legitimate (if unusual) node number — `and - # node_num` would treat it as falsy and skip the NodeDB lookup - # entirely, falling through to the formatted-number fallback - # instead of the real name. - nodes = {0: {"user": {"longName": "Zero Node"}}} - assert resolve_node_name(0, nodes_by_num=nodes) == "Zero Node" - - -# ── resolve_short_name ──────────────────────────────────────────────────────── - -class TestResolveShortName: - def test_short_name_from_db(self): - assert resolve_short_name(111111, nodes_by_num=_NODES) == "ALIC" - - def test_short_name_from_db_no_long_name(self): - assert resolve_short_name(222222, nodes_by_num=_NODES) == "BOB" - - def test_last_4_of_node_id(self): - # node 333333 has no shortName in DB; node_id="!00051615" → last 4 = "1615" - result = resolve_short_name(333333, node_id="!00051615", nodes_by_num=_NODES) - assert result == "1615" - - def test_hex_num_fallback(self): - result = resolve_short_name(12345678, nodes_by_num={}) - # 12345678 decimal = 0xbc614e; format is {:04x} (min-width 4, not truncated) - assert result == "bc614e" - - def test_unknown_fallback(self): - assert resolve_short_name(None) == "????" - - def test_short_node_id_not_truncated(self): - """A node_id shorter than 4 chars should not be truncated.""" - result = resolve_short_name(99999, node_id="!ab", nodes_by_num={}) - assert result == "!ab" - - def test_node_num_zero_is_looked_up_not_treated_as_falsy(self): - nodes = {0: {"user": {"shortName": "ZERO"}}} - assert resolve_short_name(0, nodes_by_num=nodes) == "ZERO" diff --git a/tests/test_packet_ingestor.py b/tests/test_packet_ingestor.py index fad3911..f576837 100644 --- a/tests/test_packet_ingestor.py +++ b/tests/test_packet_ingestor.py @@ -214,6 +214,20 @@ def test_node_via_mqtt_count_incremented(self): assert ing.get_node(444444).via_mqtt_count == 1 assert ing.get_node(444444).rf_count == 0 + def test_unknown_transport_increments_neither_counter(self): + # via_mqtt absent entirely (older firmware that never reports it) + # must not be misclassified as confirmed RF — `pkt.is_via_mqtt` + # coerces None to False via bool(), which used to do exactly that. + # These counters now persist across restarts, so a wrong count + # here would keep compounding every session. + ing = _make_ingestor() + pkt = dict(_FIXTURES["direct_text"]) + del pkt["viaMqtt"] + ing.ingest_raw(pkt) + node = ing.get_node(111111) + assert node.rf_count == 0 + assert node.via_mqtt_count == 0 + def test_node_last_via_mqtt_tracks_the_most_recent_packet(self): # Feeds NodeSnapshot.is_direct's via_mqtt check — a node's cumulative # via_mqtt_count/rf_count can't tell you whether the MOST RECENT @@ -248,6 +262,61 @@ def test_get_node_returns_none_for_unknown(self): assert ing.get_node(0xDEAD_BEEF) is None +# ── Seeding from persisted history ───────────────────────────────────────────── + +class TestSeedFromStore: + """seed_from_store() restores nodes from MonitorStore rows at startup. + last_snr/last_rssi/last_hops_used/last_hop_start/last_via_mqtt and the + rf_count/via_mqtt_count/position_count/telemetry_count counters used to + have no DB column at all, so they never made it into these rows and a + long-tracked node started every restart with them reset.""" + + def test_signal_hop_and_counter_fields_are_restored(self): + ing = _make_ingestor() + ing.seed_from_store( + node_rows=[{ + "node_num": 1, "node_id": "!00000001", + "last_snr": 8.25, "last_rssi": -72, + "last_hops_used": 0, "last_hop_start": 3, "last_via_mqtt": 0, + "rf_count": 4, "via_mqtt_count": 1, + "position_count": 2, "telemetry_count": 1, + }], + positions_by_num={}, + ) + node = ing.get_node(1) + assert node.last_snr == 8.25 + assert node.last_rssi == -72 + assert node.last_hops_used == 0 + assert node.last_hop_start == 3 + assert node.last_via_mqtt is False + assert node.rf_count == 4 + assert node.via_mqtt_count == 1 + assert node.position_count == 2 + assert node.telemetry_count == 1 + + def test_missing_last_via_mqtt_stays_none_not_false(self): + # A row with the column present but NULL (never observed) must not + # be conflated with an explicit False (confirmed direct RF). + ing = _make_ingestor() + ing.seed_from_store( + node_rows=[{"node_num": 2, "last_via_mqtt": None}], + positions_by_num={}, + ) + assert ing.get_node(2).last_via_mqtt is None + + def test_live_packet_before_seeding_is_not_overwritten(self): + # A node already heard live this session must keep its live data — + # seeding is startup-only backfill, not an authoritative overwrite. + ing = _make_ingestor() + ing.ingest_raw(_FIXTURES["direct_text"]) # creates node 111111 + live_snr = ing.get_node(111111).last_snr + ing.seed_from_store( + node_rows=[{"node_num": 111111, "last_snr": -99.0}], + positions_by_num={}, + ) + assert ing.get_node(111111).last_snr == live_snr + + # ── Signals emitted ─────────────────────────────────────────────────────────── class TestSignals: