176 lines
10 KiB
Python
176 lines
10 KiB
Python
"""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()
|