import csv import json import time from collections import Counter from pathlib import Path def _build_unique_path(base_path, default_suffix): base = Path(base_path) if base.suffix == "": base = base.with_suffix(default_suffix) parent = base.parent if str(base.parent) not in ("", ".") else Path.cwd() parent.mkdir(parents=True, exist_ok=True) stamp = time.strftime("%Y%m%d_%H%M%S") candidate = parent / f"{base.stem}_{stamp}{base.suffix}" attempt = 1 while candidate.exists(): candidate = parent / f"{base.stem}_{stamp}_{attempt:02d}{base.suffix}" attempt += 1 return candidate def _box_to_metrics(box): if box is None: return "", "", "", "" x1, y1, x2, y2 = [float(v) for v in box] cx = 0.5 * (x1 + x2) cy = 0.5 * (y1 + y2) w = max(0.0, x2 - x1) h = max(0.0, y2 - y1) return round(cx, 3), round(cy, 3), round(w, 3), round(h, 3) class TrackingDecisionLogger: FIELDNAMES = [ "frame_id", "timestamp_sec", "dt_sec", "source_kind", "status", "confirmed", "target_id", "target_track_score", "locked", "locked_cx", "locked_cy", "locked_w", "locked_h", "pred_cx", "pred_cy", "pred_w", "pred_h", "hit_streak", "miss_streak", "acquire_score", "preacq_hits", "switch_candidate_hits", "best_score", "have_yolo", "yolo_mode", "yolo_raw_count", "det_count", "merged_part_count", "infer_ms", "stale_lock_active", "stale_lock_reason", "fast_handoff_active", "guidance_reset_event", "used_prediction_hold", "klt_valid", "klt_quality", "klt_points", "speed", "adaptive_chase_stage", "approach_active", "close_force_fullscan", "fast_maneuver_guard", "motion_active_zones", "wavelet_active", "wavelet_energy", "wavelet_bonus", "wavelet_hits", "autogaze_ready", "autogaze_stale", "autogaze_age", "autogaze_cells", "autogaze_ms", "guidance_active", "guidance_status", "guidance_confidence", "guidance_error_x", "guidance_error_y", "guidance_cmd_x", "guidance_cmd_y", "guidance_on_target", "events", ] def __init__(self, enabled, csv_path, summary_path, flush_every=30): self.enabled = bool(enabled) self.flush_every = max(1, int(flush_every)) self.csv_path = None self.summary_path = None self._csv_file = None self._writer = None self._rows_since_flush = 0 self.prev_confirmed = None self.prev_locked = None self.prev_target_id = None self.frames = 0 self.confirmed_frames = 0 self.locked_frames = 0 self.prediction_hold_frames = 0 self.stale_lock_frames = 0 self.fast_handoff_frames = 0 self.autogaze_used_frames = 0 self.wavelet_active_frames = 0 self.guidance_active_frames = 0 self.guidance_on_target_frames = 0 self.guidance_reset_events = 0 self.guidance_confidence_sum = 0.0 self.max_hit_streak = 0 self.max_miss_streak = 0 self.klt_quality_confirmed_sum = 0.0 self.klt_quality_confirmed_count = 0 self.infer_ms_sum = 0.0 self.yolo_mode_counts = Counter() self.event_counts = Counter() if not self.enabled: return self.csv_path = _build_unique_path(csv_path, ".csv") self.summary_path = _build_unique_path(summary_path, ".json") self._csv_file = self.csv_path.open("w", newline="", encoding="utf-8") self._writer = csv.DictWriter(self._csv_file, fieldnames=self.FIELDNAMES) self._writer.writeheader() def status_line(self): if not self.enabled: return "Tracking decision log disabled" return f"Tracking decision log: csv={self.csv_path} summary={self.summary_path}" def log_frame(self, **kwargs): if not self.enabled or self._writer is None: return confirmed = bool(kwargs.get("confirmed", False)) locked = bool(kwargs.get("locked", False)) target_id = kwargs.get("target_id", None) events = [] if self.prev_confirmed is not None: if (not self.prev_confirmed) and confirmed: events.append("confirm_enter") elif self.prev_confirmed and (not confirmed): events.append("confirm_exit") if (not self.prev_locked) and locked: events.append("lock_acquired") elif self.prev_locked and (not locked): events.append("lock_lost") if ( confirmed and (self.prev_target_id is not None) and (target_id is not None) and (self.prev_target_id != target_id) ): events.append("target_switch") guidance_reset_event = str(kwargs.get("guidance_reset_event", "")).strip() if guidance_reset_event == "stale_lock_break": events.append("stale_lock_break") if bool(kwargs.get("fast_handoff_active", False)) and ("target_switch" in events): events.append("fast_handoff_switch") if guidance_reset_event: events.append("guidance_reset") locked_cx, locked_cy, locked_w, locked_h = _box_to_metrics(kwargs.get("locked_box")) pred_cx, pred_cy, pred_w, pred_h = _box_to_metrics(kwargs.get("pred_box")) row = { "frame_id": int(kwargs.get("frame_id", -1)), "timestamp_sec": round(float(kwargs.get("timestamp_sec", 0.0)), 6), "dt_sec": round(float(kwargs.get("dt_sec", 0.0)), 6), "source_kind": kwargs.get("source_kind", ""), "status": kwargs.get("status", ""), "confirmed": int(confirmed), "target_id": "" if target_id is None else int(target_id), "target_track_score": self._maybe_float(kwargs.get("target_track_score")), "locked": int(locked), "locked_cx": locked_cx, "locked_cy": locked_cy, "locked_w": locked_w, "locked_h": locked_h, "pred_cx": pred_cx, "pred_cy": pred_cy, "pred_w": pred_w, "pred_h": pred_h, "hit_streak": int(kwargs.get("hit_streak", 0)), "miss_streak": int(kwargs.get("miss_streak", 0)), "acquire_score": int(kwargs.get("acquire_score", 0)), "preacq_hits": int(kwargs.get("preacq_hits", 0)), "switch_candidate_hits": int(kwargs.get("switch_candidate_hits", 0)), "best_score": self._maybe_float(kwargs.get("best_score")), "have_yolo": int(bool(kwargs.get("have_yolo", False))), "yolo_mode": kwargs.get("yolo_mode", ""), "yolo_raw_count": int(kwargs.get("yolo_raw_count", 0)), "det_count": int(kwargs.get("det_count", 0)), "merged_part_count": int(kwargs.get("merged_part_count", 0)), "infer_ms": round(float(kwargs.get("infer_ms", 0.0)), 3), "stale_lock_active": int(bool(kwargs.get("stale_lock_active", False))), "stale_lock_reason": kwargs.get("stale_lock_reason", ""), "fast_handoff_active": int(bool(kwargs.get("fast_handoff_active", False))), "guidance_reset_event": guidance_reset_event, "used_prediction_hold": int(bool(kwargs.get("used_prediction_hold", False))), "klt_valid": int(bool(kwargs.get("klt_valid", False))), "klt_quality": round(float(kwargs.get("klt_quality", 0.0)), 4), "klt_points": int(kwargs.get("klt_points", 0)), "speed": round(float(kwargs.get("speed", 0.0)), 3), "adaptive_chase_stage": kwargs.get("adaptive_chase_stage", ""), "approach_active": int(bool(kwargs.get("approach_active", False))), "close_force_fullscan": int(bool(kwargs.get("close_force_fullscan", False))), "fast_maneuver_guard": int(bool(kwargs.get("fast_maneuver_guard", False))), "motion_active_zones": int(kwargs.get("motion_active_zones", 0)), "wavelet_active": int(bool(kwargs.get("wavelet_active", False))), "wavelet_energy": round(float(kwargs.get("wavelet_energy", 0.0)), 4), "wavelet_bonus": round(float(kwargs.get("wavelet_bonus", 0.0)), 4), "wavelet_hits": int(kwargs.get("wavelet_hits", 0)), "autogaze_ready": int(bool(kwargs.get("autogaze_ready", False))), "autogaze_stale": int(bool(kwargs.get("autogaze_stale", True))), "autogaze_age": int(kwargs.get("autogaze_age", -1)), "autogaze_cells": int(kwargs.get("autogaze_cells", 0)), "autogaze_ms": round(float(kwargs.get("autogaze_ms", 0.0)), 3), "guidance_active": int(bool(kwargs.get("guidance_active", False))), "guidance_status": kwargs.get("guidance_status", ""), "guidance_confidence": round(float(kwargs.get("guidance_confidence", 0.0)), 4), "guidance_error_x": round(float(kwargs.get("guidance_error_x", 0.0)), 4), "guidance_error_y": round(float(kwargs.get("guidance_error_y", 0.0)), 4), "guidance_cmd_x": round(float(kwargs.get("guidance_cmd_x", 0.0)), 4), "guidance_cmd_y": round(float(kwargs.get("guidance_cmd_y", 0.0)), 4), "guidance_on_target": int(bool(kwargs.get("guidance_on_target", False))), "events": "|".join(events), } self._writer.writerow(row) self._rows_since_flush += 1 if self._rows_since_flush >= self.flush_every: self._csv_file.flush() self._rows_since_flush = 0 self.frames += 1 self.confirmed_frames += int(confirmed) self.locked_frames += int(locked) self.prediction_hold_frames += int(bool(kwargs.get("used_prediction_hold", False))) self.stale_lock_frames += int(bool(kwargs.get("stale_lock_active", False))) self.fast_handoff_frames += int(bool(kwargs.get("fast_handoff_active", False))) self.autogaze_used_frames += int(bool(kwargs.get("autogaze_ready", False)) and (not bool(kwargs.get("autogaze_stale", True)))) self.wavelet_active_frames += int(bool(kwargs.get("wavelet_active", False))) self.guidance_active_frames += int(bool(kwargs.get("guidance_active", False))) self.guidance_on_target_frames += int(bool(kwargs.get("guidance_on_target", False))) self.guidance_reset_events += int(bool(guidance_reset_event)) self.guidance_confidence_sum += float(kwargs.get("guidance_confidence", 0.0)) self.max_hit_streak = max(self.max_hit_streak, int(kwargs.get("hit_streak", 0))) self.max_miss_streak = max(self.max_miss_streak, int(kwargs.get("miss_streak", 0))) self.infer_ms_sum += float(kwargs.get("infer_ms", 0.0)) if confirmed: self.klt_quality_confirmed_sum += float(kwargs.get("klt_quality", 0.0)) self.klt_quality_confirmed_count += 1 yolo_mode = str(kwargs.get("yolo_mode", "")).strip() if yolo_mode: self.yolo_mode_counts[yolo_mode] += 1 for event in events: self.event_counts[event] += 1 self.prev_confirmed = confirmed self.prev_locked = locked self.prev_target_id = target_id def close(self): if not self.enabled: return None if self._csv_file is not None: self._csv_file.flush() self._csv_file.close() self._csv_file = None avg_klt_quality_confirmed = 0.0 if self.klt_quality_confirmed_count > 0: avg_klt_quality_confirmed = self.klt_quality_confirmed_sum / float(self.klt_quality_confirmed_count) avg_infer_ms = 0.0 if self.frames > 0: avg_infer_ms = self.infer_ms_sum / float(self.frames) avg_guidance_confidence = 0.0 if self.frames > 0: avg_guidance_confidence = self.guidance_confidence_sum / float(self.frames) summary = { "frames": self.frames, "confirmed_frames": self.confirmed_frames, "recover_frames": max(0, self.frames - self.confirmed_frames), "locked_frames": self.locked_frames, "prediction_hold_frames": self.prediction_hold_frames, "stale_lock_frames": self.stale_lock_frames, "fast_handoff_frames": self.fast_handoff_frames, "autogaze_used_frames": self.autogaze_used_frames, "wavelet_active_frames": self.wavelet_active_frames, "guidance_active_frames": self.guidance_active_frames, "guidance_on_target_frames": self.guidance_on_target_frames, "guidance_reset_events": self.guidance_reset_events, "confirm_entries": int(self.event_counts.get("confirm_enter", 0)), "confirm_exits": int(self.event_counts.get("confirm_exit", 0)), "lock_acquired_events": int(self.event_counts.get("lock_acquired", 0)), "lock_lost_events": int(self.event_counts.get("lock_lost", 0)), "target_switches": int(self.event_counts.get("target_switch", 0)), "max_hit_streak": self.max_hit_streak, "max_miss_streak": self.max_miss_streak, "avg_klt_quality_confirmed": round(avg_klt_quality_confirmed, 6), "avg_guidance_confidence": round(avg_guidance_confidence, 6), "avg_infer_ms": round(avg_infer_ms, 6), "yolo_mode_counts": dict(self.yolo_mode_counts), "event_counts": dict(self.event_counts), "csv_path": str(self.csv_path), "summary_path": str(self.summary_path), } if self.summary_path is not None: with self.summary_path.open("w", encoding="utf-8") as f: json.dump(summary, f, ensure_ascii=True, indent=2) return summary @staticmethod def _maybe_float(value): if value is None: return "" return round(float(value), 6)