ilaandClaude Opus 4.8
2688f2d449
v1/report.py reads events.jsonl and lists each confirmed event with its latency (flags the 1-3s target) and screenshot, or reports a clean pass when empty. Supports T-204 latency records and false-alarm counting. 78 tests pass. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
87 lines
2.7 KiB
Python
87 lines
2.7 KiB
Python
"""Summarize confirmed fall events for review, latency records and
|
|
false-alarm counting during acceptance sessions.
|
|
|
|
Reads the JSONL written by ``alerts.EventArtifactWriter``. Use it after a test
|
|
session: run some normal activity (walking, sitting, bending, picking up) and
|
|
check that no events were recorded, or read back the latency of real falls.
|
|
|
|
python -m v1.report # scans artifacts/events
|
|
python -m v1.report artifacts/events # a source directory
|
|
python -m v1.report .../20260721/events.jsonl
|
|
"""
|
|
|
|
import json
|
|
import sys
|
|
from pathlib import Path
|
|
from typing import Dict, List
|
|
|
|
|
|
def load_events(jsonl_path: Path) -> List[Dict]:
|
|
path = Path(jsonl_path)
|
|
if not path.is_file():
|
|
return []
|
|
events = []
|
|
for line in path.read_text(encoding="utf-8").splitlines():
|
|
line = line.strip()
|
|
if line:
|
|
events.append(json.loads(line))
|
|
return events
|
|
|
|
|
|
def collect_jsonl(root: Path) -> List[Path]:
|
|
"""Return every events.jsonl under a source/day directory, or the file itself."""
|
|
|
|
path = Path(root)
|
|
if path.is_file():
|
|
return [path]
|
|
return sorted(path.glob("**/events.jsonl"))
|
|
|
|
|
|
def summarize(events: List[Dict]) -> Dict:
|
|
latencies = [float(e.get("latency_seconds", 0.0)) for e in events]
|
|
within = [lat for lat in latencies if 1.0 <= lat <= 3.0]
|
|
return {
|
|
"count": len(events),
|
|
"latencies": latencies,
|
|
"latency_in_target": len(within),
|
|
}
|
|
|
|
|
|
def format_report(jsonl_path: Path, events: List[Dict]) -> str:
|
|
summary = summarize(events)
|
|
lines = [
|
|
"{0}".format(jsonl_path),
|
|
" events: {0} (latency in 1-3s: {1}/{0})".format(
|
|
summary["count"], summary["latency_in_target"]
|
|
),
|
|
]
|
|
for event in events:
|
|
lines.append(
|
|
" {0} track={1} utc={2} latency={3:.2f}s shot={4}".format(
|
|
event.get("event_id", "?"),
|
|
event.get("track_id", "?"),
|
|
event.get("confirmed_at_utc", "?"),
|
|
float(event.get("latency_seconds", 0.0)),
|
|
event.get("screenshot", "?"),
|
|
)
|
|
)
|
|
if not events:
|
|
lines.append(" (no confirmed events — a clean pass for negative activity)")
|
|
return "\n".join(lines)
|
|
|
|
|
|
def main(argv=None) -> int:
|
|
argv = list(sys.argv[1:] if argv is None else argv)
|
|
root = Path(argv[0]) if argv else Path("artifacts/events")
|
|
files = collect_jsonl(root)
|
|
if not files:
|
|
print("no events.jsonl found under {0}".format(root))
|
|
return 0
|
|
for jsonl_path in files:
|
|
print(format_report(jsonl_path, load_events(jsonl_path)))
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|