134 lines
4.5 KiB
Python
134 lines
4.5 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"] == "初始"
|
|
|
|
|
|
def test_live_preview_does_not_overwrite_previous_finalized_queue_level():
|
|
tracker = QueueWindowTracker(
|
|
QueueConfig(
|
|
queue_time_threshold_seconds=60,
|
|
crowded_count_threshold=3,
|
|
normal_count_threshold=2,
|
|
pause_timeout_seconds=5,
|
|
),
|
|
pixel_area=(0, 0, 100, 100),
|
|
)
|
|
first_window_start = datetime(2026, 4, 15, 11, 0, tzinfo=TZ)
|
|
first_window_end = datetime(2026, 4, 15, 11, 5, tzinfo=TZ)
|
|
second_window_start = first_window_end
|
|
second_window_preview = datetime(2026, 4, 15, 11, 7, tzinfo=TZ)
|
|
second_window_end = datetime(2026, 4, 15, 11, 10, tzinfo=TZ)
|
|
|
|
normal_tracks = [
|
|
TrackObservation(
|
|
track_id=index, bbox=(0, 0, 10, 10), confidence=0.9, center=(10, 10)
|
|
)
|
|
for index in range(1, 4)
|
|
]
|
|
few_tracks = [
|
|
TrackObservation(
|
|
track_id=10, bbox=(0, 0, 10, 10), confidence=0.9, center=(10, 10)
|
|
)
|
|
]
|
|
|
|
tracker.observe(normal_tracks, first_window_start)
|
|
tracker.observe(normal_tracks, first_window_end)
|
|
first_metrics = tracker.build_queue_metrics(first_window_start, first_window_end)
|
|
|
|
assert first_metrics["queue_level"] == "normal"
|
|
assert first_metrics["status_change"] == "initial"
|
|
|
|
tracker.reset()
|
|
tracker.observe(few_tracks, second_window_start)
|
|
tracker.observe([], second_window_preview)
|
|
preview_metrics = tracker.build_queue_metrics(
|
|
second_window_start,
|
|
second_window_preview,
|
|
commit_queue_level=False,
|
|
)
|
|
|
|
assert preview_metrics["queue_level"] == "few"
|
|
assert preview_metrics["previous_queue_level"] == "normal"
|
|
assert preview_metrics["status_change"] == "queue_decreased"
|
|
|
|
final_metrics = tracker.build_queue_metrics(second_window_start, second_window_end)
|
|
|
|
assert final_metrics["queue_level"] == "few"
|
|
assert final_metrics["previous_queue_level"] == "normal"
|
|
assert final_metrics["status_change"] == "queue_decreased"
|
|
|
|
|
|
def test_queue_level_is_normal_when_count_reaches_threshold():
|
|
tracker = QueueWindowTracker(
|
|
QueueConfig(
|
|
queue_time_threshold_seconds=60,
|
|
crowded_count_threshold=3,
|
|
normal_count_threshold=2,
|
|
pause_timeout_seconds=5,
|
|
),
|
|
pixel_area=(0, 0, 100, 100),
|
|
)
|
|
start = datetime(2026, 4, 15, 11, 0, tzinfo=TZ)
|
|
end = datetime(2026, 4, 15, 11, 5, tzinfo=TZ)
|
|
tracks = [
|
|
TrackObservation(
|
|
track_id=index, bbox=(0, 0, 10, 10), confidence=0.9, center=(10, 10)
|
|
)
|
|
for index in range(1, 4)
|
|
]
|
|
|
|
tracker.observe(tracks, start)
|
|
tracker.observe(tracks, start.replace(minute=2))
|
|
tracker.observe([], end)
|
|
|
|
metrics = tracker.build_queue_metrics(start, end)
|
|
|
|
assert metrics["over_threshold_count"] == 3
|
|
assert metrics["queue_level"] == "normal"
|
|
assert metrics["queue_level_label"] == "人数正常"
|