"""September demo policy tests. Synthetic Pose only; not real-video accuracy. Run from repository root: PYTHONPATH=.:tests python -m unittest tests.test_scenario_demo_sensitivity -v """ from dataclasses import replace from itertools import count import json import os import unittest from unittest.mock import patch from app.contracts.frames import DetectionObservation from app.contracts.visualization import FrameResult, VisualizationCommand from app.fall_config import FallConfig from app.fall_runtime import FallScenario, FallSettings, project_fall, validate as validate_fall from app.fall_scenario import FallEngine from app.fight_config import FightConfig from app.fight_runtime import FightScenario, FightSettings, project_fight, validate as validate_fight from app.fight_scenario import FightEngine, _body from app.region_scenario import alerted, event_ids, visible from tests.fall_fixtures import make_frame, make_pose, Sequence from tests.test_fight import frame, person def configured_fall(**kw): return FallScenario(FallSettings.parse({'enabled': True}), **kw) def configured_fight(**kw): return FightScenario(FightSettings.parse({'enabled': True}), **kw) def down(scenario, t, tid=17, **options): f = make_frame(round(t * 1000), t) return scenario.evaluate(f, (make_pose(f, angle=70., **options),), (tid,)) def pair(scenario, t, ids=(17, 20), *, wrist=460, far=False, missing=(), gap=False): f = frame(t, frame_seq=round(t * 1000)) poses = (person(f, 450, wrist, missing=missing), person(f, 1000 if far else 650)) return scenario.evaluate(f, poses, ids, gap_before=gap) def wire(frame_value, poses, ids, rows, events, config): """Use the existing result-worker display rule and actual public serializer.""" commands = tuple(VisualizationCommand( DetectionObservation(frame_value, p.model_key, 0, p.score, p.bbox_xyxy), 'person', color='#FF3030' if alerted(row) else '#00CC66', visible=visible(p.score, row), track_id=tid, event_ids=event_ids(row), scenario=row, pose=p) for p, tid, row in zip(poses, ids, rows)) return FrameResult(frame_value, commands, events, config).snapshot() class DemoTests(unittest.TestCase): def setUp(self): # Host settings must not make policy tests order/environment dependent. guard = patch.dict(os.environ, {}, clear=True) guard.start() self.addCleanup(guard.stop) def test_effective_defaults_not_new_public_config_fields(self): fall = FallSettings.parse({}).settings_payload() fight = FightSettings.parse({}).settings_payload() self.assertEqual([fall[k] for k in ('person_score', 'joint_score', 'down_angle_deg', 'test_skip_down_hold')], [.15, .10, 50., True]) expected = dict(person_score=.15, joint_score=.10, pair_distance_ratio=1.8, pair_close_hold_seconds=.1, arm_motion_window_seconds=1., arm_motion_ratio=.1, toward_opponent_ratio=.01, required_bursts=1) self.assertEqual({k: fight[k] for k in expected}, expected) roi = {'roi', 'roi_id', 'roi_name', 'debug_roi'} self.assertEqual(set(fall), set(FallConfig().payload()) | roi) self.assertEqual(set(fight), set(FightConfig().payload()) | roi) # Pure-core baseline remains available for exact temporal regression. self.assertEqual(FallConfig().down_angle_deg, 60.) self.assertEqual(FightConfig().required_bursts, 2) def test_web_null_and_explicit_confidence(self): for name, project in (('fall', project_fall), ('fight', project_fight)): for confidence, expected in ((None, .15), (.7, .7)): with self.subTest(name=name, confidence=confidence): spec = {} project(spec, [dict(scenarioCode=name.upper(), enabled=True, confidence=confidence, rois=[])]) self.assertEqual(spec[name]['person_score'], expected) self.assertTrue(spec[name]['enabled']) def test_existing_saved_values_win_and_inactive_spec_unchanged(self): for name, project, config in (('fall', project_fall, FallConfig), ('fight', project_fight, FightConfig)): spec = {name: config().payload()} project(spec, [dict(scenarioCode=name.upper(), enabled=True, confidence=.4, rois=[])]) self.assertEqual(spec[name]['joint_score'], .15) empty = {} project(empty, [dict(scenarioCode=name.upper(), enabled=False, confidence=None, rois=[])]) self.assertEqual(empty, {}) def test_skip_hold_site_precedence_and_direct_independence(self): key = 'CONTEXTRA_FALL_SKIP_DOWN_HOLD_DEFAULT' self.assertTrue(FallSettings.parse({}).policy.test_skip_down_hold) for value in ('true', 'false'): with patch.dict(os.environ, {key: value}): self.assertEqual(FallSettings.parse({}).policy.test_skip_down_hold, value == 'true') for explicit in (True, False): c = FallSettings.parse({'test_skip_down_hold': explicit}) self.assertEqual(c.policy.test_skip_down_hold, explicit) scenario = FallScenario(FallSettings.parse({'enabled': True, 'test_skip_down_hold': False})) self.assertEqual(len(down(scenario, 0)[1]), 1) # Separate direct-down option. def test_private_environment_switches(self): with patch.dict(os.environ, {'CONTEXTRA_FALL_DEMO_DIRECT_DOWN': 'false'}): self.assertFalse(down(configured_fall(), 0)[1]) with patch.dict(os.environ, {'CONTEXTRA_FIGHT_CURRENT_FRAME_FALLBACK': 'true'}): self.assertEqual(len(pair(configured_fight(), 0)[1]), 1) def test_invalid_private_options_fail_closed(self): for key, create in ( ('CONTEXTRA_FALL_SKIP_DOWN_HOLD_DEFAULT', lambda: FallSettings.parse({})), ('CONTEXTRA_FALL_DEMO_DIRECT_DOWN', configured_fall), ('CONTEXTRA_FIGHT_CURRENT_FRAME_FALLBACK', configured_fight)): for value in ('1', 'TRUE', '', 'typo'): with self.subTest(key=key, value=value), patch.dict(os.environ, {key: value}): with self.assertRaises(ValueError): create() for key, create in ( ('CONTEXTRA_FALL_DIRECT_DOWN_COOLDOWN_SECONDS', configured_fall), ('CONTEXTRA_FIGHT_EVENT_COOLDOWN_SECONDS', configured_fight)): for value in ('nan', 'inf', '-1', '31', ''): with self.subTest(key=key, value=value), patch.dict(os.environ, {key: value}): with self.assertRaises(ValueError): create() for make in (lambda v: FallEngine(direct_down=v), lambda v: FightEngine(current_frame_fallback=v)): for v in (1, 'true'): with self.assertRaises(ValueError): make(v) for make in (lambda v: FallEngine(direct_down_cooldown_seconds=v), lambda v: FightEngine(event_cooldown_seconds=v)): with self.assertRaises(ValueError): make(True) def test_new_private_fields_are_rejected_by_settings(self): for parser, key in ((FallSettings, 'direct_down'), (FallSettings, 'direct_down_cooldown_seconds'), (FightSettings, 'current_frame_fallback'), (FightSettings, 'event_cooldown_seconds')): with self.subTest(key=key), self.assertRaises(ValueError): parser.parse({key: True}) def test_tracking_on_and_identity_requirements_preserved(self): for name, validate in (('fall', validate_fall), ('fight', validate_fight)): for extra in ({}, {'server_id': 'test', 'source_id': '77', 'tracking': {'enabled': False}}): with self.subTest(name=name, extra=extra), self.assertRaises(ValueError): validate({name: {'enabled': True}, **extra}) def test_fall_temporal_off_switch_and_initial_down(self): s = configured_fall(direct_down=False) for i in range(10): rows, events = down(s, i / 10) self.assertEqual(rows[0].state, 'UNKNOWN') self.assertFalse(events) legacy = Sequence() self.assertEqual(legacy.confirm().scenario, 'FALL') def test_fall_current_torso_boundary(self): for angle, accepted in ((49.99, False), (50., True), (80., True)): f = make_frame(0, 0) s = configured_fall() rows, events = s.evaluate(f, (make_pose(f, angle=angle),), (31,)) self.assertEqual(bool(events), accepted) self.assertEqual(alerted(rows[0]), accepted) def test_fall_score_joint_torso_and_id_gates(self): for opts, tid in (({'score': .149}, 1), ({'joint_score': .099}, 1), ({'missing': (5, 6)}, 1), ({}, None)): with self.subTest(opts=opts, tid=tid): self.assertFalse(down(configured_fall(), 0, tid, **opts)[1]) self.assertTrue(down(configured_fall(), 0, 1, score=.15, joint_score=.10)[1]) def test_fall_id_switch_cooldown_and_event_identity(self): numbers = count() s = configured_fall(event_id_factory=lambda: f'fall-{next(numbers)}') self.assertEqual(down(s, 0, 17)[1][0].event_id, 'fall-0') for t, tid in ((.1, 31), (.2, 44), (1.999, 44)): rows, events = down(s, t, tid) self.assertTrue(alerted(rows[0])) self.assertIsNone(rows[0].event_id) self.assertFalse(events) rows, events = down(s, 2., 44) self.assertEqual((events[0].track_id, events[0].event_id, events[0].frame.pts_ns), (44, 'fall-1', 2_000_000_000)) self.assertFalse(down(s, 4.1, 44)[1]) # Emitted episode is not a periodic timer. def test_fall_pending_requires_current_down_evidence(self): s = configured_fall() down(s, 0, 17) down(s, .1, 31) self.assertFalse(down(s, 2., 31, missing=(5, 6))[1]) self.assertTrue(down(s, 2.1, 31)[1]) def test_fall_normal_candidate_preempts_direct_independent_of_order(self): s = FallEngine(FallConfig(enabled=True, test_skip_down_hold=True), direct_down=True) for i in range(7): f = make_frame(i, i/10) s.evaluate(f, (make_pose(f),), (17,)) f = make_frame(7, .7) p = make_pose(f, angle=80, hip=(500, 520), span=200) r = s.evaluate(f, (p, p), (31, 17)) # Direct first, original temporal second. self.assertEqual([e.track_id for e in r.event_candidates], [17]) self.assertTrue(all(d.alert for d in r.decisions)) def test_fall_normal_events_not_limited(self): s = FallEngine(FallConfig(enabled=True, test_skip_down_hold=True), direct_down=True) for i in range(7): f = make_frame(i, i/10) s.evaluate(f, (make_pose(f), make_pose(f)), (17, 31)) f = make_frame(7, .7) p = make_pose(f, angle=80, hip=(500, 520), span=200) r = s.evaluate(f, (p, p), (17, 31)) self.assertEqual(len(r.event_candidates), 2) def test_fall_clear_new_run_and_gap(self): s = configured_fall() down(s, 0) f = make_frame(1, .1) rows, events = s.evaluate(f, (make_pose(f, angle=70),), (31,), gap_before=True) self.assertTrue(alerted(rows[0])); self.assertFalse(events) s.clear() f = replace(make_frame(0, 0), run_id='new-run', run_generation=2) self.assertEqual(len(s.evaluate(f, (make_pose(f, angle=70),), (31,))[1]), 1) def test_invalid_stale_pts_never_advances_cooldown(self): for create, poses in ( (configured_fall, lambda f: (make_pose(f, angle=70),)), (lambda: configured_fight(current_frame_fallback=True), lambda f: (person(f,450), person(f,650)))): s = create(); f = frame(0) p = poses(f); ids = tuple(range(1, len(p)+1)) self.assertTrue(s.evaluate(f, p, ids)[1]) for pts in (None, True, -1, 2**63, 0): bad = replace(frame(.1), pts_ns=pts) self.assertFalse(s.evaluate(bad, poses(bad), ids)[1]) good = frame(.2) changed = tuple(x+10 for x in ids) self.assertFalse(s.evaluate(good, poses(good), changed)[1]) def test_fight_relaxed_one_motion_two_frames(self): s = configured_fight() self.assertFalse(pair(s, 0)[1]) rows, events = pair(s, .1, wrist=490) self.assertEqual(len(events), 1) self.assertTrue(all(alerted(r) for r in rows)) self.assertEqual(events[0].track_id, 17) def test_fight_fallback_off_stationary_and_far(self): for far in (False, True): s = configured_fight() for i in range(20): self.assertFalse(pair(s, i/10, far=far)[1]) def test_fight_body_mid_same_side_cross_side(self): f = frame(0) for missing, source in (((), 'mid'), ((6, 12), 'left'), ((5,), 'right')): p = person(f, 450, missing=missing) before = p.keypoints b = _body(p, 1, '0', FightConfig()) self.assertEqual(b.torso_source, source) self.assertEqual(p.keypoints, before) self.assertIsNone(_body(person(f,450,missing=(6,11)),1,'0',FightConfig())) def test_fight_body_source_change_not_motion(self): s = configured_fight() pair(s, 0) self.assertFalse(pair(s, .1, wrist=490, missing=(5,))[1]) self.assertTrue(pair(s, .2, wrist=520, missing=(5,))[1]) def test_fight_natural_half_second_gap_and_explicit_gap(self): s = configured_fight() pair(s, 0) self.assertTrue(pair(s, .5, wrist=490)[1]) s = configured_fight() pair(s, 0) self.assertFalse(pair(s, .5, wrist=490, gap=True)[1]) self.assertTrue(pair(s, .6, wrist=520)[1]) def test_fight_current_frame_on_ignores_hold_burst_history(self): s = configured_fight(current_frame_fallback=True) rows, events = pair(s, 0) self.assertEqual(len(events), 1) self.assertTrue(all(alerted(r) for r in rows)) self.assertEqual(s.engine.snapshot().get('arm_bursts', 0), 0) def test_fight_current_frame_fallback_gates(self): for far, missing, ids in ((True, (), (1,2)), (False, (6,11), (1,2)), (False, (), (None,2))): s = configured_fight(current_frame_fallback=True) self.assertFalse(pair(s, 0, ids, far=far, missing=missing)[1]) s = configured_fight(current_frame_fallback=True) f = frame(0) p = tuple(person(f,x,missing=(9,10)) for x in (450,650)) self.assertFalse(s.evaluate(f,p,(1,2))[1]) def test_fight_changed_ids_cooldown_pending_red_and_retry(self): numbers = count() s = configured_fight(current_frame_fallback=True, event_id_factory=lambda:f'fight-{next(numbers)}') self.assertEqual(pair(s,0)[1][0].event_id, 'fight-0') for t, ids in ((.1,(31,20)), (.2,(44,20)), (1.999,(44,20))): rows, events = pair(s,t,ids) self.assertFalse(events) self.assertTrue(all(alerted(r) for r in rows)) self.assertTrue(all(not event_ids(r) for r in rows)) events = pair(s,2.,(44,20))[1] self.assertEqual(len(events),1) self.assertEqual(events[0].event_id,'fight-1') self.assertEqual(events[0].frame.pts_ns,2_000_000_000) self.assertFalse(pair(s,4.1,(44,20))[1]) def test_fight_pending_no_stale_event_after_expiry(self): s = configured_fight(current_frame_fallback=True) pair(s,0); pair(s,.1,(31,20)) f=frame(2.) p=tuple(person(f,x,missing=(9,10)) for x in (450,650)) self.assertFalse(s.evaluate(f,p,(31,20))[1]) self.assertTrue(pair(s,2.1,(31,20))[1]) def test_fight_temporal_path_also_has_cooldown(self): s = configured_fight() pair(s,0); self.assertTrue(pair(s,.1,wrist=490)[1]) pair(s,.2,(31,20)) rows, events = pair(s,.3,(31,20),wrist=490) self.assertFalse(events) self.assertTrue(all(alerted(r) for r in rows)) # No new burst edge is required after cooldown, but fresh motion is. for t in (.4,.5,.6,.7,.8,.9,1.,1.1,1.2,1.3,1.4,1.5,1.6,1.7,1.8,1.9,2.): pair(s,t,(31,20)) self.assertTrue(pair(s,2.1,(31,20),wrist=490)[1]) def test_fight_simultaneous_pairs_single_candidate(self): s = configured_fight(current_frame_fallback=True) f = frame(0) rows, events = s.evaluate(f,tuple(person(f,x) for x in (450,550,650)),(1,2,3)) self.assertEqual(len(events),1) self.assertTrue(all(alerted(r) for r in rows)) def test_fight_clear_gap_and_source_isolation(self): s = configured_fight(current_frame_fallback=True) pair(s,0) self.assertFalse(pair(s,.1,(31,20),gap=True)[1]) with self.assertRaises(ValueError): f=replace(frame(.2, frame_seq=200),source_id='88') s.evaluate(f,(person(f,450),person(f,650)),(31,20)) s.clear() self.assertTrue(pair(s,0)[1]) self.assertTrue(pair(configured_fight(current_frame_fallback=True),0)[1]) self.assertTrue(down(configured_fall(),0)[1]) # No FIGHT -> FALL limiter coupling. def test_disabled_scenarios_do_not_generate_fallback_events(self): fall = FallScenario(FallSettings.parse({})) fight = FightScenario(FightSettings.parse({}),current_frame_fallback=True) self.assertFalse(down(fall,0)[1]); self.assertFalse(pair(fight,0)[1]) def test_cooldown_red_public_serialization_no_fake_ids(self): for name, create, poses in ( ('FALL',configured_fall,lambda f:(make_pose(f,angle=70),)), ('FIGHT',lambda:configured_fight(current_frame_fallback=True), lambda f:(person(f,450),person(f,650)))): s=create(); f=frame(0); p=poses(f); ids=tuple(range(1,len(p)+1)) first_rows,first_events=s.evaluate(f,p,ids) first=wire(f,p,ids,first_rows,first_events,s.config) f=frame(.1);p=poses(f);ids=tuple(x+10 for x in ids) rows,events=s.evaluate(f,p,ids) snap=wire(f,p,ids,rows,events,s.config) self.assertFalse(snap['events']) self.assertEqual(set(first),set(snap)) for obj in snap['objects']: self.assertEqual(obj['color'],'#FF3030') self.assertTrue(obj['visible']) self.assertFalse(obj['eventIds']) self.assertTrue(all(d['eventId'] is None for d in obj['scenarioResults'])) for tree in (first,snap,s.config.settings_payload()): text=json.dumps(tree,allow_nan=False).lower() for forbidden in ('cooldown','direct_down','current_frame_fallback'): self.assertNotIn(forbidden,text) self.assertEqual(first_events[0].payload()['scenario'],name) def test_roi_subset_preserves_current_object_refs(self): roi = ((.25,0.),(.75,0.),(.75,1.),(.25,1.)) s = FallScenario(FallSettings.parse({'enabled':True, 'roi':roi})) f = make_frame(0,0) p = (make_pose(f,angle=70,box=(.01,.1,.1,.9)), make_pose(f,angle=70)) rows, events = s.evaluate(f,p,(999,17)) self.assertEqual(rows[0].state,'OUTSIDE') self.assertEqual((events[0].object_ref,events[0].track_id),('1',17)) s = FightScenario(FightSettings.parse({'enabled':True, 'roi':roi}),current_frame_fallback=True) f = frame(0) p = (person(f,1100),person(f,450),person(f,650)) rows, events = s.evaluate(f,p,(999,17,20)) self.assertEqual(rows[0][0].state,'OUTSIDE') self.assertEqual((events[0].object_ref,events[0].track_id),('1',17)) self.assertEqual(events[0].bbox,p[1].bbox_xyxy) def test_private_cooldown_duration_is_effective(self): with patch.dict(os.environ, {'CONTEXTRA_FALL_DIRECT_DOWN_COOLDOWN_SECONDS':'3.0', 'CONTEXTRA_FIGHT_EVENT_COOLDOWN_SECONDS':'3.0'}): a=configured_fall(); b=configured_fight(current_frame_fallback=True) down(a,0); pair(b,0) self.assertFalse(down(a,2.,31)[1]); self.assertFalse(pair(b,2.,(31,20))[1]) self.assertTrue(down(a,3.,31)[1]); self.assertTrue(pair(b,3.,(31,20))[1]) def test_fall_recovery_clears_suppressed_episode(self): s=configured_fall();down(s,0,17);down(s,.1,31) for i in range(2,14): f=make_frame(i*100,i/10) s.evaluate(f,(make_pose(f),),(31,)) state=s.engine.inspect(31) self.assertFalse(state['latched']) self.assertFalse(down(s,1.4,31)[1]) self.assertTrue(down(s,2.,31)[1]) def test_artifact_body_uses_existing_contract(self): from app.contracts.frames import FrameTime from app.event_artifact_contract import build_body, relative_directory timing=FrameTime(anchor_wall_time_ms=1700000000000, source_rtp_timestamp=42, source_time_wall_ms=1700000000010, sync_status='ESTIMATED') f=replace(frame(0),frame_time=timing) for name, s, poses in ( ('FALL',configured_fall(wall_ns=lambda:1700000000011000000), (make_pose(f,angle=70),)), ('FIGHT',configured_fight(current_frame_fallback=True,wall_ns=lambda:1700000000011000000), (person(f,450),person(f,650)))): ids=tuple(range(1,len(poses)+1)) event=s.evaluate(f,poses,ids)[1][0].payload() relative=relative_directory(event) body=build_body(event,{'clip':True,'thumbnail':True},None,'https://example.invalid',relative) self.assertEqual(set(body),{'serverId','events'}) row=body['events'][0] self.assertEqual(set(row),{'eventId','cameraId','eventType','detectedAtWallMs','score', 'severityCode','detail','rtpTimestamp','startWallTimeMs','clipUrl','thumbnailUrl'}) self.assertEqual(row['eventType'],name) self.assertEqual(row['detail']['trackId'],event['trackId']) self.assertEqual(row['detail']['ptsNs'],0) self.assertFalse(row['detail']['scoreMeasured']) for key in ('cooldown','direct_down','current_frame_fallback'): self.assertNotIn(key,json.dumps(body)) # The demo never fabricates synchronized source time for delivery. event['frameTime']['detectedAtWallMs']=None self.assertIsNone(build_body(event,{},None,'https://example.invalid',relative)) def test_fight_observed_quiet_rearms_after_successful_event(self): s=configured_fight() pair(s,0);self.assertTrue(pair(s,.1,wrist=490)[1]) for i in range(2,36): self.assertFalse(pair(s,i/10)[1]) self.assertTrue(pair(s,3.6,wrist=490)[1]) if __name__ == '__main__': unittest.main()