- SyntheticIdAssigner: track_id=None dets get stable negative IDs so the stabilizer hold bridges sup_det_interval gaps -> no flicker, boxes count - apply primary model class filter in merged path (UI checkboxes work) - _suppress_sacks_under_boxes: sack dets overlapping box dets dropped (models without box class label boxes as sack; stops double counting) - default web port 9000 -> 9050 (WEB_PORT env still overrides) - README: port refs, counting zone modes section, project tree - 140 tests pass (+13)
134 lines
4.5 KiB
Python
134 lines
4.5 KiB
Python
"""Tests for synthetic track-ID assignment (src/synthetic_track.py)
|
|
and sack-under-box suppression (src.pipeline)."""
|
|
|
|
import pytest
|
|
|
|
from src.interfaces import Detection
|
|
from src.pipeline import _suppress_sacks_under_boxes
|
|
from src.synthetic_track import SyntheticIdAssigner
|
|
|
|
|
|
def det(bbox, class_name="sack", track_id=None, conf=0.9):
|
|
return Detection(
|
|
bbox=tuple(bbox), confidence=conf, class_id=0,
|
|
class_name=class_name, track_id=track_id,
|
|
)
|
|
|
|
|
|
# ── SyntheticIdAssigner ─────────────────────────────────────────────────
|
|
|
|
|
|
def test_assigns_new_negative_id_to_untracked_det():
|
|
"""track_id=None det gets a fresh negative id (ByteTrack uses positive)."""
|
|
a = SyntheticIdAssigner()
|
|
out = a.assign([det((0, 0, 10, 10))])
|
|
assert out[0].track_id is not None
|
|
assert out[0].track_id < 0
|
|
assert out[0].track_id == -1
|
|
|
|
|
|
def test_same_position_reuses_id():
|
|
"""Same det next call keeps its id (registry bbox within IoU 0.3)."""
|
|
a = SyntheticIdAssigner()
|
|
first = a.assign([det((0, 0, 10, 10))])[0].track_id
|
|
second = a.assign([det((1, 1, 11, 11))])[0].track_id # 5px jitter
|
|
assert second == first
|
|
|
|
|
|
def test_moved_beyond_iou_gets_new_id():
|
|
"""Det teleports beyond IoU threshold → new id, old entry retained."""
|
|
a = SyntheticIdAssigner()
|
|
first = a.assign([det((0, 0, 10, 10))])[0].track_id
|
|
moved = a.assign([det((100, 100, 110, 110))])[0].track_id
|
|
assert moved != first
|
|
assert moved == first - 1
|
|
|
|
|
|
def test_class_mismatch_never_shares_id():
|
|
"""box next to sack must not adopt the sack's id."""
|
|
a = SyntheticIdAssigner()
|
|
sack_id = a.assign([det((0, 0, 10, 10), class_name="sack")])[0].track_id
|
|
box_id = a.assign([det((0, 0, 10, 10), class_name="box")])[0].track_id
|
|
assert box_id != sack_id
|
|
|
|
|
|
def test_real_track_id_passes_through_untouched():
|
|
"""ByteTrack det keeps its positive id; registry stays empty."""
|
|
a = SyntheticIdAssigner()
|
|
out = a.assign([det((0, 0, 10, 10), track_id=7)])
|
|
assert out[0].track_id == 7
|
|
assert a._registry == {}
|
|
|
|
|
|
def test_expiry_boundaries():
|
|
"""Entry expires after max_age (10) consecutive unmatched calls.
|
|
|
|
Note: spec bullet said "10 calls → gone"; literal `age > max_age`
|
|
(matches BboxStabilizer's `age > max_hold`) expires on the 11th miss.
|
|
9 and 10 unmatched calls both keep the entry alive.
|
|
"""
|
|
a = SyntheticIdAssigner() # max_age=10
|
|
a.assign([det((0, 0, 10, 10))])
|
|
assert -1 in a._registry
|
|
|
|
for _ in range(10):
|
|
a.assign([])
|
|
assert -1 in a._registry # still held at age 10
|
|
|
|
a.assign([])
|
|
assert -1 not in a._registry # gone at age 11
|
|
|
|
|
|
def test_fresh_det_after_expiry_gets_new_id():
|
|
"""Old id is not reused once the entry expired."""
|
|
a = SyntheticIdAssigner()
|
|
first = a.assign([det((0, 0, 10, 10))])[0].track_id
|
|
for _ in range(11):
|
|
a.assign([])
|
|
again = a.assign([det((0, 0, 10, 10))])[0].track_id
|
|
assert again != first
|
|
|
|
|
|
def test_clear_empties_registry():
|
|
a = SyntheticIdAssigner()
|
|
a.assign([det((0, 0, 10, 10)), det((50, 50, 60, 60))])
|
|
assert a._registry
|
|
a.clear()
|
|
assert a._registry == {}
|
|
|
|
|
|
# ── _suppress_sacks_under_boxes ─────────────────────────────────────────
|
|
|
|
|
|
def test_overlapping_sack_dropped_box_kept():
|
|
"""IoU > 0.5 between sack and box → sack dropped, box wins."""
|
|
box = det((0, 0, 100, 100), class_name="box")
|
|
sack = det((10, 10, 90, 90), class_name="sack") # IoU = 0.64
|
|
out = _suppress_sacks_under_boxes([box, sack])
|
|
assert [d.class_name for d in out] == ["box"]
|
|
|
|
|
|
def test_no_box_detections_unchanged():
|
|
dets = [det((0, 0, 10, 10)), det((50, 50, 60, 60))]
|
|
assert _suppress_sacks_under_boxes(dets) == dets
|
|
|
|
|
|
def test_sack_far_from_box_kept():
|
|
box = det((0, 0, 100, 100), class_name="box")
|
|
far = det((200, 200, 250, 250), class_name="sack") # IoU = 0
|
|
out = _suppress_sacks_under_boxes([box, far])
|
|
assert len(out) == 2
|
|
|
|
|
|
def test_two_sacks_unaffected():
|
|
"""Two non-overlapping sacks stay when a box is present elsewhere."""
|
|
box = det((0, 0, 100, 100), class_name="box")
|
|
s1 = det((200, 0, 250, 50), class_name="sack")
|
|
s2 = det((200, 200, 250, 250), class_name="sack")
|
|
out = _suppress_sacks_under_boxes([box, s1, s2])
|
|
assert out == [box, s1, s2]
|
|
|
|
|
|
def test_empty_input():
|
|
assert _suppress_sacks_under_boxes([]) == []
|