Add noise floor plumbing

app/main.py

@@ -39,6 +39,7 @@ from app.routers import (
     ws,
 )
 from app.security import add_optional_basic_auth_middleware
+from app.services.radio_noise_floor import start_noise_floor_sampling, stop_noise_floor_sampling
 from app.services.radio_runtime import radio_runtime as radio_manager
 from app.version_info import get_app_build_info
 
@@ -70,6 +71,7 @@ async def lifespan(app: FastAPI):
     from app.radio_sync import ensure_default_channels
 
     await ensure_default_channels()
+    await start_noise_floor_sampling()
 
     # Always start connection monitor (even if initial connection failed)
     await radio_manager.start_connection_monitor()
@@ -98,6 +100,7 @@ async def lifespan(app: FastAPI):
     await radio_manager.stop_connection_monitor()
     await stop_background_contact_reconciliation()
     await stop_message_polling()
+    await stop_noise_floor_sampling()
     await stop_periodic_advert()
     await stop_periodic_sync()
     if radio_manager.meshcore:

app/models.py

@@ -824,6 +824,27 @@ class PathHashWidthStats(BaseModel):
     triple_byte_pct: float
 
 
+class NoiseFloorSample(BaseModel):
+    timestamp: int = Field(description="Unix timestamp of the sampled reading")
+    noise_floor_dbm: int = Field(description="Noise floor in dBm")
+
+
+class NoiseFloorHistoryStats(BaseModel):
+    sample_interval_seconds: int = Field(description="Expected spacing between samples")
+    coverage_seconds: int = Field(description="How much of the last 24 hours is represented")
+    latest_noise_floor_dbm: int | None = Field(
+        default=None, description="Most recent sampled noise floor in dBm"
+    )
+    latest_timestamp: int | None = Field(
+        default=None, description="Unix timestamp of the most recent sample"
+    )
+    supported: bool | None = Field(
+        default=None,
+        description="Whether the connected radio appears to support radio stats sampling",
+    )
+    samples: list[NoiseFloorSample] = Field(default_factory=list)
+
+
 class StatisticsResponse(BaseModel):
     busiest_channels_24h: list[BusyChannel]
     contact_count: int
@@ -839,3 +860,4 @@ class StatisticsResponse(BaseModel):
     repeaters_heard: ContactActivityCounts
     known_channels_active: ContactActivityCounts
     path_hash_width_24h: PathHashWidthStats
+    noise_floor_24h: NoiseFloorHistoryStats

app/routers/statistics.py

@@ -2,6 +2,7 @@ from fastapi import APIRouter
 
 from app.models import StatisticsResponse
 from app.repository import StatisticsRepository
+from app.services.radio_noise_floor import get_noise_floor_history
 
 router = APIRouter(prefix="/statistics", tags=["statistics"])
 
@@ -9,4 +10,5 @@ router = APIRouter(prefix="/statistics", tags=["statistics"])
 @router.get("", response_model=StatisticsResponse)
 async def get_statistics() -> StatisticsResponse:
     data = await StatisticsRepository.get_all()
+    data["noise_floor_24h"] = await get_noise_floor_history()
     return StatisticsResponse(**data)

app/services/radio_noise_floor.py

@@ -0,0 +1,112 @@
+"""In-memory local-radio noise floor history sampling."""
+
+import asyncio
+import logging
+import time
+from collections import deque
+
+from meshcore import EventType
+
+from app.radio import RadioDisconnectedError, RadioOperationBusyError
+from app.services.radio_runtime import radio_runtime as radio_manager
+
+logger = logging.getLogger(__name__)
+
+NOISE_FLOOR_SAMPLE_INTERVAL_SECONDS = 300
+NOISE_FLOOR_WINDOW_SECONDS = 24 * 60 * 60
+MAX_NOISE_FLOOR_SAMPLES = 300
+
+_noise_floor_task: asyncio.Task | None = None
+_noise_floor_samples: deque[tuple[int, int]] = deque(maxlen=MAX_NOISE_FLOOR_SAMPLES)
+_noise_floor_supported: bool | None = None
+_samples_lock = asyncio.Lock()
+
+
+async def _append_sample(timestamp: int, noise_floor_dbm: int) -> None:
+    async with _samples_lock:
+        _noise_floor_samples.append((timestamp, noise_floor_dbm))
+
+
+async def sample_noise_floor_once(*, blocking: bool = False) -> None:
+    """Fetch the current radio noise floor once and record it when available."""
+    global _noise_floor_supported
+
+    if not radio_manager.is_connected:
+        return
+
+    try:
+        async with radio_manager.radio_operation("noise_floor_sample", blocking=blocking) as mc:
+            event = await mc.commands.get_stats_radio()
+    except (RadioDisconnectedError, RadioOperationBusyError):
+        return
+    except Exception as exc:
+        logger.debug("Noise floor sampling failed: %s", exc)
+        return
+
+    if event.type == EventType.ERROR:
+        _noise_floor_supported = False
+        return
+
+    if event.type != EventType.STATS_RADIO:
+        return
+
+    noise_floor = event.payload.get("noise_floor")
+    if not isinstance(noise_floor, int):
+        return
+
+    _noise_floor_supported = True
+    await _append_sample(int(time.time()), noise_floor)
+
+
+async def _noise_floor_sampling_loop() -> None:
+    while True:
+        await sample_noise_floor_once()
+        await asyncio.sleep(NOISE_FLOOR_SAMPLE_INTERVAL_SECONDS)
+
+
+async def start_noise_floor_sampling() -> None:
+    global _noise_floor_task
+    if _noise_floor_task is not None and not _noise_floor_task.done():
+        return
+    _noise_floor_task = asyncio.create_task(_noise_floor_sampling_loop())
+
+
+async def stop_noise_floor_sampling() -> None:
+    global _noise_floor_task
+    if _noise_floor_task is None:
+        return
+    if not _noise_floor_task.done():
+        _noise_floor_task.cancel()
+        try:
+            await _noise_floor_task
+        except asyncio.CancelledError:
+            pass
+    _noise_floor_task = None
+
+
+async def get_noise_floor_history() -> dict:
+    """Return the current 24-hour in-memory noise floor history snapshot."""
+    await sample_noise_floor_once(blocking=False)
+
+    now = int(time.time())
+    cutoff = now - NOISE_FLOOR_WINDOW_SECONDS
+
+    async with _samples_lock:
+        samples = [
+            {"timestamp": timestamp, "noise_floor_dbm": noise_floor_dbm}
+            for timestamp, noise_floor_dbm in _noise_floor_samples
+            if timestamp >= cutoff
+        ]
+
+    latest = samples[-1] if samples else None
+    oldest_timestamp = samples[0]["timestamp"] if samples else None
+    coverage_seconds = 0 if oldest_timestamp is None else max(0, now - oldest_timestamp)
+
+    return {
+        "sample_interval_seconds": NOISE_FLOOR_SAMPLE_INTERVAL_SECONDS,
+        "coverage_seconds": coverage_seconds,
+        "latest_noise_floor_dbm": latest["noise_floor_dbm"] if latest else None,
+        "latest_timestamp": latest["timestamp"] if latest else None,
+        "supported": _noise_floor_supported,
+        "samples": samples,
+    }

tests/test_statistics.py

@@ -1,6 +1,7 @@
 """Tests for the statistics repository and endpoint."""
 
 import time
+from unittest.mock import AsyncMock, patch
 
 import pytest
 
@@ -347,3 +348,29 @@ class TestPathHashWidthStats:
         assert breakdown["single_byte_pct"] == pytest.approx(100 / 3, rel=1e-3)
         assert breakdown["double_byte_pct"] == pytest.approx(100 / 3, rel=1e-3)
         assert breakdown["triple_byte_pct"] == pytest.approx(100 / 3, rel=1e-3)
+
+
+class TestStatisticsEndpoint:
+    @pytest.mark.asyncio
+    async def test_statistics_endpoint_includes_noise_floor_history(self, test_db, client):
+        noise_floor_history = {
+            "sample_interval_seconds": 300,
+            "coverage_seconds": 1800,
+            "latest_noise_floor_dbm": -119,
+            "latest_timestamp": 1_700_000_000,
+            "supported": True,
+            "samples": [
+                {"timestamp": 1_699_998_200, "noise_floor_dbm": -121},
+                {"timestamp": 1_700_000_000, "noise_floor_dbm": -119},
+            ],
+        }
+
+        with patch(
+            "app.routers.statistics.get_noise_floor_history",
+            new=AsyncMock(return_value=noise_floor_history),
+        ):
+            response = await client.get("/api/statistics")
+
+        assert response.status_code == 200
+        payload = response.json()
+        assert payload["noise_floor_24h"] == noise_floor_history