import unittest from dataclasses import replace from app.crowd_config import CrowdConfig, SAMPLE_LIMIT from app.crowd_scenario import Crowd from crowd_fixtures import frame,obs,observations,engine,feed,Clock class TemporalTest(unittest.TestCase): def test_initial_report_waits_for_cadence(self): e,_=engine();self.assertIsNone(feed(e,0).estimate);self.assertIsNone(feed(e,2.999).estimate) self.assertTrue(feed(e,3).reported) def test_person_head_independent(self): e,_=engine();feed(e,0,9,7);feed(e,1,2,8);r=feed(e,3,4,10) self.assertEqual((r.estimate.person_count,r.estimate.head_count),(3,9)) self.assertEqual(r.estimate.sample_count,2) def test_zero_is_valid(self): e,_=engine();feed(e,0);feed(e,1,0,0);r=feed(e,3,0,0) self.assertEqual((r.estimate.person_count,r.estimate.head_count),(0,0)) def test_missing_not_zero(self): e,_=engine();feed(e,0,10,12);feed(e,1,10,12) self.assertFalse(e.evaluate(frame(2),(),usable=False).accepted) r=feed(e,3,10,12);self.assertEqual(r.estimate.sample_count,2);self.assertEqual(r.estimate.person_count,10) def test_sparse_two_samples_allowed(self): e,_=engine();feed(e,0);feed(e,.1,2,3);r=feed(e,3,4,5) self.assertTrue(r.reported);self.assertEqual(r.estimate.person_count,3) def test_one_sample_does_not_report(self): e,_=engine();feed(e,0);r=feed(e,3);self.assertFalse(r.reported);self.assertIsNone(r.estimate) def test_left_edge_excluded(self): e,_=engine();feed(e,0,50,50);feed(e,1,2,2);r=feed(e,3,2,2) self.assertEqual(r.estimate.person_count,2);self.assertEqual(r.estimate.first_sample_pts_ns,1_000_000_000) def test_sliding_window_and_report_separate(self): e,_=engine(window_seconds=5,report_seconds=1) feed(e,0,1,1);feed(e,1,2,2);r=feed(e,2,3,3) self.assertEqual(r.estimate.sample_count,3);self.assertEqual(r.estimate.person_count,2) def test_round_uses_python_ties_even(self): e,_=engine();feed(e,0);feed(e,1,12,13);r=feed(e,3,13,14) self.assertEqual((r.estimate.person_mean,r.estimate.person_count),(12.5,12)) self.assertEqual((r.estimate.head_mean,r.estimate.head_count),(13.5,14)) def test_cached_report_not_recomputed_each_frame(self): e,_=engine();feed(e,0);feed(e,1,2,2);old=feed(e,3,2,2).estimate r=feed(e,4,10,10);self.assertIs(r.estimate,old);self.assertFalse(r.reported) def test_no_catchup_reports(self): e,_=engine();feed(e,0);self.assertFalse(feed(e,9.9).reported) self.assertFalse(feed(e,10).reported);r=feed(e,12) self.assertTrue(r.reported);self.assertEqual(e.metrics['reports'],1);self.assertEqual(e.diagnostics()['next_report_pts_ns'],15_000_000_000) def test_media_ttl_expires_insufficient_report(self): e,_=engine();feed(e,0);feed(e,1);feed(e,3) self.assertIsNotNone(feed(e,6).estimate) self.assertIsNone(feed(e,9).estimate);self.assertEqual(e.metrics['media_estimate_expired'],1) def test_idle_ttl_without_new_frames(self): e,c=engine();feed(e,0);feed(e,1);feed(e,3);c.value=6 self.assertTrue(e.expire_idle());self.assertIsNone(e.diagnostics()['estimate']);self.assertEqual(e.diagnostics()['samples'],0) def test_idle_does_not_report_or_add_zero(self): e,c=engine();feed(e,0);c.value=20;e.expire_idle() self.assertEqual(e.metrics['valid_samples'],1);self.assertEqual(e.metrics['reports'],0) def test_resume_after_idle_warms_up(self): e,c=engine();feed(e,0);feed(e,1);feed(e,3);c.value=7;e.expire_idle() r=feed(e,3.1);self.assertIsNone(r.estimate);self.assertEqual(e.diagnostics()['next_report_pts_ns'],6_100_000_000) def test_duplicate_frame_not_denominator(self): e,_=engine();feed(e,0);self.assertFalse(feed(e,1,seq=0).accepted) self.assertEqual(e.diagnostics()['samples'],1) def test_equal_pts_new_seq_rejected(self): e,_=engine();feed(e,0);self.assertFalse(feed(e,0,seq=1).accepted) def test_regressing_pts_rejected(self): e,_=engine();feed(e,2);self.assertFalse(feed(e,1,seq=3000).accepted) def test_invalid_pts_rejected(self): for pts in (None,-1,True,1.2,2**63): e,_=engine();f=frame(0,pts_ns=pts) with self.subTest(pts=pts):self.assertFalse(e.evaluate(f,()).accepted);self.assertEqual(e.diagnostics()['samples'],0) def test_skipped_source_sequence_not_missing(self): e,_=engine();feed(e,0,seq=1);feed(e,1,seq=100);r=feed(e,3,seq=300) self.assertTrue(r.reported) def test_new_run_requires_clear(self): e,_=engine();feed(e,0) with self.assertRaises(ValueError):feed(e,1,run_id='new') e.clear();self.assertIsNone(feed(e,1,run_id='new').estimate) def test_geometry_and_config_change_rejected(self): for kw in ({'frame_width':640},{'config_version':'v2'},{'source_id':'2'}): e,_=engine();feed(e,0) with self.subTest(kw=kw),self.assertRaises(ValueError):feed(e,1,**kw) def test_clear_resets_everything(self): e,_=engine();feed(e,0);feed(e,1);feed(e,3);e.clear() self.assertEqual(e.diagnostics(),dict(metrics={},samples=0,next_report_pts_ns=None,estimate=None)) def test_disabled_no_samples(self): e=Crowd(CrowdConfig());self.assertFalse(e.evaluate(frame(),()).accepted);self.assertEqual(e.diagnostics()['samples'],0) def test_history_bounded_no_silent_tail_mean(self): e,_=engine(nms_enabled=False) for i in range(SAMPLE_LIMIT+1):e.evaluate(frame(seq=i,pts_ns=i+1),()) self.assertEqual(e.diagnostics()['samples'],1);self.assertEqual(e.metrics['history_overflow_reset'],1) def test_foreign_observation_no_state_mutation(self): e,_=engine();f=frame() with self.assertRaises(ValueError):e.evaluate(f,(obs(frame(1)),)) self.assertEqual(e.diagnostics()['samples'],0) def test_bad_model_and_class(self): for changes in ({'model_key':'RFDETRMedium'},{'class_id':2},{'class_id':True}): e,_=engine();f=frame() with self.subTest(changes=changes),self.assertRaises(ValueError):e.evaluate(f,(obs(f,**changes),)) def test_over_object_bound_rejected(self): e,_=engine();f=frame() with self.assertRaises(ValueError):e.evaluate(f,(obs(f),)*301) def test_exact_object_bound_accepted(self): e,_=engine(nms_enabled=False);f=frame();r=e.evaluate(f,observations(f,150,150)) self.assertEqual((r.person_count,r.head_count),(150,150)) def test_no_image_or_frame_in_history(self): e,_=engine();feed(e,0);self.assertTrue(all(type(v) is int for s in e._samples for v in s)) def test_bad_monotonic_clock(self): e,c=engine();feed(e,0);c.value=float('nan') with self.assertRaises(ValueError):e.expire_idle() def test_processing_delay_not_window_time(self): e,c=engine();feed(e,0);c.value=2.;r=feed(e,.1) self.assertFalse(r.reported);self.assertEqual(e.diagnostics()['samples'],2)