Downsample telemetry and signal history into time buckets.

Keeps node charts bounded by limit without silently dropping the newest samples. From javastraat/meshpoint.

Co-Authored-By: Albert Einstein <javastraat@hotmail.com>

Author: KMX415

src/storage/packet_repository.py

@@ -7,6 +7,7 @@
 from src.models.packet import Packet, PacketType, Protocol
 from src.models.signal import SignalMetrics
 from src.storage.database import DatabaseManager
+from src.storage.time_bucket import bucket_seconds
 
 logger = logging.getLogger(__name__)
 
@@ -63,19 +64,31 @@ async def get_signal_history(
         limit: int = 500,
         hours: float | None = 24,
     ) -> list[dict]:
-        """RSSI/SNR samples from any packet by this node, oldest-first."""
+        """RSSI/SNR samples from any packet by this node, oldest-first.
+
+        When ``hours`` is set, samples are averaged into at most ``limit``
+        time buckets so newest points are not dropped by a plain LIMIT.
+        """
         if hours is not None and hours > 0:
             since = (
                 datetime.now(timezone.utc) - timedelta(hours=hours)
             ).isoformat()
+            span_row = await self._db.fetch_one(
+                "SELECT MIN(timestamp) AS lo, MAX(timestamp) AS hi FROM packets "
+                "WHERE source_id = ? AND rssi IS NOT NULL AND timestamp >= ?",
+                (source_id, since),
+            )
+            bucket_secs = bucket_seconds(span_row, limit, hours)
             rows = await self._db.fetch_all(
                 """
-                SELECT timestamp, rssi, snr FROM packets
+                SELECT MIN(timestamp) AS timestamp, AVG(rssi) AS rssi, AVG(snr) AS snr
+                FROM packets
                 WHERE source_id = ? AND rssi IS NOT NULL AND timestamp >= ?
+                GROUP BY CAST(strftime('%s', timestamp) AS INTEGER) / ?
                 ORDER BY timestamp ASC
                 LIMIT ?
                 """,
-                (source_id, since, limit),
+                (source_id, since, bucket_secs, limit),
             )
         else:
             rows = await self._db.fetch_all(

src/storage/telemetry_repository.py

@@ -5,6 +5,7 @@
 
 from src.models.telemetry import Telemetry
 from src.storage.database import DatabaseManager
+from src.storage.time_bucket import bucket_seconds
 
 logger = logging.getLogger(__name__)
 
@@ -49,19 +50,42 @@ async def get_history(
         limit: int = 300,
         hours: float | None = None,
     ) -> list[Telemetry]:
-        """Return telemetry oldest-first for charting (ASC)."""
+        """Return telemetry oldest-first for charting (ASC).
+
+        When ``hours`` is set, rows are averaged into at most ``limit``
+        time buckets across the real data span so newest samples are not
+        silently dropped by a plain LIMIT.
+        """
         if hours is not None and hours > 0:
             since = (
                 datetime.now(timezone.utc) - timedelta(hours=hours)
             ).isoformat()
+            span_row = await self._db.fetch_one(
+                "SELECT MIN(timestamp) AS lo, MAX(timestamp) AS hi "
+                "FROM telemetry WHERE node_id = ? AND timestamp >= ?",
+                (node_id, since),
+            )
+            bucket_secs = bucket_seconds(span_row, limit, hours)
             rows = await self._db.fetch_all(
                 """
-                SELECT * FROM telemetry
+                SELECT
+                    node_id,
+                    AVG(battery_level) AS battery_level,
+                    AVG(voltage) AS voltage,
+                    AVG(temperature) AS temperature,
+                    AVG(humidity) AS humidity,
+                    AVG(barometric_pressure) AS barometric_pressure,
+                    AVG(channel_utilization) AS channel_utilization,
+                    AVG(air_util_tx) AS air_util_tx,
+                    AVG(uptime_seconds) AS uptime_seconds,
+                    MIN(timestamp) AS timestamp
+                FROM telemetry
                 WHERE node_id = ? AND timestamp >= ?
+                GROUP BY CAST(strftime('%s', timestamp) AS INTEGER) / ?
                 ORDER BY timestamp ASC
                 LIMIT ?
                 """,
-                (node_id, since, limit),
+                (node_id, since, bucket_secs, limit),
             )
         else:
             rows = await self._db.fetch_all(

src/storage/time_bucket.py

@@ -0,0 +1,33 @@
+"""Shared time-bucketing helper for downsampled history queries.
+
+Used by TelemetryRepository.get_history() and
+PacketRepository.get_signal_history() so a long-lived node's chart data
+stays bounded to roughly ``limit`` points instead of a plain LIMIT
+silently dropping the newest rows.
+
+Credit: javastraat/meshpoint ``b10610a``.
+"""
+
+from __future__ import annotations
+
+from datetime import datetime
+
+
+def bucket_seconds(span_row: dict | None, limit: int, hours: float) -> int:
+    """Bucket width in seconds for a downsampled history query.
+
+    Derived from the actual span of matching data (``lo``/``hi``) rather
+    than the requested ``hours`` window, so over-sized request windows
+    do not crush a short real history into coarse buckets. Floored at
+    60 seconds.
+    """
+    limit = max(limit, 1)
+    lo = span_row.get("lo") if span_row else None
+    hi = span_row.get("hi") if span_row else None
+    if lo and hi:
+        span = (
+            datetime.fromisoformat(hi) - datetime.fromisoformat(lo)
+        ).total_seconds()
+        if span > 0:
+            return max(60, int(span / limit))
+    return max(60, int((hours * 3600) / limit))

tests/test_time_bucket.py

@@ -0,0 +1,69 @@
+"""Unit tests for history time-bucket width helper.
+
+Credit: javastraat/meshpoint ``b10610a``.
+"""
+
+from __future__ import annotations
+
+import unittest
+from datetime import datetime, timedelta, timezone
+
+from src.models.telemetry import Telemetry
+from src.storage.database import DatabaseManager
+from src.storage.telemetry_repository import TelemetryRepository
+from src.storage.time_bucket import bucket_seconds
+
+
+class BucketSecondsTest(unittest.TestCase):
+    def test_uses_actual_span_not_requested_hours(self):
+        lo = datetime(2026, 1, 1, tzinfo=timezone.utc)
+        hi = lo + timedelta(hours=2)
+        secs = bucket_seconds(
+            {"lo": lo.isoformat(), "hi": hi.isoformat()},
+            limit=10,
+            hours=100000,
+        )
+        self.assertEqual(secs, max(60, int((2 * 3600) / 10)))
+
+    def test_falls_back_to_hours_when_no_span(self):
+        self.assertEqual(bucket_seconds(None, limit=10, hours=1), 360)
+
+    def test_floor_at_sixty_seconds(self):
+        lo = datetime(2026, 1, 1, tzinfo=timezone.utc)
+        hi = lo + timedelta(seconds=30)
+        self.assertEqual(
+            bucket_seconds(
+                {"lo": lo.isoformat(), "hi": hi.isoformat()},
+                limit=100,
+                hours=1,
+            ),
+            60,
+        )
+
+
+class TelemetryHistoryBucketTest(unittest.IsolatedAsyncioTestCase):
+    async def asyncSetUp(self):
+        self.db = DatabaseManager(":memory:")
+        await self.db.connect()
+        self.repo = TelemetryRepository(self.db)
+
+    async def asyncTearDown(self):
+        await self.db.disconnect()
+
+    async def test_hours_path_keeps_newest_when_over_limit(self):
+        now = datetime.now(timezone.utc)
+        for i in range(20):
+            await self.repo.insert(
+                Telemetry(
+                    node_id="n1",
+                    temperature=float(i),
+                    timestamp=now - timedelta(minutes=19 - i),
+                )
+            )
+        rows = await self.repo.get_history("n1", limit=5, hours=24)
+        self.assertLessEqual(len(rows), 5)
+        self.assertGreaterEqual(rows[-1].temperature, 15.0)
+
+
+if __name__ == "__main__":
+    unittest.main()