import copy import math import unittest from engines import Engine from features import Settings,bodies from state import Episode def person(cx,tid,q=0,score=.8,dx=0,hand=None): k=[[0.,0.,0.] for _ in range(17)] for j,x,y in [(5,-20,150),(6,20,150),(11,-20,250),(12,20,250),(7,-25,190),(8,25,190),(9,-30,220),(10,30,220)]: k[j]=[cx+x+dx,y,.9] if hand is not None:k[10]=[hand+dx,200.,.9] return dict(score=score,bbox=[cx-60+dx,100.,cx+60+dx,350.],track_id=tid,query=q,keypoints=k) def row(i,kind='static',switch=False,untracked=False): dx=i*6 if kind=='translation' else 0 cx=350 if kind!='grapple' else 350+20*math.sin(i*.55) hand=(250 if i%6<3 else 320) if kind=='strike' else cx-10 if kind=='grapple' else None p=person(200,None if untracked else (3 if switch and i>=6 else 1),0,dx=dx,hand=hand) q=person(cx,None if untracked else 2,1,dx=dx) return dict(sample=i,source_frame=1+i*3,seconds=i/10,width=1000,height=800,persons=[p,q]) class EngineTest(unittest.TestCase): def run_rows(self,mode,rows): e=Engine(mode);return e,[e.step(r) for r in rows] def test_proximity_only(self): for mode in ('A','B'): _,r=self.run_rows(mode,[row(i) for i in range(60)]) self.assertFalse(any(x['events'] for x in r)) def test_rigid_translation(self): for mode in ('A','B'): _,r=self.run_rows(mode,[row(i,'translation') for i in range(35)]) self.assertFalse(any(x['events'] for x in r)) self.assertFalse(any(p['features'] and p['features']['strike'] for x in r for p in x['pairs'])) def test_strike_and_dedup(self): for mode in ('A','B'): _,r=self.run_rows(mode,[row(i,'strike') for i in range(35)]) self.assertEqual(sum(len(x['events']) for x in r),1) self.assertTrue(any('STRIKE' in x['current_rules'] for x in r)) def test_contact_branch(self): for mode in ('A','B'): _,r=self.run_rows(mode,[row(i,'grapple') for i in range(40)]) self.assertTrue(any('PUSH_GRAPPLE' in x['current_rules'] for x in r)) def test_id_change_not_forced(self): rows=[row(i,'strike',switch=True) for i in range(12)] _,a=self.run_rows('A',rows);_,b=self.run_rows('B',rows) self.assertFalse(any(x['events'] for x in a));self.assertTrue(any(x['events'] for x in b)) self.assertEqual(b[-1]['pairs'][0]['ids'],[3,2]) def test_untracked_only_b(self): rows=[row(i,'strike',untracked=True) for i in range(20)] _,a=self.run_rows('A',rows);_,b=self.run_rows('B',rows) self.assertFalse(any(x['candidate_pairs'] for x in a));self.assertTrue(any(x['events'] for x in b)) self.assertTrue(all(p['ids']==[None,None] for x in b for p in x['pairs'])) def test_missing_not_quiet(self): ep=Episode(Settings());self.assertTrue(ep.step(0,True,False)['event']) for i in range(1,20):r=ep.step(i/10,False,False) self.assertTrue(r['latched']);self.assertEqual(r['quiet_seconds'],0);self.assertFalse(r['display']) def test_observed_quiet_rearm(self): ep=Episode(Settings());ep.step(0,True,False) for i in range(1,23):r=ep.step(i/10,False,True) self.assertFalse(r['latched']);self.assertTrue(ep.step(2.3,True,False)['event']) def test_partial_joints_not_all_eight(self): rows=[row(i,'strike') for i in range(20)] for r in rows: for p in r['persons']: p['keypoints'][7][2]=p['keypoints'][8][2]=.01 _,out=self.run_rows('A',rows) self.assertTrue(any(r['events'] for r in out)) def test_low_reference_cannot_strike(self): rows=[row(i,'strike') for i in range(20)] for r in rows:r['persons'][0]['keypoints'][10][2]=.08 for mode in ('A','B'): _,out=self.run_rows(mode,rows);self.assertFalse(any(r['events'] for r in out)) def test_prefix_and_input_unchanged(self): rows=[row(i,'grapple') for i in range(30)];saved=copy.deepcopy(rows) for mode in ('A','B'): _,full=self.run_rows(mode,rows);_,prefix=self.run_rows(mode,rows[:17]);self.assertEqual(full[:17],prefix) self.assertEqual(rows,saved) def test_time_validation(self): e=Engine('A');e.step(row(0)) with self.assertRaises(ValueError):e.step(row(0)) def test_gaps_do_not_accumulate_contact(self): rows=[row(i,'grapple') for i in range(20)] for r in rows:r['seconds']=r['sample']*.3 for mode in ('A','B'): _,out=self.run_rows(mode,rows) self.assertFalse(any(r['events'] for r in out)) self.assertTrue(all(p['contact_seconds']==0 for r in out for p in r['pairs'])) def test_spatial_ambiguity_not_identity(self): rows=[row(i,'strike') for i in range(20)] for r in rows: p=copy.deepcopy(r['persons'][1]);p['query']=2;p['track_id']=None;r['persons'].append(p) _,out=self.run_rows('B',rows) self.assertTrue(any(r['ambiguous_matches'] for r in out)) self.assertFalse(any(r['events'] for r in out)) def test_threshold_and_bad_coordinates(self): r=row(0);r['persons'][0]['score']=.249999 b,reasons=bodies(r,Settings());self.assertEqual(len(b),1) r['persons'][0]['score']=.25 b,_=bodies(r,Settings());self.assertEqual(len(b),2) for j in (5,6,11,12):r['persons'][0]['keypoints'][j][0]=float('nan') b,reasons=bodies(r,Settings());self.assertEqual(len(b),1) self.assertTrue(any(r['reason']=='TORSO_ANCHOR_MISSING' for r in reasons)) def test_shared_preparation_is_immutable(self): r=row(0,untracked=True);prepared=bodies(r,Settings());saved=copy.deepcopy(prepared) Engine('A').step(r,prepared);out=Engine('B').step(r,prepared) self.assertEqual(prepared,saved);self.assertFalse(any(x['reason']=='A_REQUIRES_TRACK_ID' for x in out['rejected'])) def test_config_boundaries(self): with self.assertRaises(ValueError):Settings(window_seconds=4.) with self.assertRaises(ValueError):Settings(joint_score=.01) def test_contact_alone_can_emit(self): for mode in ('A','B'): e=Engine(mode,Settings(strike_motion=99.)) out=[e.step(row(i,'grapple')) for i in range(40)] self.assertTrue(any(r['events'] for r in out)) self.assertFalse(any('STRIKE' in r['current_rules'] for r in out)) def test_expiry_is_not_observed_quiet(self): ep=Episode(Settings());ep.step(0,True,False) r=ep.step(5.1,False,False) self.assertEqual(r['transition'],'EXPIRED_NOT_NORMAL');self.assertEqual(r['quiet_seconds'],0.) self.assertFalse(r['event']) def test_state_limit_is_explicit(self): e=Engine('A',Settings(max_states=1)) e.step(row(0));r=row(1);r['persons'][0]['track_id']=99 out=e.step(r) self.assertEqual(out['state_count'],1);self.assertEqual(out['unassigned_pairs'],1) if __name__=='__main__':unittest.main()