mirror of
https://github.com/tangerineturtle1/NX2navi.git
synced 2026-11-15 19:19:58 -04:00
fc6d84aa63
The shader reader now reads Switch 2 (NVN2) programs, whose 0xe0-byte control sections and task and mesh stages the runtime keys as it does, and runs on Python 3.12 through the zstandard package. The kit's new last step writes the 36,031 shaders of the player's game files as a seed and runs ocarina.exe --prepare-shaders, which translates them and has the GPU driver build each one's library, about 5 minutes. On a cold walk into play this took the stalls from 14 seconds of pipeline builds and 9 of translations to none. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
407 lines
20 KiB
Python
407 lines
20 KiB
Python
"""Census of every Maxwell shader in the extracted NX2navi RomFS and ExeFS, and an offline translation pass.
|
|
|
|
The inventory, the bnsh, sharcb and sarc readers and the package walk: Copyright (c) 2026 MacNativeGames contributors
|
|
(MIT; see THIRD_PARTY_NOTICES.md). This pass reads no shared shader list, runs no spirv-cross and makes no bind-state
|
|
variants; the module scan reads the NSO files under exefs/ directly;
|
|
a control section must start with CONTROL_MAGIC (sharcb_shaders says why).
|
|
|
|
Reads build/extracted/<version>/ (RomFS image plus index.json as tools/romfs_index.py reads them, and exefs/). Writes under
|
|
build/<version>/shaders/ only:
|
|
code/<key>.code translator input (graphics: SPH + instructions, compute: instructions)
|
|
jobs.json key -> job, source path, origin (bnsh|sharcb|main)
|
|
spv/<key>.spv with --translate
|
|
failures.txt with --translate, when something failed: groups by message
|
|
|
|
The key is sha256(control[:0x880] + code), the shared shader list's key. Needs Python 3.14 (compression.zstd), so run in .venv-py314:
|
|
|
|
.venv-py314\\Scripts\\python tools\\shader_census.py --version v121 --translate
|
|
"""
|
|
import argparse
|
|
import zstd_compat as zstd # compression.zstd on 3.14, the zstandard package before
|
|
import concurrent.futures
|
|
import hashlib
|
|
import json
|
|
import os
|
|
import re
|
|
import shutil
|
|
import struct
|
|
import subprocess
|
|
import sys
|
|
import time
|
|
from collections import defaultdict
|
|
from pathlib import Path
|
|
|
|
import lz4.block
|
|
|
|
ROOT = Path(__file__).resolve().parent.parent
|
|
XLATE = ROOT / 'build' / 'xlate' / ('nx2navi_xlate.exe' if os.name == 'nt' else 'nx2navi_xlate')
|
|
SPIRV_VAL = Path(r'C:\msys64\ucrt64\bin\spirv-val.exe') if os.name == 'nt' else Path(shutil.which('spirv-val') or '/usr/bin/spirv-val')
|
|
ZSTD = b'\x28\xb5\x2f\xfd'
|
|
# NVN program control section magic, u32 0x98761234
|
|
CONTROL_MAGIC = b'\x34\x12\x76\x98'
|
|
STAGE_IDS = {0: 'vs', 1: 'ps', 5: 'cs'}
|
|
MAX_WORKERS = 20
|
|
|
|
|
|
def digest(data):
|
|
return hashlib.sha256(data).hexdigest()
|
|
|
|
|
|
def checked(data, offset, size):
|
|
if offset < 0 or size < 0 or offset + size > len(data):
|
|
raise ValueError('shader container range outside file')
|
|
return data[offset:offset + size]
|
|
|
|
|
|
def sarc(data):
|
|
if data[:4] != b'SARC':
|
|
raise ValueError('expected SARC')
|
|
endian = '<' if data[6:8] == b'\xff\xfe' else '>'
|
|
header, size, start = struct.unpack_from(endian + 'HxxII', data, 4)
|
|
if size != len(data) or header < 20 or checked(data, header, 4) != b'SFAT':
|
|
raise ValueError('invalid SARC header')
|
|
count = struct.unpack_from(endian + 'H', data, header + 6)[0]
|
|
names = header + 12 + count * 16
|
|
if checked(data, names, 4) != b'SFNT':
|
|
raise ValueError('invalid SARC string table')
|
|
names += 8
|
|
result = {}
|
|
for i in range(count):
|
|
_, attribute, lo, hi = struct.unpack_from(endian + '4I', checked(data, header + 12 + i * 16, 16))
|
|
off = names + (attribute & 0xffffff) * 4
|
|
if attribute >> 24 != 1 or not names <= off < start or hi < lo or start + lo < start:
|
|
raise ValueError('invalid SARC member')
|
|
name = data[off:data.index(0, off, start)].decode('utf-8')
|
|
if name.startswith('/') or '\\' in name or any(p in ('', '.', '..') for p in name.split('/')) or name in result:
|
|
continue # names only key the members here, nothing is written under them: an odd or repeated one is skipped
|
|
result[name] = checked(data, start + lo, hi - lo)
|
|
return result
|
|
|
|
|
|
NVN2_STAGES = {0: 'vs', 1: 'ps', 5: 'cs', 6: 'ts', 7: 'ms'} # the stage number in bits 16+ of control +0x54
|
|
NVN2_CONTROL = 0xe0 # the part of an NVN2 control section the runtime keeps (runtime/gfx_shader.c, "NVN2")
|
|
NVN2_SLOTS = range(8, 96, 8) # program slots that may point at a stage record: fragment +40, task +64, mesh +72 seen
|
|
|
|
|
|
def nvn2_record(data, offset):
|
|
"""(control, code) of the stage record at offset when it is an NVN2 one: {u64, control offset, code offset,
|
|
u32 code size, u32 control size}, the control shorter than NVN1's 0x880 and starting with the magic. Else None."""
|
|
if not offset or offset + 32 > len(data):
|
|
return None
|
|
control_off, code_off, code_size, control_size = struct.unpack_from('<QQII', data, offset + 8)
|
|
if not NVN2_CONTROL <= control_size < 0x880 or control_off + control_size > len(data) or code_off + code_size > len(data):
|
|
return None
|
|
control = data[control_off:control_off + control_size]
|
|
return (control, data[code_off:code_off + code_size]) if control[:4] == CONTROL_MAGIC else None
|
|
|
|
|
|
def nvn2_program(data, program):
|
|
return any(nvn2_record(data, struct.unpack_from('<Q', data, program + r)[0]) for r in NVN2_SLOTS
|
|
if program + r + 8 <= len(data))
|
|
|
|
|
|
def nvn2_stages(data, program):
|
|
"""The stages of an NVN2 (Switch 2) program as the runtime keys them: the control section's first 0xe0 bytes in a
|
|
zeroed 0x880 one, and the code from its first byte up to the end of its c1 constants (+0x40 to +0x44, after the
|
|
code of +0x38 bytes), the c1 part included."""
|
|
for r in NVN2_SLOTS:
|
|
if program + r + 8 > len(data):
|
|
break
|
|
rec = nvn2_record(data, struct.unpack_from('<Q', data, program + r)[0])
|
|
if not rec:
|
|
continue
|
|
control, code = rec
|
|
stage = NVN2_STAGES.get(struct.unpack_from('<I', control, 0x54)[0] >> 16)
|
|
if not stage:
|
|
continue
|
|
control = control[:NVN2_CONTROL] + bytes(0x880 - NVN2_CONTROL)
|
|
size, c1_start, c1_end = struct.unpack_from('<I', control, 0x38)[0], *struct.unpack_from('<II', control, 0x40)
|
|
length = c1_end if c1_start >= size and c1_end >= c1_start and c1_end <= len(code) else size
|
|
if not 0 < length <= len(code):
|
|
continue
|
|
yield digest(control + code[:length]), stage, control, code[:length]
|
|
|
|
|
|
def bnsh_shaders(data):
|
|
"""BNSH absolute offsets are relative to each embedded BNSH, not BFSHA."""
|
|
if data[:4] != b'BNSH' or checked(data, 12, 2) != b'\xff\xfe' or checked(data, 96, 4) != b'grsc':
|
|
raise ValueError('unsupported BNSH header')
|
|
size = struct.unpack_from('<I', data, 28)[0]
|
|
data = checked(data, 0, size)
|
|
count, variations = struct.unpack_from('<IQ', data, 124)
|
|
checked(data, variations, count * 64)
|
|
for i in range(count):
|
|
program = struct.unpack_from('<Q', data, variations + i * 64 + 16)[0]
|
|
checked(data, program, 56)
|
|
if nvn2_program(data, program):
|
|
yield from nvn2_stages(data, program)
|
|
continue
|
|
for stage, relative in [('vs', 8), ('ps', 40), ('cs', 48)]:
|
|
offset = struct.unpack_from('<Q', data, program + relative)[0]
|
|
if not offset:
|
|
continue
|
|
control_off, code_off, code_size, control_size = struct.unpack_from('<QQII', checked(data, offset + 8, 24))
|
|
control = checked(data, control_off, control_size)
|
|
code = checked(data, code_off, code_size)
|
|
if len(control) < 0x880:
|
|
continue # the SDK utility BNSH uses an older, shorter control layout
|
|
control = control[:0x880] # some are 0x88c; NVN and the runtime's key read 0x880
|
|
if control[:4] != CONTROL_MAGIC:
|
|
raise ValueError('NVN control section without its magic')
|
|
c1, length = struct.unpack_from('<II', control, 0x700)
|
|
actual = struct.unpack_from('<I', control, 0x714)[0]
|
|
if actual != {'vs': 0, 'ps': 1, 'cs': 5}[stage] or not 0 <= c1 <= length <= len(code) or not length:
|
|
raise ValueError('invalid NVN shader bounds or stage')
|
|
yield digest(control + code[:length]), stage, control, code[:length]
|
|
|
|
|
|
def sharcb_shaders(data, rejected):
|
|
"""agl shader packages (.sharcb, magic BAHS): records <u64 code offset in the pool, u32 control size (0x880 or
|
|
0x88c), i32 offset of the control from the record>; a record may reuse an earlier record's control.
|
|
|
|
The scan finds records by their size word, so it also meets 16-byte groups that only look like one. Without the
|
|
magic check, 1.2.1 yields 87 such keys: each has control offset 0 (the "control" starts at the record itself, so its
|
|
first 8 bytes are the code offset) and no CONTROL_MAGIC. The game hands NVN real control sections and the runtime
|
|
key hashes those, so no draw can ever produce these keys. Their code is cut from the wrong place: they were all 67
|
|
translation failures of 1.2.1 (IndexOutOfRange in ShaderHeader.ctor and ArgumentOutOfRange in Decoder.FillBlock,
|
|
on 0 to 4,718 bytes of code). A scan without the magic check keeps them; the 7 that translate are
|
|
V rows in the shared shader list only because shader_list.py add_cache() folds the seed's own .variants files back in. A
|
|
candidate without CONTROL_MAGIC is skipped and counted in rejected[0]."""
|
|
if len(data) < 32 or data[:4] != b'BAHS':
|
|
raise ValueError('expected an agl shader package')
|
|
start, pool_size, pool = struct.unpack_from('<3I', data, 20)
|
|
if not 32 <= start <= pool or pool + pool_size > len(data):
|
|
raise ValueError('agl shader package layout outside the file')
|
|
seen = set()
|
|
for size in (b'\x80\x08\x00\x00', b'\x8c\x08\x00\x00'):
|
|
at = start
|
|
while (at := data.find(size, at, pool)) >= 0:
|
|
record, at = at - 8, at + 4
|
|
if record < start or record % 4:
|
|
continue
|
|
code_off = struct.unpack_from('<Q', data, record)[0]
|
|
control_at = record + struct.unpack_from('<i', data, record + 12)[0]
|
|
if not start <= control_at <= pool - 0x880 or code_off >= pool_size:
|
|
continue
|
|
control = data[control_at:control_at + 0x880]
|
|
if control[:4] != CONTROL_MAGIC:
|
|
rejected[0] += 1
|
|
continue
|
|
c1, length = struct.unpack_from('<II', control, 0x700)
|
|
stage = STAGE_IDS.get(struct.unpack_from('<I', control, 0x714)[0])
|
|
if not stage or not 0 < length <= pool_size - code_off or c1 > length:
|
|
continue
|
|
code = data[pool + code_off:pool + code_off + length]
|
|
key = digest(control + code)
|
|
if key not in seen:
|
|
seen.add(key)
|
|
yield key, stage, control, code
|
|
|
|
|
|
def nso_image(path):
|
|
"""The decompressed text, rodata and data segments of an NSO, concatenated (enough to find embedded BNSH files)."""
|
|
raw = path.read_bytes()
|
|
if raw[:4] != b'NSO0':
|
|
raise ValueError(f'{path} is not an NSO')
|
|
flags = struct.unpack_from('<I', raw, 0xC)[0]
|
|
parts = []
|
|
for i in range(3):
|
|
file_off, _, dec_size = struct.unpack_from('<III', raw, 0x10 + i * 0x10)
|
|
comp_size = struct.unpack_from('<I', raw, 0x60 + i * 4)[0]
|
|
data = raw[file_off:file_off + comp_size]
|
|
parts.append(lz4.block.decompress(data, uncompressed_size=dec_size) if flags & (1 << i) else data)
|
|
return b''.join(parts)
|
|
|
|
|
|
def startup_package(name):
|
|
"""The RomFS files the shared shader list's startup shaders come from; effects and UI layouts carry more."""
|
|
return (name == 'Lib/NintendoSDK/TextureShader.bnsh' or name == 'Pack/Bootup.Nin_NX_NVN.pack.zs' or
|
|
name.startswith('Shader/') and name.endswith('.zs'))
|
|
|
|
|
|
# The RomFS directories that can hold shaders: the agl and gsys packages, the effects, terrain, the UI layout archives
|
|
# and the game's own shader packs. Every shader of 1.1.0 (27,576) comes from these six; they are 1,996 entries and
|
|
# 147 MiB, 0.9% of a RomFS whose every file the walk below used to read and decompress first (65 s against 2.8 s).
|
|
SHADER_DIRS = ('Shader/', 'Effect/', 'Lib/agl/', 'Lib/Terrain/', 'Lib/gsys/', 'UI/LayoutArchive/')
|
|
|
|
|
|
def inventory(romfs, exefs, everything=True):
|
|
"""Every unique shader: key -> dict(stage, control, code, source, origin, startup). As in the shared shader list, a shader is a
|
|
startup shader when it comes from an agl package, a game module, or a startup_package() RomFS file; the others come
|
|
from effect and UI layout resources, which the list covers only for bind states players have met. Also returns the number
|
|
of .sharcb record candidates rejected for a missing control magic (see sharcb_shaders)."""
|
|
index = json.loads((romfs / 'index.json').read_text(encoding='utf-8'))
|
|
if index.get('status') != 'pass':
|
|
raise ValueError('RomFS has not passed integrity verification')
|
|
entries = {r['path']: r for r in index['files']}
|
|
shaders = {}
|
|
rejected = [0]
|
|
|
|
def add(origin, name, found, startup):
|
|
for key, stage, control, code in found:
|
|
s = shaders.setdefault(key, dict(stage=stage, control=control, code=code, source=name, origin=origin,
|
|
startup=startup))
|
|
if startup and not s['startup']:
|
|
s.update(source=name, origin=origin, startup=True)
|
|
|
|
with (romfs / 'romfs.bin').open('rb') as stream:
|
|
def read(name):
|
|
row = entries[name]
|
|
stream.seek(row['offset'])
|
|
raw = stream.read(row['size'])
|
|
if len(raw) != row['size']:
|
|
raise ValueError('short RomFS read')
|
|
return raw
|
|
|
|
dictionaries = {}
|
|
for raw in sarc(zstd.decompress(read('Pack/ZsDic.pack.zs'))).values():
|
|
d = zstd.ZstdDict(raw)
|
|
dictionaries[d.dict_id] = d
|
|
|
|
def decode(raw):
|
|
info = zstd.get_frame_info(raw)
|
|
if info.decompressed_size is None or info.decompressed_size > 512 << 20:
|
|
raise ValueError('unbounded shader package')
|
|
return zstd.decompress(raw, zstd_dict=dictionaries[info.dictionary_id] if info.dictionary_id else None)
|
|
|
|
def visit(origin, name, data, startup, depth=0):
|
|
if data[:4] == b'BAHS':
|
|
add('sharcb', name, sharcb_shaders(data, rejected), True)
|
|
return
|
|
if data[:4] == b'SARC':
|
|
if depth > 4:
|
|
return
|
|
for member, raw in sarc(data).items():
|
|
visit(origin, name + '/' + member, decode(raw) if raw[:4] == ZSTD else raw, startup, depth + 1)
|
|
return
|
|
pos = 0 # BFSHA and BNSH files, effect and layout resources: complete, independently addressed BNSH files
|
|
while (pos := data.find(b'BNSH\0\0\0\0', pos)) >= 0:
|
|
size = struct.unpack_from('<I', checked(data, pos + 28, 4))[0]
|
|
add(origin, name, bnsh_shaders(checked(data, pos, size)), startup)
|
|
pos += max(size, 8)
|
|
|
|
for name in sorted(entries):
|
|
if not everything and not name.startswith(SHADER_DIRS):
|
|
continue
|
|
raw = read(name)
|
|
if raw[:4] not in (ZSTD, b'SARC', b'FSHA', b'BNSH', b'BAHS') and b'BNSH\0\0\0\0' not in raw:
|
|
continue
|
|
visit('bnsh', name, decode(raw) if raw[:4] == ZSTD else raw, startup_package(name))
|
|
for module in sorted(exefs.iterdir()):
|
|
if module.name in ('main', 'sdk', 'subsdk0', 'rtld'):
|
|
visit('main', 'module:' + module.name, nso_image(module), True)
|
|
return shaders, rejected[0]
|
|
|
|
|
|
def write_shaders(shaders, out, everything):
|
|
"""Write code files and jobs.json; returns {key: dict(job, source, origin)}."""
|
|
(out / 'code').mkdir(parents=True, exist_ok=True)
|
|
jobs = {}
|
|
for key, s in sorted(shaders.items()):
|
|
if not s['startup'] and not everything:
|
|
continue
|
|
control, code, stage = s['control'], s['code'], s['stage']
|
|
c1, length = struct.unpack_from('<II', control, 0x700)
|
|
body = code[0 if stage == 'cs' else 0x30:length]
|
|
path = out / 'code' / f'{key}.code'
|
|
if not path.is_file() or path.read_bytes() != body:
|
|
path.write_bytes(body)
|
|
job = dict(key=key, stage=stage, code=str(path), c1=code[c1:length].hex(), debug=False)
|
|
if stage == 'cs':
|
|
job['local'] = list(struct.unpack_from('<III', control, 0x748))
|
|
jobs[key] = dict(job=job, source=s['source'], origin=s['origin'], startup=s['startup'])
|
|
(out / 'jobs.json').write_text(json.dumps(jobs, indent=0), encoding='utf-8')
|
|
return jobs
|
|
|
|
|
|
def worker(keys, jobs, spv_dir):
|
|
"""One --serve process for a slice of keys. Returns (translated, failures {key: message}, errors {key: message})."""
|
|
failures, errors, ok = {}, {}, 0
|
|
process = subprocess.Popen([str(XLATE), '--serve'], stdin=subprocess.PIPE, stdout=subprocess.PIPE,
|
|
stderr=subprocess.DEVNULL, text=True)
|
|
try:
|
|
for key in keys:
|
|
process.stdin.write(json.dumps({'job': jobs[key]['job'], 'out': str(spv_dir)}) + '\n')
|
|
process.stdin.flush()
|
|
reply = process.stdout.readline().rstrip('\n')
|
|
if not reply:
|
|
failures[key] = 'translator service exited'
|
|
break
|
|
if not reply.startswith('ok '):
|
|
failures[key] = reply.split(' ', 2)[2] if reply.count(' ') >= 2 else reply
|
|
continue
|
|
v = subprocess.run([str(SPIRV_VAL), '--target-env', 'vulkan1.3', str(spv_dir / f'{key}.spv')],
|
|
capture_output=True, text=True)
|
|
if v.returncode:
|
|
text = (v.stderr or v.stdout).strip()
|
|
errors[key] = text.splitlines()[0] if text else 'spirv-val failed'
|
|
else:
|
|
ok += 1
|
|
finally:
|
|
process.stdin.close()
|
|
process.wait()
|
|
return ok, failures, errors
|
|
|
|
|
|
def normalize(message):
|
|
return re.sub(r'\b\d+\b|0x[0-9a-fA-F]+', 'N', message)[:300]
|
|
|
|
|
|
def translate(jobs, out):
|
|
spv_dir = out / 'spv'
|
|
spv_dir.mkdir(exist_ok=True)
|
|
keys = sorted(jobs)
|
|
count = max(1, min(MAX_WORKERS, len(keys)))
|
|
slices = [keys[i::count] for i in range(count)]
|
|
ok, failures, errors = 0, {}, {}
|
|
with concurrent.futures.ThreadPoolExecutor(count) as pool:
|
|
for n, f, e in pool.map(lambda s: worker(s, jobs, spv_dir), slices):
|
|
ok += n
|
|
failures.update(f)
|
|
errors.update(e)
|
|
if failures or errors:
|
|
lines = []
|
|
for title, table in (('translate failures', failures), ('spirv-val errors', errors)):
|
|
groups = defaultdict(list)
|
|
for key, message in table.items():
|
|
groups[normalize(message)].append((key, message))
|
|
lines.append(f'== {title}: {len(table)} in {len(groups)} groups')
|
|
for _, members in sorted(groups.items(), key=lambda g: -len(g[1])):
|
|
lines.append(f'{len(members)}\t{members[0][1][:300]}')
|
|
lines.extend(f'\texample {key}' for key, _ in sorted(members)[:2])
|
|
(out / 'failures.txt').write_text('\n'.join(lines) + '\n', encoding='utf-8')
|
|
else:
|
|
(out / 'failures.txt').unlink(missing_ok=True)
|
|
return ok, len(failures), len(errors)
|
|
|
|
|
|
def main():
|
|
ap = argparse.ArgumentParser(description=__doc__.splitlines()[0])
|
|
ap.add_argument('--version', required=True, help='v121 (build/extracted/<version>)')
|
|
ap.add_argument('--all', action='store_true', help='also the layout and effect shaders MNG leaves to bind-state variants')
|
|
ap.add_argument('--translate', action='store_true', help='translate every shader and run spirv-val')
|
|
args = ap.parse_args()
|
|
t0 = time.monotonic()
|
|
extracted = ROOT / 'build' / 'extracted' / args.version
|
|
out = ROOT / 'build' / args.version / 'shaders'
|
|
out.mkdir(parents=True, exist_ok=True)
|
|
shaders, rejected = inventory(extracted / 'romfs', extracted / 'exefs')
|
|
jobs = write_shaders(shaders, out, args.all)
|
|
a, b, c = (sum(1 for j in jobs.values() if j['origin'] == o) for o in ('bnsh', 'sharcb', 'main'))
|
|
print(f'{len(shaders)} unique shaders found in all, {len(jobs)} selected; {rejected} .sharcb record candidates '
|
|
f'without a control section rejected', flush=True)
|
|
summary = f'shaders {len(jobs)} (bnsh {a}, sharcb {b}, main {c})'
|
|
print(f'extracted {summary} in {time.monotonic() - t0:.1f} s', flush=True)
|
|
if not args.translate:
|
|
print(f'{summary}, total {time.monotonic() - t0:.1f} s')
|
|
return 0
|
|
if not XLATE.is_file() or not SPIRV_VAL.is_file():
|
|
sys.exit(f'missing {XLATE} or {SPIRV_VAL}')
|
|
ok, failed, bad = translate(jobs, out)
|
|
print(f'{summary}, translated {ok}, failed {failed}, spirv-val errors {bad}, total {time.monotonic() - t0:.1f} s')
|
|
return 0
|
|
|
|
|
|
if __name__ == '__main__':
|
|
sys.exit(main())
|