aussie/EscapistsReintroduced
1
0
Fork 0
forked from escapists/EscapistsReintroduced
Code Pull Requests Activity

Observe native prison lock geometry and requalify Core loops

Browse Source
This commit is contained in:
jakub
2026-09-19 18:58:12 +02:00
parent e10b8799e9
commit b7fba094d7
39 changed files with 9884 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tools/ordinary_gate_evidence.py
+206
View File
@@ -0,0 +1,206 @@
"""Read-only native gate telemetry parsing and geometry qualification.
These functions do not send input, select actions or establish that a prisoner
approached/unlocked the gate. Raw INPUT/LOOK observations supply all joins; helper
geometry estimates are not accepted. Process/display/session provenance remains
an independent runner responsibility.
"""
import math
import re
from ordinary_look_evidence import _native, _vector, _distance, _norm
PREFIX = 'ERR GATE OBSERVE: '
NUMBER = r'-?(?:\d+(?:\.\d*)?|\.\d+)(?:[eE][+-]?\d+)?'
BOOL = r'(0|1|true|false)'
HEADER = re.compile(r'run=([A-Za-z0-9]{1,32}) sample=(\d+) real=(\d+) (.+)')
SAMPLE = re.compile(r'SAMPLE phase=(\d+) selected='+BOOL+r' context='+BOOL+
r' cachedPerformable='+BOOL+r' progressBound='+BOOL+r' performing='+BOOL+
r' progress=('+NUMBER+r') duration=('+NUMBER+r') pressed=('+NUMBER+r')')
GEOMETRY = re.compile(r'GEOMETRY body=(.+?) context=(.+?) radius=('+NUMBER+r') doorState=('+NUMBER+
r') doorControl=('+NUMBER+r') prison=(.+?) gate=(.+?) gateForward=(.+)')
CANCELLATIONS = {'explicit-scope-lost', 'original-prison-lock-unavailable', 'original-binding-changed',
'sample-or-clock-bound', 'native-player-or-prison-context-lost', 'invalid-native-action-sample'}
SCOPE = 'same-session native gate/input/camera telemetry and stable prison/lock geometry only; no approach, held-action completion, visual or multiplayer qualification'
def _require(condition, reason):
if not condition:
raise ValueError(reason)
def _nonce(nonce):
_require(type(nonce) is str and re.fullmatch(r'[A-Za-z0-9]{1,32}', nonce), 'invalid gate nonce')
def _finite(value, lower, upper, reason):
value = float(value)
_require(math.isfinite(value) and lower <= value <= upper, reason)
return value
def parse_gate_record(line, nonce):
"""Parse one native line. Unrelated -> None; invalid matching -> ValueError.
Parsing a line does not authenticate its sequence or native actor identity.
Use the full parser/join/validator before treating a snapshot as qualified.
"""
_nonce(nonce)
_require(type(line) is str, 'gate record is not text')
if PREFIX.rstrip() not in line:
return None
_require(PREFIX in line and line.count(PREFIX) == 1 and len(line) <= 4096, 'malformed gate prefix or line bound')
match = HEADER.fullmatch(line.split(PREFIX, 1)[1].strip())
_require(match is not None, 'malformed gate record')
actual, index, real, detail = match.groups()
_require(actual == nonce, 'foreign gate nonce')
index, real = int(index), int(real)
_require(1 <= index <= 450 and 0 <= real <= 2147483647, 'gate clock/sample outside bound')
record = {'sample': index, 'real_ms': real}
if detail == 'BASIS registered-prison-lock-native-interaction':
return {**record, 'kind': 'basis'}
if detail.startswith('CANCELLED '):
_require(detail[10:] in CANCELLATIONS, 'unknown gate cancellation')
return {**record, 'kind': 'cancelled', 'reason': detail[10:]}
match = SAMPLE.fullmatch(detail)
if match:
values = match.groups()
phase = int(values[0])
_require(phase in (1, 2), 'gate phase outside native observation contract')
flags = dict(zip(('selected', 'context_selected', 'cached_performable', 'progress_bound', 'performing'),
(v in ('1', 'true') for v in values[1:6])))
progress = _finite(values[6], 0, 60, 'invalid native action progress')
duration = _finite(values[7], 4, 4, 'native lock duration is not four seconds')
pressed = _finite(values[8], 0, 1, 'invalid native action input')
_require(not flags['cached_performable'] or flags['context_selected'], 'cached permission without original context')
_require(not flags['progress_bound'] or (flags['selected'] and flags['context_selected']), 'progress not bound to original action/context')
_require(flags['progress_bound'] or (not flags['performing'] and progress == 0), 'unbound native action progress')
return {**record, 'kind': 'sample', 'phase': phase, **flags,
'progress': progress, 'duration': duration, 'pressed': pressed}
match = GEOMETRY.fullmatch(detail)
if match:
body, context, radius, state, control, prison, gate, forward = match.groups()
forward = _vector(forward)
_require(.95 <= _norm(forward) <= 1.05 and math.hypot(forward[0], forward[2]) > .1,
'gate forward is not normalized or horizontal direction unusable')
radius = _finite(radius, 0, 10, 'native gate context radius outside bound')
_require(radius > 0, 'native gate context radius must be positive')
return {**record, 'kind': 'geometry', 'body': _vector(body), 'context': _vector(context), 'radius': radius,
'door_state': _finite(state, -1, 1, 'invalid native door state'),
'door_control': _finite(control, -1, 1, 'invalid native door control'),
'prison': _vector(prison), 'gate': _vector(gate), 'gate_forward': forward}
raise ValueError('unknown or malformed gate event')
def parse_gate_records(content, nonce):
"""Strict complete native gate sequence; returns records and paired snapshots.
Snapshot keys are integer sample indices. Incomplete final SAMPLE/GEOMETRY
pairs fail, so a live reader must buffer a partial pair before calling this.
"""
_nonce(nonce)
_require(type(content) is str and len(content) <= 32*1024*1024, 'gate console content outside bound')
records, snapshots = [], {}
basis = None; cancelled = None; pending = None
previous_index = previous_real = -1
for ordinal, line in enumerate(content.splitlines()):
record = parse_gate_record(line, nonce)
if record is None:
continue
_require(len(records) < 902, 'gate record count bound exceeded')
_require(cancelled is None, 'gate records after cancellation')
record['ordinal'] = ordinal
index, real, kind = record['sample'], record['real_ms'], record['kind']
_require(index >= previous_index and real >= previous_real, 'gate clocks or sample indices reversed')
previous_index, previous_real = index, real
if kind == 'basis':
_require(basis is None and not records and index == 1, 'duplicate or late native gate acknowledgement')
basis = record
elif kind == 'sample':
_require(basis is not None and pending is None and index == len(snapshots)+1, 'gate SAMPLE missing acknowledgement, duplicated or noncontiguous')
_require(index != 1 or real == basis['real_ms'], 'gate acknowledgement clock differs from first sample')
pending = record
elif kind == 'geometry':
_require(pending is not None and index == pending['sample'] and real == pending['real_ms'], 'gate GEOMETRY has no exact preceding SAMPLE')
snapshots[index] = {'sample': index, 'real_ms': real, 'action': pending, 'geometry': record}
pending = None
else:
_require(pending is None, 'gate cancellation interrupted a sample/geometry pair')
cancelled = record
records.append(record)
_require(basis is not None and snapshots and pending is None, 'missing native gate acknowledgement or complete geometry snapshots')
_require(snapshots[max(snapshots)]['real_ms']-basis['real_ms'] < 45000, 'gate observations exceed original lifetime')
return {'records': records, 'snapshots': snapshots, 'basis': basis, 'cancelled': cancelled}
def join_gate_snapshots(content, nonce):
"""Return raw gate snapshots joined with native input and camera samples.
Both native acknowledgements and INPUT READY are required. The full archived
console is expected, including the first samples, not a helper's metrics.
"""
gate = parse_gate_records(content, nonce)
poses, inputs, ready, look_cancelled = _native(content, nonce)
joined = {}
for index, snapshot in gate['snapshots'].items():
_require(index in inputs and index in poses, 'gate snapshot missing raw input or camera peer')
pose, sample = poses[index], inputs[index]
_require(snapshot['real_ms'] == pose['real_ms'] == sample['real_ms'], 'gate/input/camera timestamp mismatch')
# Gate and INPUT both read actor.GetOrigin. LOOK independently reads
# actor.GetTransform; the native18-44-37 walking capture differs by up
# to3cm at identical sample/clock. Keep that cross-API comparison bounded
# to5cm, while retaining the tighter same-API1cm agreement.
_require(_distance(snapshot['geometry']['body'], sample['origin']) <= .01 and
_distance(snapshot['geometry']['body'], pose['body']) <= .05, 'gate/input/camera body position mismatch')
joined[index] = {**snapshot, 'input': sample, 'look': pose}
return {**gate, 'snapshots': joined, 'ready': ready, 'look_cancelled': look_cancelled}
def validate_gate_geometry(content, nonce, start_sample=None, end_sample=None):
"""Qualify only telemetry geometry during an explicit inclusive sample span.
Defaults select READY through the last joined sample. Caller references are
indices into raw native records; no estimated position or metric is accepted.
"""
result = {'verified': False, 'errors': [], 'scope': SCOPE,
'approach_verified': False, 'held_action_verified': False}
try:
data = join_gate_snapshots(content, nonce)
snapshots = data['snapshots']; ready = data['ready']['sample']
if start_sample is None:
start_sample = ready
if end_sample is None:
end_sample = max(snapshots)
_require(type(start_sample) is int and type(end_sample) is int and
ready <= start_sample <= end_sample <= 450, 'invalid gate observation span')
_require(end_sample-start_sample >= 1, 'geometry stability requires at least two native samples')
_require(start_sample in snapshots and end_sample in snapshots, 'requested gate span is missing')
first = snapshots[start_sample]; initial = first['geometry']
for cancellation in (data['cancelled'], data['look_cancelled']):
_require(cancellation is None or cancellation['sample'] > end_sample, 'native context cancelled during requested gate span')
max_geometry_drift = 0.0
for index in range(start_sample, end_sample+1):
_require(index in snapshots, 'gate sample missing within requested span')
snapshot = snapshots[index]; geometry = snapshot['geometry']; action = snapshot['action']
_require(action['phase'] == 1, 'requested geometry span is not entirely PRISON phase')
_require(not snapshot['input']['inventory'], 'inventory menu open during requested gate span')
if index > start_sample:
_require(0 < snapshot['real_ms']-snapshots[index-1]['real_ms'] <= 500, 'gate sampling gap or nonadvancing clock')
for field in ('prison', 'gate', 'context'):
drift = _distance(geometry[field], initial[field]); max_geometry_drift = max(max_geometry_drift, drift)
_require(drift <= .01, 'native '+field+' geometry changed during requested span')
_require(_distance(geometry['gate_forward'], initial['gate_forward']) <= .001, 'native gate orientation changed')
_require(abs(geometry['radius']-initial['radius']) <= .001, 'native action context radius changed')
_require(abs(geometry['door_state']-initial['door_state']) <= .001 and
abs(geometry['door_control']-initial['door_control']) <= .001, 'native door state/control changed')
last = snapshots[end_sample]
result.update(verified=True, start_sample=start_sample, end_sample=end_sample,
native_duration_ms=last['real_ms']-first['real_ms'], snapshot_count=end_sample-start_sample+1,
geometry_max_drift_metres=max_geometry_drift, prison=list(initial['prison']),
gate=list(initial['gate']), context=list(initial['context']), radius=initial['radius'],
gate_forward=list(initial['gate_forward']),
initial_context_distance_metres=_distance(first['geometry']['body'], initial['context']),
final_context_distance_metres=_distance(last['geometry']['body'], initial['context']))
except (ValueError, TypeError, KeyError, OverflowError) as error:
result['errors'].append(str(error))
return result