adam/pyMC_Repeater
1
0
Fork 0
mirror of https://github.com/pyMC-dev/pyMC_Repeater.git synced 2026-08-07 17:33:16 +02:00
Code Issues Packages Projects Releases Wiki Activity

fix(companion): persist queued message signal and channel data

Browse Source
This commit is contained in:
agessaman
2026-07-15 17:06:00 -07:00
parent a208fc3763
commit 034cd6f566
4 changed files with 314 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files
repeater/companion/frame_server.py
+4
View File
@@ -116,6 +116,10 @@ class CompanionFrameServer(_BaseFrameServer):
is_channel=bool(msg_dict.get("is_channel", False)),
channel_idx=msg_dict.get("channel_idx", 0),
path_len=msg_dict.get("path_len", 0),
snr=float(msg_dict.get("snr") or 0.0),
rssi=int(msg_dict.get("rssi") or 0),
channel_data_type=int(msg_dict.get("channel_data_type") or 0),
channel_data_payload=bytes(msg_dict.get("channel_data_payload") or b""),
sender_prefix=sender_prefix,
)
repeater/data_acquisition/sqlite_handler.py
+58 -4
View File
@@ -437,6 +437,10 @@ class SQLiteHandler:
channel_idx INTEGER NOT NULL DEFAULT 0,
path_len INTEGER NOT NULL DEFAULT 0,
sender_prefix TEXT NOT NULL DEFAULT '',
snr REAL,
rssi INTEGER,
channel_data_type INTEGER,
channel_data_payload BLOB,
packet_hash TEXT,
created_at REAL NOT NULL
)
@@ -637,6 +641,41 @@ class SQLiteHandler:
)
logger.info(f"Migration '{migration_name}' applied successfully")
# Migration 12: Add signal metadata and channel-data columns to
# companion_messages. Without snr/channel_data_type/
# channel_data_payload, a message replayed from SQLite rebuilds
# with a zero SNR byte and a binary channel-data (GRP_DATA) frame
# collapses to an empty channel-text frame.
migration_name = "add_signal_and_channel_data_to_companion_messages"
existing = conn.execute(
"SELECT migration_name FROM migrations WHERE migration_name = ?",
(migration_name,),
).fetchone()
if not existing:
cursor = conn.execute("PRAGMA table_info(companion_messages)")
columns = [column[1] for column in cursor.fetchall()]
if "snr" not in columns:
conn.execute("ALTER TABLE companion_messages ADD COLUMN snr REAL")
logger.info("Added snr column to companion_messages table")
if "rssi" not in columns:
conn.execute("ALTER TABLE companion_messages ADD COLUMN rssi INTEGER")
logger.info("Added rssi column to companion_messages table")
if "channel_data_type" not in columns:
conn.execute(
"ALTER TABLE companion_messages ADD COLUMN channel_data_type INTEGER"
)
logger.info("Added channel_data_type column to companion_messages table")
if "channel_data_payload" not in columns:
conn.execute(
"ALTER TABLE companion_messages ADD COLUMN channel_data_payload BLOB"
)
logger.info("Added channel_data_payload column to companion_messages table")
conn.execute(
"INSERT INTO migrations (migration_name, applied_at) VALUES (?, ?)",
(migration_name, time.time()),
)
logger.info(f"Migration '{migration_name}' applied successfully")
conn.commit()
except Exception as e:
@@ -3334,7 +3373,8 @@ class SQLiteHandler:
cursor = conn.execute(
"""
SELECT sender_key, txt_type, timestamp, text, is_channel, channel_idx,
path_len, sender_prefix
path_len, sender_prefix, snr, rssi, channel_data_type,
channel_data_payload
FROM companion_messages WHERE companion_hash = ?
ORDER BY created_at ASC LIMIT ?
""",
@@ -3343,6 +3383,10 @@ class SQLiteHandler:
rows = [dict(row) for row in cursor.fetchall()]
for msg in rows:
msg["sender_prefix"] = bytes.fromhex(msg.get("sender_prefix") or "")
msg["snr"] = float(msg.get("snr") or 0.0)
msg["rssi"] = int(msg.get("rssi") or 0)
msg["channel_data_type"] = int(msg.get("channel_data_type") or 0)
msg["channel_data_payload"] = bytes(msg.get("channel_data_payload") or b"")
return rows
except Exception as e:
logger.error(f"Failed to load companion messages for {companion_hash}: {e}")
@@ -3382,8 +3426,9 @@ class SQLiteHandler:
"""
INSERT OR IGNORE INTO companion_messages
(companion_hash, sender_key, txt_type, timestamp, text,
is_channel, channel_idx, path_len, sender_prefix, packet_hash, created_at)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
is_channel, channel_idx, path_len, sender_prefix, snr, rssi,
channel_data_type, channel_data_payload, packet_hash, created_at)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""",
(
companion_hash,
@@ -3395,6 +3440,10 @@ class SQLiteHandler:
msg.get("channel_idx", 0),
msg.get("path_len", 0),
sender_prefix,
float(msg.get("snr") or 0.0),
int(msg.get("rssi") or 0),
int(msg.get("channel_data_type") or 0),
bytes(msg.get("channel_data_payload") or b""),
packet_hash,
time.time(),
),
@@ -3455,7 +3504,8 @@ class SQLiteHandler:
cursor = conn.execute(
"""
SELECT id, sender_key, txt_type, timestamp, text, is_channel, channel_idx,
path_len, sender_prefix
path_len, sender_prefix, snr, rssi, channel_data_type,
channel_data_payload
FROM companion_messages WHERE companion_hash = ?
ORDER BY created_at ASC LIMIT 1
""",
@@ -3466,6 +3516,10 @@ class SQLiteHandler:
return None
msg = dict(row)
msg["sender_prefix"] = bytes.fromhex(msg.get("sender_prefix") or "")
msg["snr"] = float(msg.get("snr") or 0.0)
msg["rssi"] = int(msg.get("rssi") or 0)
msg["channel_data_type"] = int(msg.get("channel_data_type") or 0)
msg["channel_data_payload"] = bytes(msg.get("channel_data_payload") or b"")
conn.execute("DELETE FROM companion_messages WHERE id = ?", (msg["id"],))
conn.commit()
return {k: v for k, v in msg.items() if k != "id"}
tests/test_companion_message_signal_persistence.py
+249
View File
@@ -0,0 +1,249 @@
"""SQLite round-trip of the queued-message signal and channel-data fields.
Companion offline-queue messages are persisted as structured fields and the
response frame is rebuilt on replay (SYNC_NEXT_MESSAGE). The parity contract is
that the frame rebuilt from a SQLite round-tripped message is byte-identical to
the frame built from the original in-memory ``QueuedMessage`` — so snr, rssi,
channel_data_type and channel_data_payload must all survive persistence.
"""
from __future__ import annotations
import sqlite3
import pytest
from openhop_core.companion.models import QueuedMessage
from openhop_core.protocol.constants import TXT_TYPE_PLAIN, TXT_TYPE_SIGNED_PLAIN
from repeater.companion.frame_server import CompanionFrameServer
from repeater.data_acquisition.sqlite_handler import SQLiteHandler
_HASH = "0x01"
def _handler(tmp_path):
return SQLiteHandler(tmp_path)
def _frame_server(handler, app_target_ver=3):
"""A CompanionFrameServer wired to ``handler`` for frame build + replay.
``_build_message_frame`` only reads ``_app_target_ver``; V3 is used so the
SNR byte is emitted in the direct and channel-text frames (the pre-V3
variants omit it).
"""
fs = CompanionFrameServer.__new__(CompanionFrameServer)
fs._app_target_ver = app_target_ver
fs.sqlite_handler = handler
fs.companion_hash = _HASH
return fs
def _qm_from_dict(d: dict) -> QueuedMessage:
"""Build the original in-memory QueuedMessage the bridge would enqueue."""
prefix = d.get("sender_prefix") or b""
return QueuedMessage(
sender_key=d.get("sender_key", b""),
txt_type=d.get("txt_type", 0),
timestamp=d.get("timestamp", 0),
text=d.get("text", ""),
is_channel=bool(d.get("is_channel", False)),
channel_idx=d.get("channel_idx", 0),
path_len=d.get("path_len", 0),
snr=float(d.get("snr") or 0.0),
rssi=int(d.get("rssi") or 0),
channel_data_type=int(d.get("channel_data_type") or 0),
channel_data_payload=bytes(d.get("channel_data_payload") or b""),
sender_prefix=bytes(prefix),
)
# Nonzero snr (negative fractional exercises the int8 wrap) and nonzero rssi.
_DIRECT_PLAIN = {
"sender_key": b"\x11" * 32,
"text": "hello world",
"timestamp": 1000,
"txt_type": TXT_TYPE_PLAIN,
"is_channel": False,
"channel_idx": 0,
"path_len": 0xFF,
"packet_hash": "direct-plain",
"snr": -6.25,
"rssi": -80,
"sender_prefix": b"",
}
_DIRECT_SIGNED = {
"sender_key": b"\x22" * 32,
"text": "signed room post",
"timestamp": 2000,
"txt_type": TXT_TYPE_SIGNED_PLAIN,
"is_channel": False,
"channel_idx": 0,
"path_len": 3,
"packet_hash": "direct-signed",
"snr": 7.5,
"rssi": -60,
"sender_prefix": b"\xaa\xbb\xcc\xdd",
}
_CHANNEL_TEXT = {
"sender_key": b"",
"text": "channel chatter",
"timestamp": 3000,
"txt_type": 0,
"is_channel": True,
"channel_idx": 2,
"path_len": 5,
"packet_hash": "channel-text",
"snr": -3.75,
"rssi": -90,
}
_CHANNEL_DATA = {
"sender_key": b"",
"text": "",
"timestamp": 0,
"txt_type": 0,
"is_channel": True,
"channel_idx": 1,
"path_len": 4,
"packet_hash": "channel-data",
"snr": -6.25,
"rssi": -70,
"channel_data_type": 0x0102,
"channel_data_payload": b"\x01\x02\x03\xff\x10",
}
@pytest.mark.parametrize(
"msg_dict",
[_DIRECT_PLAIN, _DIRECT_SIGNED, _CHANNEL_TEXT, _CHANNEL_DATA],
ids=["direct_plain", "direct_signed", "channel_text", "channel_data"],
)
def test_frame_round_trip_is_byte_identical(tmp_path, msg_dict):
h = _handler(tmp_path)
fs = _frame_server(h)
original_frame = fs._build_message_frame(_qm_from_dict(msg_dict))
assert h.companion_push_message(_HASH, dict(msg_dict))
rebuilt = fs._sync_next_from_persistence()
assert rebuilt is not None
rebuilt_frame = fs._build_message_frame(rebuilt)
assert rebuilt_frame == original_frame
def test_channel_data_survives_as_binary_frame_not_empty_text(tmp_path):
# Regression: a binary channel-data message must not collapse into an empty
# channel-text frame after the SQLite round-trip.
h = _handler(tmp_path)
fs = _frame_server(h)
assert h.companion_push_message(_HASH, dict(_CHANNEL_DATA))
rebuilt = fs._sync_next_from_persistence()
assert rebuilt.channel_data_type == 0x0102
assert rebuilt.channel_data_payload == b"\x01\x02\x03\xff\x10"
assert rebuilt.snr == pytest.approx(-6.25)
assert rebuilt.rssi == -70
def test_restart_durability_reconstructs_all_fields(tmp_path):
h = _handler(tmp_path)
assert h.companion_push_message(_HASH, dict(_DIRECT_SIGNED))
assert h.companion_push_message(_HASH, dict(_CHANNEL_DATA))
# A daemon restart: drop the handler and reopen the same database file.
del h
h2 = _handler(tmp_path)
signed = h2.companion_pop_message(_HASH)
assert signed["sender_prefix"] == b"\xaa\xbb\xcc\xdd"
assert signed["snr"] == pytest.approx(7.5)
assert signed["rssi"] == -60
assert signed["channel_data_type"] == 0
assert signed["channel_data_payload"] == b""
data = h2.companion_pop_message(_HASH)
assert data["channel_data_type"] == 0x0102
assert data["channel_data_payload"] == b"\x01\x02\x03\xff\x10"
assert data["snr"] == pytest.approx(-6.25)
assert data["rssi"] == -70
def test_load_messages_returns_new_fields_with_types(tmp_path):
h = _handler(tmp_path)
assert h.companion_push_message(_HASH, dict(_CHANNEL_DATA))
msgs = h.companion_load_messages(_HASH)
assert len(msgs) == 1
msg = msgs[0]
assert isinstance(msg["snr"], float)
assert isinstance(msg["rssi"], int)
assert isinstance(msg["channel_data_type"], int)
assert isinstance(msg["channel_data_payload"], bytes)
assert msg["snr"] == pytest.approx(-6.25)
assert msg["rssi"] == -70
assert msg["channel_data_type"] == 0x0102
assert msg["channel_data_payload"] == b"\x01\x02\x03\xff\x10"
def test_legacy_schema_migrates_and_defaults(tmp_path):
# Rewind companion_messages to the pre-change schema (sender_prefix present,
# but no snr/rssi/channel_data_type/channel_data_payload) and drop the
# migration marker, then re-run migrations against a pre-existing row.
h = _handler(tmp_path)
conn = sqlite3.connect(str(h.sqlite_path))
conn.execute(
"DELETE FROM migrations "
"WHERE migration_name = 'add_signal_and_channel_data_to_companion_messages'"
)
conn.execute("ALTER TABLE companion_messages RENAME TO companion_messages_old")
conn.execute(
"""
CREATE TABLE companion_messages (
id INTEGER PRIMARY KEY AUTOINCREMENT,
companion_hash TEXT NOT NULL,
sender_key BLOB NOT NULL,
txt_type INTEGER NOT NULL DEFAULT 0,
timestamp INTEGER NOT NULL DEFAULT 0,
text TEXT NOT NULL,
is_channel INTEGER NOT NULL DEFAULT 0,
channel_idx INTEGER NOT NULL DEFAULT 0,
path_len INTEGER NOT NULL DEFAULT 0,
sender_prefix TEXT NOT NULL DEFAULT '',
packet_hash TEXT,
created_at REAL NOT NULL
)
"""
)
conn.execute("DROP TABLE companion_messages_old")
conn.execute(
"INSERT INTO companion_messages "
"(companion_hash, sender_key, text, created_at) VALUES ('0x01', X'01', 'old', 1.0)"
)
conn.commit()
conn.close()
h2 = _handler(tmp_path) # re-runs migrations
# (a) the four columns are added.
conn = sqlite3.connect(str(h2.sqlite_path))
columns = {row[1] for row in conn.execute("PRAGMA table_info(companion_messages)")}
conn.close()
assert {"snr", "rssi", "channel_data_type", "channel_data_payload"} <= columns
# (b) the old row pops with defaults and no exception.
old = h2.companion_pop_message(_HASH)
assert old["text"] == "old"
assert old["snr"] == 0.0
assert old["rssi"] == 0
assert old["channel_data_type"] == 0
assert old["channel_data_payload"] == b""
# (c) new pushes round-trip fully through the migrated columns.
assert h2.companion_push_message(_HASH, dict(_CHANNEL_DATA))
data = h2.companion_pop_message(_HASH)
assert data["channel_data_type"] == 0x0102
assert data["channel_data_payload"] == b"\x01\x02\x03\xff\x10"
assert data["snr"] == pytest.approx(-6.25)
assert data["rssi"] == -70
tests/test_companion_settings.py
+3 -1
View File
@@ -337,7 +337,9 @@ class TestSenderPrefixPersistence:
conn = sqlite3.connect(str(h.sqlite_path))
conn.execute(
"DELETE FROM migrations "
"WHERE migration_name = 'add_sender_prefix_to_companion_messages'"
"WHERE migration_name IN ("
"'add_sender_prefix_to_companion_messages', "
"'add_signal_and_channel_data_to_companion_messages')"
)
conn.execute("ALTER TABLE companion_messages RENAME TO companion_messages_old")
conn.execute(