jakub
6a667d8deb
Include the current addon, terrain integrations, tools, regression tests and recorded development evidence.
242 lines
14 KiB
Python
242 lines
14 KiB
Python
#!/usr/bin/env python3
|
|
"""Replay delivered pursuit telemetry or compare matched dedicated-server runs.
|
|
|
|
Journal arithmetic describes reports, not hidden player locations or encounter
|
|
quality. Performance arithmetic never grants gameplay or release acceptance.
|
|
"""
|
|
import argparse
|
|
from collections import Counter, defaultdict
|
|
import copy
|
|
import csv
|
|
import hashlib
|
|
import json
|
|
import math
|
|
from pathlib import Path
|
|
import re
|
|
import statistics
|
|
|
|
import compare_performance as performance
|
|
|
|
STATES = ('QUEUED', 'RECEIVED', 'TRAVELLING', 'ARRIVED', 'ACTING', 'COMPLETED',
|
|
'BLOCKED', 'FAILED', 'SUPERSEDED', 'CONTACT_CHANGED')
|
|
INTEGER_FIELDS = ('op', 'revision', 'execution', 'unit', 'role', 'belief')
|
|
KINDS = ('OPERATION ORDER QUEUED', 'EXECUTION DELIVERED', 'OPERATION EXPIRED', 'EXECUTION NOT APPLIED')
|
|
HEADER = re.compile(r'^#\d+ t=([0-9.]+)s M\d+ (OPERATION ORDER QUEUED|EXECUTION DELIVERED|OPERATION EXPIRED|EXECUTION NOT APPLIED) (.*?) \[session=([^\]\s]+)\]\s*$')
|
|
FIELD = re.compile(r'(?<!\S)([a-z_]+)=([^\s;]+)')
|
|
|
|
|
|
def numeric(values, name, integer=False):
|
|
value = int(values[name]) if integer else float(values[name])
|
|
if value < 0 or not math.isfinite(value):
|
|
raise ValueError(f'invalid {name}')
|
|
return value
|
|
|
|
|
|
def read_events(paths):
|
|
seen = {}
|
|
for path in map(Path, paths):
|
|
for number, line in enumerate(path.read_text(encoding='utf-8', errors='replace').splitlines(), 1):
|
|
if not line.startswith('#') or not any(kind in line for kind in KINDS):
|
|
continue
|
|
try:
|
|
match = HEADER.match(line)
|
|
if not match:
|
|
raise ValueError('invalid journal header')
|
|
clock, kind, payload, session = match.groups()
|
|
fields = FIELD.findall(payload)
|
|
values = dict(fields)
|
|
if len(fields) != len(values):
|
|
raise ValueError('duplicate telemetry field')
|
|
required = INTEGER_FIELDS if kind in KINDS[:2] else INTEGER_FIELDS[:4]
|
|
item = {name: numeric(values, name, True) for name in required}
|
|
if any(item[name] < 1 for name in required if name != 'role') or item.get('role', 0) > 6:
|
|
raise ValueError('invalid identity or role')
|
|
item.update(kind=kind, session=session, journal_seconds=float(clock))
|
|
if kind == KINDS[0]:
|
|
item['queued_ms'] = numeric(values, 'queued_ms')
|
|
key = session, item['execution'], kind
|
|
elif kind == KINDS[2]:
|
|
item['deadline_ms'] = numeric(values, 'deadline_ms')
|
|
item['released_ms'] = numeric(values, 'released_ms')
|
|
if values.get('outcome') != 'unconfirmed' or item['released_ms'] < item['deadline_ms']:
|
|
raise ValueError('invalid unconfirmed-expiry record')
|
|
key = session, item['execution'], kind
|
|
else:
|
|
item['sequence'] = numeric(values, 'sequence', True)
|
|
raw_state = values['state']
|
|
state = STATES[int(raw_state)] if raw_state.isdecimal() else raw_state
|
|
if state not in STATES or item['sequence'] < 1:
|
|
raise ValueError('invalid execution state/sequence')
|
|
item['state'] = state
|
|
measured = ('observed_ms', 'received_ms', 'remaining', 'searched', 'support_until_ms') if kind == KINDS[1] else ('observed_ms', 'received_ms')
|
|
for name in measured:
|
|
item[name] = numeric(values, name)
|
|
if kind == KINDS[1]:
|
|
for name in ('contact', 'support'):
|
|
item[name] = numeric(values, name, True)
|
|
if item[name] not in (0, 1):
|
|
raise ValueError(f'invalid {name}')
|
|
if 'arrived' in values:
|
|
item['arrived'] = numeric(values, 'arrived', True)
|
|
if item['arrived'] not in (0, 1):
|
|
raise ValueError('invalid physical arrival flag')
|
|
if item['received_ms'] < item['observed_ms'] or item['searched'] > 10:
|
|
raise ValueError('impossible delivery clock/search footprint')
|
|
if item['support'] and not item['observed_ms'] < item['support_until_ms'] <= item['observed_ms'] + 20000:
|
|
raise ValueError('invalid finite support lease')
|
|
key = session, item['execution'], kind, item['sequence']
|
|
if key in seen and seen[key] != item:
|
|
raise ValueError('conflicting duplicate execution event')
|
|
seen[key] = item
|
|
except (KeyError, ValueError, IndexError) as error:
|
|
raise ValueError(f'{path}:{number}: {error}') from error
|
|
return sorted(seen.values(), key=lambda row: (row['session'], row.get('received_ms', row.get('queued_ms', row.get('released_ms'))), row['execution']))
|
|
|
|
|
|
def summary(paths, session=None):
|
|
paths = list(map(Path, paths))
|
|
events = [row for row in read_events(paths) if session is None or row['session'] == session]
|
|
if not events:
|
|
raise ValueError('no matching coordinated pursuit telemetry; current-doctrine journals do not contain these events')
|
|
sessions = []
|
|
for session_id in sorted({row['session'] for row in events}):
|
|
executions = defaultdict(list)
|
|
for event in events:
|
|
if event['session'] == session_id:
|
|
executions[event['execution']].append(event)
|
|
outcomes = Counter()
|
|
physical_arrivals = blocked = searched = supported = contacts = 0
|
|
latencies = []
|
|
arrival_delays = []
|
|
orphan_reports = 0
|
|
unconfirmed_expiries = unapplied_packets = 0
|
|
for rows in executions.values():
|
|
queued = next((row for row in rows if row['kind'] == KINDS[0]), None)
|
|
reports = sorted((row for row in rows if row['kind'] == KINDS[1]), key=lambda row: row['sequence'])
|
|
unconfirmed_expiries += any(row['kind'] == KINDS[2] for row in rows)
|
|
unapplied_packets += sum(row['kind'] == KINDS[3] for row in rows)
|
|
if queued:
|
|
for report in reports:
|
|
if any(report[key] != queued[key] for key in INTEGER_FIELDS):
|
|
raise ValueError('execution report lineage differs from its queued order')
|
|
if report['observed_ms'] < queued['queued_ms']:
|
|
raise ValueError('execution report predates its queued order')
|
|
elif reports:
|
|
orphan_reports += 1 # retained rotation can start after the original order
|
|
if reports and any(a['observed_ms'] > b['observed_ms'] or a['received_ms'] > b['received_ms'] for a, b in zip(reports, reports[1:])):
|
|
raise ValueError('execution report sequence reverses observation or delivery time')
|
|
outcomes[reports[-1]['state'] if reports else 'UNREPORTED'] += 1
|
|
arrivals = [row for row in reports if row.get('arrived', row['state'] in ('ARRIVED', 'ACTING', 'COMPLETED'))]
|
|
if arrivals:
|
|
physical_arrivals += 1
|
|
if queued:
|
|
arrival_delays.append((arrivals[0]['observed_ms'] - queued['queued_ms']) / 1000)
|
|
blocked += any(row['state'] == 'BLOCKED' for row in reports)
|
|
searched += any(row['searched'] > 0 for row in reports)
|
|
supported += any(row['support'] for row in reports)
|
|
contacts += any(row['contact'] for row in reports)
|
|
latencies.extend((row['received_ms'] - row['observed_ms']) / 1000 for row in reports)
|
|
total = len(executions)
|
|
sessions.append({'session': session_id, 'executions': total,
|
|
'last_reported_states': dict(sorted(outcomes.items())),
|
|
'physical_arrivals_reported': physical_arrivals,
|
|
'physical_arrival_fraction': physical_arrivals / total,
|
|
'blocked_at_least_once': blocked,
|
|
'failed_fraction': outcomes['FAILED'] / total,
|
|
'local_search_footprints_reported': searched,
|
|
'actual_support_reported': supported,
|
|
'contact_reported_executions': contacts,
|
|
'median_delivery_latency_seconds': statistics.median(latencies) if latencies else None,
|
|
'max_delivery_latency_seconds': max(latencies) if latencies else None,
|
|
'median_order_to_physical_arrival_seconds': statistics.median(arrival_delays) if arrival_delays else None,
|
|
'reports_missing_original_order': orphan_reports})
|
|
sessions[-1]['unconfirmed_deadline_expiries'] = unconfirmed_expiries
|
|
sessions[-1]['delivered_packets_not_applied'] = unapplied_packets
|
|
return {'schema_version': 1, 'scope': 'Delivered-report replay only. No kill, hidden-location, tactical-quality or multiplayer-acceptance inference.',
|
|
'inputs': [{'path': str(path), 'sha256': hashlib.sha256(path.read_bytes()).hexdigest()} for path in paths],
|
|
'sessions': sessions}
|
|
|
|
|
|
def average_fps(path, run):
|
|
"""Arithmetic mean of complete measured tick samples, after strict validation."""
|
|
start = run['workload']['warmup_seconds'] * 1000
|
|
end = start + run['workload']['duration_seconds'] * 1000
|
|
values = []
|
|
with path.open(newline='', encoding='utf-8') as stream:
|
|
for row in csv.DictReader(stream):
|
|
elapsed, frame = float(row['elapsed_ms']), float(row['frame_time_ms'])
|
|
if elapsed - frame >= start and elapsed <= end:
|
|
values.append(1000 / frame)
|
|
return statistics.mean(values)
|
|
|
|
|
|
def compare(manifest_path):
|
|
"""Require three pairs; only aiDoctrine may differ in recorded settings."""
|
|
path = Path(manifest_path)
|
|
raw = path.read_bytes()
|
|
data = json.loads(raw, parse_constant=performance.reject_constant, object_pairs_hook=performance.unique_object)
|
|
performance.finite_json(data)
|
|
performance.keys(data, ['schema_version', 'sample_origin', 'runs'], 'manifest')
|
|
performance.require(type(data['schema_version']) is int and data['schema_version'] == 1, 'schema_version must be 1')
|
|
performance.require(data['sample_origin'] in ('measured', 'synthetic'), 'sample_origin must be measured or synthetic')
|
|
performance.require(isinstance(data['runs'], list) and len(data['runs']) >= 6, 'three paired runs are required')
|
|
reference = None
|
|
pairs, paths, artifacts = {}, set(), {}
|
|
results = []
|
|
for run in data['runs']:
|
|
performance.validate_run(run)
|
|
performance.require(run['capture']['kind'] == 'server_tick' and run['capture']['environment'] == 'dedicated_server',
|
|
'pursuit performance target requires dedicated-server tick captures')
|
|
expected_doctrine = 0 if run['variant'] == 'baseline' else 1
|
|
doctrine = run['workload']['settings'].get('aiDoctrine')
|
|
performance.require(type(doctrine) is int and doctrine == expected_doctrine, 'baseline aiDoctrine must be 0 and successor 1')
|
|
performance.require(run['workload']['settings'].get('aiShadow', 0) == 0, 'shadow mode is not a coordinated-behavior benchmark')
|
|
comparable = copy.deepcopy({key: run[key] for key in ('hardware', 'workload', 'capture', 'engine_version')})
|
|
comparable['workload']['settings'].pop('aiDoctrine')
|
|
if reference is None:
|
|
reference = comparable
|
|
performance.require(comparable == reference, 'hardware, scenario, population, capture and remaining settings must match exactly')
|
|
pair = pairs.setdefault(run['repeat_id'], {})
|
|
performance.require(run['variant'] not in pair, 'duplicate repeat/variant')
|
|
artifact = run['artifact_id'], sorted(run['addon_ids'])
|
|
performance.require(run['variant'] not in artifacts or artifacts[run['variant']] == artifact, 'artifact changed between repeats')
|
|
artifacts[run['variant']] = artifact
|
|
sample_path = (path.parent / run['sample_file']).resolve()
|
|
performance.require(sample_path not in paths, 'sample file reused')
|
|
paths.add(sample_path)
|
|
sampled = performance.load_samples(sample_path, run)
|
|
fps = average_fps(sample_path, run)
|
|
pair[run['variant']] = fps
|
|
results.append({'repeat_id': run['repeat_id'], 'variant': run['variant'], 'average_server_fps': fps, **sampled})
|
|
performance.require(len(pairs) >= 3 and all(set(pair) == {'baseline', 'successor'} for pair in pairs.values()), 'every repeat must have both variants')
|
|
baseline = statistics.mean(pair['baseline'] for pair in pairs.values())
|
|
successor = statistics.mean(pair['successor'] for pair in pairs.values())
|
|
regression = 100 * (baseline - successor) / baseline
|
|
return {'schema_version': 1, 'sample_origin': data['sample_origin'],
|
|
'scope': 'Matched dedicated-server arithmetic only; no authenticity, tactical improvement, multiplayer or release acceptance claim.',
|
|
'manifest_sha256': hashlib.sha256(raw).hexdigest(), 'matched_conditions_except_doctrine': reference,
|
|
'runs': results, 'baseline_average_server_fps': baseline, 'coordinated_average_server_fps': successor,
|
|
'average_server_fps_regression_percent': regression, 'regression_limit_percent': 5,
|
|
'performance_target_met': regression <= 5 if data['sample_origin'] == 'measured' else None,
|
|
'release_qualified': False}
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
commands = parser.add_subparsers(dest='command', required=True)
|
|
journal = commands.add_parser('journal', help='summarize delivered pursuit reports across journal rotations')
|
|
journal.add_argument('logs', nargs='+', type=Path)
|
|
journal.add_argument('--session')
|
|
comparison = commands.add_parser('compare', help='evaluate the five-percent average FPS target on three matched dedicated-server pairs')
|
|
comparison.add_argument('manifest', type=Path)
|
|
args = parser.parse_args()
|
|
try:
|
|
report = summary(args.logs, args.session) if args.command == 'journal' else compare(args.manifest)
|
|
except (OSError, ValueError, csv.Error) as error:
|
|
parser.exit(2, f'Pursuit analysis rejected: {error}\n')
|
|
print(json.dumps(report, indent=2, allow_nan=False))
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main()
|