207 lines
12 KiB
Python
207 lines
12 KiB
Python
"""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
|