Merge original HUD installation fixes and lossless setup optimisations

This commit is contained in:
chazmm committed 2026-09-09 01:59:43 +01:00
commit b88ed0b5fd
20 files changed
+694 -80

No files matched your search

+2 -1
View File
@@ -21,7 +21,8 @@ Editable examples live in `sdk/examples/`; `tools/package_mod.py` validates and
Extract the Windows release ZIP and run `skate3rust.exe`. Select your Skate 3
Xbox 360 ISO, or select `default.xex` in an extracted game folder. Keep its
`data` folder alongside it. Setup prepares the skater, animations and all disc maps, then
starts University. No Blender, Python or Rust installation is needed.
starts University. The original scoring and session-marker HUD assets are also
exported automatically during setup. No Blender, Python or Rust installation is needed.
ISO extraction needs internet access. The first conversion can take a while.
Use an XInput controller to play. Escape opens graphics, difficulty and map
+35 -5
View File
@@ -10,6 +10,16 @@ use skate_core::{
const SCORABLE: &str = "Hash_6918469984A8C596";
const TUNING: &str = "Hash_349215E2E817703C";
fn scorable_keys(data: &Collections) -> std::collections::BTreeSet<String> {
// Setup's name dictionary may resolve the class as `scoring_trick`.
// Presence checks must use the same native identities as Collections::field.
data.entries()
.iter()
.filter(|row| crate::attrib_hash::numeric_name(&row.class_name) == SCORABLE)
.map(|row| crate::attrib_hash::numeric_name(&row.key))
.collect()
}
#[derive(Clone, Debug)]
pub struct Definition {
pub metadata: Scorable,
@@ -53,14 +63,11 @@ impl ScoringData {
}
pub fn load(data: &Collections) -> Result<Self, String> {
let mut definitions = Vec::new();
let available = scorable_keys(data);
for (id, &(identifier, class, score_type)) in IDENTIFIERS.iter().enumerate() {
// The executable enum includes unused entries without VLT records.
// Only actual records are eligible for recognition.
if !data.entries().iter().any(|r| {
r.class_name == SCORABLE
&& (r.key == identifier
|| r.key == crate::attrib_hash::numeric_name(identifier))
}) {
if !available.contains(&crate::attrib_hash::numeric_name(identifier)) {
continue;
}
let words = |name| data.words::<1>(SCORABLE, identifier, name).map(|v| v[0]);
@@ -132,6 +139,29 @@ impl ScoringData {
}
}
#[cfg(test)]
mod identity_tests {
use super::*;
#[test]
fn scoring_presence_accepts_readable_and_numeric_vault_names() {
let key = IDENTIFIERS[0].0;
let numeric_key = crate::attrib_hash::numeric_name(key);
for class in ["scoring_trick", SCORABLE, "0x6918469984a8c596"] {
for name in [key, numeric_key.as_str()] {
let data: Collections = serde_json::from_value(serde_json::json!({
"version": 1,
"collections": [{"class": class, "key": name, "parent": "",
"source": "fixture", "sha256": "", "fields": {}},
{"class": "unrelated", "key": "other", "parent": "",
"source": "fixture", "sha256": "", "fields": {}}]
})).unwrap();
assert_eq!(scorable_keys(&data), [numeric_key.clone()].into());
}
}
}
}
#[derive(Clone, Debug)]
pub struct CollectorTuning {
scalars: std::collections::BTreeMap<u16, f32>,
+2 -2
View File
@@ -211,9 +211,9 @@ fn setup(
},
));
commands.insert_resource(hud);
info!("Original scoring HUD loaded");
info!("Original scoring HUD loaded from {}", root.display());
}
Err(error) => error!("Original scoring HUD could not load: {error}"),
Err(error) => error!("Original scoring HUD could not load from {}: {error}. See docs/hud-installation.md", root.display()),
}
}
// Rasterize at output pixel resolution; retain the original 1280x720 APT
+2 -1
View File
@@ -77,7 +77,7 @@ fn load(
Ok(h) => h,
Err(e) => {
warn!(
"Original session-marker HUD unavailable at {}: {e}",
"Original session-marker HUD unavailable at {}: {e}. See docs/hud-installation.md",
folder.display()
);
return;
@@ -123,6 +123,7 @@ fn load(
error!("Invalid marker HUD geometry");
return;
}
info!("Original session-marker HUD loaded from {}", folder.display());
commands
.spawn((MarkerHudRoot, Transform::default(), Visibility::Hidden))
.with_children(|parent| {
+54
View File
@@ -0,0 +1,54 @@
# Original HUD installation
The runtime reads both caches from the selected `--assets` directory:
| HUD | Default manifest | Optional development override |
| --- | --- | --- |
| Trick scoring | `private/hud/runtime/trickdisplay.json` | `SKATE_SCORING_HUD_ROOT` |
| Session marker | `private/session-marker/hud.json` | `SKATE3_SESSION_MARKER_OVERLAY` |
Merging HUD code does not install ignored, extracted game assets. Earlier feature
launchers used private caches through environment overrides; the combined build
with only `--assets` could not find either cache. Setup now prepares both with
the existing extractors. The marker extractor defaults to the vendored UI toolkit;
the external `--ui-toolkit` argument remains supported.
The original installer calls `prepare_runtime_huds.py` after VLT conversion.
It exports only the scoring movie and marker subtree needed by the runtime,
validates both, and installs their manifests and referenced RGBA payloads.
Intermediate extraction files remain in the existing conversion workspace and
are removed by the installer's normal cleanup. No full-disc or audio extraction
stage is added, and the original map/skater/environment setup flow is retained.
For an existing installation, reuse prepared caches without reconverting maps:
```powershell
python tools/install_prepared_hud.py --assets <installed-assets> --scoring-cache <prepared-scoring-cache> --marker-cache <prepared-marker-overlay>
```
This validates both manifests and referenced texture sizes before copying any
files, checks available marker texture hashes, and verifies byte-identical copies.
It refuses differing installed files unless `--replace` is supplied. It installs
only the runtime files, preserving authored geometry, fonts, images and actions.
No copyrighted assets are committed or bundled with the source.
If no prepared caches exist, prepare only the two HUDs from the owned game folder:
```powershell
python tools/prepare_hud.py --game <owned-game> --output <installed-assets>/private/hud --collections <installed-assets>/private/stock/skater-collections.json
python tools/extract_session_marker.py --game <owned-game> --output <installed-assets>/private/session-marker
```
Startup logs report the loaded paths or a load failure. Check the executable and
asset paths in the launcher as well: `PLAY.bat` delegates to `Launch.ps1`, which
selects `bin/skate3rust.exe`, not a feature build elsewhere in `bin`.
Session-marker prompts appear while holding LB during gameplay; they are not an
always-visible panel. Scoring follows the original trickdisplay movie. Its camera
renders layer 31 offscreen before the presentation UI; the marker camera renders
layer 29 at order 1 after presentation. The existing scoring startup dependency
on `PresentationSetup` must be retained. Neither HUD uses the world mesh layers.
The installation repair and release compilation do not establish GPU visibility
or gameplay parity. Manual checks remain: resume gameplay, hold LB for the marker
panel, and perform and land a trick for the scoring display.
+48
View File
@@ -0,0 +1,48 @@
# Installer performance and fidelity
Setup keeps the same asset conversion formats, lossless compression settings,
map validation, and publication checks. The optimizations remove repeated work:
- SFIL padding checks use a view of the input instead of copying the remaining
stream for each asset. District cells share the same ATOC lookup table.
- Map preparation creates its output directories once. Direct installation
avoids temporary model/texture RX2 copies which the writer never read and
setup previously deleted. Standalone extraction still writes these sources
by default; simulation and irradiance sources remain available in both modes.
- Spawn selection consumes collision meshes during their existing decode pass,
retaining triangle order, floating-point operations, and tie behavior.
- Model serialization reads each mesh's arrays once. Prop transforms only read
arrays belonging to the selected mesh.
- The installation builds and validates its shared prop catalog once, then
passes a JSON snapshot to map workers. It lives inside the current conversion
workspace and is removed after successful installation.
- Two bounded texture-packing threads overlap lossless zlib compression, with
the original texture order and compression level. Map-worker limits retain
the existing CPU and available-memory checks.
Map entries in `maps.json` now include `phase_seconds` for extraction,
preparation, collision packaging, map writing, prop export, validation, and
hashing/cleanup. These are diagnostic timings, not runtime asset inputs.
## Verification
A Windows x64 comparison on 2026-09-09 used packaged installers, the same
extracted owned disc, all ten maps, three map workers, and the same runtime
executable. A full installation took 117.8 seconds before and 91.5 seconds after
the changes. A subsequent baseline run took 103.0 seconds, putting the observed
reduction at 11–22%, depending on the baseline run. These are local warm-cache
measurements; ISO unpacking and downloading the ISO tool were outside them.
The comparison used the same output path for both runs and checked every file
under `assets`, `maps`, and `settings`: no missing or extra files; 2,231 files
were byte-identical, including all ten maps. Five prop packages had identical
binary payloads and decoded metadata, with differences only in JSON object key
order. Prop texture metadata is now sorted to make its order deterministic.
Timing records and logs were excluded from asset equality checks.
Both complete installations passed every existing runtime asset check. The
asset-pipeline suite passed 48 tests, with one unrelated owned-customisation
fixture skipped because it is opt-in. New fixtures cover serial/parallel texture
byte equality, cubemap orientation, compressible and incompressible textures,
worker failures, spawn ties, catalog matrix precision, selective mesh reads,
and malformed or incomplete SFIL input.
+28 -4
View File
@@ -97,7 +97,8 @@ def template_meshes(raw):
def transform_mesh(arrays, index, transform):
values = {key: np.array(value, copy=True) for key, value in arrays.items() if key.endswith('_'+str(index))}
# NpzFile.items() reads every array before the filter runs.
values = {key: np.array(arrays[key], copy=True) for key in arrays if key.endswith('_'+str(index))}
key = f'vertices_{index}'
values[key] = (values[key] @ transform[:3, :3] + transform[3, :3]).astype('f4')
normal_matrix = np.linalg.inv(transform[:3, :3]).T
@@ -148,9 +149,31 @@ def catalog(cache_roots):
return templates, textures
def export(manifest_path, cache_roots, output):
district = json.loads(manifest_path.read_text())
def save_catalog(cache_roots, output):
"""Build once per installation; JSON keeps this local cache non-executable."""
templates, textures = catalog(cache_roots)
serialised = {key: dict(value, matrix=value['matrix'].tolist(),
model_matrix=value['model_matrix'].tolist(), npz=str(value['npz'].resolve()))
for key, value in templates.items()}
temporary = output.with_suffix('.new')
temporary.write_text(json.dumps(dict(version=1, templates=serialised, textures=textures)), encoding='utf-8')
temporary.replace(output)
def load_catalog(path):
data = json.loads(path.read_text(encoding='utf-8'))
if data['version'] != 1:
raise ValueError('Unsupported DMO catalog version')
for value in data['templates'].values():
value['matrix'] = np.array(value['matrix'], dtype=np.float64)
value['model_matrix'] = np.array(value['model_matrix'], dtype=np.float64)
value['npz'] = Path(value['npz'])
return data['templates'], data['textures']
def export(manifest_path, cache_roots, output, *, catalog_path=None):
district = json.loads(manifest_path.read_text())
templates, textures = catalog(cache_roots) if catalog_path is None else load_catalog(catalog_path)
placements = {}
for source in district['simulation_assets']:
raw = (manifest_path.parent/source['rx2']).read_bytes()
@@ -180,7 +203,7 @@ def export(manifest_path, cache_roots, output):
report['instances'].append(dict(item, model_asset=template['asset_id'], meshes=len(meshes)))
if models:
manifest = dict(map_name=district['map_name'], district_name=district['district_name'],
models=models, textures={key:textures[key] for key in used},
models=models, textures={key:textures[key] for key in sorted(used)},
normal_texture_policy=dict(excluded_texture_ids=[]), grind_splines=[])
(root/'manifest.json').write_text(json.dumps(manifest))
# Publish only a complete package; preserve the previous one on failure.
@@ -209,6 +232,7 @@ def prepare_catalog(game_root, work):
prepare(stream_directory=stream, output_root=root, utt_root=vendor/'utt',
district_name=stream.name, map_name=stream.name, raw_texture_cache=True)
roots.append(root)
save_catalog(roots, work/'catalog.json')
return roots
+27 -5
View File
@@ -1,6 +1,6 @@
"""Local owned-disc installation. No game content is downloaded or packaged."""
from pathlib import Path
import hashlib,json,os,shutil,subprocess,sys,urllib.request,uuid,zipfile
import hashlib,json,os,shutil,subprocess,sys,time,urllib.request,uuid,zipfile
from concurrent.futures import ThreadPoolExecutor,as_completed
from tools.owned_game.big import BigArchive
from .vlt import convert as convert_vlt
@@ -98,38 +98,57 @@ def extract(archive,destination,entries=None):
return data
def convert_map(archive,work,maps,stage,game_exe,log,report):
timings={};started=time.perf_counter()
def finished(phase):
nonlocal started
now=time.perf_counter();timings[phase]=round(now-started,3);started=now
report(f'{archive.stem}: {phase} {timings[phase]:.3f}s')
map_tools=TOOLS/'vendor/university/tools/vanilla_map_extraction/tools'
sys.path.insert(0,str(map_tools))
from prepare_hawaiian_dream import prepare
from prepare_university import EXCLUDED_NORMAL_TEXTURE_IDS
from build_retail_collision_archive import build_archive
from .map_writer import write as write_map
from .map_writer import write as write_map, SpawnSelector
district=archive.stem.removeprefix('world')
label=district.removeprefix('DIST_')
district_work=work/district
extract(archive,district_work/'raw')
finished('extract')
stream=district_work/'raw/data/content/world/stream'/district
if not stream.is_dir():raise RuntimeError('Missing district stream '+str(stream))
spawn=SpawnSelector(district)
manifest_path=prepare(stream_directory=stream,output_root=district_work/'intermediate',
utt_root=TOOLS/'vendor/utt',district_name=district,map_name=label,
package_name='Skate 3 owned disc',cache_format='skate3-rust-map-v1',
# Smaller parks keep their textures in Pres rather than a Tex stream.
texture_stream_names=('Tex',) if any(stream.glob('cTex_*.xsf')) else (),
excluded_normal_texture_ids=EXCLUDED_NORMAL_TEXTURE_IDS,raw_texture_cache=True)
excluded_normal_texture_ids=EXCLUDED_NORMAL_TEXTURE_IDS,raw_texture_cache=True,
collision_consumer=spawn.consider,
# Model/texture RX2 copies are unused by the direct writer and were
# deleted after conversion. Keep simulation and irradiance sources.
write_render_sources=False)
finished('prepare')
collision=district_work/'collision.rwcmset'
build_archive(manifest_path,collision)
finished('collision_archive')
final=maps/(label+'.skate')
write_map(manifest_path,final,collision,report)
write_map(manifest_path,final,collision,report,prepared_spawn=spawn.result(label))
finished('write_map')
from .dynamic_props import export as write_props
caches=list((work/'dmo/cache').glob('DMO_*'))
if not caches:
raise RuntimeError('Missing prepared DMO catalog')
placed, unresolved=write_props(manifest_path,caches,stage/'assets/private/native-props'/(label+'.skate'))
placed, unresolved=write_props(manifest_path,caches,stage/'assets/private/native-props'/(label+'.skate'),
catalog_path=work/'dmo/catalog.json')
finished('props')
report(f'{label}: placed {placed} authored DMO instances, {unresolved} unresolved templates')
report('Checking converted map: '+label)
run([game_exe,'--assets',stage/'assets','--map',final,'--check-assets'],log,report)
finished('validate')
entry={'name':label,'path':'maps/'+final.name,'sha256':digest(final)}
remove_intermediate(district_work,work)
finished('hash_and_cleanup')
entry['phase_seconds']=timings
return entry
@@ -181,6 +200,9 @@ def install(iso,base,game_exe,report,game_root=None):
(stock/'skater-collections.json').write_text(json.dumps(converted),encoding='utf-8')
skeleton=convert_skeleton(stock/'data/anim/OnBoard.abin')
(stock/'physics-skeletons.json').write_text(json.dumps(skeleton),encoding='utf-8')
report('Preparing original scoring and session-marker HUD assets')
run(task(TOOLS/'prepare_runtime_huds.py', '--game', game_root,
'--assets', stage/'assets', '--work', work/'hud'), log, report)
report('Preparing the skater model and textures')
manifest=json.loads((TOOLS/'default_skater_retail_manifest.json').read_text())
needed=set()
+72 -34
View File
@@ -1,6 +1,8 @@
"""Direct retail cache -> SKATE14 writer. All geometry remains in Y-up metres."""
from pathlib import Path
import io,json,struct,sys,zlib
from collections import deque
from concurrent.futures import ThreadPoolExecutor
import numpy as np
from PIL import Image
from tools.asset_pipeline.retail_material import _retail_shader_family,_retail_render_flags
@@ -14,27 +16,49 @@ def stored(f,data):
method=1
if len(packed)>=len(data):method=0;packed=data
u(f,method,len(packed));f.write(packed)
def packed_texture(root,name,entry):
if 'rgba' in entry:
width,height=entry['width'],entry['height']
pixels=np.frombuffer((root/entry['rgba']).read_bytes(),dtype=np.uint8).reshape(height,width,4)
rgba=(pixels if entry.get('cube_faces')==6 else pixels[::-1]).tobytes()
else:
with Image.open(root/entry['png']) as source:
image=source.convert('RGBA')
if entry.get('cube_faces')!=6:image=image.transpose(Image.Transpose.FLIP_TOP_BOTTOM)
width,height=image.size;rgba=image.tobytes()
result=io.BytesIO();string(result,name);u(result,width,height,1);stored(result,rgba)
return result.getvalue()
def write_textures(output,root,textures):
# zlib releases the GIL. Bound outstanding work to two textures rather
# than retaining an entire district's decoded/compressed images.
names=iter(sorted(textures))
with ThreadPoolExecutor(max_workers=2) as pool:
pending=deque()
for _ in range(2):
name=next(names,None)
if name is not None:pending.append(pool.submit(packed_texture,root,name,textures[name]))
while pending:
output.write(pending.popleft().result())
name=next(names,None)
if name is not None:pending.append(pool.submit(packed_texture,root,name,textures[name]))
def normalise(a):return a/np.maximum(np.linalg.norm(a,axis=1,keepdims=True),1e-20)
def spawn_point(manifest,root):
from retail_collision_mesh import decode_rx2_clustered_meshes
university=manifest['district_name']=='DIST_University'
best=None
for entry in manifest['simulation_assets']:
if not entry.get('collision_meshes'):continue
# Bound the exact same search before decoding distant collision
# meshes. A triangle's centroid cannot be closer than its AABB.
bounds=[mesh['bounds'] for mesh in entry['collision_meshes']]
if university:
if not any(b['minimum'][0]<=330<=b['maximum'][0] and
b['minimum'][2]<=-710<=b['maximum'][2] for b in bounds):continue
elif best is not None:
def nearest_square(b):
return sum(max(b['minimum'][j],-b['maximum'][j],0.)**2 for j in (0,2))
if min(map(nearest_square,bounds))>best[0]+.01:continue
for mesh in decode_rx2_clustered_meshes((root/entry['rx2']).read_bytes()):
class SpawnSelector:
"""Keep the original triangle search order while consuming decoded meshes."""
def __init__(self, district_name):
self.university=district_name=='DIST_University'
self.best=None
def consider(self, meshes):
university=self.university;best=self.best
for mesh in meshes:
if not mesh.triangles:continue
if university and not (mesh.bounds_min[0]<=330<=mesh.bounds_max[0] and mesh.bounds_min[2]<=-710<=mesh.bounds_max[2]):continue
if not university and best is not None:
nearest=sum(max(mesh.bounds_min[j],-mesh.bounds_max[j],0.)**2 for j in (0,2))
if nearest>best[0]+.01:continue
triangles=np.asarray([(tri.a,tri.b,tri.c) for tri in mesh.triangles],dtype=np.float64)
a,b,c=triangles[:,0],triangles[:,1],triangles[:,2]
cross=np.cross(b-a,c-a);length=np.linalg.norm(cross,axis=1)
@@ -56,10 +80,30 @@ def spawn_point(manifest,root):
points[:,1]+=1.
index=np.argmin(scores);score=scores[index]
if np.isfinite(score) and (best is None or score<best[0]):best=(score,tuple(points[index]))
if best is None:raise ValueError('No supported spawn surface in '+manifest['map_name'])
return best[1]
self.best=best
def write(manifest_path,output,collision,report=lambda _:None, *, render_only=False):
def result(self, map_name):
if self.best is None:raise ValueError('No supported spawn surface in '+map_name)
return self.best[1]
def spawn_point(manifest,root):
from retail_collision_mesh import decode_rx2_clustered_meshes
selector=SpawnSelector(manifest['district_name'])
for entry in manifest['simulation_assets']:
if not entry.get('collision_meshes'):continue
bounds=[mesh['bounds'] for mesh in entry['collision_meshes']]
if selector.university:
if not any(b['minimum'][0]<=330<=b['maximum'][0] and
b['minimum'][2]<=-710<=b['maximum'][2] for b in bounds):continue
elif selector.best is not None:
def nearest_square(b):
return sum(max(b['minimum'][j],-b['maximum'][j],0.)**2 for j in (0,2))
if min(map(nearest_square,bounds))>selector.best[0]+.01:continue
selector.consider(decode_rx2_clustered_meshes((root/entry['rx2']).read_bytes()))
return selector.result(manifest['map_name'])
def write(manifest_path,output,collision,report=lambda _:None, *, render_only=False, prepared_spawn=None):
root=manifest_path.parent;m=json.loads(manifest_path.read_text());textures=m['textures']
ids={name:i+1 for i,name in enumerate(sorted(textures))}
excluded=set(m['normal_texture_policy']['excluded_texture_ids'])
@@ -68,9 +112,13 @@ def write(manifest_path,output,collision,report=lambda _:None, *, render_only=Fa
('mat','<u4'),('decal','<f4',(2,)),('frame','i1',(4,))])
for number,model in enumerate(m['models']):
if number%100==0:report(f"Writing {m['map_name']}: model {number+1}/{len(m['models'])}")
with np.load(root/model['npz'],allow_pickle=False) as arrays:
with np.load(root/model['npz'],allow_pickle=False) as archive:
for mesh in model['meshes']:
i=mesh['index'];pos=arrays[f'vertices_{i}'];faces=arrays[f'faces_{i}'].astype('<u4')
i=mesh['index']
# NpzFile does not cache reads. Load this mesh once, including
# attributes reused by normals, UVs and tangent conversion.
arrays={key:archive[key] for key in archive.files if key.endswith('_'+str(i))}
pos=arrays[f'vertices_{i}'];faces=arrays[f'faces_{i}'].astype('<u4')
if not len(pos) or not len(faces):continue
if faces.max()>=len(pos):raise ValueError('Map face is outside vertex array')
shader=mesh.get('shader_name') or '';roles=mesh.get('retail_texture_ids',{})
@@ -115,7 +163,7 @@ def write(manifest_path,output,collision,report=lambda _:None, *, render_only=Fa
record['frame'][:,:3]=np.rint(np.clip(binormal,-1,1)*127).astype('i1');record['frame'][:,3]=np.rint(sign*127).astype('i1')
if not np.isfinite(pos).all():raise ValueError('Non-finite map geometry')
vertices.write(record.tobytes());indices.write((faces+nv).astype('<u4').tobytes());nv+=len(pos);ni+=faces.size;nm+=1
report('Selecting starting position: '+m['map_name']);spawn=(0.,0.,0.) if render_only else spawn_point(m,root)
report('Selecting starting position: '+m['map_name']);spawn=(0.,0.,0.) if render_only else (prepared_spawn if prepared_spawn is not None else spawn_point(m,root))
# Match the supplied exporter's environment defaults. Native sky shaders
# remain a separate runtime feature; no geometry is synthesized here.
environment=[.10,.36,.75,.64,.82,1.,.18,.24,.30,0.,11.,0.,18.,0.,
@@ -125,17 +173,7 @@ def write(manifest_path,output,collision,report=lambda _:None, *, render_only=Fa
with output.open('wb') as f:
f.write(b'SKATE14\0');u(f,0x12345678);string(f,m['map_name']);floats(f,*spawn,0.,*environment)
u(f,nm,len(ids),nv,ni,0,len(rails),0,0,0);f.write(mats.getvalue())
for name in sorted(ids):
entry=textures[name]
if 'rgba' in entry:
width,height=entry['width'],entry['height']
pixels=np.frombuffer((root/entry['rgba']).read_bytes(),dtype=np.uint8).reshape(height,width,4)
rgba=(pixels if entry.get('cube_faces')==6 else pixels[::-1]).tobytes()
else:
image=Image.open(root/entry['png']).convert('RGBA')
if entry.get('cube_faces')!=6:image=image.transpose(Image.Transpose.FLIP_TOP_BOTTOM)
width,height=image.size;rgba=image.tobytes()
string(f,name);u(f,width,height,1);stored(f,rgba)
write_textures(f,root,textures)
stored(f,vertices.getvalue());stored(f,indices.getvalue());stored(f,b'')
for rail in rails:
string(f,f"{rail['asset_id']}_{rail['section_index']}_{rail['rail_index']}")
+31 -1
View File
@@ -1,8 +1,11 @@
import struct
import unittest
from unittest.mock import patch
from pathlib import Path
from tempfile import TemporaryDirectory
import numpy as np
from .dynamic_props import locators, transform_mesh, template_meshes
from .dynamic_props import locators, transform_mesh, template_meshes, save_catalog, load_catalog
def resource(kind, payload):
@@ -16,6 +19,33 @@ def resource(kind, payload):
class DynamicPropsTests(unittest.TestCase):
def test_transform_does_not_read_other_meshes(self):
class LazyArrays(dict):
def __getitem__(self, key):
if key.endswith('_1'):
raise AssertionError('Unrelated mesh was decompressed')
return super().__getitem__(key)
arrays = LazyArrays(vertices_0=np.zeros((3, 3)), faces_0=np.array([[0, 1, 2]]), vertices_1=None)
result = transform_mesh(arrays, 0, np.eye(4))
np.testing.assert_array_equal(result['faces_0'], [[0, 1, 2]])
def test_catalog_roundtrip_preserves_double_precision_and_bindings(self):
template = dict(matrix=np.arange(16, dtype=float).reshape(4, 4)/7,
model_matrix=np.eye(4), npz=Path('model.npz'),
meshes=[dict(retail_texture_ids={'diffuse': '0xf123456789abcdef'})],
asset_id='original', mesh_info=[112])
textures={'0xf123456789abcdef': dict(width=4, height=4, rgba='original.rgba')}
with TemporaryDirectory() as work:
path=Path(work)/'catalog.json'
with patch('tools.asset_pipeline.dynamic_props.catalog', return_value=({'template':template}, textures)):
save_catalog([], path)
templates, actual_textures=load_catalog(path)
actual=templates['template']
for key in ('matrix','model_matrix'):
self.assertEqual(actual[key].tobytes(),template[key].tobytes())
self.assertEqual(actual['meshes'],template['meshes'])
self.assertEqual(actual_textures,textures)
def test_locator_id_and_row_matrix(self):
payload = bytearray(166)
struct.pack_into('>5I', payload, 0, 0, 1, 1, 32, 160)
+63
View File
@@ -0,0 +1,63 @@
import io
import struct
import tempfile
import unittest
import zlib
from pathlib import Path
from types import SimpleNamespace
import numpy as np
from PIL import Image
from .map_writer import SpawnSelector, write_textures
class MapWriterTests(unittest.TestCase):
def test_parallel_textures_match_serial_format_exactly(self):
with tempfile.TemporaryDirectory() as work:
root=Path(work);textures={};expected=io.BytesIO()
rng=np.random.default_rng(23)
for name, cube, png in [('z-cube', True, False), ('a-rgba', False, False), ('m-png', False, True), ('b-flat', False, False)]:
pixels=rng.integers(0,256,(24 if cube else 4,4,4),dtype=np.uint8)
if name=='b-flat':pixels[:]=0
path=root/(name+('.png' if png else '.rgba'))
if png:Image.fromarray(pixels).save(path)
else:path.write_bytes(pixels.tobytes())
textures[name]=dict(width=4,height=len(pixels),cube_faces=6 if cube else 1,
**{'png' if png else 'rgba':path.name})
for name,entry in sorted(textures.items()):
if 'png' in entry:
with Image.open(root/entry['png']) as image:pixels=np.array(image.convert('RGBA'))
else:pixels=np.frombuffer((root/entry['rgba']).read_bytes(),dtype=np.uint8).reshape(entry['height'],4,4)
raw=(pixels if entry['cube_faces']==6 else pixels[::-1]).tobytes()
packed=zlib.compress(raw,1);method=1
if len(packed)>=len(raw):packed=raw;method=0
encoded=name.encode()
expected.write(struct.pack('<I',len(encoded))+encoded)
expected.write(struct.pack('<5I',4,entry['height'],1,method,len(packed))+packed)
actual=io.BytesIO();write_textures(actual,root,textures)
self.assertEqual(actual.getvalue(),expected.getvalue())
empty=io.BytesIO();write_textures(empty,root,{})
self.assertEqual(empty.getvalue(),b'')
def test_texture_failure_is_propagated(self):
with tempfile.TemporaryDirectory() as work:
with self.assertRaises(FileNotFoundError):
write_textures(io.BytesIO(),Path(work),{'missing':dict(rgba='missing',width=4,height=4)})
def test_spawn_streaming_preserves_ties_and_university_height(self):
def mesh(x,y,z):
return SimpleNamespace(bounds_min=(x-10,y,z-10),bounds_max=(x+10,y,z+10),triangles=[
SimpleNamespace(a=(x-10,y,z-10),b=(x,y,z+10),c=(x+10,y,z-10))])
selector=SpawnSelector('DIST_Test')
selector.consider([mesh(0,5,0),mesh(0,20,0)])
self.assertEqual(selector.result('test'),(0.,6.,-10/3))
selector.consider([mesh(500,30,500)])
self.assertEqual(selector.result('test'),(0.,6.,-10/3))
university=SpawnSelector('DIST_University')
university.consider([mesh(330,100,-710),mesh(330,132,-710),mesh(0,132,0)])
self.assertEqual(university.result('University'),(330.,133.,-710.))
with self.assertRaisesRegex(ValueError,'No supported spawn'):
SpawnSelector('DIST_Test').result('test')
if __name__=='__main__':unittest.main()
+51
View File
@@ -0,0 +1,51 @@
import struct
import sys
import unittest
from pathlib import Path
from unittest.mock import patch
sys.path.insert(0,str(Path(__file__).resolve().parents[1]/'vendor/university/tools/vanilla_map_extraction/tools'))
from skate3_streams import AssetRecord, StreamFormatError, read_sfil
class StreamTests(unittest.TestCase):
def fixture(self, tail=b'\0'*128):
raw=bytearray(384)
raw[:4]=b'SFIL';struct.pack_into('>I',raw,16,128)
struct.pack_into('>Q3I',raw,128,123,32,128,256)
raw[256:288]=bytes(range(32))
record=AssetRecord(123,0,32,128,128,0,4,b'')
return bytes(raw)+tail,record
def test_raw_assets_padding_and_shared_index(self):
class NoSuffixCopy(bytes):
def __getitem__(self,key):
if isinstance(key,slice) and key.start and key.stop is None:
raise AssertionError('Copied the entire remaining stream')
return super().__getitem__(key)
raw,record=self.fixture()
with patch.object(Path,'read_bytes',return_value=NoSuffixCopy(raw)):
normal=read_sfil('fixture.xsf',[record])
shared=read_sfil('fixture.xsf',[record],record_index={123:record})
self.assertEqual(normal,shared)
self.assertEqual(normal[0].data,bytes(range(32)))
self.assertEqual(normal[0].source_offset,128)
def test_truncated_nonzero_tail_is_still_rejected(self):
raw,record=self.fixture(b'\0\1')
with patch.object(Path,'read_bytes',return_value=raw):
with self.assertRaisesRegex(StreamFormatError,'truncated'):
read_sfil('fixture.xsf',[record])
def test_missing_and_unknown_records_are_still_rejected(self):
raw,record=self.fixture()
missing=AssetRecord(456,0,32,128,128,0,4,b'')
with patch.object(Path,'read_bytes',return_value=raw):
with self.assertRaisesRegex(StreamFormatError,'missing'):
read_sfil('fixture.xsf',[record,missing])
self.assertEqual(len(read_sfil('fixture.xsf',[record,missing],require_all_records=False)),1)
with self.assertRaisesRegex(StreamFormatError,'absent'):
read_sfil('fixture.xsf',[missing])
if __name__=='__main__':unittest.main()
+19 -12
View File
@@ -9,6 +9,7 @@ import hashlib
import json
from pathlib import Path
import sys
import importlib
# Presentation calibration against the supplied retail HUD captures. These are
# host compositor adjustments, not recovered APT constants. Keep the original
@@ -17,10 +18,11 @@ PANEL_OPACITY = 0.75
SHADOW_COVERAGE_GAMMA = 2.2
def compile_hud(cache_root: Path, output: Path) -> None:
from skate3_ui_extract.scene_graph import AssetCache, SceneFlattener, transform_point
from skate3_ui_extract.bitmap_font import measure_bitmap_text
from skate3_ui_extract.timeline import playback_frame
def compile_hud(cache_root: Path, output: Path, toolkit="vendor.skate3_ui") -> None:
scene = importlib.import_module(toolkit + ".scene_graph")
AssetCache, SceneFlattener, transform_point = scene.AssetCache, scene.SceneFlattener, scene.transform_point
measure_bitmap_text = importlib.import_module(toolkit + ".bitmap_font").measure_bitmap_text
playback_frame = importlib.import_module(toolkit + ".timeline").playback_frame
cache = AssetCache(cache_root)
bundle = "data/fe/source/screens/hud2/hudphonelist"
labels = json.loads((cache_root / "metadata/languages/english_global.json").read_text())
@@ -148,21 +150,26 @@ def compile_hud(cache_root: Path, output: Path) -> None:
def prepare(game: Path, output: Path, toolkit="vendor.skate3_ui"):
extract_project = importlib.import_module(toolkit + ".project").extract_project
cache = output / "source-cache"
extract_project(game, cache, include_dynamic=True, prefixes=(
"data/fe/source/screens/hud2/hudphonelist", "data/fe/source/controls/button_item2",
"data/fe/source/images/buttons/xbox360"), update=True)
compile_hud(cache, output, toolkit)
def main():
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--game", type=Path, required=True)
parser.add_argument("--ui-toolkit", type=Path, required=True)
parser.add_argument("--ui-toolkit", type=Path, help="Optional external toolkit; defaults to the vendored extractor")
parser.add_argument("--output", type=Path, required=True)
args = parser.parse_args()
if args.output.resolve().is_relative_to(args.game.resolve()):
parser.error("Output must be outside the owned game source")
sys.path.insert(0, str(args.ui_toolkit.resolve()))
from skate3_ui_extract.project import extract_project
cache = args.output / "source-cache"
extract_project(args.game, cache, include_dynamic=True, prefixes=(
"data/fe/source/screens/hud2/hudphonelist", "data/fe/source/controls/button_item2",
"data/fe/source/images/buttons/xbox360"), update=True)
compile_hud(cache, args.output)
if args.ui_toolkit:
sys.path.insert(0, str(args.ui_toolkit.resolve()))
prepare(args.game, args.output, "skate3_ui_extract" if args.ui_toolkit else "vendor.skate3_ui")
print(f"Private original session-marker HUD: {args.output / 'hud.json'}")
+90
View File
@@ -0,0 +1,90 @@
"""Install already extracted original HUD caches into an existing asset root.
Copies only runtime manifests and referenced RGBA files, byte for byte. No game,
disc extraction, map conversion or renderer is started. Both caches are checked
before writing; existing differing files require an explicit --replace.
"""
from pathlib import Path
import argparse
import hashlib
import json
import shutil
def digest(path):
return hashlib.sha256(path.read_bytes()).hexdigest()
def contained(root, relative):
path = (root / relative).resolve()
if not path.is_relative_to(root.resolve()) or path == root.resolve():
raise ValueError(f"HUD path escapes cache: {relative}")
return path
def runtime_files(root, marker=False):
manifest = "hud.json" if marker else "runtime/trickdisplay.json"
data = json.loads((root / manifest).read_text(encoding="utf-8"))
if data.get("version") != 1:
raise ValueError(f"Unsupported HUD version: {root}")
textures = []
if marker:
if data.get("canvas") != [1280, 720] or not data.get("meshes"):
raise ValueError("Invalid session-marker HUD")
textures = [(t["file"], t["width"], t["height"], t.get("output_sha256"))
for t in data["textures"]]
else:
if data.get("format") != "skate3-scoring-hud" or data.get("unresolved_fonts"):
raise ValueError("Invalid scoring HUD or unresolved original fonts")
for shape in data["shapes"].values():
for primitive in shape:
t = primitive["texture"]
textures.append((t["rgba"], t["width"], t["height"], None))
for font in data["fonts"].values():
size = font["definition"]["textures"][0]
textures.append((font["texture"], size["width"], size["height"], None))
files = {manifest: digest(root / manifest)}
for relative, width, height, expected in textures:
path = contained(root, relative)
if width <= 0 or height <= 0 or path.stat().st_size != width * height * 4:
raise ValueError(f"Invalid RGBA size: {path}")
actual = digest(path)
if expected and expected != actual:
raise ValueError(f"Original HUD texture hash mismatch: {path}")
files[relative] = actual
return files
def install(assets, scoring_cache, marker_cache, replace=False):
plan = []
for source, name, marker in [(scoring_cache, "hud", False),
(marker_cache, "session-marker", True)]:
destination = assets / "private" / name
for relative, sha256 in runtime_files(source, marker).items():
target = contained(destination, relative)
if target.exists() and digest(target) != sha256 and not replace:
raise ValueError(f"Different installed HUD file (use --replace): {target}")
plan.append((contained(source, relative), target, sha256))
for source, target, sha256 in plan:
if target.exists() and digest(target) == sha256:
continue
target.parent.mkdir(parents=True, exist_ok=True)
shutil.copyfile(source, target)
if digest(target) != sha256:
raise ValueError(f"Installed HUD hash mismatch: {target}")
return {str(target.relative_to(assets.resolve())): sha for _, target, sha in plan}
def main():
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--assets", type=Path, required=True)
parser.add_argument("--scoring-cache", type=Path, required=True)
parser.add_argument("--marker-cache", type=Path, required=True)
parser.add_argument("--replace", action="store_true")
args = parser.parse_args()
files = install(args.assets, args.scoring_cache, args.marker_cache, args.replace)
print(json.dumps({"assets": str(args.assets.resolve()), "files": files}, indent=2))
if __name__ == "__main__":
main()
+4 -4
View File
@@ -24,11 +24,12 @@ def font_mapping(collections: Path, cache: AssetCache) -> dict:
data = json.loads(collections.read_text(encoding='utf-8'))
result = {}
for row in data['collections']:
if row['class'] != 'Hash_FECFBCAF356518C4':
class_id = int(row['class'][5:], 16) if row['class'].startswith('Hash_') else hash64(row['class'])
if class_id != 0xFECFBCAF356518C4:
continue
def field(name):
return row['fields'].get(name, row['fields'].get(f'Hash_{hash64(name):016X}'))
apt_name = row['fields']['Hash_340832CCFD9FDEB4']['data']
apt_name = field('AptName')['data']
filename = field('FileName')['data']
matches = [r for r in cache.manifest['bundles']
if 'font' in r and Path(r['name']).name == filename]
@@ -48,8 +49,7 @@ def font_mapping(collections: Path, cache: AssetCache) -> dict:
def prepare(game: Path, output: Path, collections: Path):
extract_project(game, output, prefixes=("data/fe/source/screens/hud2/",
"data/fe/source/controls/"), update=True)
extract_project(game, output, prefixes=("data/fe/source/screens/hud2/trickdisplay2",), update=True)
cache = AssetCache(output)
mappings = font_mapping(collections,cache)
name = "data/fe/source/screens/hud2/trickdisplay2"
+31
View File
@@ -0,0 +1,31 @@
"""Export the two HUDs used by the engine, as one automatic setup stage.
Only runtime manifests and referenced RGBA payloads enter the installation.
Extraction intermediates stay in setup's conversion workspace and are removed
by its normal successful-install cleanup. No audio or full-disc extraction.
"""
import argparse
from pathlib import Path
from prepare_hud import prepare as prepare_scoring
from extract_session_marker import prepare as prepare_marker
from install_prepared_hud import install
def prepare(game, assets, work):
scoring = work / 'scoring'
marker = work / 'session-marker'
prepare_scoring(game, scoring, assets / 'private/stock/skater-collections.json')
prepare_marker(game, marker)
files = install(assets, scoring, marker)
print(f'Original runtime HUDs ready: {len(files)} verified files', flush=True)
return files
if __name__ == '__main__':
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument('--game', type=Path, required=True)
parser.add_argument('--assets', type=Path, required=True)
parser.add_argument('--work', type=Path, required=True)
args = parser.parse_args()
prepare(args.game, args.assets, args.work)
+37
View File
@@ -0,0 +1,37 @@
import json
from pathlib import Path
import tempfile
import unittest
from prepare_hud import font_mapping
from asset_pipeline.vlt import hash64
class Cache:
manifest = {'bundles': [{'name': 'fonts/original', 'font': {'family': 'Original family'}}]}
def add_font_alias(self, name, target, provenance):
self.alias = (name, target)
class FontIdentityTests(unittest.TestCase):
def test_readable_and_numeric_exports_resolve_the_same_original_font(self):
with tempfile.TemporaryDirectory() as directory:
path = Path(directory) / 'collections.json'
for named_class in (True, False):
for named_fields in (True, False):
fields = {'AptName': {'data': 'Original APT name'}, 'FileName': {'data': 'original'}}
fields.update({key: {'data': '3F800000'} for key in ('ScaleX', 'ScaleY', 'OffsetX', 'OffsetY')})
if not named_fields:
fields = {f'Hash_{hash64(key):016X}': value for key, value in fields.items()}
path.write_text(json.dumps({'collections': [{
'class': 'fe_fonts' if named_class else 'Hash_FECFBCAF356518C4',
'key': 'original', 'sha256': '', 'fields': fields}]}))
cache = Cache()
result = font_mapping(path, cache)
self.assertEqual(cache.alias, ('Original APT name', 'Original family'))
self.assertEqual(result['Original APT name']['native_layout']['ScaleX'], 1.0)
if __name__ == '__main__':
unittest.main()
+74
View File
@@ -0,0 +1,74 @@
import hashlib
import json
from pathlib import Path
import tempfile
import unittest
from install_prepared_hud import install
class HudInstallationTests(unittest.TestCase):
def setUp(self):
self.temp = tempfile.TemporaryDirectory()
self.addCleanup(self.temp.cleanup)
root = Path(self.temp.name)
self.assets, self.score, self.marker = [root / p for p in ("assets", "score", "marker")]
(self.score / "runtime").mkdir(parents=True)
self.marker.mkdir()
self.pixels = bytes([12, 34, 56, 78])
for cache in (self.score, self.marker):
(cache / "atlas.rgba").write_bytes(self.pixels)
self.score_json = {"format": "skate3-scoring-hud", "version": 1,
"shapes": {"1": [{"texture": {"rgba": "atlas.rgba", "width": 1, "height": 1}}]},
"fonts": {"native": {"texture": "atlas.rgba", "definition": {
"textures": [{"width": 1, "height": 1}]}}}}
(self.score / "runtime/trickdisplay.json").write_text(json.dumps(self.score_json))
self.marker_json = {"version": 1, "canvas": [1280, 720], "meshes": [{}],
"textures": [{"file": "atlas.rgba", "width": 1, "height": 1,
"output_sha256": hashlib.sha256(self.pixels).hexdigest()}]}
self.write_marker()
def write_marker(self):
(self.marker / "hud.json").write_text(json.dumps(self.marker_json))
def run_install(self, **kwargs):
return install(self.assets, self.score, self.marker, **kwargs)
def test_preserves_bytes_and_repeat_install_is_a_noop(self):
files = self.run_install()
before = {p: (self.assets / p).stat().st_mtime_ns for p in files}
self.assertEqual(files, self.run_install())
self.assertEqual(before, {p: (self.assets / p).stat().st_mtime_ns for p in files})
self.assertEqual((self.assets / "private/hud/atlas.rgba").read_bytes(), self.pixels)
self.assertEqual((self.assets / "private/session-marker/hud.json").read_bytes(),
(self.marker / "hud.json").read_bytes())
def test_invalid_second_cache_does_not_partially_install_first(self):
(self.marker / "atlas.rgba").write_bytes(b"bad")
with self.assertRaisesRegex(ValueError, "RGBA size"):
self.run_install()
self.assertFalse(self.assets.exists())
def test_hash_and_path_failures(self):
(self.marker / "atlas.rgba").write_bytes(b"fake")
with self.assertRaisesRegex(ValueError, "hash mismatch"):
self.run_install()
self.marker_json["textures"][0]["file"] = "../outside.rgba"
self.write_marker()
with self.assertRaisesRegex(ValueError, "escapes cache"):
self.run_install()
self.assertFalse(self.assets.exists())
def test_differing_existing_file_requires_replace(self):
self.run_install()
target = self.assets / "private/hud/atlas.rgba"
target.write_bytes(b"keep")
with self.assertRaisesRegex(ValueError, "Different installed"):
self.run_install()
self.assertEqual(target.read_bytes(), b"keep")
self.run_install(replace=True)
self.assertEqual(target.read_bytes(), self.pixels)
if __name__ == "__main__":
unittest.main()
@@ -90,10 +90,10 @@ def _default_stream_directory(workspace: Path) -> Path:
)
def _write_rx2(output_root: Path, category: str, asset: StreamAsset) -> Path:
def _write_rx2(output_root: Path, category: str, asset: StreamAsset, *, write: bool = True) -> Path:
target = output_root / "rx2" / category / f"{asset.record.asset_id:016X}.rx2"
target.parent.mkdir(parents=True, exist_ok=True)
target.write_bytes(asset.data)
if write:
target.write_bytes(asset.data)
return target
@@ -407,12 +407,19 @@ def prepare(
grind_coordinate_mode: str = "world_space",
excluded_static_model_asset_ids: tuple[str, ...] = (),
raw_texture_cache: bool = False,
collision_consumer=None,
write_render_sources: bool = True,
) -> Path:
sys.path.insert(0, str(utt_root))
import rx2_parser
from mdl_parser import parse_rx2 as parse_model
output_root.mkdir(parents=True, exist_ok=True)
# Create each output directory once, rather than thousands of failing
# mkdir calls against directories already created for previous assets.
for directory in ("rx2/textures", "rx2/models", "rx2/presentation_other",
"rx2/simulation", "textures/by_asset", "models"):
(output_root / directory).mkdir(parents=True, exist_ok=True)
presentation_assets = load_district_stream(
stream_directory,
"Pres",
@@ -476,7 +483,7 @@ def prepare(
asset_id = asset.record.asset_id
source_file = asset.source_path.name
if asset.record.asset_type == ASSET_TYPE_TEXTURE:
rx2_path = _write_rx2(output_root, "textures", asset)
rx2_path = _write_rx2(output_root, "textures", asset, write=write_render_sources)
base_texture_id = _texture_id(asset.data, asset_id)
parsed = rx2_parser.parse_rx2(asset.data)
if not parsed.textures:
@@ -507,7 +514,6 @@ def prepare(
/ "by_asset"
/ f"{asset_id:016X}_{texture_index}{'.rgba' if raw_texture_cache else '.png'}"
)
png_path.parent.mkdir(parents=True, exist_ok=True)
if raw_texture_cache:png_path.write_bytes(texture.rgba)
else:texture.save_png(png_path)
new_entry = {
@@ -556,14 +562,13 @@ def prepare(
known_digests.add(texture_digest)
textures[texture_id] = new_entry
elif asset.record.asset_type == ASSET_TYPE_MODEL:
rx2_path = _write_rx2(output_root, "models", asset)
rx2_path = _write_rx2(output_root, "models", asset, write=write_render_sources)
parsed = parse_model(
asset.data,
source_path=rx2_path,
strict=False,
)
npz_path = output_root / "models" / f"{asset_id:016X}.npz"
npz_path.parent.mkdir(parents=True, exist_ok=True)
arrays: dict[str, numpy.ndarray] = {}
mesh_entries: list[dict[str, object]] = []
source_material_groups = _group_material_parameters(parsed.materials)
@@ -791,6 +796,8 @@ def prepare(
rx2_path = _write_rx2(output_root, "simulation", asset)
asset_grinds = decode_grind_splines(asset.data)
asset_collision_meshes = decode_rx2_clustered_meshes(asset.data)
if collision_consumer is not None:
collision_consumer(asset_collision_meshes)
grind_spline_asset_count += bool(asset_grinds)
collision_mesh_asset_count += bool(asset_collision_meshes)
collision_mesh_count += len(asset_collision_meshes)
@@ -196,13 +196,15 @@ def read_sfil(
records: list[AssetRecord],
*,
require_all_records: bool = True,
record_index: dict[int, AssetRecord] | None = None,
) -> list[StreamAsset]:
source_path = Path(path)
source = source_path.read_bytes()
if len(source) < SFIL_HEADER_SIZE or source[:4] != b"SFIL":
raise StreamFormatError(f"{source_path} is not an SFIL stream")
by_id = {record.asset_id: record for record in records}
by_id = record_index if record_index is not None else {record.asset_id: record for record in records}
source_view = memoryview(source)
first_asset_offset = struct.unpack_from(">I", source, 16)[0]
if first_asset_offset < SFIL_HEADER_SIZE or first_asset_offset > len(source):
raise StreamFormatError(
@@ -214,10 +216,12 @@ def read_sfil(
seen_ids: set[int] = set()
while cursor < len(source):
if not any(source[cursor:]):
# bytes[cursor:] copied the entire remaining stream for every asset.
# A view preserves the same padding check without quadratic copying.
if not any(source_view[cursor:]):
break
if cursor + SFIL_SECTION_DATA_OFFSET > len(source):
if any(source[cursor:]):
if any(source_view[cursor:]):
raise StreamFormatError(f"truncated SFIL asset at 0x{cursor:X}")
break
@@ -288,7 +292,7 @@ def read_sfil(
)
cursor += asset_stride
missing = set(by_id) - seen_ids
missing = set(by_id) - seen_ids if require_all_records else set()
if require_all_records and missing:
missing_text = ", ".join(f"0x{asset_id:016X}" for asset_id in sorted(missing))
raise StreamFormatError(f"ATOC assets missing from SFIL: {missing_text}")
@@ -323,11 +327,13 @@ def load_district_stream(
)
assets_by_id: dict[int, StreamAsset] = {}
record_index = {record.asset_id: record for record in records}
for stream_file in stream_files:
for asset in read_sfil(
stream_file,
records,
require_all_records=False,
record_index=record_index,
):
asset_id = asset.record.asset_id
previous = assets_by_id.get(asset_id)