You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
 

139 lines
7.5 KiB

"""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()