skye.yue
4e69eca7cb
chore: update docker-compose for additional host configuration test: add unit tests for LineCrossCounter and WindowIdentityResolver functionality
44 lines
1.5 KiB
Python
44 lines
1.5 KiB
Python
import numpy as np
|
|
|
|
from src.people_flow.counting import LineCrossCounter
|
|
from src.people_flow.models import CountingConfig, TrackObservation
|
|
from src.people_flow.window_identity import WindowIdentityResolver
|
|
|
|
|
|
def _observation(track_id: int, center_y: float) -> TrackObservation:
|
|
return TrackObservation(
|
|
track_id=track_id,
|
|
bbox=(0, 0, 10, 10),
|
|
confidence=0.9,
|
|
center=(5.0, center_y),
|
|
)
|
|
|
|
|
|
def test_line_cross_counter_dedupes_reentry_with_same_person_key():
|
|
counter = LineCrossCounter(
|
|
(0.0, 5.0, 10.0, 5.0), CountingConfig(crossing_tolerance=0.0)
|
|
)
|
|
|
|
counter.update([_observation(1, 2.0)], person_keys={1: "person:1"})
|
|
first_events = counter.update([_observation(1, 8.0)], person_keys={1: "person:1"})
|
|
counter.update([_observation(2, 2.0)], person_keys={2: "person:1"})
|
|
second_events = counter.update([_observation(2, 8.0)], person_keys={2: "person:1"})
|
|
|
|
assert len(first_events) == 1
|
|
assert second_events == []
|
|
assert counter.total_people == 1
|
|
assert len(counter.crossings) == 1
|
|
assert counter.crossings[0].track_id == 1
|
|
|
|
|
|
def test_window_identity_resolver_matches_reentry_after_track_switch():
|
|
resolver = WindowIdentityResolver(similarity_threshold=0.97)
|
|
frame = np.zeros((16, 16, 3), dtype=np.uint8)
|
|
frame[:, :] = (30, 60, 180)
|
|
|
|
first_keys = resolver.resolve(frame, [_observation(1, 2.0)])
|
|
resolver.resolve(frame, [])
|
|
second_keys = resolver.resolve(frame, [_observation(2, 8.0)])
|
|
|
|
assert first_keys[1] == second_keys[2]
|