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