feat(v1): add event-log summary tool for acceptance evidence
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>
This commit is contained in:
@@ -0,0 +1,86 @@
|
||||
"""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())
|
||||
Reference in New Issue
Block a user