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

Capture current mission, AI, map and administration development

Browse Source 
Include the current addon, terrain integrations, tools, regression tests and recorded development evidence.
This commit is contained in:
jakub
2026-10-13 22:52:53 +02:00
parent a951a96edc
commit 6a667d8deb
8104 changed files with 2124945 additions and 2061 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tools/ai_pursuit_replay.py
+241
View File
@@ -0,0 +1,241 @@
#!/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()
tools/build_atlas_model.py
+125
View File
@@ -0,0 +1,125 @@
"""Create the atlas's own solid mesh. Run with Blender --background --python.
Coordinates below are metres in Enfusion's X/right, Y/front, Z/up frame.
The printable face retains the existing 744x1024 layout and its left gutter.
The rigid book and separate turning sheet use the native atlas hand graph.
"""
from pathlib import Path
import math
import bpy
ROOT = Path(__file__).resolve().parents[1]
OUT = ROOT / 'addons/core/Assets/Items/ERR_Atlas'
OUT.mkdir(parents=True, exist_ok=True)
bpy.ops.object.select_all(action='SELECT')
bpy.ops.object.delete(use_global=False)
def xyz(p):
# Place the rigid paper just in front of the stock folded-map grip.
return (p[0] + .015, -p[2], p[1] + .018)
materials = {}
for name, color, metal in [('AtlasPrint', (.85,.8,.7,1), 0),
('AtlasPaper', (.73,.70,.62,1), 0),
('AtlasCover', (.48,.018,.012,1), 0),
('AtlasWire', (.8,.8,.78,1), .65)]:
mat = bpy.data.materials.new(name)
mat.diffuse_color = color
mat.use_nodes = True
bsdf = mat.node_tree.nodes.get('Principled BSDF')
bsdf.inputs['Base Color'].default_value = color
bsdf.inputs['Metallic'].default_value = metal
bsdf.inputs['Roughness'].default_value = .4 if metal else .8
materials[name] = mat
def mesh(name, vertices, faces, material, uvs=None):
data = bpy.data.meshes.new(name)
data.from_pydata([xyz(v) for v in vertices], [], faces)
data.update()
obj = bpy.data.objects.new(name, data)
bpy.context.collection.objects.link(obj)
data.materials.append(materials[material])
uv = data.uv_layers.new(name='UVMap')
for polygon in data.polygons:
for loop in polygon.loop_indices:
v = data.loops[loop].vertex_index
uv.data[loop].uv = uvs[v] if uvs else (0,0)
return obj
def box(name, lo, hi, material):
x,y,z = lo; X,Y,Z = hi
return mesh(name, [(x,y,z),(X,y,z),(X,Y,z),(x,Y,z),
(x,y,Z),(X,y,Z),(X,Y,Z),(x,Y,Z)],
[(0,3,2,1),(4,5,6,7),(0,1,5,4),(3,7,6,2),
(0,4,7,3),(1,2,6,5)], material)
# The paper mesh used a folded sheet. This replacement has separate covers,
# a 10 mm block of leaves and actual wire geometry with depth and silhouette.
left, right, bottom, top = -.083, .108, -.1485, .1485
box('BackCover', (left-.002,-.012,bottom-.002), (right+.002,-.010,top+.002), 'AtlasCover')
box('PageBlock', (left,-.010,bottom), (right,-.0005,top), 'AtlasPaper')
# Small real edge seams catch light at grazing angles without one mesh per leaf.
for n in range(12):
y = -.0096+n*.00075
box('PageEdge_%02d'%n, (left,y,bottom-.00025), (right+.0003,y+.00016,top+.00025), 'AtlasPaper')
box('FrontCover', (left-.002,-.0005,bottom-.002), (right+.002,.0005,top+.002), 'AtlasCover')
mesh('PrintedSheet', [(left,.0006,bottom),(right,.0006,bottom),
(right,.0006,top),(left,.0006,top)], [(0,3,2,1)], 'AtlasPrint',
[(80/744,1),(1,1),(1,0),(80/744,0)])
# Continuous helix around the spine; the upper arc passes through the gutter
# and the lower arc wraps behind the page block. No painted fake rings.
vertices=[]; faces=[]
turns=14; steps=turns*32; sides=6
for i in range(steps+1):
angle=i/32*2*math.pi
z=bottom+.009+(top-bottom-.018)*i/steps
for j in range(sides):
tube=j/sides*2*math.pi
radius=.0088+.0009*math.cos(tube)
vertices.append((left+.002+radius*math.cos(angle),
-.005+radius*math.sin(angle), z+.0009*math.sin(tube)))
for i in range(steps):
for j in range(sides):
a=i*sides+j; b=i*sides+(j+1)%sides
faces.append((a,b,b+sides,a+sides))
wire=mesh('SpiralWire',vertices,faces,'AtlasWire')
for poly in wire.data.polygons: poly.use_smooth=True
# A simple inventory collider; no dynamic page physics or runtime allocations.
collider=box('UBX_Atlas', (left,-.013,bottom), (right,.004,top), 'AtlasPaper')
collider['usage']='Geometry'
collider.hide_render=True
bpy.context.scene.unit_settings.system='METRIC'
bpy.context.scene.unit_settings.scale_length=1
bpy.ops.object.select_all(action='SELECT')
source = ROOT / 'source/atlas'
source.mkdir(parents=True, exist_ok=True)
bpy.ops.wm.save_as_mainfile(filepath=str(source/'ERR_Atlas.blend'))
bpy.ops.export_scene.fbx(filepath=str(OUT/'ERR_Atlas.fbx'), use_selection=True,
object_types={'MESH'}, apply_unit_scale=True, apply_scale_options='FBX_SCALE_ALL',
axis_forward='-Z', axis_up='Y', bake_space_transform=True,
use_mesh_modifiers=True, use_custom_props=True, add_leaf_bones=False, bake_anim=False)
print('ERR ATLAS MODEL EXPORTED', OUT)
# Separate lightweight turning leaf. Its origin is the spine hinge; runtime
# rotates this visual only, keeping the held book and gripping hand steady.
bpy.ops.object.select_all(action='SELECT')
bpy.ops.object.delete(use_global=False)
verts=[]; faces=[]; uvs=[]
for i in range(17):
u=i/16
for z,v in [(bottom,1),(top,0)]:
# Subtract the book placement offset: this mesh is hinge-local.
verts.append((.191*u-.015, .004*math.sin(math.pi*u)-.018, z))
uvs.append((80/744+(1-80/744)*u,v))
for i in range(16):
faces.append((2*i,2*i+1,2*i+3,2*i+2))
leaf=mesh('TurningLeaf',verts,faces,'AtlasPrint',uvs)
back=mesh('LeafBack',verts,[tuple(reversed(f)) for f in faces],'AtlasPaper',uvs)
bpy.ops.object.select_all(action='SELECT')
bpy.ops.wm.save_as_mainfile(filepath=str(source/'ERR_AtlasLeaf.blend'))
bpy.ops.export_scene.fbx(filepath=str(OUT/'ERR_AtlasLeaf.fbx'),use_selection=True,
object_types={'MESH'},apply_unit_scale=True,apply_scale_options='FBX_SCALE_ALL',
axis_forward='-Z',axis_up='Y',bake_space_transform=True,
use_mesh_modifiers=True,use_custom_props=True,add_leaf_bones=False,bake_anim=False)
tools/check.py
+48 -2
View File
File diff suppressed because one or more lines are too long
tools/dedicated_native.py
+1 -1
View File
@@ -49,7 +49,7 @@ def sandbox(runtime, source, run):
PROJECTS = {
'core': ('2528466DE220D91C', '{AD62322D5213C440}Missions/ERR_Everon_Base.conf'),
'core': ('2528466DE220D91C', '{C1E20667DCE324ED}Missions/ERR_Everon.conf'),
'chat': ('2528466DE220D91C', '{C1E20667DCE324ED}Missions/ERR_Everon.conf'),
}
ADDON_GUIDS = {'core': PROJECTS['core'][0]}
tools/dedicated_preflight.py
+1 -1
View File
@@ -11,7 +11,7 @@ import shutil
from source_manifest import addon_manifest
APP_ID = '1874900'
HEADERS = {'core': 'Missions/ERR_Everon_Base.conf', 'chat': 'Missions/ERR_Everon.conf'}
HEADERS = {'core': 'Missions/ERR_Everon.conf', 'chat': 'Missions/ERR_Everon.conf'}
SYSTEMS = '{C50579A2EC48E8EF}Configs/Systems/MissionSystems.conf'
WORLD = '{B6C516F86302C679}Worlds/ERR_Everon.ent'
tools/desktop/kwin-right-monitor.js
-40
View File
@@ -1,40 +0,0 @@
// Session-local placement for Enfusion/Reforger test windows only.
var targetName = "HDMI-A-1";
var placing = false;
function isTestWindow(w) {
var app = String(w.resourceClass).toLowerCase();
return app === "steam_app_1874910" || app.indexOf("armareforger") >= 0 || String(w.caption).indexOf("Enfusion Workbench") >= 0;
}
function place(w) {
if (placing || !isTestWindow(w)) return;
placing = true;
var outputs = workspace.screens;
for (var i = 0; i < outputs.length; i++) {
if (outputs[i].name !== targetName) continue;
w.fullScreen = false;
w.setMaximize(false, false);
workspace.sendClientToScreen(w, outputs[i]);
var a = outputs[i].geometry;
var g = w.frameGeometry;
w.frameGeometry = {x: a.x + 20, y: a.y + 30,
width: Math.min(g.width, a.width - 40), height: Math.min(g.height, a.height - 60)};
print("ERRI_DISPLAY " + w.resourceClass + " -> " + w.output.name + " " + w.frameGeometry.x);
placing = false;
return;
}
w.minimized = true;
placing = false;
print("ERRI_DISPLAY target unavailable; test window minimized");
}
workspace.windowAdded.connect(function(w) {
place(w);
w.captionChanged.connect(function() { place(w); });
w.frameGeometryChanged.connect(function() { place(w); });
w.fullScreenChanged.connect(function() { place(w); });
});
var windows = workspace.windowList();
// Existing user editors are deliberately left untouched.
for (var i = 0; i < windows.length; i++) {
if (String(windows[i].caption).indexOf("Reintroduced") >= 0) place(windows[i]);
}
print("ERRI_DISPLAY guard active for " + targetName);
tools/generate_map_icons.py
+70
View File
@@ -0,0 +1,70 @@
#!/usr/bin/env python3
"""Build original vector site symbols, review PNGs and the game's single atlas.
Requires rsvg-convert and Pillow. Runtime texture import is done in Workbench.
"""
from pathlib import Path
import io
import subprocess
from PIL import Image, ImageDraw, ImageFont
ROOT = Path(__file__).resolve().parents[1]
OUT = ROOT / 'art/map-markers'
RUNTIME = ROOT / 'addons/core/UI/Textures/ERR'
VIOLET = '#B277F0'
# All symbols use a 64-unit grid and a generous quiet border for small map sizes.
SYMBOLS = [
('unknown', 'Unknown location', 'mark-question', '<path d="M24 25c0-11 19-11 19 0 0 7-11 7-11 14"/><circle cx="32" cy="47" r="2.6" fill="black" stroke="none"/>'),
('radio', 'Radio station', 'radio', '<path d="M25 49l7-25 7 25M23 49h18M28 40h8M26 46h12M23 20q-6 7 0 14M41 20q6 7 0 14M17 15q-11 12 0 24M47 15q11 12 0 24"/><circle cx="32" cy="22" r="3" fill="black" stroke="none"/>'),
('extraction', 'Extraction pickup', 'helicopters', '<path d="M13 19h34M31 19v8M18 45h30M25 38v7M41 38v7M20 33l-9-3v-7M10 25h6"/><path d="M21 29h19q9 0 11 9H25q-6 0-4-9Z" fill="black" stroke="none"/><path d="M39 30v5h7" stroke="#B277F0" stroke-width="2.5"/>'),
('camp', 'Camp / outpost', 'building', '<path d="M12 48l20-32 20 32ZM29 20l-4-6M35 20l4-6"/><path d="M24 48l8-20 8 20Z" fill="black" stroke="none"/>'),
('supplies', 'Supply depot', 'supplies', '<rect x="16" y="24" width="32" height="25" rx="2"/><path d="M23 24v25M41 24v25M15 29h34M26 24v-7h12v7"/><path d="M29 35h6v7h-6Z" fill="black" stroke="none"/>'),
('mortar', 'Mortar position', 'artilleryFireMission', '<path d="M19 16l6-4 19 30-6 4Z" fill="black" stroke="none"/><path d="M30 29L19 48M31 32l6 16M14 49h31M39 47h11"/>'),
('hospital', 'Field hospital', 'medical', '<path d="M26 15h12v11h11v12H38v11H26V38H15V26h11Z" fill="black" stroke="none"/>'),
('prison', 'Prison', 'prison', '<rect x="17" y="16" width="30" height="33" rx="2"/><path d="M24 17v31M32 17v31M40 17v31M18 27h28M18 39h28"/>'),
('roadblock', 'Roadblock', 'roadblock', '<path d="M16 36v14M48 36v14"/><rect x="12" y="23" width="40" height="13" rx="1"/><path d="M15 34l10-10M29 34l10-10M43 34l7-7"/>'),
('factory', 'Industrial site', 'factory', '<path d="M15 48V31l11-8v8l11-8v25ZM38 48V16h9l2 32ZM15 48h34"/><path d="M20 39h2M29 39h2"/>'),
('minefield', 'Minefield', 'minefield', '<path d="M31 14L11 49h42L33 14Z"/><path d="M32 26v11"/><circle cx="32" cy="43" r="2" fill="black" stroke="none"/>'),
('crash', 'Aircraft wreck', 'crash', '<path d="M34 15l-3 15-16 6v5l16-3 1 10-5 4M38 19l-1 11 12 11M37 41l2 9M19 17l7 6M46 15l-3 7M48 49l5 3"/>'),
]
def svg(symbol):
return f'<svg xmlns="http://www.w3.org/2000/svg" width="256" height="256" viewBox="0 0 64 64"><circle cx="32" cy="32" r="30" fill="{VIOLET}"/><g fill="none" stroke="#000000" stroke-width="3.2" stroke-linecap="round" stroke-linejoin="round">{symbol}</g></svg>\n'
def main():
OUT.mkdir(parents=True, exist_ok=True)
RUNTIME.mkdir(parents=True, exist_ok=True)
atlas = Image.new('RGBA', (512, 512))
sheet = Image.new('RGB', (1200, 1050), '#f2f0f5')
draw = ImageDraw.Draw(sheet)
font_path = '/usr/share/fonts/TTF/DejaVuSans.ttf'
title = ImageFont.truetype(font_path, 32)
label = ImageFont.truetype(font_path, 19)
note = ImageFont.truetype(font_path, 16)
draw.text((40, 26), 'CO10 ESCAPE / MAP MARKERS', font=title, fill='#21152f')
draw.text((40, 74), 'Violet circles · black symbols · transparent outside · actual 36 px samples below', font=note, fill='#544662')
definitions = []
for i, (name, caption, image_name, symbol) in enumerate(SYMBOLS):
source = OUT / (name + '.svg')
source.write_text(svg(symbol))
png = subprocess.check_output(['rsvg-convert', str(source)])
(OUT / (name + '.png')).write_bytes(png)
icon = Image.open(io.BytesIO(png)).convert('RGBA')
x, y = i % 4 * 128, i // 4 * 128
atlas.alpha_composite(icon.resize((128, 128), Image.Resampling.LANCZOS), (x, y))
definitions.append(f' ImageSetDefClass "{image_name}" {{\n Pos {x} {y}\n Size 128 128\n Flags 0 0\n }}')
sx, sy = 40 + i % 4 * 290, 135 + i // 4 * 295
sheet.paste(icon.resize((160, 160), Image.Resampling.LANCZOS), (sx + 50, sy), icon.resize((160, 160), Image.Resampling.LANCZOS))
draw.text((sx + 130, sy + 178), caption, font=label, fill='#21152f', anchor='mt')
small = icon.resize((36, 36), Image.Resampling.LANCZOS)
for j, bg in enumerate(('#d5d3b2', '#303b32')):
px = sx + 82 + j * 55
draw.rectangle((px-4, sy+215, px+39, sy+258), fill=bg)
sheet.paste(small, (px, sy + 219), small)
atlas.save(RUNTIME / 'SiteMarkers.png')
sheet.save(OUT / 'preview.png')
(RUNTIME / 'SiteMarkers.imageset').write_text('ImageSetClass {\n RefSize 512 512\n Textures {\n ImageSetTextureClass {\n screenSize FHD\n size 512 512\n path "{ADEE022D9F06C05A}UI/Textures/ERR/SiteMarkers.edds"\n }\n }\n Images {\n' + '\n'.join(definitions) + '\n }\n}\n')
print(OUT / 'preview.png')
if __name__ == '__main__':
main()
tools/headless_game.py
+12
View File
@@ -313,6 +313,9 @@ def prefix_process_candidate(proc):
# exact native foreground-service signature, after all runtime checks.
if comm == 'apache2' and args == ['apache2', '-DFOREGROUND']:
return False
# GnuPG's non-dumpable smart-card helper, started by the user's agent.
if comm == 'scdaemon' and args == ['scdaemon', '--multi-server']:
return False
# A readable native executable is conclusive; otherwise require an absolute
# Linux argv[0] corroborated by the kernel's (possibly truncated) comm name.
if executable.startswith('/'):
@@ -321,6 +324,15 @@ def prefix_process_candidate(proc):
basename = Path(args[0]).name
if comm and (comm == basename or comm == basename[:15]):
return False
# A confirmed zombie has already exited and cannot own an active Wine
# prefix. Its empty cmdline/exe otherwise looks like an unclassified process.
if not args and not executable:
try:
status = (proc/'status').read_text()
except FileNotFoundError:
status = ''
if re.search(r'^State:\s+Z\s+\(zombie\)\s*$', status, re.M):
return False
argv0 = args[0] if args else ''
refuse(f'Cannot classify possible Wine/runtime process {proc.name} (comm={comm!r}, argv0={argv0!r})')
tools/headless_workbench.py
+45 -3
View File
@@ -14,7 +14,14 @@ p.add_argument('--steam', type=Path, default=Path.home()/'.local/share/Steam')
p.add_argument('--external-addons', type=Path, help='Installed RHS/WCS/BMP source root; expose only audited closure in a separate Chat profile')
p.add_argument('--compat', type=Path, required=True, help='Disposable cloned Proton compatdata directory outside Steam')
p.add_argument('--profile', type=Path, required=True, help='Separate disposable successor Workbench profile')
p.add_argument('--smoke', choices=['escape-route', 'optional-site-skip', 'infantry-combat', 'medical-combat', 'hunter-faction', 'unconscious-camera', 'town-garrisons', 'site-garrisons', 'radio-interaction', 'enemy-loadout', 'aggression', 'air-logistics', 'air-assault', 'air-recon', 'airfield-markers', 'opening-tools', 'mortar-station', 'generated-sites', 'hq-local-handoff', 'infantry-transfer', 'hq-body', 'convoy', 'supply-convoys', 'hq-memory', 'hq-awareness', 'aircraft-field', 'aircraft-route', 'native-end-active', 'native-end-preparing', 'aircraft-control', 'aircraft-cancel', 'aircraft-translate', 'map-ownership', 'mortar-refresh', 'mortar-salvo', 'mortar-courier', 'station-teardown', 'prison-breach', 'powered-flight', 'powered-translate', 'local-admission', 'mortar-expiry', 'mortar-transfer', 'mortar-fire', 'patrol-hq', 'victory', 'lost-helicopter', 'player-death', 'radio-network', 'hq-response', 'mortar-inventory', 'startup-failure', 'site-survey', 'cache-loss', 'extraction-capacity', 'early-helicopter-loss', 'extraction-startup-failure', 'extraction-boundary-failure', 'transport-loss', 'transport-survey', 'lifecycle', 'objective-ui', 'recovery', 'prisoner-kit', 'medical', 'casualty-load', 'casualty-direct', 'casualty-bench', 'casualty-remove', 'environment', 'environment-restart', 'patrols', 'radio-document-loss', 'radio-document-loss-partial', 'guard-cancel', 'casualty-service'], default='victory')
p.add_argument('--smoke', choices=['coordinated-ai', 'crash-start', 'crash-geometry', 'normal-escape-route', 'escape-route', 'optional-site-skip', 'infantry-combat', 'medical-combat', 'hunter-faction', 'unconscious-camera', 'objective-map', 'town-garrisons', 'site-garrisons', 'radio-interaction', 'enemy-loadout', 'aggression', 'air-logistics', 'air-assault', 'air-recon', 'airfield-markers', 'opening-tools', 'mortar-station', 'generated-sites', 'hq-local-handoff', 'infantry-transfer', 'hq-body', 'convoy', 'supply-convoys', 'supply-travel', 'mounted-ammo', 'alarm-hearing', 'hq-memory', 'hq-awareness', 'aircraft-field', 'aircraft-route', 'native-end-active', 'native-end-preparing', 'aircraft-control', 'aircraft-cancel', 'aircraft-translate', 'map-ownership', 'mortar-refresh', 'mortar-salvo', 'mortar-courier', 'station-teardown', 'inventory-hitzone', 'prison-breach', 'powered-flight', 'powered-translate', 'local-admission', 'mortar-expiry', 'mortar-transfer', 'mortar-fire', 'patrol-hq', 'victory', 'lost-helicopter', 'player-death', 'radio-network', 'hq-response', 'mortar-inventory', 'startup-failure', 'site-survey', 'cache-loss', 'extraction-capacity', 'early-helicopter-loss', 'extraction-startup-failure', 'extraction-boundary-failure', 'transport-loss', 'transport-survey', 'lifecycle', 'objective-ui', 'recovery', 'prisoner-kit', 'medical', 'casualty-load', 'casualty-direct', 'casualty-bench', 'casualty-remove', 'environment', 'environment-restart', 'patrols', 'radio-document-loss', 'radio-document-loss-partial', 'guard-cancel', 'casualty-service'], default='victory')
p.add_argument('--trace-random-radius', action='store_true', help='Workbench-only read-only zero-radius caller traces; one normal Chat round, capped at sixteen records')
p.add_argument('--maintenance-refuel', action='store_true', help='Opt-in normal-route check: original held Southern fuel can transfers finite fuel through its native UAZ action; one arranged tank depletion, no stock creation')
p.add_argument('--mounted-swap-eligibility', action='store_true', help='Read-only mounted-ammo variant: one exact finite cargo/mounted magazine CanSwap call; no transfer or radio proof')
p.add_argument('--mounted-bmp', action='store_true', help='Read-only original BMP slot4 complete mounted counts and its natural HQ radio delivery; isolates unrelated opening hunters before unlock')
p.add_argument('--mine-stock', action='store_true', help='Read-only mounted-ammo check of exact callback-proved finite PMN4 cargo before and after natural radio delivery')
p.add_argument('--mounted-courier', action='store_true', help='Mounted-ammo variant: intentionally deplete one original installed magazine, then require natural HQ radio, finite pickup, native exchange and original-seat return')
p.add_argument('--mounted-intercept', action='store_true', help='After convoy driving, two synthetic witness samples exercise real HQ radio/planning and original escort detour/hold/expiry/resume')
p.add_argument('--convoy-cover', action='store_true', help='After convoy driving, arrange the existing prisoner to verify native sighting/radio HOLD and RESUME')
p.add_argument('--field-terrain-audit', action='store_true', help='After a fixed aircraft-field survey refusal, log one read-only 1681-node coarse terrain grid; never selects or qualifies a corridor')
p.add_argument('--field-service-cancel', action='store_true', help='Explicit aircraft-field variant: Stop the original retained service during measured cruise; requires physical cancellation landing')
@@ -24,10 +31,27 @@ p.add_argument('--flight-autohover', action='store_true', help='Opt-in owned nat
p.add_argument('--flight-action-inputs', action='store_true', help='Experimental exact-AI ActionManager inputs; Core powered-flight fixture only')
p.add_argument('--flight-input-diagnostics', action='store_true', help='Opt-in read-only powered-flight input/application and native rotor signal observations')
p.add_argument('--recon-takeover-only', action='store_true', help='Air-recon fixture: only native faction handover/release and explicit aircraft deletion')
p.add_argument('--body-recovery-only', action='store_true', help='Air-logistics fixture: only physical corpse inspection and returned document intelligence')
p.add_argument('--coordinated-blocked-only', action='store_true', help='Coordinated fixture: native movement, cancellation and unreachable-route failure only; explicitly excludes suppression qualification')
p.add_argument('--rounds', type=int, default=2)
p.add_argument('--timeout', type=int, default=360)
p.add_argument('--project', choices=['core', 'chat'], default='chat', help='Diagnostic fixture family (both load the combined mod)')
a = p.parse_args()
if a.smoke == 'coordinated-ai' and (a.project != 'core' or a.rounds != 1 or a.timeout < 720): p.error('coordinated-ai requires Core, one round and timeout at least720 seconds')
if a.coordinated_blocked_only and a.smoke != 'coordinated-ai': p.error('--coordinated-blocked-only requires --smoke coordinated-ai')
if a.smoke == 'inventory-hitzone' and (a.project != 'core' or a.rounds != 1 or a.timeout < 240): p.error('inventory-hitzone requires --project core --rounds 1 and timeout at least 240 seconds')
if a.smoke == 'crash-start' and (a.project != 'chat' or a.rounds != 1 or a.timeout < 240): p.error('crash-start requires Chat, one round and timeout at least 240 seconds')
if a.smoke == 'crash-geometry' and (a.project != 'chat' or a.rounds != 1 or a.timeout < 180): p.error('crash-geometry requires Chat, one round and timeout at least180 seconds')
if a.trace_random_radius and (a.smoke != 'normal-escape-route' or a.project != 'chat' or a.rounds != 1 or a.timeout < 420): p.error('--trace-random-radius requires Chat normal-escape-route, one round and timeout at least 420 seconds')
if a.smoke == 'objective-map' and (a.project != 'core' or a.rounds != 1): p.error('objective-map requires --project core --rounds 1')
if a.smoke == 'unconscious-camera' and a.project != 'core': p.error('unconscious-camera requires --project core to keep the general smoke fixture inactive')
if a.mine_stock:
if a.smoke != 'mounted-ammo' or a.project != 'chat' or a.rounds != 1 or a.timeout < 360:
p.error('--mine-stock requires --smoke mounted-ammo --project chat --rounds 1 and timeout at least 360 seconds')
if a.mounted_courier or a.mounted_swap_eligibility:
p.error('--mine-stock cannot combine with --mounted-courier or --mounted-swap-eligibility')
if a.maintenance_refuel and (a.smoke != 'normal-escape-route' or a.project != 'chat' or a.rounds != 1 or a.timeout < 420): p.error('--maintenance-refuel requires Chat normal-escape-route, one round and timeout at least 420 seconds')
if a.body_recovery_only and (a.smoke != 'air-logistics' or a.project != 'core' or a.rounds != 1): p.error('--body-recovery-only requires Core air-logistics and one round')
if a.recon_takeover_only and (a.smoke != 'air-recon' or a.project != 'core' or a.rounds != 1): p.error('--recon-takeover-only requires Core air-recon and one round')
steam, compat, profile = a.steam.resolve(), a.compat.resolve(), a.profile.resolve()
if not 1 <= a.rounds <= 10 or not 60 <= a.timeout <= 1800:
@@ -41,9 +65,18 @@ if a.smoke == 'mortar-courier' and (a.project != 'chat' or a.rounds != 1 or a.ti
if a.smoke in {'mortar-refresh', 'mortar-salvo', 'mortar-expiry', 'mortar-transfer', 'mortar-fire'} and (a.project != 'chat' or a.rounds != 1 or a.timeout < 400):
p.error('mortar-refresh/mortar-salvo/mortar-expiry/mortar-transfer/mortar-fire requires --project chat --rounds 1 --timeout at least 400')
if a.smoke in {'aircraft-field', 'aircraft-route', 'aircraft-control', 'aircraft-cancel', 'aircraft-translate'} and (a.project != 'chat' or a.rounds != 1 or a.timeout < 420): p.error('aircraft-field/aircraft-route/aircraft-control/aircraft-cancel/aircraft-translate requires Chat, one round, and at least 420 seconds')
if a.mounted_swap_eligibility and a.smoke != 'mounted-ammo': p.error('--mounted-swap-eligibility requires --smoke mounted-ammo')
if a.mounted_courier and (a.smoke != 'mounted-ammo' or a.project != 'chat' or a.rounds != 1 or a.timeout < 780): p.error('--mounted-courier requires --smoke mounted-ammo --project chat --rounds 1 and timeout at least780 seconds')
if a.mounted_courier and a.mounted_swap_eligibility: p.error('--mounted-courier and --mounted-swap-eligibility are mutually exclusive')
if a.mounted_intercept and (a.smoke != 'convoy' or a.project != 'chat' or a.rounds != 1 or a.timeout < 900): p.error('--mounted-intercept requires Chat --smoke convoy, one round and timeout at least900 seconds')
if a.mounted_intercept and a.convoy_cover: p.error('--mounted-intercept and --convoy-cover are mutually exclusive')
if a.convoy_cover and a.smoke != 'convoy': p.error('--convoy-cover requires --smoke convoy')
if a.convoy_cover and a.timeout < 900: p.error('--convoy-cover requires timeout at least900 seconds')
if a.smoke == 'supply-travel' and (a.project != 'chat' or a.rounds != 1 or a.timeout < 480): p.error('supply-travel requires Chat, one round, and at least480 seconds')
if a.smoke == 'supply-convoys' and (a.project != 'chat' or a.rounds != 1 or a.timeout < 900): p.error('supply-convoys requires Chat, one round, and at least900 seconds')
if a.smoke == 'alarm-hearing' and (a.project != 'chat' or a.rounds != 1 or a.timeout < 360): p.error('alarm-hearing requires Chat, exactly one round, and timeout at least360 seconds')
if a.smoke == 'mounted-ammo' and (a.project != 'chat' or a.rounds != 1 or a.timeout < 360): p.error('mounted-ammo requires Chat, exactly one round, and timeout at least360 seconds')
if a.mounted_bmp and (a.smoke != 'mounted-ammo' or a.project != 'chat' or a.rounds != 1 or a.timeout < 360 or a.mine_stock or a.mounted_courier or a.mounted_swap_eligibility): p.error('--mounted-bmp requires Chat mounted-ammo, one round, timeout at least360 seconds, without mine/courier/swap flags')
if a.smoke == 'convoy' and (a.project != 'chat' or a.rounds != 1 or a.timeout < 600): p.error('convoy requires Chat, exactly one round, and timeout at least600 seconds')
if a.smoke == 'infantry-transfer' and (a.project != 'chat' or a.rounds != 1 or a.timeout < 360): p.error('infantry-transfer requires Chat, one round and timeout at least360 seconds')
if a.smoke == 'hq-local-handoff' and (a.project != 'chat' or a.rounds != 1 or a.timeout < 360): p.error('hq-local-handoff requires Chat, one round and timeout at least360 seconds')
@@ -107,6 +140,13 @@ for proc in Path('/proc').iterdir():
except (FileNotFoundError, PermissionError, ProcessLookupError): pass
command = ['gamescope', '--backend', 'headless', '-W', '1280', '-H', '720', '-w', '1280', '-h', '720', '-r', '30', '--', str(proton), 'run', str(wb/'ArmaReforgerWorkbenchSteamDiag.exe'), '-gproj', 'Z:'+str(ROOT/'addons/core/addon.gproj'), '-addonsDir', 'Z:'+str(ROOT/'addons'), '-profile', profile_arg, '-logconsole', '-wbModule=WorldEditor', '-load', 'Worlds/ERR_Everon.ent', '-plugin=ERR_ValidateWorldPlugin', '-errPlay', '1', '-errSmoke', '1', '-errSmokeOutcome', a.smoke, '-noThrow', '-disableCrashReporter']
if a.convoy_cover: command += ['-errConvoyCover', '1']
if a.mounted_intercept: command += ['-errMountedIntercept', '1']
if a.mounted_swap_eligibility: command += ['-errMountedSwapEligibility', '1']
if a.mounted_courier: command += ['-errMountedCourier', '1']
if a.mounted_bmp: command += ['-errMountedBMP', '1']
if a.mine_stock: command += ['-errMineStock', '1']
if a.maintenance_refuel: command += ['-errMaintenanceRefuel', '1']
if a.trace_random_radius: command += ['-errTraceRandomRadius', '1']
if a.field_service_cancel: command += ['-errSmokeFieldServiceCancel', '1']
if a.field_terrain_audit: command += ['-errSmokeFieldTerrainAudit', '1']
if a.route_service_cancel: command += ['-errSmokeRouteServiceCancel', '1']
@@ -115,11 +155,13 @@ if a.flight_input_diagnostics: command += ['-errFlightInputDiag', '1']
if a.flight_action_inputs: command += ['-errFlightActionInputs', '1']
if a.flight_autohover: command += ['-errFlightAutohover', '1']
if a.recon_takeover_only: command += ['-errReconTakeoverOnly', '1']
if a.body_recovery_only: command += ['-errBodyRecoveryOnly', '1']
if a.coordinated_blocked_only: command += ['-errCoordinatedBlockedOnly', '1']
if a.project == 'core':
if a.smoke not in {'optional-site-skip', 'infantry-combat', 'medical-combat', 'hunter-faction', 'unconscious-camera', 'town-garrisons', 'site-garrisons', 'radio-interaction', 'enemy-loadout', 'aggression', 'air-logistics', 'air-assault', 'air-recon', 'airfield-markers', 'opening-tools', 'mortar-station', 'generated-sites', 'native-end-active', 'native-end-preparing', 'station-teardown', 'prison-breach', 'powered-flight', 'powered-translate', 'local-admission', 'victory', 'startup-failure', 'site-survey', 'cache-loss', 'extraction-capacity', 'early-helicopter-loss', 'extraction-startup-failure', 'extraction-boundary-failure', 'transport-loss', 'transport-survey', 'lifecycle', 'objective-ui', 'recovery', 'prisoner-kit', 'medical', 'casualty-load', 'casualty-direct', 'casualty-bench', 'casualty-remove', 'environment', 'environment-restart', 'patrols', 'radio-document-loss', 'radio-document-loss-partial', 'guard-cancel', 'casualty-service'}: p.error('Unsupported Core-only fixture')
if a.smoke not in {'coordinated-ai', 'optional-site-skip', 'infantry-combat', 'medical-combat', 'hunter-faction', 'unconscious-camera', 'objective-map', 'town-garrisons', 'site-garrisons', 'radio-interaction', 'enemy-loadout', 'aggression', 'air-logistics', 'air-assault', 'air-recon', 'airfield-markers', 'opening-tools', 'mortar-station', 'generated-sites', 'native-end-active', 'native-end-preparing', 'station-teardown', 'inventory-hitzone', 'prison-breach', 'powered-flight', 'powered-translate', 'local-admission', 'victory', 'startup-failure', 'site-survey', 'cache-loss', 'extraction-capacity', 'early-helicopter-loss', 'extraction-startup-failure', 'extraction-boundary-failure', 'transport-loss', 'transport-survey', 'lifecycle', 'objective-ui', 'recovery', 'prisoner-kit', 'medical', 'casualty-load', 'casualty-direct', 'casualty-bench', 'casualty-remove', 'environment', 'environment-restart', 'patrols', 'radio-document-loss', 'radio-document-loss-partial', 'guard-cancel', 'casualty-service'}: p.error('Unsupported Core-only fixture')
command[command.index('-plugin=ERR_ValidateWorldPlugin')] = '-plugin=ERR_ValidateBaseWorldPlugin'
command[command.index('-errSmoke')] = '-errBaseSmoke'
if a.smoke in {'station-teardown', 'prison-breach', 'powered-flight', 'powered-translate', 'local-admission', 'cache-loss', 'extraction-capacity', 'early-helicopter-loss', 'extraction-startup-failure', 'extraction-boundary-failure', 'transport-loss', 'transport-survey', 'lifecycle', 'objective-ui', 'recovery', 'prisoner-kit', 'medical', 'casualty-load', 'casualty-direct', 'casualty-bench', 'casualty-remove', 'environment', 'environment-restart', 'patrols', 'radio-document-loss', 'radio-document-loss-partial', 'guard-cancel', 'casualty-service'} and a.project != 'core': p.error('This fixture requires --project core')
if a.smoke in {'station-teardown', 'inventory-hitzone', 'prison-breach', 'powered-flight', 'powered-translate', 'local-admission', 'cache-loss', 'extraction-capacity', 'early-helicopter-loss', 'extraction-startup-failure', 'extraction-boundary-failure', 'transport-loss', 'transport-survey', 'lifecycle', 'objective-ui', 'recovery', 'prisoner-kit', 'medical', 'casualty-load', 'casualty-direct', 'casualty-bench', 'casualty-remove', 'environment', 'environment-restart', 'patrols', 'radio-document-loss', 'radio-document-loss-partial', 'guard-cancel', 'casualty-service'} and a.project != 'core': p.error('This fixture requires --project core')
if a.smoke in {'startup-failure', 'site-survey', 'transport-survey', 'extraction-startup-failure', 'extraction-boundary-failure'} and (a.project != 'core' or a.rounds != 1): p.error('This diagnostic requires --project core --rounds 1')
env = os.environ.copy()
env.update(STEAM_COMPAT_DATA_PATH=str(compat), STEAM_COMPAT_CLIENT_INSTALL_PATH=str(steam), WINEPREFIX=str(compat/'pfx'))
tools/import_settings.py
+12 -1
View File
@@ -3,6 +3,15 @@
import argparse,json,os,tempfile,sys
from pathlib import Path
SCHEMA={
'aiDoctrine':(0,0,1),
'aiShadow':(0,0,1),
'planeAltitudeMetres':(250,150,500),
'planeReplacementSeconds':(60,30,1800),
'planeSectorSeconds':(120,30,600),
'planeReconEnabled':(1,0,1),
'roadblockCount':(4,0,12),
'civilianTrafficCount':(3,0,6),
'startMode':(0,0,1),
'keepWeaponOnUnconscious':(1,0,1),
'reportIntervalSeconds':(10,5,60),
'supportSquadLimit':(4,0,8),
@@ -14,13 +23,15 @@ SCHEMA={
'timeMultiplier':(1,1,24),
'startDateCode':(0,0,1501231),
'weatherProfile':(0,0,3),
'viewDistanceMetres':(0,0,10000),
'automaticEquipmentChancePercent':(25,0,100),
'antiAirChancePercent':(7,0,100),
'antiPersonnelMineChancePercent':(60,0,100),
'automaticConvoyEnabled':(1,0,1),
'convoyCount':(3,0,20),
'heliTimerMinutes':(15,1,1440),
}
AI_KEYS=('reportIntervalSeconds','supportSquadLimit','responseRadiusMetres','antiAirChancePercent','antiPersonnelMineChancePercent','automaticConvoyEnabled','heliTimerMinutes')
AI_KEYS=('aiDoctrine','aiShadow','planeAltitudeMetres','planeReplacementSeconds','planeSectorSeconds','planeReconEnabled','reportIntervalSeconds','supportSquadLimit','responseRadiusMetres','automaticEquipmentChancePercent','antiAirChancePercent','antiPersonnelMineChancePercent','automaticConvoyEnabled','heliTimerMinutes')
INTEL_KEYS=('aiIntelChancePercent','intelRevealChancePercent')
def merge(ai=None,intel=None):
result={key:spec[0] for key,spec in SCHEMA.items()};deferred=[]
tools/parse_hq_performance.py
+108
View File
@@ -0,0 +1,108 @@
#!/usr/bin/env python3
"""Summarize ERR PERF windows from performance logs (also accepts older console logs).
Script wall_ms_per_s measures scoped elapsed time, not CPU utilization.
Parent/child scopes overlap; engine values are sampled gauges, not scopes.
"""
import argparse
import csv
import math
import re
import sys
from collections import defaultdict
from pathlib import Path
FIELDS = (
'run', 'window', 'elapsed_ms', 'final', 'phase_start', 'phase_end',
'metric', 'samples', 'sum', 'min', 'max', 'zero', 'over50', 'over100',
)
INTEGER_FIELDS = ('window', 'elapsed_ms', 'phase_start', 'phase_end',
'samples', 'zero', 'over50', 'over100')
SCRIPT_METRICS = {
'mission_tick_ms', 'maintenance_ms', 'civilians_ms', 'forces_ms',
'observers_ms', 'hq_tick_ms', 'listeners_ms', 'hq_prep_ms',
'hq_hunters_ms', 'hq_garrisons_ms', 'hq_convoys_ms', 'hq_bodies_ms',
'hq_radio_index_ms', 'hq_reports_ms', 'hq_planner_ms', 'hq_support_ms',
'hq_journal_ms', 'garrison_work_ms', 'performance_flush_ms',
}
def read_rows(paths):
seen = {}
for path in paths:
with path.open(encoding='utf-8', errors='replace') as stream:
for number, line in enumerate(stream, 1):
marker = line.find('ERR PERF run=')
if marker < 0:
continue
values = dict(re.findall(r'(\w+)=([^\s]+)', line[marker:]))
try:
row = {key: values[key] for key in FIELDS}
for key in INTEGER_FIELDS:
row[key] = int(row[key])
if row[key] < 0:
raise ValueError(f'negative {key}')
for key in ('sum', 'min', 'max'):
row[key] = float(row[key])
if not math.isfinite(row[key]) or row[key] < 0:
raise ValueError(f'invalid {key}')
if row['elapsed_ms'] == 0 or row['samples'] == 0:
raise ValueError('empty window/sample count')
if row['min'] > row['max']:
raise ValueError('min exceeds max')
if not 0 <= row['over100'] <= row['over50'] <= row['samples']:
raise ValueError('invalid threshold counts')
if row['zero'] > row['samples']:
raise ValueError('invalid zero count')
except (KeyError, ValueError) as error:
raise ValueError(f'{path}:{number}: malformed performance row: {error}') from error
key = row['run'], row['window'], row['metric']
if key in seen:
if seen[key] != row:
raise ValueError(f'{path}:{number}: conflicting duplicate {key}')
continue
seen[key] = row
return list(seen.values())
def main():
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument('logs', nargs='+', type=Path)
parser.add_argument('--csv', action='store_true', help='output individual window rows as CSV')
args = parser.parse_args()
try:
rows = read_rows(args.logs)
if not rows:
raise ValueError('no ERR PERF measurement rows found')
except (OSError, ValueError) as error:
parser.error(str(error))
if args.csv:
writer = csv.DictWriter(sys.stdout, fieldnames=FIELDS)
writer.writeheader()
writer.writerows(rows)
return
grouped = defaultdict(list)
for row in rows:
grouped[row['run'], row['metric']].append(row)
print('Scope times overlap: do not add mission_tick, hq_tick and HQ children.')
print('Wall timings have 1 ms resolution; zero does not prove zero cost.')
print('Engine metrics are ~1 Hz samples of engine-reported values, NOT AI-only time.')
previous_run = None
for (run, metric), items in sorted(grouped.items()):
if run != previous_run:
print(f'\nRun {run}')
print(f'{"metric":26s} {"samples":>8s} {"mean":>10s} {"max":>10s} {"zero":>8s} {"wall_ms/s":>10s}')
previous_run = run
count = sum(item['samples'] for item in items)
total = sum(item['sum'] for item in items)
elapsed = sum(item['elapsed_ms'] for item in items)
maximum = max(item['max'] for item in items)
zero = sum(item['zero'] for item in items)
rate = f'{1000 * total / elapsed:.3f}' if metric in SCRIPT_METRICS else '-'
print(f'{metric:26s} {count:8d} {total / count:10.3f} {maximum:10.3f} {zero:8d} {rate:>10s}')
print('\nMeans/rates include only windows containing that metric; use --csv to compare matching windows and player counts.')
print('No native-AI timing or predicted offload speedup is inferred.')
if __name__ == '__main__':
main()
tools/resource_references.py
+36
View File
@@ -0,0 +1,36 @@
"""Hard resource references, with one fail-closed audited optional identity test.
This is not a general Enforce parser or a new external dependency allowlist.
Changing any byte of the audited source requires reviewing this exception again.
"""
import hashlib
import re
OPTIONAL_FILE = 'addons/core/Scripts/Game/ERR/00_Core/ERR_ReaperRHSCompat.c'
OPTIONAL_SHA256 = 'bcdf98c05ac6cdf0c7eaa854531b02e9ee4f53fbb346ae01d58114e572e7dd4f'
OPTIONAL_RESOURCE = '{574AA64002B8E308}Prefabs/Characters/REAPER_SpectatorCameraHelper.et'
OPTIONAL_METHOD = ''' protected static bool CameraSlot(BaseContainer slot)
{
if (!slot || slot.GetName() != "SpectatorCameraHelper") return false;
ResourceName prefab;
if (!slot.Get("Prefab", prefab)) return false;
return prefab == "{574AA64002B8E308}Prefabs/Characters/REAPER_SpectatorCameraHelper.et";
}'''
def hard_resource_references(relative_path, text):
"""Exclude only the pinned comparison occurrence; keep all other checks."""
identity_offset = -1
if (str(relative_path) == OPTIONAL_FILE
and hashlib.sha256(text.encode('utf-8')).hexdigest() == OPTIONAL_SHA256
and text.count(OPTIONAL_METHOD) == 1
and text.count(OPTIONAL_RESOURCE) == 1):
identity_offset = text.index(OPTIONAL_METHOD) + OPTIONAL_METHOD.index(OPTIONAL_RESOURCE)
for match in re.finditer(r'\{([A-F0-9]{16})\}([^"\n]+)', text):
if match.start() != identity_offset:
guid, resource = match.groups()
# Enforce strings that build a graph escape its closing quote. The
# escape belongs to the string syntax, not to the resource path.
if str(relative_path).endswith('.c') and resource.endswith('\\') and text[match.end():match.end()+1] == '"':
resource = resource[:-1]
yield guid, resource
tools/sample_server_memory.py
+89
View File
@@ -0,0 +1,89 @@
#!/usr/bin/env python3
"""Read-only Linux Reforger memory sampler; one bounded JSONL file, no game RPCs.
Run on the server host: python3 sample_server_memory.py --output /tmp/escape-memory.jsonl
Records all Reforger processes so a host OOM can be distinguished from a container
limit. Stops after 24 hours by default. Never reads command arguments or secrets.
"""
import argparse
import datetime
import json
from pathlib import Path
import time
def fields(text):
result = {}
for line in text.splitlines():
key, sep, value = line.partition(':')
if sep and value.split() and value.split()[0].isdigit():
result[key] = int(value.split()[0])
return result
def snapshot(proc=Path('/proc'), cgroups=Path('/sys/fs/cgroup')):
memory = fields((proc / 'meminfo').read_text())
result = {'utc': datetime.datetime.now(datetime.timezone.utc).isoformat(),
'host_kb': {k: memory.get(k) for k in ('MemTotal', 'MemAvailable', 'SwapTotal', 'SwapFree')},
'processes': []}
for directory in proc.iterdir():
if not directory.name.isdigit():
continue
try:
if 'ArmaReforgerServer' not in (directory / 'exe').readlink().name:
continue
status = fields((directory / 'status').read_text())
# PID + kernel start tick distinguish process restarts/PID reuse.
start = (directory / 'stat').read_text().rsplit(')', 1)[1].split()[19]
row = {'pid': int(directory.name), 'start_ticks': int(start),
'kb': {k: status.get(k) for k in ('VmRSS', 'VmHWM', 'VmSize', 'RssAnon', 'RssFile', 'VmSwap')}}
for line in (directory / 'cgroup').read_text().splitlines():
if not line.startswith('0::'):
continue
group = cgroups / line[3:].lstrip('/')
row['cgroup'] = line[3:]
for name in ('memory.current', 'memory.max', 'memory.peak', 'memory.events', 'memory.stat'):
path = group / name
if not path.exists():
continue
text = path.read_text().strip()
if name in ('memory.events', 'memory.stat'):
stats = dict(part.split() for part in text.splitlines())
if name == 'memory.stat':
stats = {k: v for k, v in stats.items() if k in ('anon', 'file', 'kernel', 'sock')}
row[name] = {k: int(v) for k, v in stats.items()}
else:
row[name] = int(text) if text.isdigit() else text
result['processes'].append(row)
except (OSError, ValueError, IndexError):
# Processes may exit at any point while /proc is being read.
continue
return result
def append(path, record, limit=2 * 1024 * 1024):
if path.exists() and path.stat().st_size >= limit:
path.replace(path.with_name(path.name + '.1'))
with path.open('a') as stream:
stream.write(json.dumps(record, separators=(',', ':')) + '\n')
def main():
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument('--output', type=Path, required=True)
parser.add_argument('--interval', type=int, default=60)
parser.add_argument('--duration', type=int, default=86400)
args = parser.parse_args()
if args.interval < 30 or not 1 <= args.duration <= 172800:
parser.error('interval must be >=30 seconds; duration must be 1..172800 seconds')
deadline = time.monotonic() + args.duration
while True:
append(args.output, snapshot())
remaining = deadline - time.monotonic()
if remaining <= 0:
break
time.sleep(min(args.interval, remaining))
if __name__ == '__main__':
main()
tools/source_bundle.py
+37 -5
View File
@@ -16,6 +16,7 @@ sys.dont_write_bytecode = True
from source_manifest import addon_manifest
from project_schema import parse_project
from resource_references import hard_resource_references
from external_addons import policy as external_policy, direct_for, resource_allowed, ALLOWED
from release_check import audit, read_json, evidence_path
@@ -24,7 +25,15 @@ BASE = '58D0FB3206B6F859'
PROJECTS = {'core': ('ERR_Core', '2528466DE220D91C', ())}
DIRECTORIES = {'Scripts', 'Prefabs', 'Configs', 'Missions', 'Worlds', 'UI', 'Materials', 'AI'}
SUFFIXES = {'.c', '.et', '.conf', '.ent', '.layer', '.layout', '.emat', '.bt', '.meta', '.gproj'}
SUFFIXES = {'.c', '.et', '.conf', '.ent', '.layer', '.layout', '.emat', '.bt', '.meta', '.gproj', '.imageset', '.png', '.edds'}
# Locally authored atlas, with its editable FBX and native import result.
# Keep binary-model admission scoped to this reviewed asset, not arbitrary Assets/.
ATLAS_ROOT = 'addons/core/Assets/Items/ERR_Atlas/'
ATLAS_FILES = {name + suffix for name in ('ERR_Atlas', 'ERR_AtlasLeaf')
for suffix in ('.fbx', '.txo', '.xob', '.xob.meta')}
ATLAS_FILES.update('AtlasTurn' + suffix for suffix in ('.agr', '.agr.meta', '.agf', '.agf.meta'))
ATLAS_FILES.update(name + suffix for name in ('AtlasPrint', 'AtlasPaper', 'AtlasCover', 'AtlasWire')
for suffix in ('.emat', '.emat.meta'))
EXCLUDED = {'.git', '.build', '.local', '__pycache__', 'profiles', 'profile', 'cache',
'runtime', 'evidence', 'vendor', 'vanilla', 'original', 'node_modules', '.agents', '.codex'}
README = '''# DEVELOPMENT SOURCE BUNDLE — Co10 Escape Reintroduced
@@ -41,7 +50,9 @@ Open addons/core/addon.gproj and use Missions/ERR_Everon.conf.
Scenario: {C1E20667DCE324ED}Missions/ERR_Everon.conf
Use the installed engine's normal project/scenario workflow. This archive has not
been compiled by its bundler; no native caches or binaries are included.
been compiled by its bundler; no native caches or executable binaries are included.
Authored UI textures include their PNG sources and Workbench-imported EDDS files.
The authored atlas includes FBX/TXO model sources and its imported XOB resource.
Target: Everon, up to 12 players. The mod requires the listed external addons. This remains development source.
Manual prison/scavenging/radio/extraction play, remote joins/reconnects and
@@ -105,6 +116,8 @@ def inventory(root):
folded.add(name.casefold())
if parts[2] == 'addon.gproj' and len(parts) == 3:
pass
elif name.startswith(ATLAS_ROOT) and name[len(ATLAS_ROOT):] in ATLAS_FILES:
pass
elif len(parts) < 4 or parts[2] not in DIRECTORIES or path.suffix not in SUFFIXES - {'.gproj'}:
raise ValueError(f'unreviewed addon source/resource format: {name}')
result[name] = digest
@@ -143,7 +156,26 @@ def validate(payload, root):
suffix = PurePosixPath(name).suffix
if suffix == '.meta':
continue
text = data.decode('utf-8') # Current reviewed handoff contains text assets only.
if name.startswith(ATLAS_ROOT) and suffix in {'.fbx', '.xob', '.txo'}:
stem = str(PurePosixPath(name).with_suffix(''))
if any(stem + required not in payload for required in ('.xob', '.xob.meta', '.fbx')):
raise ValueError(f'atlas model lacks source/runtime/metadata: {name}')
signatures = {'.fbx': b'Kaydara FBX Binary', '.xob': b'FORM', '.txo': b'$object'}
if not data.startswith(signatures[suffix]) or (suffix == '.xob' and data[8:12] != b'XOB9'):
raise ValueError(f'invalid atlas model signature: {name}')
continue
if suffix in {'.png', '.edds'}:
if not name.startswith('addons/core/UI/Textures/'):
raise ValueError(f'image outside authored UI textures: {name}')
signature = b'\x89PNG\r\n\x1a\n' if suffix == '.png' else b'DDS '
if not data.startswith(signature):
raise ValueError(f'invalid UI texture signature: {name}')
# Workbench attaches PNG import settings to the EDDS runtime resource.
runtime = str(PurePosixPath(name).with_suffix('.edds'))
if runtime not in payload or runtime + '.meta' not in payload:
raise ValueError(f'UI texture lacks imported runtime/metadata: {name}')
continue
text = data.decode('utf-8')
if re.search(r'\b(?:ESCT\w*|SCRTC_\w+|SCRT_\w+|EAI_\w+)\b', text):
raise ValueError(f'original ecosystem dependency: {name}')
if suffix not in {'.c', '.layer', '.gproj'} and name + '.meta' not in payload:
@@ -151,13 +183,13 @@ def validate(payload, root):
owner = name.split('/')[1]
visible = {owner, *PROJECTS[owner][2]}
if 'core' in visible: visible.update(ALLOWED)
for guid, resource in re.findall(r'\{([A-F0-9]{16})\}([^"\n]+)', text):
for guid, resource in hard_resource_references(name, text):
if guid in resources:
path, target = resources[guid]
if path != resource or target not in visible:
raise ValueError(f'resource outside project dependency closure: {name}')
elif vanilla.get(resource) != guid and not resource_allowed(external,visible,guid,resource):
raise ValueError(f'unresolved or nonvanilla resource: {name}')
raise ValueError(f'unresolved or nonvanilla resource: {name}: {{{guid}}}{resource}')
return projects
tools/terrain_adapter.py
+101
View File
@@ -0,0 +1,101 @@
#!/usr/bin/env python3
"""Export a terrain scenario using Escape's universal TerrainConnector prefab.
Copy the exported Missions/ and Worlds/ into the terrain author's addon and add
Escape as a dependency. --standalone creates a separate addon instead. Neither
mode modifies the terrain source or generates navigation.
"""
import argparse, hashlib, json, math, re
from pathlib import Path
ROOT=Path(__file__).resolve().parents[1]
CORE='2528466DE220D91C'; BASE='58D0FB3206B6F859'
CONNECTOR='{8FB1559AE3F29722}Prefabs/ERR/TerrainConnector.et'
PROJECTS=('Soldiers','BTRlike','LowRes')
RESOURCE=re.compile(r'\{([A-F0-9]{16})\}([A-Za-z0-9_./-]+)')
def guid(label):return hashlib.sha256(('ERR_TERRAIN_V1:'+label).encode()).hexdigest()[:16].upper()
def resource(value,suffix):
m=RESOURCE.fullmatch(value) if isinstance(value,str) else None
if not m or any(part in ('', '.', '..') for part in m[2].split('/')) or not m[2].endswith(suffix):raise ValueError('Invalid resource: '+str(value))
return value
def validate(profile):
if not isinstance(profile,dict):raise ValueError('Terrain profile must be an object')
if set(profile)-{'id','title','addon','world','navigation','ai_world','inherit_ai_world','airfields','notes'}:raise ValueError('Unknown terrain profile fields')
if not re.fullmatch(r'[a-z][a-z0-9_]{0,39}',profile.get('id','')):raise ValueError('Invalid terrain id')
if not re.fullmatch(r'[A-Za-z0-9 _-]{1,48}',profile.get('title','')):raise ValueError('Invalid terrain title')
if not re.fullmatch('[A-F0-9]{16}',profile.get('addon','')):raise ValueError('Invalid terrain addon GUID')
resource(profile.get('world'),'.ent')
nav=profile.get('navigation')
if not isinstance(nav,dict) or set(nav)-set(PROJECTS):raise ValueError('Unknown navigation project')
for files in nav.values():
if not isinstance(files,list) or not len(files)==1:raise ValueError('Navigation needs one .nmn file or native .conf per project')
for f in files:resource(f,('.nmn','.conf'))
if 'ai_world' in profile:
resource(profile['ai_world'],'.et')
if nav:raise ValueError('Use a terrain AI world prefab OR navigation resources, not both')
if type(profile.get('inherit_ai_world',False)) is not bool:raise ValueError('inherit_ai_world must be a boolean')
if profile.get('inherit_ai_world') and (nav or profile.get('ai_world')):raise ValueError('Inherited terrain AI world must not be duplicated by adapter navigation')
anchors=profile.get('airfields',[])
if not isinstance(anchors,list) or len(anchors)>16:raise ValueError('At most 16 airfield centres are supported')
for anchor in anchors:
if not isinstance(anchor,list) or len(anchor)!=3 or any(type(v) not in (int,float) or not math.isfinite(v) or abs(v)>1000000 for v in anchor):raise ValueError('Airfield centres must be finite [x,y,z] coordinates')
return profile
def meta(path,identity,kind='ResourceClass'):
return f'MetaFileClass {{\n Name "{{{identity}}}{path}"\n Configurations {{\n {kind} PC {{}}\n {kind} HEADLESS : PC {{}}\n }}\n}}\n'
def files(profile, standalone=False):
validate(profile);name=profile['id'];title=profile['title'];identity=guid(name+':addon')
world=f'Worlds/ERR_{name}.ent';mission=f'Missions/ERR_{name}.conf'
nav=profile['navigation'];ready=profile.get('inherit_ai_world',False) or bool(profile.get('ai_world')) or all(p in nav for p in PROJECTS)
deps=list(dict.fromkeys([BASE,CORE,profile['addon']]))
result={}
if standalone:
result['addon.gproj']='GameProject {\n ID "ERR_Terrain_'+name+'"\n GUID "'+identity+'"\n TITLE "Escape - '+title+'"\n Dependencies {\n'+''.join(' "'+d+'"\n' for d in deps)+' }\n}\n'
result[world]='SubScene {\n Parent "'+profile['world']+'"\n}\n'
result[world+'.meta']=meta(world,guid(name+':world'),'ENTResourceClass')
result[mission]=f'SCR_MissionHeader {{\n World "{{{guid(name+":world")}}}{world}"\n SystemsConfig "{{C50579A2EC48E8EF}}Configs/Systems/MissionSystems.conf"\n m_sName "Co10 Escape Reintroduced - {title}"\n m_sAuthor "Escape Reintroduced"\n m_sDescription "Escape captivity, gather intelligence, find the radio and flee by helicopter."\n m_sGameMode "Escape"\n m_iPlayerCount 12\n m_eSaveTypes 0\n m_bShowInScenarioMenu {int(ready)}\n}}\n'
result[mission+'.meta']=meta(mission,guid(name+':mission'))
# All maps reference the same prefab. Only navigation and airport hints vary.
layer='ERR_GameMode Reintroduced : "'+CONNECTOR+'" {}\n'
ai_world=profile.get('ai_world','{E0A05C76552E7F58}Prefabs/AI/SCR_AIWorld.et')
ai='SCR_AIWorld AIWorld : "'+ai_world+'" {\n'
if nav:
ai+=' components {\n'
components=['5584F30E67F617AD','5584F30EEFEE1223','5C8C9B750D124A63']
settings=['NavmeshWorld','ChimeraNavmeshWorld','NavmeshWorld']
settings_ids=['50FC63BEBE3973C5','50FC63BEBE3973C5','5C8C9B750B60C6E2']
config_ids=['60CA9BE5536BF701','60CA9BE5536BF701','5C90BD0EC793647D']
for i,project in enumerate(PROJECTS):
if project not in nav:continue
navfile=nav[project][0]
config=(f': "{navfile}" {{}}' if navfile.endswith('.conf') else f'{{\n NavmeshFile "{navfile}"\n }}')
ai+=f' NavmeshWorldComponent "{{{components[i]}}}" {{\n NavmeshSettings {settings[i]} "{{{settings_ids[i]}}}" {{\n NavmeshFilesConfig BaseNavmeshFilesConfig "{{{config_ids[i]}}}" {config}\n }}\n }}\n'
ai+=' }\n'
ai+='}\n'
if profile.get('inherit_ai_world'):ai=''
layer+=ai
for index,anchor in enumerate(profile.get('airfields',[])):
coords=' '.join(format(v,'.9g') for v in anchor)
layer+=f'GenericEntity ERR_Airfield_{index} {{\n coords {coords}\n}}\n'
result[f'Worlds/ERR_{name}_Layers/default.layer']=layer
result['terrain-adapter.json']=json.dumps({'schema':2,'profile':profile,'mode':'standalone' if standalone else 'embedded','addon':identity if standalone else profile['addon'],'required_dependencies':deps if standalone else [BASE,CORE],'connector':CONNECTOR,'scenarioId':'{'+guid(name+':mission')+'}'+mission,'world':world,'navigation_configured':ready,'qualification':'unverified'},indent=2)+'\n'
return result
def generate(profile,destination,standalone=False):
rendered=files(profile,standalone=standalone);destination=Path(destination)
if destination.exists():raise ValueError('Output already exists; use a new destination to preserve authored/navigation data')
destination.mkdir(parents=True)
for name,content in rendered.items():
path=destination/name;path.parent.mkdir(parents=True,exist_ok=True);path.write_text(content)
return json.loads(rendered['terrain-adapter.json'])
def main():
p=argparse.ArgumentParser(description=__doc__);p.add_argument('profile',type=Path);p.add_argument('--output',type=Path,required=True)
p.add_argument('--standalone',action='store_true',help='Create a separate addon project instead of files to include in the terrain addon')
a=p.parse_args()
try:report=generate(json.loads(a.profile.read_text()),a.output,standalone=a.standalone)
except (OSError,ValueError) as e:p.error(str(e))
print(json.dumps(report,indent=2))
if not a.standalone:print('Copy Missions/ and Worlds/ into the terrain addon, add Escape '+CORE+' to its project dependencies, then publish that terrain. Do not also load a standalone adapter with the same scenario resources.')
if not report['navigation_configured']:print('Navigation references missing: supply the published terrain addon’s AI-world prefab or its Soldiers/BTRlike/LowRes resources before running Escape.')
if __name__=='__main__':main()
tools/welcome_voice.py
+329
View File
@@ -0,0 +1,329 @@
#!/usr/bin/env python3
"""Generate Google welcome speech and serve cached, authenticated briefings.
Credentials use Google's Application Default Credentials and never enter the
addon or HTTP response. Public playback never calls the paid synthesis API;
the optional authenticated bridge generates and caches terrain/weather text.
"""
import argparse
import base64
import io
import json
import os
import hashlib
import hmac
import re
import threading
import time
from pathlib import Path
import tempfile
import wave
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from urllib.parse import urlsplit
WELCOME = ("Welcome to Escape Reintroduced. Work together, stay alert, "
"and find your way off the island. Good luck.")
ENDPOINT = "https://texttospeech.googleapis.com/v1/text:synthesize"
MAX_AUDIO = 2 * 1024 * 1024
VETERAN_PROMPT = ("Speak as a fictional older war veteran, about sixty-five: a weathered, "
"slightly gravelly male voice, calm authority and dry warmth. "
"A concise field briefing to fellow soldiers, not a movie trailer. "
"Use a steady natural pace, short pauses, no shouting, music or sound effects. "
"Read only the supplied text, exactly; do not add details.")
class SpeechHTTPError(RuntimeError):
def __init__(self, status):
self.status = status
super().__init__(f"Google speech request failed (HTTP {status}); no retry")
def validate_wav(data):
if not 44 <= len(data) <= MAX_AUDIO:
raise ValueError("Clip must be a WAV smaller than 2 MiB")
try:
with wave.open(io.BytesIO(data), "rb") as wav:
if (wav.getcomptype() != "NONE" or wav.getsampwidth() != 2
or wav.getnchannels() != 1
or not 8000 <= wav.getframerate() <= 48000
or not 0 < wav.getnframes() / wav.getframerate() <= 30):
raise ValueError("Expected mono 16-bit PCM, 8–48 kHz, up to 30 seconds")
expected = wav.getnframes() * 2
if len(wav.readframes(wav.getnframes())) != expected:
raise ValueError("Truncated audio")
except (wave.Error, EOFError) as exc:
raise ValueError("Invalid PCM WAV") from exc
return data
def synthesize(session, endpoint, voice, language, text=WELCOME, model="", prompt=""):
# Reject credentials in URLs: providers authenticate through ADC headers.
url = urlsplit(endpoint)
if url.scheme != "https" or not url.hostname or url.username or url.password or url.query or url.fragment:
raise ValueError("Speech endpoint must be an HTTPS URL without credentials or query parameters")
request = {"input": {"text": text},
"voice": {"languageCode": language, "name": voice},
"audioConfig": {"audioEncoding": "LINEAR16", "sampleRateHertz": 24000}}
if model:
request["voice"]["modelName"] = model
request["input"]["prompt"] = prompt
# Explicitly disable redirects to avoid forwarding credentials to another host.
with session.post(endpoint, json=request, timeout=60, stream=True, allow_redirects=False) as response:
if response.status_code != 200:
raise SpeechHTTPError(response.status_code)
chunks = []
size = 0
for chunk in response.iter_content(65536):
size += len(chunk)
if size > MAX_AUDIO * 2:
raise ValueError("Speech response too large")
chunks.append(chunk)
try:
body = json.loads(b"".join(chunks))
data = base64.b64decode(body["audioContent"], validate=True)
except (ValueError, KeyError, TypeError) as exc:
raise ValueError("Invalid speech response") from exc
return validate_wav(data)
def save_clip(path, data):
validate_wav(data)
path.parent.mkdir(parents=True, exist_ok=True)
with tempfile.NamedTemporaryFile(dir=path.parent, delete=False) as output:
temporary = Path(output.name)
try:
with temporary.open("wb") as output:
output.write(data)
output.flush()
os.fsync(output.fileno())
os.replace(temporary, path)
finally:
temporary.unlink(missing_ok=True)
class BriefingCache:
"""Authenticated server generation; public playback is cache-only."""
def __init__(self, directory, token, generate, identity, clock=time.monotonic):
if not 32 <= len(token) <= 256:
raise ValueError("Bridge token must contain 32–256 characters")
self.directory, self.token, self.generate = Path(directory), token, generate
self.identity, self.clock = identity, clock
self.lock = threading.Lock()
self.last_attempt = -float("inf")
self.directory.mkdir(parents=True, exist_ok=True)
def prepare(self, request):
if not isinstance(request, dict):
raise ValueError("Expected object")
token = request.get("token", "")
if not isinstance(token, str) or not hmac.compare_digest(token.encode(), self.token.encode()):
raise PermissionError("Unauthorized")
text = request.get("text")
if not isinstance(text, str) or not 1 <= len(text) <= 800 or any(ord(c) < 32 for c in text):
raise ValueError("Invalid briefing")
key = hashlib.sha256(json.dumps([self.identity, text], ensure_ascii=False).encode()).hexdigest()
path = self.directory / (key + ".wav")
with self.lock:
if path.exists():
validate_wav(path.read_bytes())
return key
# Limit billable work even for authorized requests; failed attempts count.
now = self.clock()
if now - self.last_attempt < 60:
raise BlockingIOError("Synthesis rate limited")
if len(list(self.directory.glob("*.wav"))) >= 256:
raise BlockingIOError("Cache full; remove unused cached clips")
self.last_attempt = now
save_clip(path, self.generate(text))
return key
def read(self, key):
if not re.fullmatch(r"[0-9a-f]{64}", key):
raise FileNotFoundError()
return validate_wav((self.directory / (key + ".wav")).read_bytes())
def read_request_body(headers, stream, limit=8192):
"""Read fixed-length or chunked HTTP bodies without trusting their sizes."""
transfer = headers.get("Transfer-Encoding", "").strip().lower()
length = headers.get("Content-Length")
if not transfer:
if length is None or not re.fullmatch(r"[0-9]+", length):
raise ValueError("Invalid size")
size = int(length)
if not 0 < size <= limit:
raise ValueError("Invalid size")
data = stream.read(size)
if len(data) != size:
raise ValueError("Invalid HTTP body framing")
return data
if transfer != "chunked" or length is not None:
raise ValueError("Invalid HTTP body framing")
data = bytearray()
framing = 0
while True:
line = stream.readline(130)
framing += len(line)
if not line.endswith(b"\r\n") or len(line) > 128 or framing > 4096:
raise ValueError("Invalid HTTP body framing")
size_text = line[:-2].split(b";", 1)[0]
if not re.fullmatch(rb"[0-9a-fA-F]{1,8}", size_text):
raise ValueError("Invalid HTTP body framing")
size = int(size_text, 16)
if size == 0:
while True:
trailer = stream.readline(1026)
framing += len(trailer)
if not trailer.endswith(b"\r\n") or len(trailer) > 1024 or framing > 4096:
raise ValueError("Invalid HTTP body framing")
if trailer == b"\r\n":
if not data:
raise ValueError("Invalid size")
return bytes(data)
if b":" not in trailer:
raise ValueError("Invalid HTTP body framing")
if len(data) + size > limit:
raise ValueError("Invalid size")
chunk = stream.read(size)
if len(chunk) != size or stream.read(2) != b"\r\n":
raise ValueError("Invalid HTTP body framing")
data.extend(chunk)
def handler_for(data, briefings=None):
validate_wav(data)
encoded = base64.b64encode(data)
class CachedClip(BaseHTTPRequestHandler):
def setup(self):
super().setup()
self.connection.settimeout(15)
def do_POST(self):
if self.path != "/" or briefings is None:
self.send_error(404)
return
try:
body = read_request_body(self.headers, self.rfile)
key = briefings.prepare(json.loads(body))
payload = json.dumps({"clip": key}).encode()
except PermissionError:
print("TTS request rejected: authentication failed (403)", flush=True)
self.send_error(403)
return
except (ValueError, TypeError) as error:
# Only fixed diagnostics: never print the request, token,
# provider response or arbitrary exception message.
reasons = {
"Invalid size": "invalid_body_size",
"Invalid HTTP body framing": "invalid_http_body_framing",
"Expected object": "json_object_required",
"Invalid briefing": "invalid_briefing_text",
"Clip must be a WAV smaller than 2 MiB": "audio_size_invalid",
"Expected mono 16-bit PCM, 8–48 kHz, up to 30 seconds": "audio_format_or_duration_invalid",
"Truncated audio": "audio_truncated",
"Invalid PCM WAV": "audio_format_invalid",
"Invalid speech response": "provider_response_invalid",
"Speech response too large": "provider_response_too_large",
}
reason = reasons.get(str(error), "invalid_data")
if isinstance(error, json.JSONDecodeError):
reason = "invalid_json"
print(f"TTS request rejected: {reason} (400)", flush=True)
self.send_error(400)
return
except BlockingIOError:
print("TTS request deferred: synthesis limit reached (429)", flush=True)
self.send_error(429)
return
except Exception as error:
reason = "service_or_provider_failure"
if isinstance(error, SpeechHTTPError):
reason = "provider_http_" + str(int(error.status))
print(f"TTS request failed: {reason} (503)", flush=True)
self.send_error(503) # Never expose provider errors or secrets.
return
self.send_response(200)
self.send_header("Content-Type", "application/json")
self.send_header("Content-Length", str(len(payload)))
self.end_headers()
self.wfile.write(payload)
print("TTS request completed (200)", flush=True)
def do_GET(self):
match = re.fullmatch(r"/clips/([0-9a-f]{64})\.b64", self.path)
if match and briefings:
try:
payload = base64.b64encode(briefings.read(match[1]))
except (OSError, ValueError):
self.send_error(404)
return
elif self.path in ("/welcome.wav", "/welcome.b64"):
payload = encoded if self.path == "/welcome.b64" else data
else:
self.send_error(404)
return
self.send_response(200)
self.send_header("Content-Type", "text/plain" if self.path.endswith(".b64") else "audio/wav")
self.send_header("Content-Length", str(len(payload)))
self.end_headers()
self.wfile.write(payload)
def log_message(self, *_):
pass # Do not log arbitrary request paths/headers.
return CachedClip
def main():
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--output", type=Path, required=True)
parser.add_argument("--input-wav", type=Path, help="Use an existing clip without any API request")
parser.add_argument("--regenerate", action="store_true", help="Replace a cached clip with one new synthesis request")
parser.add_argument("--project", help="Google Cloud quota project")
parser.add_argument("--voice", default="en-US-Chirp3-HD-Algenib")
parser.add_argument("--language", default="en-US")
parser.add_argument("--model", default="", help="Optional prompt-controlled model, e.g. gemini-2.5-flash-tts with --voice Algenib")
parser.add_argument("--text", default=WELCOME, help="Text for the standalone cached clip")
parser.add_argument("--briefing-cache", type=Path, help="Enable authenticated terrain/weather generation")
parser.add_argument("--bridge-token-file", type=Path, help="Secret shared with the game server; never sent to clients")
parser.add_argument("--endpoint", default=ENDPOINT)
parser.add_argument("--serve", action="store_true")
parser.add_argument("--bind", default="127.0.0.1")
parser.add_argument("--port", type=int, default=18947)
args = parser.parse_args()
if args.briefing_cache and (not args.serve or not args.bridge_token_file):
parser.error("--briefing-cache requires --serve and --bridge-token-file")
def generate(text):
import google.auth
from google.auth.transport.requests import AuthorizedSession
credentials, _ = google.auth.default(
scopes=["https://www.googleapis.com/auth/cloud-platform"], quota_project_id=args.project)
with AuthorizedSession(credentials, max_refresh_attempts=0) as session:
return synthesize(session, args.endpoint, args.voice, args.language, text, args.model, VETERAN_PROMPT)
if args.input_wav:
save_clip(args.output, args.input_wav.read_bytes())
elif args.regenerate or not args.output.exists():
save_clip(args.output, generate(args.text))
data = validate_wav(args.output.read_bytes())
print(f"Welcome clip ready ({len(data)} bytes). This is synthetic speech.")
if args.serve:
briefings = None
if args.briefing_cache:
briefings = BriefingCache(args.briefing_cache, args.bridge_token_file.read_text().strip(),
generate, [args.endpoint, args.voice, args.language, args.model, VETERAN_PROMPT])
with ThreadingHTTPServer((args.bind, args.port), handler_for(data, briefings)) as server:
print(f"Serving cached audio on port {args.port}; authenticated briefing generation={briefings is not None}.", flush=True)
server.serve_forever()
if __name__ == "__main__":
try:
main()
except KeyboardInterrupt:
pass
except SpeechHTTPError as exc:
raise SystemExit(str(exc)) from None
except Exception as exc:
# Provider/network exceptions can contain URLs or credential details.
raise SystemExit(f"Welcome voice failed ({type(exc).__name__}). Check credentials, project/API access, endpoint and WAV format.") from None
tools/welcome_voice_requirements.txt
+2
View File
@@ -0,0 +1,2 @@
google-auth>=2.40,<3
requests>=2.32,<3