import bootstrap import copy from dataclasses import replace import json import unittest from inputs import OUT from integrated import Integrated,DEFAULT,gates from evaluate_integrated import group_states from features import Settings from state import Episode def feature(**kw): f=dict(actor=0,limb='LW',kind='hand',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=.3,response_motion=.03,response_cosine=.5) 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()] if fs is None else fs) return dict(seconds=t,pairs=[p],ambiguous=0,pairs_over_cap=0) class ContractTests(unittest.TestCase): def test_direct_and_curve_are_one_observation(self): e=Integrated();v=e.step(row(0.0));self.assertFalse(v['hit']);self.assertEqual(v['pairs'][0]['counts'],[1,0]) self.assertGreater(len(v['pairs'][0]['passing'][0]['rules']),1) self.assertTrue(e.step(row(.1))['hit']) def test_lag_and_limb_duplicates_do_not_add_counts(self): e=Integrated();v=e.step(row(0.,[feature(),feature(elapsed=.2),feature(limb='RW')])) self.assertEqual(v['pairs'][0]['counts'],[1,0]);self.assertFalse(v['hit']) def test_subject_counts_not_merged(self): e=Integrated();e.step(row(0.));v=e.step(row(.1,[feature(actor=1)])) self.assertEqual(v['pairs'][0]['counts'],[1,1]);self.assertFalse(v['hit']) def test_pairs_not_merged(self): e=Integrated();e.step(row(0.));self.assertFalse(e.step(row(.1,token='S2'))['hit']) def test_no_pair_single_person(self): r=row(0.);r['pairs']=[];self.assertFalse(Integrated().step(r)['hit']) def test_standing_not_motion(self): e=Integrated() for i in range(15):self.assertFalse(e.step(row(i/10,[feature(speed=0,motion=0,approach=0,extension=0)]))['raw']) def test_local_suppression_not_global(self): e=Integrated();r=row(0.);q=copy.deepcopy(r['pairs'][0]);q['token']='S2';q['orders']=[2,3];q['features']=[feature(vertical=.5)] r['pairs'].append(q);e.step(r);r['seconds']=.1;v=e.step(r) self.assertTrue(v['hit']);self.assertFalse(v['pairs'][1]['raw']) def test_low_walking_arm_rejected(self): g=gates(feature(vertical=.4),DEFAULT);self.assertEqual(g['rules'],[]);self.assertIn('LOW_ENDPOINT',g['suppression_evidence']) def test_body_drop_rejected(self): g=gates(feature(shoulder_vy=1.,hip_vy=1.,root_vy=2.),DEFAULT) self.assertIn('BODY_DROP_DOMINANT',g['suppression_evidence']);self.assertFalse(g['rules']) def test_hand_contact_rejected(self): f=feature(vertical=-.1,hand_contact=.02,target_hand_offset=-.1) g=gates(f,DEFAULT,seed=dict(delta=[-.08,0])) self.assertTrue(g['suppression_applied']);self.assertIn('ARM_DIRECT',g['rules']) self.assertNotIn('ARM_RETURN',g['rules']) def test_contact_only_without_response_is_insufficient(self): f=feature(speed=.3,near=.04,delta=[-.04,0],cosine=-.9,approach=-.01,response_speed=None) self.assertNotIn('CLOSE_REVERSAL',gates(f,DEFAULT,contact=dict(delta=[.1,0]))['rules']) def test_seed_requires_current_motion(self): e=Integrated();e.step(row(0.));v=e.step(row(.1,[feature(speed=0,motion=0,approach=0,cosine=-1)])) self.assertFalse(v['hit']);self.assertFalse(v['raw']) def test_return_does_not_renew_seed(self): e=Integrated();e.step(row(0.)) for i in range(1,8): v=e.step(row(i/10,[feature(approach=-.03,cosine=-.9,extension=-.04,angular=0,delta=[-.08,0])])) self.assertFalse(v['raw']) def test_gap_reset_and_unknown(self): e=Integrated();e.step(row(0.));e.step(row(.1));v=e.step(row(.5,reset=True)) self.assertFalse(v['hit']);self.assertEqual(v['pairs'][0]['counts'],[1,0]) v=e.step(row(.6,[]));self.assertFalse(v['observable']);self.assertFalse(v['quiet']) def test_retention_does_not_create_evidence(self): e=Integrated();e.step(row(0.));e.step(row(.1));r=row(.2,[]);r['pairs'][0]['candidate_state']='GEOMETRY_HOLD' self.assertFalse(e.step(r)['hit']) def test_window_expiry(self): e=Integrated();e.step(row(0.));self.assertFalse(e.step(row(1.1))['hit']) def test_score_threshold_constants(self): from geometry import Geometry,point self.assertEqual(Geometry().person_score,.25);self.assertEqual(Geometry().joint_score,.1) self.assertTrue(point([2,2,.1],10,10));self.assertFalse(point([2,2,.099],10,10)) def test_feature_boundary(self): f=feature(speed=DEFAULT['hand_speed'],near=DEFAULT['hand_near'],vertical=DEFAULT['hand_vertical'], cosine=DEFAULT['direction'],approach=DEFAULT['approach']*.1,extension=-.01) self.assertIn('ARM_DIRECT',gates(f,DEFAULT)['rules']) def test_foot_needs_raised_geometry(self): self.assertFalse(gates(feature(kind='foot',limb='LF',vertical=.5),DEFAULT)['rules']) self.assertIn('LEG_DIRECT',gates(feature(kind='foot',limb='LF',vertical=.1),DEFAULT)['rules']) def test_contact_requires_reversal(self): seed=dict(time=0,delta=[.1,0]) f=feature(speed=.3,near=.04,delta=[-.04,0],cosine=-.9,approach=-.01) self.assertIn('CLOSE_REVERSAL',gates(f,DEFAULT,contact=seed)['rules']) self.assertNotIn('CLOSE_REVERSAL',gates(dict(f,delta=[.04,0]),DEFAULT,contact=seed)['rules']) def test_labels_do_not_affect_decisions(self): a=Integrated();b=Integrated() for i in range(8): r=row(i/10);q=copy.deepcopy(r);q.update(reference='positive_reference',source_frame=999,clip='answer') self.assertEqual(a.step(r),b.step(q)) def test_prefix_causality(self): a=Integrated();b=Integrated();prefix=[row(i/10) for i in range(5)] x=[a.step(r) for r in prefix];y=[b.step(r) for r in prefix] b.step(row(.5,[feature(speed=0)]));self.assertEqual(x,y) def test_episode_display_does_not_change_events(self): old=Episode(Settings());new=Episode(replace(Settings(),display_hold=1.)) for i in range(60): hit=i in (0,2,8,40);a=old.step(i/10,hit,not hit);b=new.step(i/10,hit,not hit) for k in ('event','latched','quiet_seconds','transition'):self.assertEqual(a[k],b[k]) self.assertEqual(new.last_event,4.) def test_yellow_expires_and_red_preempts(self): ep=Episode(replace(Settings(),display_hold=1.));self.assertTrue(ep.step(0,True,False)['event']) self.assertTrue(ep.step(.9,False,False)['display']);self.assertFalse(ep.step(1.1,False,False)['display']) v=ep.step(1.2,True,False);self.assertTrue(v['display']);self.assertFalse(v['event']) def test_group_hold_does_not_extend_itself(self): e=Integrated(dict(min_samples=1));r=row(0.);v=e.step(r);h={} self.assertEqual(group_states(r,v,0,h)[0]['state'],'RED_CURRENT') r=row(.5,[]);v=e.step(r);self.assertEqual(group_states(r,v,.5,h)[0]['state'],'YELLOW_HOLD') r=row(1.1,[]);v=e.step(r);self.assertEqual(group_states(r,v,1.1,h)[0]['state'],'BLUE_CANDIDATE') def test_uncertain_group_not_painted(self): e=Integrated(dict(min_samples=1));h={};r=row(0.);group_states(r,e.step(r),0,h) r=row(.2,[],token='S2',reset=True);self.assertEqual(group_states(r,e.step(r),.2,h)[0]['state'],'BLUE_CANDIDATE') if __name__=='__main__': suite=unittest.defaultTestLoader.loadTestsFromTestCase(ContractTests) result=unittest.TextTestRunner(verbosity=2).run(suite) (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(0 if result.wasSuccessful() else 1)