From 91af4324deec7db55a453d6fc2d3403ea52c2102 Mon Sep 17 00:00:00 2001 From: MarekWo Date: Tue, 14 Jul 2026 19:01:54 +0200 Subject: [PATCH] feat(observer): wire packet capture into DeviceManager and app startup DeviceManager hands device identity (name/pubkey/self_info/device_info) to the observer after connect and feeds every RX_LOG_DATA packet into handle_raw_packet() before the GRP_TXT echo gate, wrapped in its own try/except so observer failures can never affect chat. create_app() builds the ObserverManager and links it with the DeviceManager. Co-Authored-By: Claude Fable 5 --- app/device_manager.py | 29 +++++++++++++++++++++++++++-- app/main.py | 8 +++++++- 2 files changed, 34 insertions(+), 3 deletions(-) diff --git a/app/device_manager.py b/app/device_manager.py index 60fe016..e18023a 100644 --- a/app/device_manager.py +++ b/app/device_manager.py @@ -190,10 +190,11 @@ class DeviceManager: dm.stop() # disconnect and stop background thread """ - def __init__(self, config, db, socketio=None): + def __init__(self, config, db, socketio=None, observer=None): self.config = config self.db = db self.socketio = socketio + self.observer = observer # ObserverManager (packet capture -> MQTT) self.mc = None # meshcore.MeshCore instance self._loop = None # asyncio event loop (in background thread) self._thread = None # background thread @@ -396,6 +397,7 @@ class DeviceManager: # Fetch device_info for max_channels and FIRMWARE_VER_CODE. # fw_ver_code gates feature support: CMD_SEND_RAW_PACKET needs ≥13 # (companion-v1.16.0), set_path_hash_mode needs ≥10, etc. + dev_info = {} try: dev_info_event = await self.mc.commands.send_device_query() if dev_info_event and hasattr(dev_info_event, 'payload'): @@ -414,6 +416,19 @@ class DeviceManager: except Exception as e: logger.warning(f"Could not fetch device_info: {e}") + # Hand device identity to the observer and start it if enabled + if self.observer: + try: + self.observer.set_identity( + self._device_name, + self._self_info.get('public_key', ''), + self_info=self._self_info, + device_info=dev_info, + ) + self.observer.ensure_started() + except Exception as e: + logger.warning(f"Observer start failed (non-fatal): {e}") + # Workaround: meshcore lib 2.2.21 has a bug where list.extend() # return value (None) corrupts reader.channels for idx >= 20. # Pre-allocate the channels list to max_channels to avoid this. @@ -1233,7 +1248,8 @@ class DeviceManager: Firmware sends LOG_DATA (0x88) packets for every repeated radio frame. Payload format: header(1) [transport_code(4)] path_len(1) path(N) pkt_payload(rest) - We only process GRP_TXT (payload_type=0x05) for channel message echoes. + Every packet is handed to the observer (MQTT capture) when active; + only GRP_TXT (payload_type=0x05) continues into echo detection. """ try: import io @@ -1244,6 +1260,15 @@ class DeviceManager: if not payload_hex: return + if self.observer: + try: + self.observer.handle_raw_packet( + payload_hex, snr=data.get('snr'), + rssi=data.get('rssi'), + payload_length=data.get('payload_length')) + except Exception: + logger.debug("Observer packet handling failed", exc_info=True) + pkt = bytes.fromhex(payload_hex) pbuf = io.BytesIO(pkt) diff --git a/app/main.py b/app/main.py index 2f4c4dd..42a6206 100644 --- a/app/main.py +++ b/app/main.py @@ -19,6 +19,7 @@ from app.config import config, runtime_config from app.database import Database from app.device_manager import DeviceManager, parse_meshcore_uri from app.log_handler import MemoryLogHandler +from app.observer import ObserverManager from app.routes.views import views_bp from app.routes.api import api_bp from app.version import VERSION_STRING, GIT_BRANCH @@ -291,9 +292,14 @@ def create_app(): except Exception as e: logger.warning(f"Failed to migrate .read_status.json: {e}") + # Observer: MQTT packet capture (starts once the device connects) + observer_manager = ObserverManager(db, socketio) + app.observer_manager = observer_manager + # v2: Initialize and start device manager - device_manager = DeviceManager(config, db, socketio) + device_manager = DeviceManager(config, db, socketio, observer=observer_manager) app.device_manager = device_manager + observer_manager.device_manager = device_manager # Start device connection in background (non-blocking) device_manager.start()