Document real-world peak detector validation

2026-09-04 08:49:12 +02:00
parent 5068c95158
commit 1b8fe600d5
5 changed files with 135 additions and 17 deletions
@@ -0,0 +1,82 @@
+#!/usr/bin/env python3
+"""Validate the peak detector against an ignored local Bento 2 capture.
+
+This developer utility reads no network data and does not modify the capture.
+Run it from the repository root when
+data/raw/enlyze/bento2-roll-weight-extended.raw.json is available.
+"""
+
+from __future__ import annotations
+
+import json
+from datetime import datetime
+from pathlib import Path
+
+from production_analytics.calculations import (
+    PeakCycleDetector,
+    PeakDetectorConfig,
+    TimeSeriesSample,
+)
+
+
+CAPTURE = Path("data/raw/enlyze/bento2-roll-weight-extended.raw.json")
+VARIABLE_UUID = "059c0015-9aff-4270-a4ce-f21b83f99378"
+EXPECTED_PEAKS = (
+    ("2026-09-04T00:46:50+00:00", 936.0),
+    ("2026-09-04T00:57:10+00:00", 919.0),
+    ("2026-09-04T01:07:20+00:00", 935.0),
+    ("2026-09-04T01:17:20+00:00", 930.0),
+    ("2026-09-04T01:27:20+00:00", 930.0),
+    ("2026-09-04T01:37:30+00:00", 920.0),
+    ("2026-09-04T01:48:00+00:00", 909.0),
+    ("2026-09-04T02:14:20+00:00", 1195.0),
+    ("2026-09-04T02:24:10+00:00", 921.0),
+    ("2026-09-04T02:35:40+00:00", 930.0),
+    ("2026-09-04T02:46:10+00:00", 939.0),
+    ("2026-09-04T02:56:20+00:00", 912.0),
+    ("2026-09-04T03:06:30+00:00", 914.0),
+)
+
+
+def main() -> int:
+    if not CAPTURE.is_file():
+        print(f"Bento 2 capture is not available: {CAPTURE}")
+        return 1
+
+    payload = json.loads(CAPTURE.read_text(encoding="utf-8"))
+    data = payload["response"]["body"]["data"]
+    value_index = data["columns"].index(VARIABLE_UUID)
+    samples = (
+        TimeSeriesSample(
+            datetime.fromisoformat(record[0].replace("Z", "+00:00")),
+            float(record[value_index]),
+        )
+        for record in data["records"]
+        if record[value_index] is not None
+    )
+    detector = PeakCycleDetector(
+        PeakDetectorConfig(
+            min_peak=800.0,
+            drop_ratio=0.75,
+            hold_seconds=60.0,
+            max_sample_gap_seconds=20.0,
+        )
+    )
+    peaks = list(detector.process_many(samples))
+    observed = tuple((peak.peak_timestamp.isoformat(), peak.value) for peak in peaks)
+
+    print(f"Detected peaks: {len(peaks)}")
+    for peak in peaks:
+        print(
+            f"peak={peak.value:.0f} kg at {peak.peak_timestamp.isoformat()} "
+            f"confirmed={peak.confirmed_at.isoformat()}"
+        )
+    if observed != EXPECTED_PEAKS:
+        print("Validation failed: detected peaks differ from the expected extended-window result.")
+        return 2
+    print("Validation passed: detected peaks match the 13 expected completed peaks.")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())