class Sam2Mot(HybridBaseTracker):
"""Hybrid bbox + mask tracker with three-stage matching, COI, and frame-out recovery.
This tracker uses externally provided segmentation masks for mask-IoU-based
association. Despite the name (for continuity with the paper), it does **not**
require SAM2 – any source of per-detection masks works (Mask R-CNN, etc.).
"""
supports_masks = True
supports_obb = True
def __init__(
self,
# Base tracker params
det_thresh: float = 0.3,
max_age: int = 60,
min_hits: int = 1,
iou_threshold: float = 0.3,
per_class: bool = False,
# Sam2Mot-specific params
tolerance_frames: int = 30,
memory_window: int = 25,
cost_weight: float = 0.5,
tau_r: float = 0.8,
tau_p: float = 0.5,
tau_s: float = 0.3,
density_threshold: float = 0.9,
second_stage_iou_threshold: float = 0.3,
frame_out_d_thre: float = 0.6,
miou_threshold: float = 0.8,
untracked_ratio_threshold: float = 0.5,
new_track_thresh: float = 0.5,
obb_theta_damping: float = 0.8,
**kwargs,
):
super().__init__(
det_thresh=det_thresh,
max_age=max_age,
min_hits=min_hits,
iou_threshold=iou_threshold,
per_class=per_class,
**kwargs,
)
self.tolerance_frames = tolerance_frames
self.memory_window = memory_window
self.cost_weight = cost_weight
self.density_threshold = density_threshold
self.second_stage_iou_threshold = second_stage_iou_threshold
self.frame_out_d_thre = frame_out_d_thre
self.new_track_thresh = new_track_thresh
self.obb_theta_damping = float(np.clip(obb_theta_damping, 0.0, 1.0))
self.trajectory_manager = _TrajectoryManager(
tau_r=tau_r,
tau_p=tau_p,
tau_s=tau_s,
tolerance_frames=tolerance_frames,
untracked_ratio_threshold=untracked_ratio_threshold,
)
self.coi = _CrossObjectInteraction(miou_threshold=miou_threshold)
# Internal state
self._tracks: List[_Track] = []
self._next_id = 1
LOGGER.info(
f"Sam2Mot: det_thresh={det_thresh}, tolerance_frames={tolerance_frames}, "
f"cost_weight={cost_weight}, density_threshold={density_threshold}, "
f"miou_threshold={miou_threshold}, obb_theta_damping={self.obb_theta_damping}"
)
def reset(self):
"""Reset tracker state."""
self._reset_common_state()
self._tracks = []
self._next_id = 1
# ------------------------------------------------------------------
# Core update
# ------------------------------------------------------------------
def _damped_obb_update(self, measurement: np.ndarray, reference: np.ndarray | None) -> np.ndarray:
"""Align an OBB measurement and damp its angular correction."""
if reference is None:
updated = np.asarray(measurement, dtype=np.float32).copy().reshape(5)
updated[4] = float(normalize_angle(updated[4]))
return updated
ref = np.asarray(reference, dtype=np.float32).reshape(5)
updated = align_obb_measurement(measurement, ref)
theta_delta = float(normalize_angle(float(updated[4]) - float(ref[4])))
theta_gain = 1.0 - self.obb_theta_damping
updated[4] = float(normalize_angle(float(ref[4]) + theta_gain * theta_delta))
return updated
@staticmethod
def _append_track_history(track: _Track) -> None:
"""Append geometry using the common AABB-4/OBB-corners-8 display contract."""
if track.history_observations is None:
return
if track.obb is not None:
geometry, track._plot_angle = smooth_obb_corners(track.obb, track._plot_angle)
else:
geometry = np.asarray(track.bbox, dtype=np.float32).reshape(-1)[:4]
track.history_observations.append(np.asarray(geometry, dtype=np.float32).copy())
def _track_detections(self, dets: np.ndarray, img: np.ndarray, embs: np.ndarray = None, masks: np.ndarray = None):
"""Process one frame.
Args:
dets: (N, 6) detections [x1, y1, x2, y2, conf, cls].
img: Current frame (H, W, 3).
embs: Ignored (no ReID).
masks: (N, H, W) binary masks aligned to dets.
Returns:
Tuple of (tracks_array, output_masks):
tracks_array: (M, 8) [x1, y1, x2, y2, id, conf, cls, det_ind]
output_masks: (M, H, W) or None
"""
self.frame_count += 1
frame_id = self.frame_count
H, W = img.shape[:2]
batch = self.make_detection_batch(dets, masks=masks)
det_inds = batch.det_inds
# Mask operations use enclosing AABBs while OBB geometry is preserved for output.
det_obbs = batch.boxes.copy() if self.is_obb and len(batch) else None
det_bboxes = xywha_to_xyxy(det_obbs) if det_obbs is not None else batch.boxes.copy()
det_confs = batch.confs
det_classes = batch.clss.astype(int)
n_dets = len(batch)
# Masks array (may be at a different resolution than the image)
det_masks = masks if (masks is not None and len(masks) == n_dets) else None
if det_masks is not None:
mH, mW = det_masks.shape[1], det_masks.shape[2]
else:
mH, mW = H, W
# Letterbox-aware scale factors: image coords -> mask coords
# Masks are in letterboxed model space (square with padding), not proportional to image
scale = min(mH / H, mW / W)
self._mask_scale = scale
self._mask_pad_x = (mW - int(W * scale)) / 2.0
self._mask_pad_y = (mH - int(H * scale)) / 2.0
# Update existing track states
for track in self._tracks:
track.prev_bbox = track.bbox.copy() if track.bbox is not None else None
track.age += 1
active_tracks = [t for t in self._tracks if t.state != TrackState.LOST]
# --- Identify frame-out candidates ---
# Only move to frame-out after a long gap (10+ frames unmatched)
frame_out_tracks = []
normal_tracks = []
for t in active_tracks:
if (
t.last_matched_frame is not None
and t.last_matched_frame <= frame_id - 10
and not t.is_dense
and t.age > 1
):
t.state = TrackState.FRAME_OUT
t.mask = None
frame_out_tracks.append(t)
else:
normal_tracks.append(t)
# === Stage 1+2: Two-stage matching on normal tracks ===
all_matches, unmatched_dets, unmatched_trk_indices, second_stage_matches = self._two_stage_matching(
det_bboxes,
det_confs,
normal_tracks,
det_masks=det_masks,
det_obbs=det_obbs,
)
# Apply matches
matched_track_ids = set()
tracks_need_reconstruction = []
for det_idx, trk_idx in all_matches:
track = normal_tracks[trk_idx]
bbox = det_bboxes[det_idx]
conf = det_confs[det_idx]
density = self._compute_density(det_idx, det_bboxes)
track.last_matched_density = density
track.is_dense = density > self.frame_out_d_thre
track.last_matched_frame = frame_id
track.last_matched_bbox = bbox.copy()
matched_track_ids.add(track.id)
is_second_stage = (det_idx, trk_idx) in set(second_stage_matches)
if is_second_stage:
if density >= self.density_threshold:
# Skip reconstruction for dense second-stage
pass
else:
tracks_need_reconstruction.append((track, det_idx))
else:
# Crop mask to detection bbox region (in mask coordinates)
if track.mask is not None and det_masks is not None and det_idx < len(det_masks):
x1 = max(0, int(bbox[0] * self._mask_scale + self._mask_pad_x))
y1 = max(0, int(bbox[1] * self._mask_scale + self._mask_pad_y))
x2 = min(mW, int(bbox[2] * self._mask_scale + self._mask_pad_x))
y2 = min(mH, int(bbox[3] * self._mask_scale + self._mask_pad_y))
cropped = np.zeros_like(track.mask)
cropped[y1:y2, x1:x2] = track.mask[y1:y2, x1:x2]
track.mask = cropped
# Check if quality reconstruction needed
if track.state == TrackState.PENDING and conf > self.trajectory_manager.tau_r:
if density < self.density_threshold:
tracks_need_reconstruction.append((track, det_idx))
# Update velocity after resolving equivalent OBB forms. This
# prevents a width/height swap from becoming a spurious pi/2 turn.
if det_obbs is not None and track.obb is not None:
aligned_obb = self._damped_obb_update(det_obbs[det_idx], track.obb)
new_vel = aligned_obb - track.obb
new_vel[4] = float(normalize_angle(new_vel[4]))
else:
aligned_obb = None
new_vel = bbox - track.bbox
if track.velocity is not None:
track.velocity = 0.6 * track.velocity + 0.4 * new_vel
else:
track.velocity = new_vel
track.obb = aligned_obb
track.bbox = xywha_to_xyxy(aligned_obb)[0] if aligned_obb is not None else bbox.copy()
track.last_matched_bbox = track.bbox.copy()
track.last_matched_obb = None if aligned_obb is None else aligned_obb.copy()
track.confidence = conf
track.conf_history.append(conf)
track.last_seen_frame = frame_id
track.lost_frames = 0
track.cls = det_classes[det_idx]
track.det_ind = int(det_inds[det_idx])
# Assign mask from detection
if det_masks is not None and det_idx < len(det_masks):
track.mask = det_masks[det_idx]
# Update state
new_state = self.trajectory_manager.classify_state(conf)
if new_state != TrackState.LOST:
track.state = new_state
self._append_track_history(track)
# --- Cross-Object Interaction ---
if len(active_tracks) > 1:
coi_skip_ids = self.coi.detect_and_resolve(active_tracks)
for track in active_tracks:
if track.id in coi_skip_ids and track.skip_memory_current:
track.mask = None
track.skip_memory_current = False
# Reconstruct tracks that need it
for track, det_idx in tracks_need_reconstruction:
if det_masks is not None and det_idx < len(det_masks):
track.mask = det_masks[det_idx]
track.state = TrackState.RELIABLE
if det_obbs is not None:
# Matched tracks have already received their single damped
# geometry update above; reconstruction refreshes the mask and
# confidence without applying the same angle correction twice.
if track.obb is None:
track.obb = self._damped_obb_update(det_obbs[det_idx], None)
track.bbox = xywha_to_xyxy(track.obb)[0]
track.last_matched_obb = track.obb.copy()
else:
track.bbox = det_bboxes[det_idx].copy()
track.obb = None
track.last_matched_bbox = track.bbox.copy()
track.confidence = det_confs[det_idx]
track.conf_history.append(det_confs[det_idx])
track.det_ind = int(det_inds[det_idx])
# Increment lost frames for unmatched tracks
for t in self._tracks:
if t.id not in matched_track_ids:
t.lost_frames += 1
if t.lost_frames > self.trajectory_manager.tolerance_frames:
t.state = TrackState.LOST
# === Stage 3: Frame-out recovery ===
if frame_out_tracks and unmatched_dets:
fo_matches = self._frame_out_matching(
det_bboxes,
unmatched_dets,
frame_out_tracks,
det_obbs=det_obbs,
)
for det_idx, fo_track in fo_matches:
bbox = det_bboxes[det_idx]
conf = det_confs[det_idx]
density = self._compute_density(det_idx, det_bboxes)
fo_track.state = TrackState.RELIABLE
if det_obbs is not None and fo_track.obb is not None:
previous_obb = fo_track.obb.copy()
fo_track.obb = self._damped_obb_update(det_obbs[det_idx], previous_obb)
new_velocity = fo_track.obb - previous_obb
new_velocity[4] = float(normalize_angle(new_velocity[4]))
if fo_track.velocity is not None:
fo_track.velocity = 0.6 * fo_track.velocity + 0.4 * new_velocity
else:
fo_track.velocity = new_velocity
fo_track.bbox = xywha_to_xyxy(fo_track.obb)[0]
else:
fo_track.bbox = bbox.copy()
fo_track.obb = self._damped_obb_update(det_obbs[det_idx], None) if det_obbs is not None else None
fo_track.confidence = conf
fo_track.conf_history.append(conf)
fo_track.last_seen_frame = frame_id
fo_track.lost_frames = 0
fo_track.last_matched_frame = frame_id
fo_track.last_matched_bbox = bbox.copy()
fo_track.last_matched_obb = None if fo_track.obb is None else fo_track.obb.copy()
fo_track.last_matched_density = density
fo_track.is_dense = density > self.frame_out_d_thre
fo_track.cls = det_classes[det_idx]
fo_track.det_ind = int(det_inds[det_idx])
self._append_track_history(fo_track)
if det_masks is not None and det_idx < len(det_masks):
fo_track.mask = det_masks[det_idx]
matched_track_ids.add(fo_track.id)
unmatched_dets = [d for d in unmatched_dets if d != det_idx]
# === Add new tracks for unmatched detections ===
if unmatched_dets:
tracked_masks_list = [t.mask for t in self._tracks if t.mask is not None and t.state != TrackState.LOST]
guard_bboxes = []
for t in active_tracks:
if t.mask is None or not np.any(t.mask):
gb = t.last_matched_bbox if t.last_matched_bbox is not None else t.bbox
if gb is not None:
guard_bboxes.append(gb)
elif t.is_dense and t.last_matched_bbox is not None:
guard_bboxes.append(t.last_matched_bbox)
untracked = self.trajectory_manager.compute_untracked_mask(
(mH, mW),
tracked_masks_list,
guard_bboxes,
scale=(self._mask_scale, self._mask_pad_y, self._mask_pad_x),
)
for det_idx in unmatched_dets:
bbox = det_bboxes[det_idx]
conf = det_confs[det_idx]
# Only create new tracks from high-confidence detections
if conf < self.new_track_thresh:
continue
if not self.trajectory_manager.should_add_detection(
bbox, untracked, scale=(self._mask_scale, self._mask_pad_y, self._mask_pad_x)
):
continue
density = self._compute_density(det_idx, det_bboxes)
mask = det_masks[det_idx] if (det_masks is not None and det_idx < len(det_masks)) else None
new_obb = self._damped_obb_update(det_obbs[det_idx], None) if det_obbs is not None else None
new_track = _Track(
id=self._next_id,
bbox=bbox.copy(),
mask=mask,
confidence=conf,
state=TrackState.RELIABLE,
lost_frames=0,
age=1,
conf_history=deque(maxlen=self.memory_window),
last_seen_frame=frame_id,
init_frame=frame_id,
last_matched_frame=frame_id,
last_matched_bbox=bbox.copy(),
last_matched_density=density,
is_dense=density > self.frame_out_d_thre,
cls=det_classes[det_idx],
det_ind=int(det_inds[det_idx]),
obb=new_obb,
last_matched_obb=None if new_obb is None else new_obb.copy(),
history_observations=deque(maxlen=self.max_obs),
)
new_track.conf_history.append(conf)
self._append_track_history(new_track)
self._tracks.append(new_track)
matched_track_ids.add(self._next_id)
self._next_id += 1
# === Remove dead tracks ===
self._tracks = [t for t in self._tracks if not self.trajectory_manager.should_remove(t)]
# === Build output ===
output_tracks = []
output_masks_list = []
for track in self._tracks:
if track.id not in matched_track_ids:
continue
if track.age < self.min_hits and self.frame_count > self.min_hits:
continue
box = track.obb if self.is_obb else track.bbox
output_tracks.append(
self.format_output_row(
box,
track.id,
track.confidence,
track.cls,
track.det_ind,
)
)
output_masks_list.append(track.mask)
if output_tracks:
self.active_tracks = [track for track in self._tracks if track.id in matched_track_ids]
tracks_array = self.format_output_rows(output_tracks, dtype=np.float32)
# Build output masks at mask resolution (not image resolution)
has_any_mask = any(m is not None and m.shape == (mH, mW) and np.any(m) for m in output_masks_list)
if has_any_mask:
out_masks = np.zeros((len(output_masks_list), mH, mW), dtype=np.uint8)
for i, m in enumerate(output_masks_list):
if m is not None and m.shape == (mH, mW):
out_masks[i] = m
return tracks_array, out_masks
return tracks_array, None
else:
self.active_tracks = []
return self.empty_output(dtype=np.float32), None
# ------------------------------------------------------------------
# Matching helpers
# ------------------------------------------------------------------
@staticmethod
def _iou_matrix(bboxes_a: np.ndarray, bboxes_b: np.ndarray) -> np.ndarray:
"""Compute IoU matrix between two sets of bboxes (vectorized).
Args:
bboxes_a: (M, 4) array [x1, y1, x2, y2]
bboxes_b: (N, 4) array [x1, y1, x2, y2]
Returns:
(M, N) IoU matrix
"""
# Broadcast: (M, 1, 4) vs (1, N, 4)
a = bboxes_a[:, None, :] # (M, 1, 4)
b = bboxes_b[None, :, :] # (1, N, 4)
ix1 = np.maximum(a[..., 0], b[..., 0])
iy1 = np.maximum(a[..., 1], b[..., 1])
ix2 = np.minimum(a[..., 2], b[..., 2])
iy2 = np.minimum(a[..., 3], b[..., 3])
inter = np.maximum(0, ix2 - ix1) * np.maximum(0, iy2 - iy1)
area_a = (a[..., 2] - a[..., 0]) * (a[..., 3] - a[..., 1])
area_b = (b[..., 2] - b[..., 0]) * (b[..., 3] - b[..., 1])
union = area_a + area_b - inter
return inter / np.maximum(union, 1e-6)
def _association_similarity(
self,
det_bboxes: np.ndarray,
tracks: List[_Track],
det_indices: list[int] | np.ndarray,
track_indices: list[int] | np.ndarray,
*,
det_masks: np.ndarray | None,
det_obbs: np.ndarray | None,
use_last: bool = False,
) -> np.ndarray:
"""Return OBB/AABB geometry fused with mask IoU when available."""
det_indices = np.asarray(det_indices, dtype=np.int64)
track_indices = np.asarray(track_indices, dtype=np.int64)
if not len(det_indices) or not len(track_indices):
return np.empty((len(det_indices), len(track_indices)), dtype=np.float32)
selected_tracks = [tracks[int(index)] for index in track_indices]
if self.is_obb and det_obbs is not None:
predicted = []
for track in selected_tracks:
source = track.last_matched_obb if use_last else track.obb
if source is None:
predicted.append(np.zeros(5, dtype=np.float32))
continue
box = np.asarray(source, dtype=np.float32).copy()
if not use_last and track.velocity is not None and len(track.velocity) == 5:
box += track.velocity
box[2:4] = np.maximum(box[2:4], 1e-4)
box[4] = float(normalize_angle(box[4]))
predicted.append(box)
geometry = AssociationFunction.iou_batch_obb(det_obbs[det_indices], np.asarray(predicted))
else:
predicted = np.asarray(
[
(track.last_matched_bbox if use_last else track.bbox)
+ (
track.velocity
if not use_last and track.velocity is not None and len(track.velocity) == 4
else 0
)
for track in selected_tracks
],
dtype=np.float32,
)
geometry = self._iou_matrix(det_bboxes[det_indices], predicted)
if det_masks is None or self.cost_weight <= 0:
return geometry
similarity = geometry.copy()
for row, det_index in enumerate(det_indices):
detection_mask = det_masks[int(det_index)]
for col, track in enumerate(selected_tracks):
if detection_mask is None or track.mask is None or detection_mask.shape != track.mask.shape:
continue
mask_iou = self.coi.mask_iou(detection_mask, track.mask)
similarity[row, col] = (1.0 - self.cost_weight) * geometry[row, col] + self.cost_weight * mask_iou
return similarity
def _two_stage_matching(
self,
det_bboxes: np.ndarray,
det_confs: np.ndarray,
tracks: List[_Track],
det_masks=None,
det_obbs: np.ndarray | None = None,
):
"""Two-stage matching: high-conf first, then low-conf on remaining tracks."""
n_dets = len(det_bboxes)
n_trks = len(tracks)
if n_dets == 0 or n_trks == 0:
return [], list(range(n_dets)), list(range(n_trks)), []
# Split detections into high and low confidence
high_conf_mask = det_confs >= self.det_thresh
high_inds = np.where(high_conf_mask)[0]
low_inds = np.where(~high_conf_mask)[0]
matches_all = []
matched_dets = set()
matched_trks = set()
# --- Pass 1: Match high-confidence detections ---
if len(high_inds) > 0:
similarity = self._association_similarity(
det_bboxes,
tracks,
high_inds,
list(range(n_trks)),
det_masks=det_masks,
det_obbs=det_obbs,
)
cost = 1.0 - similarity
row_ind, col_ind = linear_sum_assignment(cost)
for r, c in zip(row_ind, col_ind):
if similarity[r, c] >= self.iou_threshold:
orig_det = high_inds[r]
matches_all.append((int(orig_det), c))
matched_dets.add(int(orig_det))
matched_trks.add(c)
# --- Pass 2: Match low-confidence detections to remaining tracks ---
unmatched_trks_pass1 = [j for j in range(n_trks) if j not in matched_trks]
if len(low_inds) > 0 and unmatched_trks_pass1:
similarity2 = self._association_similarity(
det_bboxes,
tracks,
low_inds,
unmatched_trks_pass1,
det_masks=det_masks,
det_obbs=det_obbs,
)
cost2 = 1.0 - similarity2
r2, c2 = linear_sum_assignment(cost2)
for ri, ci in zip(r2, c2):
if similarity2[ri, ci] >= self.second_stage_iou_threshold:
orig_det = low_inds[ri]
orig_trk = unmatched_trks_pass1[ci]
matches_all.append((int(orig_det), orig_trk))
matched_dets.add(int(orig_det))
matched_trks.add(orig_trk)
unmatched_dets = [i for i in range(n_dets) if i not in matched_dets]
unmatched_trks = [j for j in range(n_trks) if j not in matched_trks]
# Stage 2: last_matched_bbox for still-unmatched tracks (recovery)
second_stage_matches = []
if unmatched_dets and unmatched_trks:
valid_trks = [
(idx, tracks[idx])
for idx in unmatched_trks
if (
tracks[idx].last_matched_obb is not None
if self.is_obb
else tracks[idx].last_matched_bbox is not None
)
]
if valid_trks:
valid_indices = [index for index, _ in valid_trks]
similarity2 = self._association_similarity(
det_bboxes,
tracks,
unmatched_dets,
valid_indices,
det_masks=det_masks,
det_obbs=det_obbs,
use_last=True,
)
cost2 = 1.0 - similarity2
r2, c2 = linear_sum_assignment(cost2)
matched_dets_s2 = set()
matched_trks_s2 = set()
for ri, ci in zip(r2, c2):
if similarity2[ri, ci] >= self.second_stage_iou_threshold:
orig_det = unmatched_dets[ri]
orig_trk = valid_trks[ci][0]
second_stage_matches.append((orig_det, orig_trk))
matched_dets_s2.add(orig_det)
matched_trks_s2.add(orig_trk)
unmatched_dets = [d for d in unmatched_dets if d not in matched_dets_s2]
unmatched_trks = [t for t in unmatched_trks if t not in matched_trks_s2]
all_matches = matches_all + second_stage_matches
return all_matches, unmatched_dets, unmatched_trks, second_stage_matches
def _frame_out_matching(
self,
det_bboxes: np.ndarray,
unmatched_dets: List[int],
frame_out_tracks: List[_Track],
*,
det_obbs: np.ndarray | None = None,
) -> List[Tuple[int, _Track]]:
"""Stage 3: Match unmatched detections to frame-out tracks."""
if not unmatched_dets or not frame_out_tracks:
return []
if self.is_obb and det_obbs is not None:
has_geometry = np.array([track.last_matched_obb is not None for track in frame_out_tracks])
track_boxes = np.asarray(
[
track.last_matched_obb if track.last_matched_obb is not None else np.zeros(5)
for track in frame_out_tracks
],
dtype=np.float32,
)
similarity = AssociationFunction.iou_batch_obb(det_obbs[unmatched_dets], track_boxes)
else:
has_geometry = np.array([track.last_matched_bbox is not None for track in frame_out_tracks])
track_boxes = np.asarray(
[
track.last_matched_bbox if track.last_matched_bbox is not None else np.zeros(4)
for track in frame_out_tracks
],
dtype=np.float32,
)
similarity = self._iou_matrix(det_bboxes[unmatched_dets], track_boxes)
similarity[:, ~has_geometry] = 0
cost = 1.0 - similarity
row_ind, col_ind = linear_sum_assignment(cost)
results = []
for r, c in zip(row_ind, col_ind):
if similarity[r, c] >= self.iou_threshold:
results.append((unmatched_dets[r], frame_out_tracks[c]))
return results
# ------------------------------------------------------------------
# Utility
# ------------------------------------------------------------------
@staticmethod
def _bbox_iou(a: np.ndarray, b: np.ndarray) -> float:
x1 = max(a[0], b[0])
y1 = max(a[1], b[1])
x2 = min(a[2], b[2])
y2 = min(a[3], b[3])
if x2 <= x1 or y2 <= y1:
return 0.0
inter = (x2 - x1) * (y2 - y1)
area_a = (a[2] - a[0]) * (a[3] - a[1])
area_b = (b[2] - b[0]) * (b[3] - b[1])
union = area_a + area_b - inter
return float(inter) / max(float(union), 1e-6)
def _compute_density(self, target_idx: int, all_bboxes: np.ndarray) -> float:
"""Compute overlap density for a detection relative to all others (vectorized)."""
bbox = all_bboxes[target_idx]
x1, y1, x2, y2 = bbox
area = max((x2 - x1) * (y2 - y1), 1e-6)
# Vectorized intersection computation
ix1 = np.maximum(x1, all_bboxes[:, 0])
iy1 = np.maximum(y1, all_bboxes[:, 1])
ix2 = np.minimum(x2, all_bboxes[:, 2])
iy2 = np.minimum(y2, all_bboxes[:, 3])
inter = np.maximum(0, ix2 - ix1) * np.maximum(0, iy2 - iy1)
inter[target_idx] = 0 # exclude self
return float(inter.sum() / area)