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.

120 lines
3.1 KiB
Python

from dataclasses import dataclass
@dataclass(frozen=True)
class HandoffDecision:
stale_lock_active: bool
allow_fast_handoff: bool
required_switch_hits: int
disable_prediction_hold: bool
reasons: list[str]
def reason_text(self) -> str:
return "|".join(self.reasons)
def evaluate_stale_lock(
*,
confirmed,
have_fresh_candidate,
target_has_fresh_update,
candidate_dist_px,
pred_diag_px,
miss_streak,
klt_valid,
klt_quality,
klt_iou,
candidate_motion_ok,
target_motion_ok,
stale_break_dist_diag,
stale_break_min_miss,
stale_break_klt_quality,
stale_break_klt_iou,
):
reasons = []
if not confirmed or not have_fresh_candidate:
return HandoffDecision(False, False, 0, False, [])
far_enough = candidate_dist_px >= max(40.0, stale_break_dist_diag * max(pred_diag_px, 1.0))
weak_anchor = (
(not klt_valid)
or (klt_quality < stale_break_klt_quality)
or (klt_iou < stale_break_klt_iou)
or (miss_streak >= stale_break_min_miss)
or (candidate_motion_ok and not target_motion_ok)
or (not target_has_fresh_update)
)
if not target_has_fresh_update:
reasons.append("target_not_fresh")
if not klt_valid:
reasons.append("klt_invalid")
if klt_quality < stale_break_klt_quality:
reasons.append("weak_klt_quality")
if klt_iou < stale_break_klt_iou:
reasons.append("weak_klt_iou")
if miss_streak >= stale_break_min_miss:
reasons.append("miss_streak")
if candidate_motion_ok and not target_motion_ok:
reasons.append("motion_conflict")
stale = bool(far_enough and weak_anchor)
return HandoffDecision(
stale_lock_active=stale,
allow_fast_handoff=stale,
required_switch_hits=0,
disable_prediction_hold=stale,
reasons=reasons if stale else [],
)
def compute_fast_handoff_hits(
*,
stale_lock_active,
motion_switch_mode,
approach_extra_hits,
fast_maneuver_extra_hits,
default_switch_hits,
fast_handoff_hits,
motion_switch_hits,
):
if stale_lock_active:
return int(fast_handoff_hits)
hits = int(default_switch_hits) + int(approach_extra_hits) + int(fast_maneuver_extra_hits)
if motion_switch_mode:
hits = max(hits, int(motion_switch_hits))
return int(hits)
def should_override_guidance(
*,
confirmed,
target_track_missing,
have_fresh_candidate,
candidate_matches_target,
miss_streak,
override_miss_ge,
):
return bool(
confirmed
and target_track_missing
and have_fresh_candidate
and (not candidate_matches_target)
and (int(miss_streak) >= int(override_miss_ge))
)
def pick_guidance_override_box(*, track_candidate_box, raw_yolo_box):
if track_candidate_box is not None:
return track_candidate_box
return raw_yolo_box
def update_guidance_override_latch(*, new_box, prev_box, prev_ttl, max_ttl):
if new_box is not None:
return new_box, int(max(0, max_ttl))
if prev_box is not None and int(prev_ttl) > 0:
return prev_box, int(prev_ttl) - 1
return None, 0