99 lines
3.2 KiB
Python
99 lines
3.2 KiB
Python
"""Smoke tests for LineCrossCounter + MultiClassLineCounter (src/counting.py). Pure-Python: needs only numpy."""
|
|
|
|
from src.counting import LineCrossCounter, MultiClassLineCounter
|
|
from src.interfaces import Detection
|
|
|
|
|
|
def _det(tid, y1, cx=500.0, class_name="sack"):
|
|
return Detection(
|
|
bbox=(cx - 20, y1, cx + 20, y1 + 60),
|
|
confidence=0.9,
|
|
class_id=0,
|
|
class_name=class_name,
|
|
track_id=tid,
|
|
)
|
|
|
|
|
|
def _counter():
|
|
return LineCrossCounter(line_y=100, line_x_start=0, line_x_end=1000, margin=20)
|
|
|
|
|
|
def test_loading_above_then_below():
|
|
c = _counter()
|
|
assert c.update([_det(1, y1=10)]) == [] # above line
|
|
events = c.update([_det(1, y1=150)]) # below line
|
|
assert len(events) == 1
|
|
assert events[0]["direction"] == "loading"
|
|
assert c.loading_count == 1
|
|
|
|
|
|
def test_below_first_counts_as_unloading_not_loading():
|
|
c = _counter()
|
|
assert c.update([_det(2, y1=150)]) == [] # appeared below first
|
|
assert c.update([_det(2, y1=10)]) != [] # moved above => unloading
|
|
assert c.unloading_count == 1
|
|
assert c.loading_count == 0
|
|
|
|
|
|
def test_track_counted_once_per_direction():
|
|
c = _counter()
|
|
c.update([_det(3, y1=10)])
|
|
c.update([_det(3, y1=150)])
|
|
c.update([_det(3, y1=10)])
|
|
c.update([_det(3, y1=150)])
|
|
assert c.loading_count == 1 # track_id counted once for loading
|
|
|
|
|
|
def test_outside_x_bounds_skipped():
|
|
c = _counter()
|
|
c.update([_det(4, y1=10, cx=500.0)])
|
|
assert c.update([_det(4, y1=150, cx=5000.0)]) == []
|
|
assert c.loading_count == 0
|
|
|
|
|
|
def test_net_and_reset():
|
|
c = _counter()
|
|
c.update([_det(5, y1=10)])
|
|
c.update([_det(5, y1=150)])
|
|
assert c.net_count == 1
|
|
c.reset()
|
|
assert (c.loading_count, c.unloading_count, c.net_count) == (0, 0, 0)
|
|
|
|
|
|
def _mcounter():
|
|
return MultiClassLineCounter(line_y=100, line_x_start=0, line_x_end=1000, margin=20)
|
|
|
|
|
|
def test_multiclass_sack_and_box_independent():
|
|
c = _mcounter()
|
|
c.update([_det(1, y1=10, class_name="sack"), _det(7, y1=10, class_name="box")])
|
|
events = c.update([_det(1, y1=150, class_name="sack"), _det(7, y1=150, class_name="box")])
|
|
assert len(events) == 2 # flat list, callsites unaffected
|
|
assert {e["class_name"] for e in events} == {"sack", "box"}
|
|
assert c.loading_count == 1
|
|
assert c.box_loading_count == 1
|
|
assert c.unloading_count == 0 and c.box_unloading_count == 0
|
|
|
|
|
|
def test_multiclass_shared_track_id_no_collision():
|
|
# sack track 9 and box track 9 are different objects; both must count
|
|
c = _mcounter()
|
|
c.update([_det(9, y1=10, class_name="sack"), _det(9, y1=10, class_name="box")])
|
|
c.update([_det(9, y1=150, class_name="sack"), _det(9, y1=150, class_name="box")])
|
|
assert c.loading_count == 1
|
|
assert c.box_loading_count == 1
|
|
|
|
|
|
def test_multiclass_geometry_sync_and_reset():
|
|
c = _mcounter()
|
|
c.line_y = 200
|
|
assert c._sack.line_y == 200 and c._box.line_y == 200
|
|
c.line_x_start = 10
|
|
c.line_x_end = 900
|
|
assert (c._sack.line_x_start, c._box.line_x_end) == (10, 900)
|
|
c.update([_det(1, y1=100, class_name="sack")])
|
|
c.update([_det(1, y1=250, class_name="sack")])
|
|
assert c.loading_count == 1
|
|
c.reset()
|
|
assert (c.loading_count, c.box_loading_count, c.box_net_count) == (0, 0, 0)
|