"""Synthetic stationary-camera/action simulations only; no native input backend.""" import copy import math from pathlib import Path import sys import unittest sys.path.insert(0,str(Path(__file__).resolve().parents[1]/'tools')) import prison_alignment_proof as policy class Scene: def __init__(self): self.now=31.;self.next_sample=31.1;self.angle=126.154 self.geometry={'nonce':'N','basis':'registered-prison-lock-native-interaction', 'prison':[6090.,211.444,7160.],'gate':[6090.,211.344,7154.], 'gate_forward':[0.,0.,1.],'context':[6092.77,213.226,7154.],'radius':2.} self.origin=[6092.42,211.297,7155.81];self.camera=[6092.42,212.921,7155.71] self.gain=.085;self.delay=.15;self.guard_cost=.08;self.pending=[];self.motions=[];self.notifications=[] self.select=True;self.bound=True;self.input=[0.,0.,0.];self.velocity=[0.,0.,0.] self.pressed=0.;self.progress=0.;self.performing=False;self.cancel=False;self.stale=False self.phase=1;self.door=0.;self.focus=True;self.ignore=False;self.reverse=False self.snapshot=self.make();self.history=[copy.deepcopy(self.snapshot)];self.last_read=309 self.prior=copy.deepcopy(self.snapshot) response=math.radians(24*.085) self.calibration={'nonce':'N','completed':True,'pixels':24,'baseline_direction':[0.,0.,1.], 'response_direction':[math.sin(response),0.,math.cos(response)], 'observer_start_real_ms':30000,'ready_real_ms':32000,'ready_monotonic':2., 'observer_deadline_monotonic':44.5,'alignment_deadline_monotonic':39.} self.report={'approach_proof':{'completed':True,'nonce':'N','geometry':copy.deepcopy(self.geometry), 'final_sample':310,'samples':[self.prior]}} self.accepted={'verified':True,'last_sample':310} def clock(self):return self.now def make(self): r=math.radians(self.angle) desired=math.degrees(math.atan2(self.geometry['context'][0]-self.camera[0],self.geometry['context'][2]-self.camera[2])) chosen=self.select and abs(policy.delta(desired,self.angle))<=2.5 return {'sample':int(round(self.now*10)),'real_ms':30000+int(round(self.now*1000)), 'received_monotonic':self.now-1 if self.stale else self.now, 'origin':list(self.origin),'camera':list(self.camera),'direction':[math.sin(r),0.,math.cos(r)], 'body_forward':[.807436,0.,-.589955],'input':list(self.input),'forward':0.,'velocity':list(self.velocity), 'inventory':False,'phase':self.phase,'door_state':self.door,'door_control':0.,'geometry':copy.deepcopy(self.geometry), 'selected':chosen,'context_selected':chosen,'cached_performable':chosen,'progress_bound':chosen and self.bound, 'performing':self.performing,'pressed':self.pressed,'progress':self.progress,'duration':4.} def sleep(self,seconds): end=self.now+seconds while self.now < end-1e-8: self.now+=min(.005,end-self.now) while self.pending and self.pending[0][0]<=self.now:self.angle+=self.pending.pop(0)[1] if self.now >= self.next_sample-1e-8: self.snapshot=self.make();self.history.append(copy.deepcopy(self.snapshot));self.next_sample+=.1 def check(self): if not self.focus:raise policy.Refused('focus lost') self.sleep(self.guard_cost) def idle(self):return None def move(self,dx,dy,guard,*,deadline): if dy or not 1<=abs(dx)<=128:raise AssertionError('Only bounded horizontal relative moves permitted') guard.check();guard.check() if self.now>=deadline:raise policy.Refused('adapter deadline') self.motions.append((dx,dy,self.now,deadline)) if not self.ignore:self.pending.append((self.now+self.delay,dx*self.gain*(-1 if self.reverse else 1))) def observe(self): if self.cancel:raise policy.Refused('native camera identity changed') batch=[s for s in self.history if s['sample']>self.last_read] or [self.snapshot] self.last_read=batch[-1]['sample'];return copy.deepcopy(batch) def run(self): return policy.perform_alignment(self,self.idle,self,self.observe,self.geometry,self.calibration,self.accepted,self.report, clock=self.clock,sleep=self.sleep,notify=self.notifications.append) class AlignmentTests(unittest.TestCase): def test_actual_pose_bearing_aligns_without_movement_or_interaction(self): scene=Scene();result=scene.run() self.assertTrue(result['completed']);self.assertLessEqual(len(scene.motions),6) self.assertGreater(len(scene.motions),0);self.assertLess(scene.now,39) self.assertEqual(scene.origin,[6092.42,211.297,7155.81]) a,b=result['selection_samples'];samples={s['sample']:s for s in result['samples']} self.assertEqual(b,a+1);self.assertGreaterEqual(samples[b]['real_ms']-samples[a]['real_ms'],100) self.assertTrue(all(samples[i]['progress_bound'] for i in (a,b))) self.assertTrue(all(m[1]==0 for m in scene.motions)) def test_delayed_turn_requires_real_rotation_before_settling(self): scene=Scene();scene.delay=.55 result=scene.run();self.assertTrue(result['completed']) self.assertTrue(all(m['response_sample']>m['after_sample'] for m in result['motions'])) def test_aligned_missing_selection_refuses_without_movement_fallback(self): scene=Scene();scene.angle=168.433;scene.select=False scene.snapshot=scene.make();scene.history=[copy.deepcopy(scene.snapshot)] with self.assertRaises(policy.Refused):scene.run() self.assertEqual(scene.motions,[]);self.assertLessEqual(scene.now,39.2) self.assertFalse(scene.report['alignment_proof']['completed']) def test_unbound_progress_is_not_zero_progress_proof(self): scene=Scene();scene.bound=False with self.assertRaises(policy.Refused):scene.run() self.assertFalse(scene.report['alignment_proof']['completed']) def test_original_observer_deadline_wins_and_is_not_renewed(self): scene=Scene();scene.calibration['observer_deadline_monotonic']=31.2 with self.assertRaises(policy.Refused):scene.run() self.assertEqual(scene.motions,[]) self.assertLessEqual(scene.report['alignment_proof']['deadline_monotonic'],31.2) scene=Scene();scene.calibration['alignment_deadline_monotonic']=31.2 with self.assertRaises(policy.Refused):scene.run() self.assertEqual(scene.motions,[]) def test_expired_original_native_samples_refuse_even_with_future_host_deadline(self): scene=Scene();scene.snapshot['real_ms']=75000;scene.history=[copy.deepcopy(scene.snapshot)] with self.assertRaisesRegex(policy.Refused,'lifetime'):scene.run() self.assertEqual(scene.motions,[]) def test_prior_approach_and_same_run_are_required(self): for fault in ('verified','sample','nonce'): scene=Scene() if fault=='verified':scene.accepted['verified']=False elif fault=='sample':scene.accepted['last_sample']=309 else:scene.calibration['nonce']='other' with self.subTest(fault=fault),self.assertRaises(policy.Refused):scene.run() self.assertEqual(scene.motions,[]) def test_movement_and_action_in_intermediate_batch_abort(self): for field,value in (('origin',[6092.6,211.297,7155.81]),('pressed',1.),('progress',.1),('performing',True),('input',[0,0,1])): scene=Scene();scene.sleep(.3) scene.history[1][field]=value with self.subTest(field=field),self.assertRaises(policy.Refused):scene.run() self.assertEqual(scene.motions,[]) def test_overshoot_reverse_and_missing_rotation_refuse(self): for fault in ('overshoot','reverse','ignore'): scene=Scene() if fault=='overshoot':scene.gain=.5 elif fault=='reverse':scene.reverse=True else:scene.ignore=True with self.subTest(fault=fault),self.assertRaises(policy.Refused):scene.run() self.assertEqual(len(scene.motions),1) def test_stale_camera_or_focus_loss_never_authorizes_input(self): for fault in ('stale','cancel','focus'): scene=Scene() if fault=='focus':scene.focus=False else:setattr(scene,fault,True);scene.sleep(.1) with self.subTest(fault=fault),self.assertRaises(policy.Refused):scene.run() self.assertEqual(scene.motions,[]) def test_measured_slow_guards_and_delayed_response_fit_without_renewal(self): scene=Scene();scene.guard_cost=.18;scene.delay=.55 result=scene.run() self.assertTrue(result['completed']);self.assertLessEqual(scene.now-31,8) self.assertEqual(result['deadline_monotonic'],39) self.assertEqual(len(scene.motions),5) def test_diagonal_body_drift_over_five_centimetres_refuses(self): scene=Scene();scene.sleep(.1) scene.history[-1]['origin'][0]+=.04;scene.history[-1]['origin'][1]+=.04 with self.assertRaisesRegex(policy.Refused,'body moved'):scene.run() self.assertEqual(scene.motions,[]) def test_transient_yaw_overshoot_in_intermediate_batch_refuses(self): scene=Scene() def move(dx,dy,guard,*,deadline): guard.check();scene.angle+=20;scene.sleep(.2);scene.angle-=10 scene.motions.append((dx,dy,scene.now,deadline)) scene.move=move with self.assertRaisesRegex(policy.Refused,'turn bound'):scene.run() self.assertEqual(len(scene.motions),1) def test_selected_but_wrong_bearing_is_not_success(self): scene=Scene();scene.ignore=True original=scene.observe def falsely_selected(): batch=original() for s in batch: s['selected']=s['context_selected']=s['cached_performable']=s['progress_bound']=True return batch scene.observe=falsely_selected with self.assertRaises(policy.Refused):scene.run() self.assertFalse(scene.report['alignment_proof']['completed']) def test_retry_same_report_refuses(self): scene=Scene();scene.report['alignment_proof']={} with self.assertRaisesRegex(policy.Refused,'already exists'):scene.run() self.assertEqual(scene.motions,[]) if __name__=='__main__':unittest.main()