import copy import json import unittest from common import OUT,read,save from fall_candidate import features,SpatialState C=read(OUT/'conditions.json') def person(kind='down',tid=1): k=[[0.,0.,0.] for _ in range(17)] if kind=='down': box=[100,100,300,180];xy={5:(120,130),6:(120,150),11:(170,130),12:(170,150),13:(215,130),14:(215,150),15:(270,130),16:(270,150)} elif kind=='standing': box=[100,100,180,300];xy={5:(130,130),6:(150,130),11:(130,190),12:(150,190),13:(130,240),14:(150,240),15:(130,290),16:(150,290)} elif kind=='bend': box=[80,100,240,300];xy={5:(100,145),6:(120,165),11:(170,155),12:(190,175),13:(170,215),14:(190,235),15:(170,275),16:(190,295)} else:raise ValueError(kind) for j,(x,y) in xy.items():k[j]=[float(x),float(y),.9] return dict(query=0,score=.7,track_id=tid,bbox=box,keypoints=k) class Tests(unittest.TestCase): def test_current_lying(self): self.assertTrue(features(person(),C)['hit']) def test_single_standing(self): self.assertFalse(features(person('standing'),C)['hit']) def test_bending_support(self): p=features(person('bend'),C);self.assertTrue(p['candidate']);self.assertFalse(p['hit']);self.assertTrue(p['standing_support']) def test_no_tracker_required(self): self.assertTrue(features(person(tid=None),C)['hit']) def test_id_switch_invariant(self): a=features(person(tid=1),C);b=features(person(tid=81),C) a.pop('track_id');b.pop('track_id');self.assertEqual(a,b) def test_missing_is_not_normal(self): p=person() for j in (13,14,15,16):p['keypoints'][j][2]=0 r=features(p,C);self.assertFalse(r['hit']);self.assertFalse(r['observable']) def test_sparse_same_side(self): p=person() for j in (6,12,14,16):p['keypoints'][j][2]=0 self.assertTrue(features(p,C)['hit']) def test_person_boundary(self): p=person();p['score']=.25;self.assertTrue(features(p,C)['hit']) p['score']=.249999;self.assertFalse(features(p,C)['eligible']) def test_joint_boundary(self): p=person() for j in (5,6,11,12,13,14,15,16):p['keypoints'][j][2]=.1 self.assertTrue(features(p,C)['hit']) for j in (11,12):p['keypoints'][j][2]=.099999 self.assertFalse(features(p,C)['observable']) def test_invalid_coordinate(self): p=person() for j in (11,12):p['keypoints'][j][0]=float('nan') self.assertFalse(features(p,C)['observable']) def test_local_not_global_veto(self): a=features(person(),C);b=features(person('bend'),C) s=SpatialState(C,'local');r=s.evaluate(0,[a,b],monotonic_now=0) self.assertTrue(r['hit']);self.assertIsNotNone(r['event']) def test_no_cross_person_joints(self): a=person();b=person() for j in (13,14,15,16):a['keypoints'][j][2]=0 for j in (5,6):b['keypoints'][j][2]=0 s=SpatialState(C,'no-mix');r=s.evaluate(0,[features(a,C),features(b,C)],monotonic_now=0) self.assertFalse(r['hit']) def test_rearm_two_seconds(self): s=SpatialState(C,'rearm');a=features(person(),C) s.evaluate(0,[a],monotonic_now=0) self.assertFalse(s.evaluate(1.999,[],monotonic_now=1.999)['rearmed']) r=s.evaluate(2.,[],monotonic_now=2.);self.assertTrue(r['rearmed']);self.assertIsNone(r['event']) def test_hold_not_rearm_evidence(self): s=SpatialState(C,'hold');a=features(person(),C);b=copy.deepcopy(a);b['hit']=False s.evaluate(0,[a],monotonic_now=0) r=s.evaluate(1.,[b],monotonic_now=1.);self.assertTrue(r['display_hold']);self.assertEqual(r['last_hit'],0) r=s.evaluate(1.001,[b],monotonic_now=1.001);self.assertFalse(r['display']);self.assertIsNone(r['event']) def test_continuous_cooldown(self): s=SpatialState(C,'continuous');a=features(person(),C) first=s.evaluate(0,[a],monotonic_now=0)['event'] self.assertIsNone(s.evaluate(9.999,[a],monotonic_now=9.999)['event']) last=s.evaluate(10,[a],monotonic_now=10)['event'];self.assertNotEqual(first['id'],last['id']) def test_blocked_does_not_latch(self): s=SpatialState(C,'blocked');a=features(person(),C) s.evaluate(0,[a],monotonic_now=0);s.evaluate(2,[],monotonic_now=2) r=s.evaluate(3,[a],monotonic_now=3);self.assertFalse(r['latched']);self.assertEqual(r['last_issued'],0) self.assertIsNotNone(s.evaluate(10,[a],monotonic_now=10)['event']) def test_no_scheduled_old_hit(self): s=SpatialState(C,'no-queue');a=features(person(),C) s.evaluate(0,[a],monotonic_now=0);s.evaluate(3,[a],monotonic_now=3) self.assertIsNone(s.evaluate(11,[],monotonic_now=11)['event']) def test_missing_and_long_gap(self): s=SpatialState(C,'gap');s.evaluate(0,[features(person(),C)],monotonic_now=0) r=s.evaluate(100,[],monotonic_now=100);self.assertFalse(r['latched']);self.assertFalse(r['display']);self.assertIsNone(r['event']) def test_out_of_order_and_duplicate(self): s=SpatialState(C,'order');s.evaluate(5,[],monotonic_now=0) for pts in (5,4): with self.assertRaises(ValueError):s.evaluate(pts,[],monotonic_now=1) def test_display_not_old_bbox(self): s=SpatialState(C,'bbox');s.evaluate(0,[features(person(),C)],monotonic_now=0) r=s.evaluate(.1,[],monotonic_now=.1);self.assertTrue(r['association_uncertain']);self.assertEqual(r['colors'],[]) def test_display_ambiguous(self): s=SpatialState(C,'ambiguous');a=features(person(),C);s.evaluate(0,[a],monotonic_now=0) b=copy.deepcopy(a);b['hit']=False;c=copy.deepcopy(b);c['query']=1 r=s.evaluate(.1,[b,c],monotonic_now=.1);self.assertTrue(r['association_uncertain']);self.assertTrue(all(x['state']!='YELLOW_HOLD' for x in r['colors'])) def test_red_overrides_hold(self): s=SpatialState(C,'red');a=features(person(),C);s.evaluate(0,[a],monotonic_now=0) b=copy.deepcopy(a);b['hit']=False;s.evaluate(.1,[b],monotonic_now=.1) r=s.evaluate(.2,[a],monotonic_now=.2);self.assertEqual(r['colors'][0]['state'],'RED_CURRENT') def test_run_reset(self): a=features(person(),C) for identity in ('run-a','run-b'): self.assertIsNotNone(SpatialState(C,identity).evaluate(0,[a],monotonic_now=0)['event']) def test_mock_same_id_deduplication(self): event=SpatialState(C,'delivery').evaluate(0,[features(person(),C)],monotonic_now=0)['event'] ledger={};attempts=0 def accept(item,fail=False): nonlocal attempts attempts+=1 if fail:return False ledger.setdefault(item['id'],item);return True self.assertFalse(accept(event,fail=True));self.assertTrue(accept(event));self.assertTrue(accept(event)) self.assertEqual(attempts,3);self.assertEqual(len(ledger),1) if __name__=='__main__': result=unittest.TextTestRunner(verbosity=2).run(unittest.defaultTestLoader.loadTestsFromTestCase(Tests)) save(OUT/'reports/synthetic_validation.json',dict(status='PASS' if result.wasSuccessful() else 'FAIL', tests=result.testsRun,failures=len(result.failures),errors=len(result.errors), external_event_delivery=False,limits='Mock duplicate delivery only; does not validate operational HTTP/DB')) raise SystemExit(0 if result.wasSuccessful() else 1)