83 lines
2.6 KiB
Python
83 lines
2.6 KiB
Python
#!/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())
|