"""UTF-8 BOM CSV logs for tracking frames, events, and summaries.""" from __future__ import annotations import csv import json from datetime import datetime from pathlib import Path from typing import TextIO from .config_loader import TrackingLoggingSettings from .exceptions import TrackingLogError from .models import FrameInfo, TrackEvent, TrackedPerson, TrackState class TrackingLogger: """Own the three Phase 2 CSV streams and prevent duplicate summaries.""" FRAME_FIELDS = ["experiment_id", "frame_number", "timestamp", "elapsed_time_sec", "track_id", "is_confirmed", "confidence", "bbox_x1", "bbox_y1", "bbox_x2", "bbox_y2", "bbox_width", "bbox_height", "center_x", "center_y", "foot_x", "foot_y", "inside_evaluation_region", "track_age_frames", "missed_frames", "movement_direction", "trigger_crossed"] SUMMARY_FIELDS = ["experiment_id", "track_id", "first_seen_time", "last_seen_time", "first_seen_frame", "last_seen_frame", "duration_sec", "total_visible_frames", "total_missed_frames", "max_missed_frames", "entered_evaluation_region", "entry_time", "entry_frame", "exited_evaluation_region", "exit_time", "exit_frame", "dwell_time_sec", "trigger_crossed", "trigger_crossing_time", "trigger_crossing_frame", "trigger_crossing_direction", "movement_direction", "trajectory_points", "valid_for_counting", "invalid_reason"] EVENT_FIELDS = ["experiment_id", "event_type", "track_id", "frame_number", "timestamp", "elapsed_time_sec", "x", "y", "direction", "metadata_json"] def __init__(self, config: TrackingLoggingSettings, experiment_id: str, run_stamp: str | None = None) -> None: self.config = config self.experiment_id = experiment_id self.run_stamp = run_stamp or datetime.now().strftime("%Y%m%d_%H%M%S") self.directory = Path(config.directory) self._files: dict[str, TextIO] = {} self._writers: dict[str, csv.DictWriter[str]] = {} self._summary_ids: set[int] = set() self._frames_since_flush = 0 def open(self) -> None: if not self.config.enabled: return self.directory.mkdir(parents=True, exist_ok=True) try: self._open_stream("frames", "tracking_frames", self.FRAME_FIELDS, self.config.log_frame_tracks) self._open_stream("events", "events", self.EVENT_FIELDS, self.config.log_events) self._open_stream("summary", "track_summary", self.SUMMARY_FIELDS, self.config.log_summaries) except OSError as exc: self.close() raise TrackingLogError(f"Failed to open tracking logs: {exc}") from exc def _open_stream(self, key: str, suffix: str, fields: list[str], enabled: bool) -> None: if not enabled: return path = self.directory / f"{self.run_stamp}_{self.experiment_id}_{suffix}.csv" stream = path.open("w", encoding="utf-8-sig", newline="") writer = csv.DictWriter(stream, fieldnames=fields) writer.writeheader() self._files[key], self._writers[key] = stream, writer def write_frame_tracks( self, frame_info: FrameInfo, tracked_persons: list[TrackedPerson], track_states: dict[int, TrackState] | None = None, ) -> None: writer = self._writers.get("frames") if writer is None: return states = track_states or {} for person in tracked_persons: bbox, state = person.bbox, states.get(person.track_id) writer.writerow({ "experiment_id": self.experiment_id, "frame_number": frame_info.frame_number, "timestamp": frame_info.timestamp_iso, "elapsed_time_sec": f"{frame_info.elapsed_time_sec:.6f}", "track_id": person.track_id, "is_confirmed": person.is_confirmed, "confidence": f"{person.confidence:.6f}", "bbox_x1": bbox.x1, "bbox_y1": bbox.y1, "bbox_x2": bbox.x2, "bbox_y2": bbox.y2, "bbox_width": bbox.width, "bbox_height": bbox.height, "center_x": person.center_x, "center_y": person.center_y, "foot_x": person.foot_x, "foot_y": person.foot_y, "inside_evaluation_region": state.current_inside_region if state else person.inside_evaluation_region, "track_age_frames": person.track_age_frames, "missed_frames": person.missed_frames, "movement_direction": state.movement_direction if state else "unknown", "trigger_crossed": state.trigger_crossed if state else False, }) self._frames_since_flush += 1 if self._frames_since_flush >= self.config.flush_interval_frames: self.flush() def write_events(self, events: list[TrackEvent]) -> None: writer = self._writers.get("events") if writer is None: return for event in events: writer.writerow({ "experiment_id": self.experiment_id, "event_type": event.event_type, "track_id": event.track_id, "frame_number": event.frame_number, "timestamp": event.timestamp_iso, "elapsed_time_sec": f"{event.elapsed_time_sec:.6f}", "x": event.x, "y": event.y, "direction": event.direction or "", "metadata_json": json.dumps(event.metadata, ensure_ascii=False), }) def write_summary(self, state: TrackState) -> None: writer = self._writers.get("summary") if writer is None or state.track_id in self._summary_ids: return writer.writerow({ "experiment_id": self.experiment_id, "track_id": state.track_id, "first_seen_time": state.first_seen_time, "last_seen_time": state.last_seen_time, "first_seen_frame": state.first_seen_frame, "last_seen_frame": state.last_seen_frame, "duration_sec": f"{state.duration_sec:.6f}", "total_visible_frames": state.total_visible_frames, "total_missed_frames": state.total_missed_frames, "max_missed_frames": state.max_missed_frames, "entered_evaluation_region": state.entered_evaluation_region, "entry_time": state.entry_time or "", "entry_frame": state.entry_frame if state.entry_frame is not None else "", "exited_evaluation_region": state.exited_evaluation_region, "exit_time": state.exit_time or "", "exit_frame": state.exit_frame if state.exit_frame is not None else "", "dwell_time_sec": f"{state.dwell_time_sec:.6f}", "trigger_crossed": state.trigger_crossed, "trigger_crossing_time": state.trigger_crossing_time or "", "trigger_crossing_frame": state.trigger_crossing_frame if state.trigger_crossing_frame is not None else "", "trigger_crossing_direction": state.trigger_crossing_direction or "", "movement_direction": state.movement_direction, "trajectory_points": len(state.trajectory), "valid_for_counting": state.valid_for_counting, "invalid_reason": state.invalid_reason, }) self._summary_ids.add(state.track_id) def flush(self) -> None: for stream in self._files.values(): stream.flush() self._frames_since_flush = 0 def close(self) -> None: for stream in self._files.values(): stream.close() self._files.clear() self._writers.clear() def __enter__(self) -> "TrackingLogger": self.open() return self def __exit__(self, *_args: object) -> None: self.close()