237 lines
14 KiB
Python
237 lines
14 KiB
Python
"""Opt-in stationary yaw alignment policy, with no input backend or keyboard send.
|
|
|
|
perform_alignment consumes the same validated ordered native snapshot batches as
|
|
prison_approach_proof plus camera and flat native selected/context_selected/
|
|
cached_performable/progress_bound/performing/pressed/progress/duration fields.
|
|
The adapter pins the original console, body, camera and gate identities and
|
|
rejects cancellation. No field here permits native identity re-adoption.
|
|
|
|
require_idle() is read-only. mouse.move(dx,0,guard,deadline=absolute) is an explicit
|
|
alignment-only capability with existing ownership/focus/current+projected-pointer
|
|
checks and a final absolute deadline check before its relative input call.
|
|
notify({kind:'alignment-pending',deadline_monotonic:...}) requires an independent
|
|
hard supervisor; alignment-finished is only a policy result, not final run proof.
|
|
No F, W, wheel, pitch, focus, camera write or native connection exists here.
|
|
"""
|
|
import copy
|
|
import math
|
|
|
|
from prison_approach_proof import vector, number, yaw, delta, length, horizontal, target_for
|
|
|
|
|
|
class Refused(ValueError):
|
|
pass
|
|
|
|
|
|
def need(value, reason):
|
|
if not value:
|
|
raise Refused(reason)
|
|
|
|
|
|
def perform_alignment(guard, require_idle, mouse, observe, geometry, calibration, approach_evidence, report, *, clock, sleep, notify):
|
|
return _perform_alignment(guard, require_idle, mouse, observe, geometry, calibration, approach_evidence, report, clock=clock, sleep=sleep, notify=notify)
|
|
|
|
|
|
def perform_facing(guard, require_idle, mouse, observe, geometry, calibration, approach_evidence, report, *, clock, sleep, notify, profile="standard"):
|
|
"""Separate facing-only outcome; requires exact disabled-door/lock diagnostics."""
|
|
return _perform_alignment(guard, require_idle, mouse, observe, geometry, calibration, approach_evidence, report, clock=clock, sleep=sleep, notify=notify, facing=True, profile=profile)
|
|
|
|
|
|
def _perform_alignment(guard, require_idle, mouse, observe, geometry, calibration, approach_evidence, report, *, clock, sleep, notify, facing=False, profile="standard"):
|
|
need(type(profile) is str and profile in ('standard', 'coarse') and (facing or profile == 'standard'), 'Unknown or non-facing coarse profile')
|
|
coarse = profile == 'coarse'
|
|
need(not coarse or (report.get('coarse_facing_requested') is True and report.get('coarse_approach_requested') is True and report.get('selection_requested') is True), 'Coarse facing requires explicit matching capabilities')
|
|
max_pixels, desired_angle, max_angle = (384, 25., 35.) if coarse else (128, 10., 15.)
|
|
proof_key = 'facing_proof' if facing else 'alignment_proof'
|
|
event_prefix = 'facing' if facing else 'alignment'
|
|
need(proof_key not in report, 'Alignment attempt already exists')
|
|
pinned = copy.deepcopy(geometry)
|
|
target_for(pinned) # Existing measured authored-enclosure compatibility gate.
|
|
approach = report.get('approach_proof', {})
|
|
need(approach_evidence.get('verified') is True and approach.get('completed') is True and
|
|
approach.get('nonce') == pinned['nonce'] and approach.get('geometry') == pinned and
|
|
approach_evidence.get('last_sample') == approach.get('final_sample'),
|
|
'Alignment requires independently completed same-run approach')
|
|
need(not coarse or approach.get('profile') == 'coarse', 'Coarse facing requires the accepted coarse approach')
|
|
prior = approach.get('samples', [])
|
|
need(prior and prior[-1]['sample'] == approach['final_sample'], 'Original approach final body sample unavailable')
|
|
body = vector(prior[-1]['origin'])
|
|
need(calibration.get('completed') is True and calibration.get('nonce') == pinned['nonce'] and calibration.get('pixels') == 24,
|
|
'Alignment requires accepted same-run yaw calibration')
|
|
gain = delta(yaw(calibration['response_direction']), yaw(calibration['baseline_direction']))/24
|
|
need(.01 <= abs(gain) <= .5, 'Native yaw calibration outside bounded range')
|
|
queued = number(calibration['observer_start_real_ms']); ready = number(calibration['ready_real_ms'])
|
|
ready_at = number(calibration['ready_monotonic']); started = clock()
|
|
need(0 <= queued and 2000 <= ready-queued < 45000 and 0 <= ready_at <= started, 'Original observation clocks invalid')
|
|
original_deadline = ready_at+(45000-(ready-queued))/1000-.5
|
|
supplied = number(calibration.get('observer_deadline_monotonic', original_deadline))
|
|
need(supplied <= original_deadline, 'Supplied observer deadline renews original lifetime')
|
|
stage_supplied = number(calibration.get('alignment_deadline_monotonic', started+8))
|
|
need(stage_supplied <= started+8, 'Supplied alignment deadline exceeds stage cap')
|
|
deadline = min(started+8, original_deadline, supplied, stage_supplied)
|
|
need(started < deadline, 'No original observation time remains for alignment')
|
|
state = {'requested': True, 'completed': False, 'nonce': pinned['nonce'], 'geometry': pinned,
|
|
'started_monotonic': started, 'finished_monotonic': None, 'deadline_monotonic': deadline,
|
|
'initial_sample': None, 'final_sample': None, 'selection_samples': [], 'motions': [], 'samples': []}
|
|
report[proof_key] = state
|
|
if facing:
|
|
if coarse: state['profile'] = profile
|
|
state['facing_samples'] = []
|
|
del state['selection_samples']
|
|
previous = None; active_turn = None; requests = 0; pixels = 0
|
|
|
|
def neutral(s):
|
|
return abs(s['forward']) <= .01 and length(s['input']) <= .01 and length(s['velocity']) <= .05
|
|
|
|
def bearing(s):
|
|
camera = s['camera']; context = pinned['context']
|
|
need(.25 <= horizontal(camera, context) <= 5, 'Native camera/handle bearing is degenerate or out of scope')
|
|
return math.degrees(math.atan2(context[0]-camera[0], context[2]-camera[2]))
|
|
|
|
def consume(s):
|
|
nonlocal previous
|
|
s = copy.deepcopy(s)
|
|
index = s['sample']; real = number(s['real_ms']); receipt = number(s['received_monotonic'])
|
|
need(type(index) is int and approach['final_sample'] <= index <= 450, 'Alignment sample precedes approach or exceeds bound')
|
|
need(0 <= receipt <= clock() and queued <= real < queued+45000, 'Original native observation lifetime exceeded')
|
|
need(s['geometry'] == pinned and s['phase'] == 1 and s['inventory'] is False,
|
|
'Original prison geometry, phase or UI changed')
|
|
need(number(s['door_state']) == 0 and number(s['door_control']) == 0, 'Original gate was operated')
|
|
for k in ('origin', 'camera', 'direction', 'body_forward', 'input', 'velocity'):
|
|
s[k] = vector(s[k])
|
|
s['forward'] = number(s['forward']); yaw(s['direction']); yaw(s['body_forward'])
|
|
need(neutral(s) and math.dist(body, s['origin']) <= .05,
|
|
'Original stopped body moved during alignment')
|
|
for k in ('selected', 'context_selected', 'cached_performable', 'progress_bound', 'performing'):
|
|
need(type(s[k]) is bool, 'Malformed native action flag')
|
|
need(number(s['pressed']) == 0 and number(s['progress']) == 0 and s['performing'] is False and number(s['duration']) == 4,
|
|
'Gate interaction began during yaw-only alignment')
|
|
bearing(s)
|
|
if facing and s['context_selected']:
|
|
from prison_action_selection_proof import choices_for
|
|
actions = choices_for(s, pinned)
|
|
if 'actions' not in state: state['actions'] = actions
|
|
need(state['actions'] == actions, 'Original facing action identities/order changed')
|
|
if active_turn is not None:
|
|
before_yaw, sign = active_turn
|
|
change = delta(yaw(s['direction']), before_yaw)
|
|
need(abs(change) <= max_angle, 'Native yaw exceeded alignment turn bound')
|
|
if abs(change) >= .05:
|
|
need(change*sign > 0, 'Native yaw reversed alignment request')
|
|
if previous:
|
|
need(index >= previous['sample'] and real >= previous['real_ms'], 'Alignment native samples reversed')
|
|
if index == previous['sample']:
|
|
need(s == previous, 'Repeated alignment sample changed')
|
|
else:
|
|
need(index == previous['sample']+1 and 0 < real-previous['real_ms'] <= 500, 'Alignment native sample gap')
|
|
if previous is None or index != previous['sample']:
|
|
state['samples'].append(copy.deepcopy(s))
|
|
previous = s
|
|
if state['initial_sample'] is None:
|
|
state['initial_sample'] = index
|
|
state['final_sample'] = index
|
|
return s
|
|
|
|
def read():
|
|
need(clock() < deadline, 'Frozen alignment deadline exceeded')
|
|
guard.check(); require_idle()
|
|
batch = observe()
|
|
need(type(batch) is list and 1 <= len(batch) <= 450, 'Missing ordered alignment snapshot batch')
|
|
for raw in batch:
|
|
s = consume(raw)
|
|
need(0 <= clock()-s['received_monotonic'] <= .5, 'Latest native alignment observation is stale')
|
|
need(clock() < deadline, 'Frozen alignment deadline exceeded during observation')
|
|
return s
|
|
|
|
def stable(first, second):
|
|
return (second['sample'] == first['sample']+1 and second['real_ms']-first['real_ms'] >= 100 and
|
|
abs(delta(yaw(second['direction']), yaw(first['direction']))) <= .2)
|
|
|
|
def selected(s):
|
|
if facing:
|
|
if not s['context_selected'] or not s['cached_performable'] or abs(delta(bearing(s), yaw(s['direction']))) > 3:
|
|
return False
|
|
from prison_action_selection_proof import choices_for
|
|
choices_for(s, pinned)
|
|
return True
|
|
return s['selected'] and s['context_selected'] and s['cached_performable'] and s['progress_bound'] and abs(delta(bearing(s), yaw(s['direction']))) <= 3
|
|
|
|
def selection_pair():
|
|
samples = state['samples']
|
|
if len(samples) >= 2 and stable(samples[-2], samples[-1]) and selected(samples[-2]) and selected(samples[-1]):
|
|
return [samples[-2]['sample'], samples[-1]['sample']]
|
|
return []
|
|
|
|
try:
|
|
notify({'kind': event_prefix+'-pending', 'deadline_monotonic': deadline})
|
|
initial = read()
|
|
# First receipt cannot extend the frozen original READY-derived lease.
|
|
# Deduct the explicit 500ms observation/dispatch reserve, not just age.
|
|
deadline = min(deadline, initial['received_monotonic']+(queued+45000-initial['real_ms'])/1000-.5)
|
|
state['deadline_monotonic'] = deadline
|
|
need(clock() < deadline, 'Fresh native sample leaves no alignment time')
|
|
notify({'kind': event_prefix+'-pending', 'deadline_monotonic': deadline})
|
|
current = initial
|
|
while clock() < deadline:
|
|
pair = selection_pair()
|
|
if pair:
|
|
if not facing:
|
|
current = read()
|
|
pair = selection_pair()
|
|
if not pair:
|
|
continue
|
|
# Facing already has a fully consumed, guarded stable pair.
|
|
# Return its exact span; selection still consumes all later
|
|
# history and rechecks ownership before any wheel request.
|
|
require_idle()
|
|
completed_at = clock()
|
|
if facing:
|
|
need(started <= completed_at and 0 <= completed_at-state['samples'][-1]['received_monotonic'] <= .5,
|
|
'Final native alignment observation is stale or clock reversed')
|
|
need(completed_at < deadline, 'Frozen alignment deadline exceeded at completion')
|
|
state['facing_samples' if facing else 'selection_samples'] = pair; state['completed'] = True
|
|
return state
|
|
error = delta(bearing(current), yaw(current['direction']))
|
|
if abs(error) <= 3:
|
|
# Reached measured bearing; allow bounded native selection update.
|
|
# Never substitute additional movement, pitch or interaction.
|
|
sleep(.025); current = read(); continue
|
|
need(requests < 6, 'Alignment relative-request budget exhausted')
|
|
need(len(state['samples']) >= 2 or clock()+.1 < deadline, 'Insufficient time for stable initial pose')
|
|
if len(state['samples']) < 2 or not stable(state['samples'][-2], state['samples'][-1]):
|
|
sleep(.025); current = read(); continue
|
|
before = read()
|
|
# Always propose from the final fresh camera/handle bearing.
|
|
error = delta(bearing(before), yaw(before['direction']))
|
|
if abs(error) <= 3:
|
|
current = before; continue
|
|
desired = max(-desired_angle, min(desired_angle, .75*error))
|
|
dx = max(-max_pixels, min(max_pixels, int(round(desired/gain))))
|
|
if dx == 0: dx = 1 if desired/gain > 0 else -1
|
|
need(pixels+abs(dx) <= 768, 'Alignment cumulative pixel budget exhausted')
|
|
pixels += abs(dx) # Retire the attempted input before any uncertain native write.
|
|
active_turn = (yaw(before['direction']), dx*gain)
|
|
mouse.move(dx, 0, guard, deadline=deadline)
|
|
requests += 1
|
|
motion = {'dx': dx, 'dy': 0, 'monotonic': clock(), 'after_sample': before['sample'], 'response_sample': None}
|
|
state['motions'].append(motion)
|
|
end = min(deadline, clock()+1.5)
|
|
while clock() < end:
|
|
current = read()
|
|
change = delta(yaw(current['direction']), active_turn[0])
|
|
samples = state['samples']
|
|
if abs(change) >= .05 and len(samples) >= 2 and stable(samples[-2], samples[-1]):
|
|
need(clock() <= end, 'Native yaw response exceeded frozen turn interval')
|
|
motion['response_sample'] = current['sample']; active_turn = None
|
|
break
|
|
sleep(.025)
|
|
else:
|
|
raise Refused('Signed native yaw response and settling missing')
|
|
raise Refused('Exact lock selection missing before frozen alignment deadline')
|
|
except Exception as error:
|
|
state['error'] = str(error)
|
|
raise
|
|
finally:
|
|
state['finished_monotonic'] = clock()
|
|
notify({'kind': event_prefix+'-finished', 'successful': state['completed']})
|