"""CPU-only deterministic temporal cases. No SDK, real sockets, DB or GPU.""" from dataclasses import replace import math import unittest from app.contracts.frames import FrameEnvelope, DetectionObservation from app.loitering import Loitering, LoiteringConfig ROI = ((.2,.2),(.8,.2),(.8,.8),(.2,.8)) INSIDE = (.3,.3,.5,.7) OUTSIDE = (.85,.3,.95,.7) class Clock: value = 0. def __call__(self): return self.value class Harness: def __init__(self, **kw): self.clock=Clock(); self.ids=iter(('event-1','event-2','event-3','event-4')) self.engine=Loitering(LoiteringConfig(enabled=True,roi=ROI,**kw),clock=self.clock, wall_ns=lambda:1758000000000000000,event_id_factory=lambda:next(self.ids)) self.seq=0 def step(self, t, *, tid=1, score=.9, box=INSIDE, missing=False, model='RFDETRMedium', cls=0): self.seq+=1 pts=None if t is None else round(t*1e9) f=FrameEnvelope('server','77','stream','run',1,self.seq,pts,1758000000000,1280,720) obs=() if missing else (DetectionObservation(f,model,cls,score,box),) return self.engine.evaluate(f,obs,() if missing else (tid,)) def span(self, lo, hi, **kw): events=[] for i in range(round(lo*10),round(hi*10)+1): _,ev=self.step(i/10,**kw);events.extend(ev) return events class ConfigTests(unittest.TestCase): def test_defaults(self): cfg=LoiteringConfig(); self.assertFalse(cfg.enabled) self.assertEqual((cfg.dwell_seconds,cfg.grace_seconds,cfg.rearm_seconds),(5,1,1)) self.assertEqual((cfg.point_x,cfg.point_y,cfg.confidence),(.5,1,.5)) self.assertEqual(LoiteringConfig.parse(cfg.payload()),cfg) def test_numbers(self): for field in ('dwell_seconds','grace_seconds','rearm_seconds','retention_seconds','confidence','point_x','point_y'): for v in (float('nan'),float('inf'),-1,True,'1',None): with self.subTest(field=field,value=v),self.assertRaises(ValueError):LoiteringConfig(**{field:v}) def test_unknown(self): for value in (None,[],{'dwellSeconds':5}): with self.subTest(value=value),self.assertRaises(ValueError):LoiteringConfig.parse(value) def test_bounds(self): for kw in ({'retention_seconds':1},{'max_states':0},{'max_states':4097}, {'dwell_seconds':0}, {'rearm_seconds':3601}): with self.subTest(kw=kw),self.assertRaises(ValueError):LoiteringConfig(**kw) def test_polygons(self): for poly in ((),((0,0),(1,1)),((0,0),(1,1),(1,0),(0,1)),((0,0),(.5,.5),(1,1)),((0,0),(2,0),(1,1)),((0,0),(1,0),(0,0))): with self.subTest(poly=poly),self.assertRaises(ValueError):LoiteringConfig(roi=poly) def test_mutable_roi_copied(self): poly=[list(x) for x in ROI]; cfg=LoiteringConfig(roi=poly);poly[0][0]=.9 self.assertEqual(cfg.roi,ROI) class TemporalTests(unittest.TestCase): def test_five_seconds_and_latch(self): h=Harness(); self.assertFalse(h.span(0,4.9)) d,e=h.step(5);self.assertEqual(d[0].state,'LOITERING');self.assertEqual(len(e),1) self.assertEqual(e[0].ai_recorded_at_wall_ms,1758000000000) self.assertFalse(h.span(5.1,30));self.assertEqual(h.engine.states[1].event_id,'event-1') def test_rearm(self): h=Harness();self.assertEqual(len(h.span(0,5)),1) h.span(5.1,6.1,box=OUTSIDE);self.assertIsNone(h.engine.states[1].event_id) self.assertFalse(h.span(6.2,11.1));self.assertEqual(h.step(11.2)[1][0].event_id,'event-2') def test_short_outside_pauses(self): h=Harness();h.span(0,2);h.span(2.1,2.5,box=OUTSIDE);h.step(2.6) self.assertEqual(h.engine.states[1].dwell_ns,2_000_000_000) self.assertFalse(h.span(2.7,5.5));self.assertEqual(len(h.step(5.6)[1]),1) def test_missing_not_added(self): h=Harness();h.span(0,2);h.step(2.1,missing=True);h.step(2.5) self.assertEqual(h.engine.states[1].dwell_ns,2_000_000_000) self.assertFalse(h.span(2.6,5.4));self.assertEqual(len(h.step(5.5)[1]),1) def test_low_score_not_added_inclusive_threshold(self): h=Harness();h.span(0,2,score=.5);h.step(2.1,score=.49);h.step(2.5,score=.5) self.assertEqual(h.engine.states[1].dwell_ns,2_000_000_000) self.assertEqual(len(h.span(2.6,5.5,score=.5)),1) def test_long_gap_resets_dwell(self): h=Harness();h.span(0,4);h.step(5.1) self.assertEqual(h.engine.states[1].dwell_ns,0) def test_grace_equality(self): h=Harness();h.step(0);h.step(.5,missing=True);h.step(1) self.assertEqual(h.engine.states[1].dwell_ns,0) h.step(2);self.assertEqual(h.engine.states[1].dwell_ns,1_000_000_000) def test_absence_does_not_rearm_fired_visit(self): h=Harness();h.span(0,5);h.step(10) self.assertEqual(h.engine.states[1].event_id,'event-1') self.assertFalse(h.span(10.1,15));self.assertEqual(h.step(15.1)[0][0].event_id,'event-1') def test_bad_pts_breaks_interval(self): for t in (None,-1,1.5): with self.subTest(t=t): h=Harness();h.span(0,2);d,_=h.step(t);self.assertEqual(d[0].state,'TIME_UNAVAILABLE') h.step(2.5);self.assertEqual(h.engine.states[1].dwell_ns,2_000_000_000) def test_same_pts_no_increment(self): h=Harness();h.step(0);h.step(0);self.assertEqual(h.engine.states[1].dwell_ns,0) def test_wall_clock_not_dwell(self): h=Harness();h.step(0);h.clock.value=20;h.step(.1) self.assertEqual(h.engine.states[1].dwell_ns,100_000_000) def test_no_id(self): h=Harness();d,e=h.step(0,tid=None);self.assertEqual(d[0].state,'PENDING_ID');self.assertFalse(h.engine.states) def test_new_id(self): h=Harness();h.span(0,4);d,e=h.step(4.1,tid=2) self.assertFalse(e);self.assertEqual(h.engine.states[2].dwell_ns,0) def test_expire(self): h=Harness();h.step(0);h.clock.value=60;h.engine.expire();self.assertFalse(h.engine.states) def test_missing_heartbeats_do_not_refresh_retention(self): h=Harness();h.step(0);h.clock.value=59;h.step(1,missing=True) h.clock.value=60;h.engine.expire();self.assertFalse(h.engine.states) def test_disabled(self): h=Harness();h.engine=Loitering(LoiteringConfig()) self.assertEqual(h.step(0),((None,),()));self.assertFalse(h.engine.states) def test_wrong_model_and_class(self): for kw in ({'model':'RFDETRFireSmoke'},{'model':'unknown-provider'},{'cls':2}): h=Harness();self.assertEqual(h.step(0,**kw),((None,),()));self.assertFalse(h.engine.states) def test_shared_person_models_are_supported(self): from app.pose_config import PERSON_MODELS for model in PERSON_MODELS: h=Harness();decisions,events=h.step(0,model=model) self.assertIsNotNone(decisions[0]);self.assertFalse(events);self.assertTrue(h.engine.states) def test_capacity(self): h=Harness(max_states=1);h.step(0) with self.assertRaises(RuntimeError):h.step(.1,tid=2) self.assertEqual(h.engine.last_seq,1) def test_boundary_and_bottom_center(self): h=Harness();d,_=h.step(0,box=(.3,.1,.5,.2));self.assertTrue(d[0].inside) d,_=h.step(.1,box=(.3,.3,.5,.9));self.assertFalse(d[0].inside) def test_duplicate_or_foreign_frame_no_mutation(self): h=Harness();h.step(0);seq=h.engine.last_seq f=FrameEnvelope('server','77','stream','run',1,seq,0,0,1280,720) o=DetectionObservation(f,'RFDETRMedium',0,.9,INSIDE) with self.assertRaises(ValueError):h.engine.evaluate(f,(o,),(1,)) f=replace(f,frame_seq=seq+1);o=replace(o,frame=f) with self.assertRaises(ValueError):h.engine.evaluate(f,(o,o),(1,1)) with self.assertRaises(ValueError):h.engine.evaluate(replace(f,run_generation=2),(),()) self.assertEqual(h.engine.last_seq,seq) def test_invalid_track_types(self): for tid in (True,1.1,[],{},'', 'a'*257): h=Harness() with self.subTest(tid=tid),self.assertRaises(ValueError):h.step(0,tid=tid) def test_clear_allows_new_generation(self): h=Harness();h.span(0,5);h.engine.clear() self.assertFalse(h.engine.states);self.assertIsNone(h.engine.identity);self.assertEqual(h.engine.last_seq,-1) def test_source_frame_seq_gaps_are_sampling_not_missing(self): h=Harness();h.step(0);h.seq+=20;h.step(.1) self.assertEqual(h.engine.states[1].dwell_ns,100_000_000) if __name__=='__main__':unittest.main()