Add radio model and stats display. Closes #64

app/radio.py

@@ -131,6 +131,11 @@ class RadioManager:
         self._setup_lock: asyncio.Lock | None = None
         self._setup_in_progress: bool = False
         self._setup_complete: bool = False
+        self.device_info_loaded: bool = False
+        self.max_contacts: int | None = None
+        self.device_model: str | None = None
+        self.firmware_build: str | None = None
+        self.firmware_version: str | None = None
         self.max_channels: int = 40
         self.path_hash_mode: int = 0
         self.path_hash_mode_supported: bool = False
@@ -488,6 +493,11 @@ class RadioManager:
             await self._disable_meshcore_auto_reconnect(mc)
             self._meshcore = None
             self._setup_complete = False
+            self.device_info_loaded = False
+            self.max_contacts = None
+            self.device_model = None
+            self.firmware_build = None
+            self.firmware_version = None
             self.max_channels = 40
             self.path_hash_mode = 0
             self.path_hash_mode_supported = False

app/routers/health.py

@@ -11,18 +11,34 @@ from app.services.radio_runtime import radio_runtime as radio_manager
 router = APIRouter(tags=["health"])
 
 
+class RadioDeviceInfoResponse(BaseModel):
+    model: str | None = None
+    firmware_build: str | None = None
+    firmware_version: str | None = None
+    max_contacts: int | None = None
+    max_channels: int | None = None
+
+
 class HealthResponse(BaseModel):
     status: str
     radio_connected: bool
     radio_initializing: bool = False
     radio_state: str = "disconnected"
     connection_info: str | None
+    radio_device_info: RadioDeviceInfoResponse | None = None
     database_size_mb: float
     oldest_undecrypted_timestamp: int | None
     fanout_statuses: dict[str, dict[str, str]] = {}
     bots_disabled: bool = False
 
 
+def _clean_optional_str(value: object) -> str | None:
+    if not isinstance(value, str):
+        return None
+    cleaned = value.strip()
+    return cleaned or None
+
+
 async def build_health_data(radio_connected: bool, connection_info: str | None) -> dict:
     """Build the health status payload used by REST endpoint and WebSocket broadcasts."""
     db_size_mb = 0.0
@@ -48,22 +64,12 @@ async def build_health_data(radio_connected: bool, connection_info: str | None)
         pass
 
     setup_in_progress = getattr(radio_manager, "is_setup_in_progress", False)
-    if not isinstance(setup_in_progress, bool):
-        setup_in_progress = False
-
     setup_complete = getattr(radio_manager, "is_setup_complete", radio_connected)
-    if not isinstance(setup_complete, bool):
-        setup_complete = radio_connected
     if not radio_connected:
         setup_complete = False
 
     connection_desired = getattr(radio_manager, "connection_desired", True)
-    if not isinstance(connection_desired, bool):
-        connection_desired = True
-
     is_reconnecting = getattr(radio_manager, "is_reconnecting", False)
-    if not isinstance(is_reconnecting, bool):
-        is_reconnecting = False
 
     radio_initializing = bool(radio_connected and (setup_in_progress or not setup_complete))
     if not connection_desired:
@@ -77,12 +83,26 @@ async def build_health_data(radio_connected: bool, connection_info: str | None)
     else:
         radio_state = "disconnected"
 
+    radio_device_info = None
+    device_info_loaded = getattr(radio_manager, "device_info_loaded", False)
+    if radio_connected and device_info_loaded:
+        radio_device_info = {
+            "model": _clean_optional_str(getattr(radio_manager, "device_model", None)),
+            "firmware_build": _clean_optional_str(getattr(radio_manager, "firmware_build", None)),
+            "firmware_version": _clean_optional_str(
+                getattr(radio_manager, "firmware_version", None)
+            ),
+            "max_contacts": getattr(radio_manager, "max_contacts", None),
+            "max_channels": getattr(radio_manager, "max_channels", None),
+        }
+
     return {
         "status": "ok" if radio_connected and not radio_initializing else "degraded",
         "radio_connected": radio_connected,
         "radio_initializing": radio_initializing,
         "radio_state": radio_state,
         "connection_info": connection_info,
+        "radio_device_info": radio_device_info,
         "database_size_mb": db_size_mb,
         "oldest_undecrypted_timestamp": oldest_ts,
         "fanout_statuses": fanout_statuses,

app/services/radio_lifecycle.py

@@ -7,6 +7,21 @@ POST_CONNECT_SETUP_TIMEOUT_SECONDS = 300
 POST_CONNECT_SETUP_MAX_ATTEMPTS = 2
 
 
+def _clean_device_string(value: object) -> str | None:
+    if not isinstance(value, str):
+        return None
+    cleaned = value.strip()
+    return cleaned or None
+
+
+def _decode_fixed_string(raw: bytes, start: int, length: int) -> str | None:
+    if len(raw) < start:
+        return None
+    return _clean_device_string(
+        raw[start : start + length].decode("utf-8", "ignore").replace("\0", "")
+    )
+
+
 async def run_post_connect_setup(radio_manager) -> None:
     """Run shared radio initialization after a transport connection succeeds."""
     from app.event_handlers import register_event_handlers
@@ -78,26 +93,66 @@ async def run_post_connect_setup(radio_manager) -> None:
                     return await _original_handle_rx(data)
 
                 reader.handle_rx = _capture_handle_rx
+                radio_manager.device_info_loaded = False
+                radio_manager.max_contacts = None
+                radio_manager.device_model = None
+                radio_manager.firmware_build = None
+                radio_manager.firmware_version = None
                 radio_manager.max_channels = 40
                 radio_manager.path_hash_mode = 0
                 radio_manager.path_hash_mode_supported = False
                 try:
                     device_query = await mc.commands.send_device_query()
-                    if device_query and "max_channels" in device_query.payload:
-                        radio_manager.max_channels = max(
-                            1, int(device_query.payload["max_channels"])
-                        )
-                    if device_query and "path_hash_mode" in device_query.payload:
-                        radio_manager.path_hash_mode = device_query.payload["path_hash_mode"]
+                    payload = (
+                        device_query.payload
+                        if device_query is not None and isinstance(device_query.payload, dict)
+                        else {}
+                    )
+
+                    payload_max_contacts = payload.get("max_contacts")
+                    if isinstance(payload_max_contacts, int):
+                        radio_manager.max_contacts = max(1, payload_max_contacts)
+
+                    payload_max_channels = payload.get("max_channels")
+                    if isinstance(payload_max_channels, int):
+                        radio_manager.max_channels = max(1, payload_max_channels)
+
+                    radio_manager.device_model = _clean_device_string(payload.get("model"))
+                    radio_manager.firmware_build = _clean_device_string(payload.get("fw_build"))
+                    radio_manager.firmware_version = _clean_device_string(payload.get("ver"))
+
+                    fw_ver = payload.get("fw ver")
+                    payload_reports_device_info = isinstance(fw_ver, int) and fw_ver >= 3
+                    if payload_reports_device_info:
+                        radio_manager.device_info_loaded = True
+
+                    if "path_hash_mode" in payload and isinstance(payload["path_hash_mode"], int):
+                        radio_manager.path_hash_mode = payload["path_hash_mode"]
                         radio_manager.path_hash_mode_supported = True
-                    elif _captured_frame:
-                        # Raw-frame fallback:
-                        # byte 1 = fw_ver, byte 3 = max_channels, byte 81 = path_hash_mode
+
+                    if _captured_frame:
+                        # Raw-frame fallback / completion:
+                        # byte 1 = fw_ver, byte 2 = max_contacts/2, byte 3 = max_channels,
+                        # bytes 8:20 = fw_build, 20:60 = model, 60:80 = ver, byte 81 = path_hash_mode
                         raw = _captured_frame[-1]
                         fw_ver = raw[1] if len(raw) > 1 else 0
-                        if fw_ver >= 3 and len(raw) >= 4:
-                            radio_manager.max_channels = max(1, raw[3])
-                        if fw_ver >= 10 and len(raw) >= 82:
+                        if fw_ver >= 3:
+                            radio_manager.device_info_loaded = True
+                            if radio_manager.max_contacts is None and len(raw) >= 3:
+                                radio_manager.max_contacts = max(1, raw[2] * 2)
+                            if len(raw) >= 4 and not isinstance(payload_max_channels, int):
+                                radio_manager.max_channels = max(1, raw[3])
+                            if radio_manager.firmware_build is None:
+                                radio_manager.firmware_build = _decode_fixed_string(raw, 8, 12)
+                            if radio_manager.device_model is None:
+                                radio_manager.device_model = _decode_fixed_string(raw, 20, 40)
+                            if radio_manager.firmware_version is None:
+                                radio_manager.firmware_version = _decode_fixed_string(raw, 60, 20)
+                        if (
+                            not radio_manager.path_hash_mode_supported
+                            and fw_ver >= 10
+                            and len(raw) >= 82
+                        ):
                             radio_manager.path_hash_mode = raw[81]
                             radio_manager.path_hash_mode_supported = True
                             logger.warning(
@@ -114,6 +169,17 @@ async def run_post_connect_setup(radio_manager) -> None:
                         logger.info("Path hash mode: %d (supported)", radio_manager.path_hash_mode)
                     else:
                         logger.debug("Firmware does not report path_hash_mode")
+                    if radio_manager.device_info_loaded:
+                        logger.info(
+                            "Radio device info: model=%s build=%s version=%s max_contacts=%s max_channels=%d",
+                            radio_manager.device_model or "unknown",
+                            radio_manager.firmware_build or "unknown",
+                            radio_manager.firmware_version or "unknown",
+                            radio_manager.max_contacts
+                            if radio_manager.max_contacts is not None
+                            else "unknown",
+                            radio_manager.max_channels,
+                        )
                     logger.info("Max channel slots: %d", radio_manager.max_channels)
                 except Exception as exc:
                     logger.debug("Failed to query device info capabilities: %s", exc)