118 lines
3.4 KiB
Python
Executable File
118 lines
3.4 KiB
Python
Executable File
"""Build and write the per-camera count JSON including the 4-camera total."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
|
|
from chicken_counter.types import BatchReport, CameraBatchResult
|
|
|
|
|
|
def build_camera_report_entry(
|
|
item: CameraBatchResult,
|
|
*,
|
|
output_dir: Path | None = None,
|
|
) -> dict:
|
|
if item.skipped:
|
|
return {
|
|
"skipped": True,
|
|
"skip_reason": item.skip_reason,
|
|
"total_entered": 0,
|
|
}
|
|
|
|
pipeline = item.pipeline
|
|
if pipeline is None:
|
|
return {}
|
|
|
|
source_name = Path(pipeline.source_video).name
|
|
return {
|
|
"total_entered": pipeline.total_entered_count,
|
|
"source_video": source_name,
|
|
"vis_video": _relative_output_path(pipeline.vis_video_path, output_dir),
|
|
"compressed_video": _relative_output_path(item.compressed_video_path, output_dir),
|
|
"compressed_size_mb": item.compressed_size_mb,
|
|
"frames_processed": pipeline.frames_processed,
|
|
"stopped_reason": pipeline.stopped_reason,
|
|
"elapsed_seconds": round(pipeline.elapsed_seconds, 1),
|
|
}
|
|
|
|
|
|
def build_batch_report(
|
|
date: str,
|
|
results: list[CameraBatchResult],
|
|
*,
|
|
output_dir: Path | None = None,
|
|
) -> BatchReport:
|
|
cameras: dict[str, dict] = {}
|
|
total_sum = 0
|
|
|
|
for item in results:
|
|
camera_entry = build_camera_report_entry(item, output_dir=output_dir)
|
|
if not camera_entry:
|
|
continue
|
|
cameras[item.camera_id] = camera_entry
|
|
if not item.skipped:
|
|
total_sum += camera_entry["total_entered"]
|
|
|
|
return BatchReport(
|
|
date=date,
|
|
generated_at=datetime.now(timezone.utc).isoformat(),
|
|
cameras=cameras,
|
|
total_entered_sum=total_sum,
|
|
)
|
|
|
|
|
|
def write_batch_report(report: BatchReport, output_path: str | Path) -> Path:
|
|
path = Path(output_path)
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_text(json.dumps(report.__dict__, indent=2), encoding="utf-8")
|
|
print(f"[report] wrote {path}")
|
|
return path
|
|
|
|
|
|
def write_camera_report(
|
|
date: str,
|
|
item: CameraBatchResult,
|
|
output_dir: str | Path,
|
|
) -> Path:
|
|
output_path = Path(output_dir)
|
|
output_path.mkdir(parents=True, exist_ok=True)
|
|
camera_entry = build_camera_report_entry(item, output_dir=output_path)
|
|
payload = {
|
|
"date": date,
|
|
"camera_id": item.camera_id,
|
|
"generated_at": datetime.now(timezone.utc).isoformat(),
|
|
**camera_entry,
|
|
}
|
|
path = output_path / f"{item.camera_id}_counts_{date}.json"
|
|
path.write_text(json.dumps(payload, indent=2), encoding="utf-8")
|
|
print(f"[report] wrote {path}")
|
|
return path
|
|
|
|
|
|
def persist_batch_reports(
|
|
date: str,
|
|
results: list[CameraBatchResult],
|
|
output_dir: str | Path,
|
|
) -> Path:
|
|
output_path = Path(output_dir)
|
|
latest = results[-1]
|
|
write_camera_report(date, latest, output_path)
|
|
aggregate_path = output_path / f"counts_{date}.json"
|
|
report = build_batch_report(date, results, output_dir=output_path)
|
|
write_batch_report(report, aggregate_path)
|
|
return aggregate_path
|
|
|
|
|
|
def _relative_output_path(path: str | Path | None, base_dir: Path | None = None) -> str | None:
|
|
if path is None:
|
|
return None
|
|
resolved = Path(path)
|
|
if base_dir is not None:
|
|
try:
|
|
return resolved.relative_to(base_dir).as_posix()
|
|
except ValueError:
|
|
pass
|
|
return resolved.name
|