You cannot select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
348 lines
14 KiB
Python
348 lines
14 KiB
Python
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)
|