49 lines
1.7 KiB
Python
49 lines
1.7 KiB
Python
from datetime import datetime
|
|
from zoneinfo import ZoneInfo
|
|
|
|
from src.people_flow.models import QueueConfig, TrackObservation
|
|
from src.people_flow.queue_analytics import QueueWindowTracker
|
|
|
|
TZ = ZoneInfo("Asia/Shanghai")
|
|
|
|
|
|
def test_queue_window_tracker_builds_crowded_report():
|
|
tracker = QueueWindowTracker(
|
|
QueueConfig(
|
|
queue_time_threshold_seconds=300,
|
|
crowded_count_threshold=5,
|
|
normal_count_threshold=2,
|
|
pause_timeout_seconds=5,
|
|
),
|
|
pixel_area=(0, 0, 100, 100),
|
|
)
|
|
start = datetime(2026, 4, 15, 11, 0, tzinfo=TZ)
|
|
crowded_tracks = [
|
|
TrackObservation(
|
|
track_id=index, bbox=(0, 0, 10, 10), confidence=0.9, center=(10, 10)
|
|
)
|
|
for index in range(1, 7)
|
|
]
|
|
short_tracks = [
|
|
TrackObservation(
|
|
track_id=index, bbox=(0, 0, 10, 10), confidence=0.9, center=(10, 10)
|
|
)
|
|
for index in range(7, 9)
|
|
]
|
|
|
|
tracker.observe(crowded_tracks, start)
|
|
tracker.observe(crowded_tracks, start.replace(minute=6))
|
|
tracker.observe(crowded_tracks + short_tracks, start.replace(minute=27))
|
|
tracker.observe(short_tracks, start.replace(minute=30))
|
|
|
|
queue_metrics = tracker.build_queue_metrics(start, start.replace(minute=30))
|
|
|
|
assert queue_metrics["over_threshold_count"] == 6
|
|
assert queue_metrics["under_threshold_count"] == 2
|
|
assert queue_metrics["queue_level"] == "crowded"
|
|
assert queue_metrics["queue_level_label"] == "人多"
|
|
assert queue_metrics["previous_queue_level"] is None
|
|
assert queue_metrics["previous_queue_level_label"] == ""
|
|
assert queue_metrics["status_change"] == "initial"
|
|
assert queue_metrics["status_change_label"] == "初始"
|