import unittest from dataclasses import replace from app.contracts.visualization import FrameResult, VisualizationCommand from app.crowd_config import CROWD_COLORS from app.crowd_result import CrowdResultProcessor from app.detection_nms import detection_nms_indices from app.pose_nms import pose_nms_indices from app.intrusion import Intrusion, IntrusionConfig from app.loitering import Loitering, LoiteringConfig from crowd_fixtures import frame,obs,observations,engine,feed class DetectionTest(unittest.TestCase): def test_confidence_inclusive(self): e,_=engine();f=frame();r=e.evaluate(f,(obs(f,0,.45),obs(f,1,.45))) self.assertEqual((r.person_count,r.head_count),(1,1)) def test_below_confidence_removed(self): e,_=engine();f=frame();self.assertEqual(e.evaluate(f,(obs(f,score=.4499),)).indices,()) def test_separate_confidences(self): e,_=engine(person_confidence=.8,head_confidence=.3);f=frame() self.assertEqual(e.evaluate(f,(obs(f,0,.5),obs(f,1,.5))).indices,(1,)) def test_zero_area_excluded_before_nms(self): e,_=engine();f=frame();self.assertEqual(e.evaluate(f,(obs(f,box=(.1,.1,.1,.2)),)).indices,()) def test_same_class_suppressed(self): e,_=engine();f=frame();self.assertEqual(e.evaluate(f,(obs(f,score=.8),obs(f,score=.9))).indices,(1,)) def test_different_classes_never_suppress(self): e,_=engine();f=frame();self.assertEqual(e.evaluate(f,(obs(f,0),obs(f,1))).indices,(0,1)) def test_nms_disabled(self): e,_=engine(nms_enabled=False);f=frame();self.assertEqual(e.evaluate(f,(obs(f),obs(f))).person_count,2) def test_nms_tie_preserves_original_index(self): f=frame();self.assertEqual(detection_nms_indices((obs(f),obs(f))),(0,)) def test_shared_nms_api_matches_pose(self): f=frame();a=(obs(f),obs(f),obs(f,1)) self.assertEqual(detection_nms_indices(a),pose_nms_indices(a));self.assertEqual(pose_nms_indices(a),(0,2)) def test_nms_frame_and_model_guard(self): f=frame() for other in (obs(frame(1)),obs(f,model_key='other')): with self.subTest(other=other),self.assertRaises(ValueError):detection_nms_indices((obs(f),other)) def test_nms_iou_boundary(self): f=frame();a=obs(f,score=.9,box=(0,0,1,1)) self.assertEqual(detection_nms_indices((a,obs(f,score=.8,box=(0,0,.85,1)))),(0,1)) self.assertEqual(detection_nms_indices((a,obs(f,score=.8,box=(0,0,.85001,1)))),(0,)) def test_full_frame_empty_roi(self): e,_=engine(roi=[]);f=frame();self.assertEqual(e.evaluate(f,observations(f,2,3)).head_count,3) def test_different_roi_anchors(self): e,_=engine(roi=((0,0),(1,0),(1,.5),(0,.5)));f=frame() r=e.evaluate(f,(obs(f,0,box=(.1,.2,.3,.8)),obs(f,1,box=(.1,.2,.3,.8)))) self.assertEqual(r.indices,(1,)) # head center y=.5 boundary inside; person bottom y=.8 outside def test_roi_boundary_inside(self): e,_=engine(roi=((0,0),(1,0),(1,.5),(0,.5)));f=frame() self.assertEqual(e.evaluate(f,(obs(f,box=(.1,.1,.3,.5)),)).person_count,1) def test_roi_outside_excluded_count_and_draw(self): e,_=engine(roi=((0,0),(.5,0),(.5,1),(0,1)));f=frame() r=CrowdResultProcessor(e).process(f,(obs(f,box=(.6,.1,.8,.3)),)) self.assertEqual(r.snapshot()['objects'],[]);self.assertEqual(e._samples[-1][1:],(0,0)) def test_nms_precedes_roi(self): e,_=engine(roi=((0,0),(.5,0),(.5,1),(0,1)));f=frame() r=e.evaluate(f,(obs(f,score=.8,box=(.3,.1,.7,.5)),obs(f,score=.9,box=(.31,.1,.71,.5)))) self.assertEqual(r.indices,());self.assertEqual(e.metrics['nms_suppressed'],1) class OutputTest(unittest.TestCase): def ready(self): e,c=engine();r=CrowdResultProcessor(e,camera_id=1,scenario_revision='rev') for t in (0,1,3):f=frame(t);result=r.process(f,observations(f,2,3)) return r,result,c def test_separate_labels(self): _,r,_=self.ready();labels=[c.label_text for c in r.commands] self.assertEqual(labels.count('person | P≈2'),1);self.assertEqual(labels.count('head | H≈3'),1) self.assertFalse(any('P≈' in s and 'H≈' in s for s in labels)) def test_current_highest_confidence_anchors(self): p,_,_=self.ready();f=frame(4) r=p.process(f,(obs(f,0,.7,box=(.1,.1,.2,.2)),obs(f,0,.9,box=(.4,.4,.5,.5)),obs(f,1,.8))) self.assertEqual(r.commands[0].label_text,'person');self.assertEqual(r.commands[1].label_text,'person | P≈2') def test_warmup_no_estimate_label(self): e,_=engine();f=frame();r=CrowdResultProcessor(e).process(f,observations(f)) self.assertEqual([c.label_text for c in r.commands],['person','head']) def test_current_frame_and_boxes_only(self): p,_,_=self.ready();f=frame(4);o=obs(f,box=(.8,.8,.9,.9));r=p.process(f,(o,)) self.assertEqual(len(r.commands),1);self.assertIs(r.commands[0].detection,o);self.assertEqual(r.frame,f) def test_no_synthetic_objects_when_zero(self): p,_,_=self.ready();r=p.process(frame(4),());self.assertEqual(r.commands,());self.assertEqual(r.snapshot()['objects'],[]) def test_absent_class_has_no_label(self): p,_,_=self.ready();f=frame(4);r=p.process(f,(obs(f,1),)) self.assertEqual([c.label_text for c in r.commands],['head | H≈3']) def test_colors_and_no_event_or_tracking(self): _,r,_=self.ready();self.assertEqual(r.events,()) for c in r.commands: self.assertEqual(c.color,CROWD_COLORS[c.detection.class_id]);self.assertIsNone(c.track_id) self.assertEqual(c.event_ids,());self.assertIsNone(c.scenario) def test_existing_object_keys_exact(self): _,r,_=self.ready();wire=r.snapshot() expected={'objectRef','classId','modelKey','confidence','bbox','trackId','eventIds','labelText','color','visible'} self.assertEqual(set(wire['objects'][0]),expected);self.assertEqual(wire['events'],[]) self.assertNotIn('nms',wire);self.assertNotIn('poseDisplay',wire);self.assertNotIn('crowdCount',wire) def test_public_config_has_no_new_timer_count_keys(self): _,r,_=self.ready();p=r.snapshot()['scenarioConfig']['CROWD'] self.assertEqual(set(p),{'enabled','roi_id','roi_name','roi','debug_roi','confidence'}) def test_frame_wire_equals_existing_snapshot_except_existing_optional_fields(self): _,r,_=self.ready();plain=FrameResult(r.frame,r.commands,provider_model=r.provider_model,camera_id=1).snapshot() wire=r.snapshot() for key in ('scenarioConfig','scenarioRevision','events'):wire.pop(key) self.assertEqual(wire,plain) def test_rejected_packet_produces_no_frame(self): p,_,_=self.ready();self.assertIsNone(p.process(frame(3),()));self.assertIsNone(p.process(frame(4),(),usable=False)) def test_camera_invalid_before_state_change(self): e,_=engine();p=CrowdResultProcessor(e,camera_id=2) with self.assertRaises(ValueError):p.process(frame(),()) self.assertEqual(e.diagnostics()['samples'],0) def test_close_clears_and_prevents_reuse(self): p,_,_=self.ready();p.close();self.assertIsNone(p.engine.diagnostics()['estimate']) with self.assertRaises(RuntimeError):p.process(frame(4),()) def test_stale_label_removed_on_next_current_frame(self): p,_,c=self.ready();c.value=6;p.expire_idle();f=frame(4) r=p.process(f,observations(f));self.assertEqual([x.label_text for x in r.commands],['person','head']) def test_baseline_plain_wire_fixture(self): import json from pathlib import Path f=frame();o=obs(f,model_key='RFDETRMedium') r=FrameResult(f,(VisualizationCommand(o,'person'),),provider_model='RFDETRMedium') expected=json.loads((Path(__file__).parent/'fixtures/crowd_legacy_frame.json').read_text()) self.assertEqual(r.snapshot(),expected) class LegacyCoreSmokeTest(unittest.TestCase): def test_intrusion_original_strict_threshold(self): e=Intrusion(IntrusionConfig(enabled=True));f=frame();o=obs(f,score=.5,model_key='RFDETRMedium') d,events=e.evaluate(f,(o,),(1,));self.assertEqual(d[0].state,'LOW_CONFIDENCE');self.assertEqual(events,()) def test_loitering_original_event_and_latch(self): e=Loitering(LoiteringConfig(enabled=True),clock=lambda:0,wall_ns=lambda:1_000_000_000,event_id_factory=lambda:'test-event') found=[] for t in range(7): f=frame(t);d,events=e.evaluate(f,(obs(f,score=.5,model_key='RFDETRMedium'),),(1,));found.extend(events) self.assertEqual(len(found),1);self.assertEqual(found[0].event_id,'test-event')