"""FALL v1 core tests; standard-library CPU tests, no SDK/network/DB.""" from dataclasses import asdict, replace import math import unittest from app.fall_config import FallConfig from app.fall_geometry import extract_features from app.fall_scenario import FallEngine from app.pose_config import PoseConfig from fall_fixtures import make_frame, make_pose, Sequence class ConfigTests(unittest.TestCase): def test_defaults(self): c = FallConfig() self.assertFalse(c.enabled) self.assertEqual((c.person_score,c.joint_score,c.upright_angle_deg,c.down_angle_deg),(.25,.15,35,60)) self.assertEqual((c.transition_window_seconds,c.transition_to_down_seconds,c.down_hold_seconds),(.8,.8,.7)) self.assertEqual((c.missing_grace_seconds,c.recovery_hold_seconds,c.confirmed_retention_seconds),(.5,1,5)) def test_roundtrip(self): c = FallConfig(enabled=True, max_states=4) self.assertEqual(c, FallConfig.parse(c.payload())) def test_unknown_and_nonmapping(self): for value in (None, [], {'junk':1}, {'min_joints':10}): with self.subTest(value=value), self.assertRaises(ValueError): FallConfig.parse(value) def test_bad_numeric_types(self): for name in ('person_score','joint_score','down_angle_deg','hip_drop_ratio','transition_to_down_seconds'): for val in (True, '0.5', float('nan'), float('inf'), -float('inf')): with self.subTest(name=name,val=val), self.assertRaises(ValueError): FallConfig(**{name:val}) def test_angle_relationships(self): for kw in ({'upright_angle_deg':60},{'down_angle_deg':30},{'knee_guard_torso_max_deg':60}, {'torso_delta_deg':0},{'knee_guard_angle_deg':180}): with self.subTest(kw=kw), self.assertRaises(ValueError): FallConfig(**kw) def test_ratios(self): for kw in ({'hip_drop_ratio':0},{'hip_drop_ratio':-1},{'span_collapse_ratio':1}, {'span_collapse_ratio':0},{'person_score':1.1},{'joint_score':-.1}): with self.subTest(kw=kw), self.assertRaises(ValueError): FallConfig(**kw) def test_durations(self): for name in [n for n in FallConfig().__dict__ if n.endswith('_seconds')]: for value in (0,-1,1e-10,3601): with self.subTest(name=name,value=value), self.assertRaises(ValueError): FallConfig(**{name:value}) def test_caps_and_window_relationship(self): for kw in ({'max_states':0},{'max_states':True},{'history_capacity':1}, {'history_capacity':4097},{'reference_window_seconds':.1}, {'confirmed_retention_seconds':.5},{'missing_grace_seconds':1.}): with self.subTest(kw=kw), self.assertRaises(ValueError): FallConfig(**kw) def test_disabled_does_not_allocate(self): e=FallEngine(); f=make_frame(0,0) r=e.evaluate(f,(make_pose(f),),(1,)) self.assertEqual(r.reason,'DISABLED');self.assertEqual(e.status()['states'],0) class GeometryTests(unittest.TestCase): def feature(self, **kw): f=make_frame(0,0) return extract_features(make_pose(f,**kw),FallConfig()) def test_aspect_correction(self): f=make_frame(0,0,width=1280,height=720) self.assertAlmostEqual(extract_features(make_pose(f,angle=45),FallConfig()).torso_angle,45) def test_bbox_irrelevant(self): for box in ((.1,.1,.2,.9),(.1,.1,.9,.2),(0,0,1,1)): self.assertEqual(self.feature(box=box),self.feature()) def test_midpoint(self): g=self.feature() self.assertEqual((g.torso_source,g.hip_source,g.span_source),('mid','mid','mid_ankle')) self.assertAlmostEqual(g.span,400) def test_same_side_fallback(self): self.assertEqual(self.feature(missing=(6,12)).torso_source,'left') self.assertEqual(self.feature(missing=(5,11)).torso_source,'right') def test_cross_side_forbidden(self): self.assertIsNone(self.feature(missing=(6,11)).torso_angle) def test_less_than_ten_joints(self): self.assertIsNotNone(self.feature().torso_angle) # fixture has only 8 scored joints self.assertEqual(PoseConfig().min_joints,10) def test_many_joints_but_no_torso(self): f=make_frame(0,0);p=make_pose(f) points=list(p.keypoints) for i in range(17): points[i]=(.5,.5,.9) for i in (5,6): points[i]=(.5,.5,0) self.assertIsNone(extract_features(replace(p,keypoints=tuple(points)),FallConfig()).torso_angle) def test_person_score_gate_and_boundary(self): self.assertIsNone(self.feature(score=.249).torso_angle) self.assertIsNotNone(self.feature(score=.25).torso_angle) def test_joint_score_boundary(self): self.assertIsNone(self.feature(joint_score=.149).torso_angle) self.assertIsNotNone(self.feature(joint_score=.15).torso_angle) def test_knee_guard(self): self.assertFalse(self.feature().knee_guard) self.assertTrue(self.feature(knee_bent=True).knee_guard) self.assertFalse(self.feature(angle=50,knee_bent=True).knee_guard) def test_knee_unavailable(self): self.assertIsNone(self.feature(missing=(13,14)).knee_guard) def test_one_knee_suffices(self): self.assertTrue(self.feature(knee_bent=True,missing=(14,)).knee_guard) def test_ankle_fallback_has_new_signature(self): self.assertEqual(self.feature(missing=(15,16)).span_source,'mid_knee') def test_nose_wrist_outliers_ignored(self): f=make_frame(0,0);p=make_pose(f);points=list(p.keypoints) for index in (0,1,2,3,4,7,8,9,10): points[index]=(999,-999,1) self.assertEqual(extract_features(p,FallConfig()),extract_features(replace(p,keypoints=tuple(points)),FallConfig())) def test_no_upper_span_unavailable(self): self.assertIsNone(self.feature(missing=(5,6)).span) def test_raw_pose_unmodified(self): f=make_frame(0,0);p=make_pose(f,joint_score=.1);before=asdict(p) extract_features(p,FallConfig());self.assertEqual(asdict(p),before) class TemporalTests(unittest.TestCase): def test_initial_arming(self): s=Sequence();self.assertEqual(s.step(0).decisions[0].state,'UNKNOWN') for i in range(1,5): self.assertEqual(s.step(i/10).decisions[0].state,'UNKNOWN') self.assertEqual(s.step(.5).decisions[0].state,'UPRIGHT') def test_initial_lying(self): s=Sequence() for i in range(30): r=s.down(i/10);self.assertFalse(r.event_candidates);self.assertEqual(r.decisions[0].state,'UNKNOWN') def test_initial_visible_knee_hint_prevents_arming(self): s=Sequence() for i in range(20): self.assertEqual(s.step(i/10,knee_bent=True).decisions[0].state,'UNKNOWN') def test_unavailable_knee_not_standing_proof_or_veto(self): s=Sequence() for i in range(6): r=s.step(i/10,missing=(13,14)) self.assertEqual(r.decisions[0].state,'UPRIGHT') self.assertIsNone(r.decisions[0].evidence.geometry.knee_guard) def test_reference_only_after_arming(self): s=Sequence();s.step(0);self.assertIsNone(s.engine.inspect(1)['scale']) for i in range(1,6):s.step(i/10) self.assertAlmostEqual(s.engine.inspect(1)['scale'][1],400) def test_reference_median(self): s=Sequence() for i,span in enumerate((400,400,600,400,400,400,400)): s.step(i/10,span=span) self.assertAlmostEqual(s.engine.inspect(1)['scale'][1],400) def test_confirmation_exact_point_seven(self): s=Sequence();s.arm() for t in (.7,.8,.9,1.,1.1,1.2,1.3,1.39):self.assertFalse(s.down(t).event_candidates) r=s.down(1.4);self.assertEqual(len(r.event_candidates),1) self.assertEqual(r.decisions[0].down_hold_ns,700_000_000) def test_candidate_current_identity_and_bbox(self): s=Sequence();e=s.confirm() self.assertEqual((e.frame.frame_seq,e.frame.pts_ns,e.object_ref,e.track_id),(14,1_400_000_000,'0',1)) self.assertEqual(e.scenario,'FALL');self.assertEqual(e.bbox_xyxy,(.1,.1,.9,.9)) def test_episode_once(self): s=Sequence();s.confirm() for i in range(15,100):self.assertFalse(s.down(i/10).event_candidates) self.assertEqual(s.engine.status()['events_created'],1) def test_recovery_and_second_episode(self): s=Sequence();s.confirm() for i in range(15,25): r=s.step(i/10);self.assertTrue(r.decisions[0].alert) self.assertFalse(s.step(2.5).decisions[0].alert) s.step(2.6) events=[] for i in range(27,35): events.extend(s.down(i/10).event_candidates) self.assertEqual(len(events),1);self.assertEqual(events[0].episode,2) def test_reference_frozen_during_fall(self): s=Sequence();s.arm();before=s.engine.inspect(1)['references'] for i in range(7,16):s.down(i/10) self.assertEqual(s.engine.inspect(1)['references'],before) def test_short_missing_freezes_and_excludes_edges(self): s=Sequence();s.arm();s.down(.7);s.down(.8);s.down(.9) for t in (1.,1.1):s.step(t,present=False) self.assertEqual(s.engine.inspect(1)['down_hold_ns'],200_000_000) s.down(1.2) self.assertEqual(s.engine.inspect(1)['down_hold_ns'],200_000_000) for t in (1.3,1.4,1.5,1.6):self.assertFalse(s.down(t).event_candidates) self.assertTrue(s.down(1.7).event_candidates) def test_long_down_feature_unavailable_resets(self): s=Sequence();s.arm();s.down(.7);s.down(.8) for t in (.9,1.,1.1,1.2,1.3):s.down(t,missing=(5,6)) self.assertEqual(s.engine.inspect(1)['state'],'DOWN_CANDIDATE') s.down(1.31,missing=(5,6)) self.assertEqual(s.engine.inspect(1)['state'],'UNKNOWN') self.assertEqual(s.engine.inspect(1)['down_hold_ns'],0) def test_long_whole_track_missing_resets_unconfirmed(self): s=Sequence();s.arm();s.down(.7) s.step(1.3,present=False) self.assertEqual(s.engine.inspect(1)['state'],'UNKNOWN') def test_recovery_short_missing_freezes(self): s=Sequence();s.confirm();s.step(1.5);s.step(1.6) s.step(1.7,missing=(5,6));s.step(1.8) self.assertEqual(s.engine.inspect(1)['recovery_hold_ns'],100_000_000) def test_recovery_long_feature_unavailable_latch_preserved(self): s=Sequence();s.confirm();s.step(1.5);s.step(1.6) for t in (1.7,1.8,1.9,2.,2.1,2.11):s.step(t,missing=(5,6)) state=s.engine.inspect(1) self.assertEqual(state['state'],'FALL_CONFIRMED');self.assertTrue(state['latched']) self.assertEqual(state['recovery_hold_ns'],0) def test_opposite_down_posture_resets(self): s=Sequence();s.arm();s.down(.7);s.down(.8) s.step(.9,angle=40) self.assertEqual(s.engine.inspect(1)['down_hold_ns'],0) def test_recovery_down_resets(self): s=Sequence();s.confirm();s.step(1.5);s.step(1.6);s.down(1.7) self.assertEqual(s.engine.inspect(1)['state'],'FALL_CONFIRMED') self.assertEqual(s.engine.inspect(1)['recovery_hold_ns'],0) def test_explicit_gap_does_not_count_bridge(self): s=Sequence();s.arm();s.down(.7);s.down(.8);s.down(1.,gap=True) self.assertEqual(s.engine.inspect(1)['down_hold_ns'],100_000_000) def test_invalid_pts_no_state_advance(self): s=Sequence();s.confirm();before=s.engine.inspect(1) r=s.step(None) self.assertFalse(r.accepted);self.assertEqual(before,s.engine.inspect(1)) def test_invalid_pts_boundaries(self): s=Sequence();s.arm();before=s.engine.inspect(1) for v in (None,-1,True,1.25,2**63): f=make_frame(100,0,pts_ns=v) self.assertFalse(s.engine.evaluate(f,(make_pose(f),),(1,)).accepted) self.assertEqual(before,s.engine.inspect(1)) def test_duplicate_regressing_and_stale_seq(self): s=Sequence();s.arm();before=s.engine.inspect(1) for seq,t in ((10,.6),(10,.5),(6,.8)): f=make_frame(seq,t) self.assertFalse(s.engine.evaluate(f,(make_pose(f),),(1,)).accepted) self.assertEqual(before,s.engine.inspect(1)) def test_declared_invalid_gap_not_credited_on_return(self): s=Sequence();s.arm();s.down(.7);s.down(.8);s.step(None);s.down(1.) self.assertEqual(s.engine.inspect(1)['down_hold_ns'],100_000_000) def test_latch_not_cleared_by_feature_loss(self): s=Sequence();s.confirm() for i in range(15,90):s.step(i/10,missing=(5,6)) self.assertTrue(s.engine.inspect(1)['latched']) # person remains present def test_retention_media_boundary(self): s=Sequence();s.confirm();s.step(6.399,present=False) self.assertIsNotNone(s.engine.inspect(1)) s.step(6.4,present=False);self.assertIsNone(s.engine.inspect(1)) self.assertEqual(s.engine.status()['events_created'],1) def test_reappearing_after_retention_is_new_unknown(self): s=Sequence();s.confirm();r=s.down(6.4) self.assertEqual(r.decisions[0].state,'UNKNOWN');self.assertFalse(r.event_candidates) def test_new_track_no_state_transfer(self): s=Sequence();s.confirm();r=s.down(1.5,track_id=31) self.assertEqual(r.decisions[0].state,'UNKNOWN');self.assertTrue(s.engine.inspect(1)['latched']) def test_missing_id_no_allocation(self): s=Sequence();r=s.step(0,track_id=None) self.assertEqual(r.decisions[0].state,'PENDING_ID');self.assertEqual(s.engine.status()['states'],0) def test_foreign_run_config_geometry_rejected_without_mutation(self): s=Sequence();s.arm();before=s.engine.inspect(1) for kw in ({'run_id':'other'},{'source_id':'88'},{'run_generation':2}, {'frame_width':2000},{'config_version':'new'}): f=make_frame(100,2,**kw) with self.subTest(kw=kw),self.assertRaises(ValueError):s.engine.evaluate(f,(make_pose(f),),(1,)) self.assertEqual(before,s.engine.inspect(1)) def test_batch_association_rejected_before_mutation(self): s=Sequence();s.arm();before=s.engine.inspect(1);f=make_frame(7,.7);p=make_pose(f) for poses,ids in (((p,p),(1,1)),((p,),()),((p,),(False,)), ((replace(p,model_key='RFDETRMedium'),),(1,)),((replace(p,frame=make_frame(8,.8)),),(1,))): with self.assertRaises(ValueError):s.engine.evaluate(f,poses,ids) self.assertEqual(before,s.engine.inspect(1)) def test_capacity_reject_new_not_evict_existing(self): s=Sequence(max_states=1);s.confirm();r=s.step(1.5,track_id=2) self.assertEqual(r.decisions[0].state,'CAPACITY');self.assertTrue(s.engine.inspect(1)['latched']) self.assertEqual(s.engine.status()['states'],1) def test_history_bounded(self): s=Sequence(history_capacity=16) for i in range(500):s.step(i/100) state=s.engine.inspect(1) self.assertLessEqual(max(state['history_sizes']),16) self.assertLessEqual(state['reference_samples'],16) def test_clear_allows_new_run(self): s=Sequence();s.confirm();s.engine.clear();self.assertEqual(s.engine.status()['states'],0) r=s.step(0,frame_changes={'run_id':'new','run_generation':2}) self.assertTrue(r.accepted);self.assertEqual(r.decisions[0].state,'UNKNOWN') if __name__=='__main__':unittest.main()