"""FALL-WIRE-01: no GPU, external requests or persistent runtime writes.""" from copy import deepcopy from dataclasses import replace import json from pathlib import Path import threading import time from types import SimpleNamespace import unittest from unittest.mock import Mock, patch from app.fall_runtime import FallSettings, FallScenario from app.pose_config import PoseConfig, POSE_MODEL, execution_plan, planned_profile from app.region_scenario import compose from app.secure_control import ControlService, ControlError, Start from app.stream_settings import patch_settings, revision from app.frame_ingress import FrameIngress from app.det_result_path import DetResultPath from app.metadata_samples import MetadataSamples from app.event_artifact_contract import build_body, relative_directory, media_path from control_sidecar.main import validate_event from fall_fixtures import make_frame, make_pose from test_loitering_settings import FakeManager, spec, request from test_loitering_connection import CopiedDetection, Sink from loitering_contract_helpers import frame as anchored_frame class SettingsTests(unittest.TestCase): def test_web_selects_pose_once_and_returns_to_medium(self): template = json.loads(Path('configs/control_templates/medium.json').read_text()) m = FakeManager() s = ControlService(m, 'server', template).start_camera(Start(json.dumps(request('FALL','INTRUSION','LOITERING')))) self.assertEqual(planned_profile(s)['roles'], ['pose']) self.assertEqual(execution_plan(s)['model'], POSE_MODEL) self.assertEqual(s['pose'], template['pose']) self.assertEqual(s['fall']['person_score'], .25) self.assertTrue(s['tracking']['enabled']) self.assertFalse(s['secure_control_projection']['skippedScenarios']) m = FakeManager(s); service = ControlService(m,'server',template) self.assertEqual(s,service.start_camera(Start(json.dumps(request('FALL','INTRUSION','LOITERING'))))) off = service.start_camera(Start(json.dumps(request('LOITERING')))) self.assertEqual(execution_plan(off)['model'], 'RFDETRMedium') self.assertEqual(off['pose'], template['pose']) def test_explicit_pose_display_survives_fall_off(self): s = dict(spec(), pose={'show_joints':True}, fall={'enabled':True}) s = ControlService(FakeManager(s),'server',spec()).start_camera(Start(json.dumps(request()))) self.assertTrue(s['pose']['show_joints']) self.assertEqual(execution_plan(s)['model'], POSE_MODEL) def test_fire_combination_no_save(self): m=FakeManager(spec()); before=deepcopy(m.group.spec) with self.assertRaisesRegex(ControlError,'NEEDS_MULTI_PROVIDER_RUNTIME_CONFIRMATION'): ControlService(m,'server',spec()).start_camera(Start(json.dumps(request('FALL','FIRE')))) self.assertFalse(m.applied);self.assertEqual(before,m.group.spec) def test_roi_empty_aliases_and_independent_confidence(self): values=[] for mode in ('missing','null','empty'): d=request('FALL','LOITERING') for row in d['scenarios']: row['confidence']=.35 if row['scenarioCode']=='FALL' else .8 if mode=='missing':row.pop('rois') elif mode=='null':row['rois']=None values.append(ControlService(FakeManager(),'server',spec()).start_camera(Start(json.dumps(d)))) self.assertEqual(values[0],values[1]);self.assertEqual(values[1],values[2]) self.assertEqual(values[0]['fall']['person_score'],.35) self.assertEqual(values[0]['loitering']['confidence'],.8) def test_patch_noop_timers_and_tracking_guard(self): s=dict(spec(),fall={'enabled':True,'down_hold_seconds':.9}) n=patch_settings(s,{'fall':{'person_score':.4}}) self.assertEqual(n['fall']['down_hold_seconds'],.9) self.assertEqual(revision(n),revision(patch_settings(n,{'fall':{'person_score':.4}}))) with self.assertRaises(ValueError):patch_settings(n,{'tracking':{'enabled':False}}) for roi in (False,0,{},'bad'): with self.subTest(roi=roi),self.assertRaises(ValueError):FallSettings.parse({'roi':roi}) class AdapterTests(unittest.TestCase): def test_outside_roi_cannot_arm_or_emit(self): a=FallScenario(FallSettings.parse({'enabled':True,'roi':[[0,0],[.2,0],[.2,1],[0,1]]})) for i in range(15): f=make_frame(i,i/10);p=make_pose(f,**({} if i<7 else dict(angle=80,hip=(500,520),span=200))) d,e=a.evaluate(f,(p,),(1,)) self.assertEqual(d[0].state,'OUTSIDE');self.assertFalse(e) self.assertEqual(a.engine.status()['states'],0) def test_event_identity_and_episode_uuid_once(self): factory=Mock(return_value='fall-fixture-1') a=FallScenario(FallSettings.parse({'enabled':True}),event_id_factory=factory,wall_ns=lambda:1780000000000000000) events=[] for i in range(18): f=make_frame(i,i/10,frame_time=anchored_frame(i,i/10).frame_time) p=make_pose(f,**({} if i<7 else dict(angle=80,hip=(500,520),span=200))) d,e=a.evaluate(f,(p,),(1,));events.extend(e) factory.assert_called_once();self.assertEqual(len(events),1) event=events[0].payload();event['cameraId']=77;event['scenarioConfig']={'FALL':a.config.payload()} self.assertEqual(event['frameSeq'],14);self.assertEqual(event['objectRef'],'0') self.assertEqual(event['trackId'],1);self.assertEqual(event['modelKey'],POSE_MODEL) relative=relative_directory(event) body=build_body(event,{'clip':True,'thumbnail':True},None,'https://fixture.invalid/artifacts',relative) validate_event(body,'test-server') self.assertIn('_FALL_',relative) self.assertEqual(d[0].event_id,'fall-fixture-1') a.clear();self.assertFalse(a._events);self.assertEqual(a.engine.status()['states'],0) class GapTests(unittest.TestCase): def test_selected_ordinal_not_decoded_sequence(self): i=FrameIngress('test-server','77','test-stream','run-a',1) i.track_result_gaps=True i._record_selection(0,True,'selected') i._record_selection(1,False,'sampling_drop') i._record_selection(3,True,'selected') self.assertFalse(i.result_gap(make_frame(0,0))) self.assertFalse(i.result_gap(make_frame(3,.1))) i._record_selection(6,True,'selected');i._record_selection(9,True,'selected') self.assertTrue(i.result_gap(make_frame(9,.3))) i._record_selection(10,False,'invalid_pts');i._record_selection(12,True,'selected') self.assertTrue(i.result_gap(make_frame(12,.4))) class WorkerTests(unittest.TestCase): def make(self,capacity=32,current=lambda:True,capture=None): s=dict(spec(),pose={'enabled':True},intrusion={'enabled':True}, loitering={'enabled':True,'dwell_seconds':.5}) engine=FallScenario(FallSettings.parse({'enabled':True})) tracker=Mock(update=Mock(side_effect=lambda rows:[1]*len(rows))) ingress=SimpleNamespace(identity=('test-server','77','test-stream','run-a',1)) errors=Mock();sink=Sink() with patch.object(MetadataSamples,'from_environment',return_value=None): p=DetResultPath(ingress,tracker,POSE_MODEL,.5,Mock(add=Mock(return_value=1)),current,errors, pose_config=PoseConfig(enabled=True),intrusion=compose(s),fall=engine,publisher=sink, capacity=capacity) p.artifacts=capture self.addCleanup(lambda:p.close(2)) return p,tracker,sink,errors def copied(self,i,t,down=False): f=make_frame(i,t) p=make_pose(f,**(dict(angle=80,hip=(500,520),span=200) if down else {})) rows=(CopiedDetection(score=.3,bbox_xyxy=p.bbox_xyxy,keypoints=p.keypoints), CopiedDetection(score=.9,bbox_xyxy=p.bbox_xyxy,keypoints=p.keypoints)) return f,SimpleNamespace(width=1000,height=1000,frame_number=i,pts=f.pts_ns,detections=rows) def feed(self,p,i,t,down=False): n=p.status().get('processed',0);p.submit(*self.copied(i,t,down)) end=time.monotonic()+2 while p.status().get('processed',0)==n and not p.status()['error'] and time.monotonic()=7) self.assertEqual(t.update.call_count,15) self.assertTrue(all(len(call.args[0])==1 for call in t.update.call_args_list)) snap=r.snapshot();self.assertEqual(set(snap['scenarioConfig']),{'FALL','LOITERING','INTRUSION'}) event=next(x for x in snap['events'] if x['scenario']=='FALL') self.assertEqual(event['objectRef'],'1');self.assertEqual(event['bbox'],snap['objects'][0]['bbox']) self.assertEqual(len(snap['objects'][0]['scenarioResults']),3) self.assertEqual(len(snap['objects'][0]['eventIds']),3) self.assertEqual(snap['objects'][0]['color'],'#FF3030') self.assertEqual(s.rows[-1],snap);e.assert_not_called() def test_queue_replacement_marks_next_frame(self): p,_,_,_=self.make(capacity=1) p.submit(*self.copied(0,0));p.submit(*self.copied(3,.1)) self.assertTrue(p._queue[0][-1]);self.assertEqual(p.status()['queue_drop'],1) def test_stop_after_admission_preserves_handoff(self): active=[True];capture=Mock(submit=Mock(return_value=True),close=Mock(return_value=True)) p,_,_,_=self.make(current=lambda:active[0],capture=capture);p.start() for i in range(14):self.feed(p,i,i/10,i>=7) entered=threading.Event();release=threading.Event();original=p.fall.evaluate def evaluate(*args,**kw): entered.set();release.wait(2);return original(*args,**kw) p.fall.evaluate=evaluate p.submit(*self.copied(14,1.4,True));self.assertTrue(entered.wait(1)) active[0]=False closed=threading.Thread(target=lambda:p.close(3));closed.start() release.set();closed.join(3);self.assertFalse(closed.is_alive()) self.assertTrue(any(any(e.payload()['scenario']=='FALL' for e in call.args[0].events) for call in capture.submit.call_args_list)) self.assertEqual(p.fall.engine.status()['states'],0) self.assertFalse(p.submit(*self.copied(15,1.5,True))) if __name__ == '__main__':unittest.main()