import bootstrap from copy import deepcopy from dataclasses import replace import json import unittest from data_robust import OUT,read from approach_link import ApproachLink,admit,follow_check from anchor_features import AnchorBuilder,measure from features import Settings from state import Episode from evaluate_integrated import group_states C=read(OUT/'candidates.json');CFG=C['config'];LINK=C['link'] def feature(t=.1,**kw): f=dict(actor=0,limb='LW',kind='hand',root=5,end=9,speed=.8,near=.05,vertical=.05, cosine=.9,approach=.06,elapsed=.1,extension=.04,angular=20.,motion=.08,delta=[.08,0], target_region='torso',hand_contact=.5,target_hand_offset=.5,shoulder_vy=0.,hip_vy=0.,root_vy=0., torso_horizontal=0.,response_speed=None,response_motion=.03,response_cosine=.5, input_start=t-.1,input_end=t,anchor_reason='OBSERVED',primary_anchor='T5:6', anchors={'T5:6':dict(start=.12,end=.05)},old_relative=[0,0],current_relative=[.08,0], relative_motion=.08,relative_delta=[.08,0],measurement_refs=['T1','T2'],measurement_previous_refs=['T1','T2']) f.update(kw);return f def row(t,fs=None,token='S1',reset=False): p=dict(token=token,refs=['T1','T2'],orders=[0,1],mode='STABLE_T_AND_GEOMETRY', candidate_state='CURRENT_NEAR',hold_age=0.,reset=reset,features=[feature(t)] if fs is None else fs) return dict(seconds=t,pairs=[p],ambiguous=0,pairs_over_cap=0) def recoil(t,**kw): f=feature(t,cosine=-.9,approach=-.03,extension=-.03,angular=0,delta=[-.08,0], anchors={'T5:6':dict(start=.05,end=.09)},relative_delta=[-.08,0],old_relative=[.08,0],current_relative=[0,0]) f.update(kw);return f class Tests(unittest.TestCase): def engine(self):return ApproachLink(CFG,LINK) def test_weak_first_does_not_license_recoil(self): e=self.engine();f=feature(.1,anchors={'T5:6':dict(start=.06,end=.05)},relative_motion=.03,relative_delta=[.03,0]) x=e.step(row(.1,[f]));self.assertTrue(x['raw']);self.assertFalse(x['pairs'][0]['diagnostics'][0]['seed']) self.assertFalse(e.step(row(.2,[recoil(.2)]))['hit']) def test_clear_approach_and_return(self): e=self.engine();v=e.step(row(.1));self.assertTrue(v['pairs'][0]['diagnostics'][0]['seed']) v=e.step(row(.2,[recoil(.2)]));self.assertTrue(v['hit']);self.assertIn('ARM_RETURN',v['pairs'][0]['rules']) def test_weak_direct_path_not_blanket_blocked(self): e=self.engine() for t in (.1,.2):v=e.step(row(t,[feature(t,relative_motion=.03,anchors={'T5:6':dict(start=.06,end=.05)})])) self.assertTrue(v['hit']);self.assertIn('ARM_DIRECT',v['pairs'][0]['rules']) def test_return_must_move_away_from_observed_minimum(self): e=self.engine();e.step(row(.1));f=recoil(.2,anchors={'T5:6':dict(start=.05,end=.055)}) v=e.step(row(.2,[f]));self.assertFalse(v['raw']);self.assertEqual(v['pairs'][0]['diagnostics'][0]['follow_link']['reason'],'FOLLOW_PATH_NOT_CONFIRMED') def test_return_cannot_include_preapproval_interval(self): e=self.engine();e.step(row(.1));f=recoil(.2,input_start=0.,elapsed=.2) self.assertFalse(e.step(row(.2,[f]))['raw']) def test_alternative_lag_can_supply_valid_return(self): e=self.engine();e.step(row(.1));fs=[recoil(.2,input_start=0.,elapsed=.2),recoil(.2)] v=e.step(row(.2,fs));self.assertTrue(v['hit']);self.assertEqual(v['pairs'][0]['counts'],[2,0]) def test_continue_approach(self): e=self.engine();e.step(row(.1)) f=feature(.2,approach=.001,angular=0,extension=-.02,anchors={'T5:6':dict(start=.06,end=.03)}) v=e.step(row(.2,[f]));self.assertTrue(v['hit']);self.assertEqual(v['pairs'][0]['diagnostics'][0]['follow_link']['reason'],'OBSERVED_CONTINUATION') def test_new_anchor_cannot_borrow_old_seed(self): e=self.engine();e.step(row(.1));f=recoil(.2,primary_anchor='H0',anchors={'H0':dict(start=.05,end=.09)}) self.assertFalse(e.step(row(.2,[f]))['raw']) def test_other_limb_cannot_borrow_seed(self): e=self.engine();e.step(row(.1));self.assertFalse(e.step(row(.2,[recoil(.2,limb='RW',root=6,end=10)]))['raw']) def test_other_actor_cannot_borrow_seed(self): e=self.engine();e.step(row(.1));self.assertFalse(e.step(row(.2,[recoil(.2,actor=1)]))['raw']) def test_partner_change_breaks_first_link(self): e=self.engine();e.step(row(.1));r=row(.2,[recoil(.2)]);r['pairs'][0]['refs']=['T1','T3'] self.assertFalse(e.step(r)['raw']) def test_reset_breaks_link(self): e=self.engine();e.step(row(.1));self.assertFalse(e.step(row(.2,[recoil(.2)],reset=True))['raw']) def test_absent_limb_and_expiry(self): e=self.engine();e.step(row(.1));v=e.step(row(.2,[]));self.assertFalse(v['observable']);self.assertFalse(v['quiet']) self.assertFalse(e.step(row(.5,[recoil(.5)]))['raw']) def test_follow_does_not_refresh_first_time(self): e=self.engine();e.step(row(.1)) for i in range(2,9):v=e.step(row(i/10,[recoil(i/10)])) self.assertFalse(v['raw']) def test_corrob_distinct_observations(self): pending={};f=feature(.1,relative_motion=.03,relative_delta=[.03,0],current_relative=[.03,0],anchors={'T5:6':dict(start=.12,end=.105)}) seed,_=admit(f,dict(seed=True),pending,CFG,LINK,.1);self.assertIsNone(seed) seed,d=admit(deepcopy(f),dict(seed=True),pending,CFG,LINK,.1);self.assertIsNone(seed);self.assertEqual(d['distinct_observations'],1) g=feature(.2,relative_motion=.04,relative_delta=[.04,0],old_relative=[.03,0],current_relative=[.07,0],anchors={'T5:6':dict(start=.105,end=.085)}) seed,d=admit(g,dict(seed=True),pending,CFG,LINK,.2);self.assertIsNotNone(seed);self.assertEqual(d['status'],'CORROBORATED_APPROACH') def test_pending_anchor_separated(self): p={};f=feature(.1,relative_motion=.03,anchors={'T5:6':dict(start=.06,end=.05)}) admit(f,dict(seed=True),p,CFG,LINK,.1) g=feature(.2,primary_anchor='H0',relative_motion=.03,anchors={'H0':dict(start=.06,end=.05)}) s,d=admit(g,dict(seed=True),p,CFG,LINK,.2);self.assertIsNone(s);self.assertEqual(d['distinct_observations'],1) def test_one_person_and_stationary_neighbors(self): e=self.engine();r=row(.1);r['pairs']=[];self.assertFalse(e.step(r)['hit']) for i in range(2,20):self.assertFalse(e.step(row(i/10,[feature(i/10,speed=0,motion=0,approach=0,extension=0)]))['raw']) def test_rules_lags_and_actor_counting(self): e=self.engine();v=e.step(row(.1,[feature(.1),feature(.1,elapsed=.2,input_start=-.1)])) self.assertEqual(v['pairs'][0]['counts'],[1,0]);self.assertFalse(v['hit']) self.assertFalse(e.step(row(.2,[feature(.2,actor=1)]))['hit']) def test_local_link_failure_not_global_veto(self): e=self.engine();r=row(.1);q=deepcopy(r['pairs'][0]);q['token']='S2';q['orders']=[2,3];q['features']=[];r['pairs'].append(q) e.step(r);r['seconds']=.2;r['pairs'][0]['features']=[feature(.2)];self.assertTrue(e.step(r)['hit']) def test_foot_direct_and_linked_return(self): e=self.engine();f=feature(.1,kind='foot',limb='LF',root=11,end=15) e.step(row(.1,[f]));g=recoil(.2,kind='foot',limb='LF',root=11,end=15) self.assertIn('LEG_RETURN',e.step(row(.2,[g]))['pairs'][0]['rules']) def test_close_branch_remains_separate(self): e=self.engine();e.step(row(.1)) for i in range(2,5): f=recoil(i/10,speed=.3,near=.04,response_speed=.3,delta=[.04 if i%2 else -.04,0]) v=e.step(row(i/10,[f])) self.assertIn('CLOSE_REVERSAL',v['pairs'][0]['rules']) def test_labels_do_not_affect_results(self): a=self.engine();b=self.engine() for i in range(1,5): r=row(i/10);s=deepcopy(r);s.update(clip='answer',reference='negative_reference',source_frame=999) self.assertEqual(a.step(r),b.step(s)) def test_prefix_causality(self): a=self.engine();b=self.engine();rows=[row(i/10) for i in range(1,5)] x=[a.step(r) for r in rows];y=[b.step(r) for r in rows];b.step(row(.5,[]));self.assertEqual(x,y) def test_first_l02_initialization(self):self.assertFalse(self.engine().step(row(105.352669))['hit']) def test_same_target_measurement(self): person=dict(height=100.,ref='T1',joints={'5':[0,0,1.],'9':[0,0,1.]}) target=dict(height=100.,ref='T2',joints={'0':[10,0,1.],'5':[50,0,1.],'6':[50,10,1.]}) new=deepcopy(person);new['joints']['9']=[8,0,1.] result=measure(dict(actor=0,root=5,end=9),[person,target],[new,target]) self.assertEqual(result['primary_anchor'],'H0');self.assertAlmostEqual(result['anchors']['H0']['start'],.035) self.assertEqual(result['anchors']['H0']['end'],0.) def test_missing_anchor_not_filled(self): person=dict(height=100.,ref='T1',joints={'5':[0,0,1.],'9':[0,0,1.]}) target=dict(height=100.,ref='T2',joints={}) r=measure(dict(actor=0,root=5,end=9),[person,target],[person,target]) self.assertEqual(r['anchors'],{});self.assertIsNone(r['primary_anchor']) def test_target_switch_not_false_approach(self): a=dict(height=100.,ref='T1',joints={'5':[0,0,1.],'9':[0,0,1.]}) old=dict(height=100.,ref='T2',joints={'0':[10,0,1.],'1':[100,0,1.]}) new=dict(height=100.,ref='T2',joints={'0':[90,0,1.],'1':[1,0,1.]}) r=measure(dict(actor=0,root=5,end=9),[a,old],[a,new]) self.assertEqual(r['primary_anchor'],'H0') self.assertGreater(r['anchors']['H0']['end'],r['anchors']['H0']['start']) self.assertEqual(r['anchors']['H1']['end'],0.) def test_yellow_and_event_separate(self): e=Episode(replace(Settings(),display_hold=1.));self.assertTrue(e.step(0,True,False)['event']) self.assertTrue(e.step(.9,False,False)['display']);self.assertFalse(e.step(1.1,False,False)['display']) r=e.step(1.2,True,False);self.assertTrue(r['display']);self.assertFalse(r['event']) def test_group_hold_not_extended(self): e=self.engine();h={};r=row(.1);group_states(r,e.step(r),.1,h);r=row(.2) self.assertEqual(group_states(r,e.step(r),.2,h)[0]['state'],'RED_CURRENT') r=row(.7,[]);self.assertEqual(group_states(r,e.step(r),.7,h)[0]['state'],'YELLOW_HOLD') r=row(1.3,[]);self.assertEqual(group_states(r,e.step(r),1.3,h)[0]['state'],'BLUE_CANDIDATE') def test_score_constants(self): from geometry import Geometry,point self.assertEqual((Geometry().person_score,Geometry().joint_score),(.25,.1)) self.assertTrue(point([1,1,.1],10,10));self.assertFalse(point([1,1,.099],10,10)) if __name__=='__main__': result=unittest.TextTestRunner(verbosity=2).run(unittest.defaultTestLoader.loadTestsFromTestCase(Tests)) (OUT/'reports/unit_tests.json').write_text(json.dumps(dict(tests=result.testsRun,failures=len(result.failures),errors=len(result.errors),status='PASS' if result.wasSuccessful() else 'FAIL'),indent=2)) raise SystemExit(not result.wasSuccessful())