mirror of
https://github.com/SK8-ENGINE/skate-3-rust-engine.git
synced 2026-10-26 19:47:44 +00:00
Setup: extract and decode the disc's game audio (new audio group)
- audio_formats.py: EA SNR (EA-XMA) stream scanner with block-chain validation, SPLC bank streams and patch tree (records, layers, containers; verified on all 20 banks), .grain files (speed-sweep recordings), loopable speed bands, gain-free 5->2 downmix. - audio_export.py: decodes only what the engine plays (22 ambience beds, 14 grains x 6 speed bands, wheel spins, 23 sample banks) to PCM16 WAV in assets/private/audio with audio_manifest.json (v3, incl. SPLC patch trees). Decoder: pinned vgmstream r2117 (SHA-256 checked) via the existing dependency() helper. - Registered as optional asset group `audio` (versions, receipts, asset_exports, install); a failure records audio-availability.json and the game runs silent. Other groups' fingerprints unchanged. - .gitignore: retail audio extensions and *.wav (Skyline mod WAVs excepted). Tests updated for the new group.
This commit is contained in:
10 files changed
+633
-4
No files matched your search
@@ -93,3 +93,20 @@ def environment(game_root, stage, work, report, log, converted=None, game_exe=No
|
||||
report('Preparing global foliage backdrops')
|
||||
from .backdrop import convert as write_backdrops
|
||||
attempt('backdrops',lambda assets:write_backdrops(game_root,assets,converted))
|
||||
|
||||
|
||||
def audio(game_root, stage, work, report, log, tools):
|
||||
private=stage/"assets/private"
|
||||
from .audio_export import VGMSTREAM_SHA, VGMSTREAM_URL, convert
|
||||
from .optional_content import CONTENT_ERRORS, note
|
||||
availability=private/'audio-availability.json'
|
||||
report('Preparing game audio')
|
||||
try:
|
||||
try:vgmstream=engine.dependency(tools,'vgmstream-cli',VGMSTREAM_URL,VGMSTREAM_SHA,report)
|
||||
except OSError as error:raise RuntimeError(f'Could not download the audio decoder: {error}') from error
|
||||
convert(game_root,private,work/'audio',vgmstream,report,log)
|
||||
except CONTENT_ERRORS as error:
|
||||
if (private/'audio').exists():shutil.rmtree(private/'audio')
|
||||
note(availability,'Game audio',error,report=report)
|
||||
return
|
||||
availability.unlink(missing_ok=True)
|
||||
@@ -0,0 +1,201 @@
|
||||
"""Decode the owned disc's game audio into PCM16 WAV for the engine (assets/private/audio).
|
||||
|
||||
Only what the engine plays is exported (crates/skate-game/src/game_audio/): the
|
||||
ambience beds, rolling grains, wheel spins and the sample banks below. Audio is
|
||||
written at its original level: nothing is normalised or boosted, and surround
|
||||
ambience is downmixed to stereo by a weighted average (audio_formats.py).
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import shutil
|
||||
import subprocess
|
||||
import wave
|
||||
from pathlib import Path
|
||||
|
||||
from tools.owned_game.big import BigArchive
|
||||
from .audio_formats import downmix_pcm16, grain, loop_bands, scan_snr, splc_patches, splc_streams, standalone
|
||||
|
||||
VERSION = 3
|
||||
|
||||
# Each rolling grain is a recording that sweeps from slow to fast rolling (about
|
||||
# 10 dB louder and brighter by the end). It is cut into this many speed bands,
|
||||
# each made loopable, so the engine can play the band matching board speed.
|
||||
GRAIN_BANDS = 6
|
||||
LOOP_CROSSFADE = 0.08 # seconds
|
||||
|
||||
# Pinned decoder (EA-XMA needs FFmpeg's XMA2 decoder, which this build bundles).
|
||||
# The SHA-256 matches the digest GitHub publishes for the release asset.
|
||||
VGMSTREAM_URL = 'https://github.com/vgmstream/vgmstream/releases/download/r2117/vgmstream-win64.zip'
|
||||
VGMSTREAM_SHA = '6c4a8a3813864fefed081bbd337dbc0ad93bf88e0b92f5db98d7ab258b22dc6c'
|
||||
|
||||
# Per-map ambience beds (ambience.big + headers in ambienceresident.big).
|
||||
AMBIENCE = (
|
||||
'01_dt_apt', '02_dt_less_busy', '03_dt_rez', '04_dt_main', '05_dt_parks', '06_dt_open',
|
||||
'07_univ_mt_high', '08_univ_mt_low', '09_univ_campus', '10_univ_housing',
|
||||
'11_indu_shipyard', '12_indu_drydock', '13_indu_quarry', '14_reclaimed_a', '15_indu_new_factory',
|
||||
'16_spillway', '17_space_park', '18_skate_school', '19_reclaimed_b', '20_indu_old_factory',
|
||||
'21_interior_arena_amb', '22_interior_tunnel_amb',
|
||||
)
|
||||
|
||||
# Sample banks from audiofiles.big (data/audio/<name>).
|
||||
BANKS = (
|
||||
'Skate_Collisions.bnk', 'Skate_Metal.bnk', 'sk8_foley.bnk', 'Sk82_Whsh_Bys.bnk',
|
||||
'Sk8_Air_Flip_Tricks.abk', 'GRINDS.abk', 'board_scrapes.abk', 'Brd_Squeaks.abk',
|
||||
'WHEEL_SKID_BANK.abk', 'FOOT_DRAG.abk', 'fstep_skateshoe1_sm.abk', 'Bodyslide.abk',
|
||||
'Rolling_Rattles.abk', 'Seams_Bank.abk', 'PatchBank_Rolling_Surfaces.abk', 'sense_of_speed.abk',
|
||||
'water_misc.abk', 'Foley_Cloth.abk',
|
||||
# Water emitters on the map's water (game_audio/water.rs) and splash tails.
|
||||
'water_lapping.abk', 'water_lapping_pond.abk', 'fountains_waterlaps_left.abk',
|
||||
'water_fountain.abk', 'ocean_wave_small.abk',
|
||||
)
|
||||
|
||||
# vgmstream stops accepting input files somewhere between 600 and 1100 arguments.
|
||||
_BATCH = 256
|
||||
|
||||
|
||||
def _decode(vgmstream: Path, directory: Path, names: list[str], log) -> None:
|
||||
"""Decode standalone `<name>` files in `directory` to `<name>.wav` beside them."""
|
||||
for start in range(0, len(names), _BATCH):
|
||||
batch = names[start:start + _BATCH]
|
||||
# Imported lazily: install imports this module's group function.
|
||||
from . import install as engine
|
||||
child = engine.spawn([vgmstream, '-i', '-o', '?f.wav', *batch], cwd=directory,
|
||||
stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
|
||||
with child as process:
|
||||
output = process.stdout.read()
|
||||
if process.wait():
|
||||
log.write(output)
|
||||
raise RuntimeError(f'vgmstream failed on {directory.name}')
|
||||
for name in batch:
|
||||
if not (directory/(name + '.wav')).is_file():
|
||||
raise RuntimeError(f'vgmstream did not decode {directory.name}/{name}')
|
||||
|
||||
|
||||
def _wav_info(path: Path) -> dict:
|
||||
with wave.open(str(path)) as source:
|
||||
return {'channels': source.getnchannels(), 'sample_rate': source.getframerate(),
|
||||
'seconds': round(source.getnframes() / source.getframerate(), 4)}
|
||||
|
||||
|
||||
def _stereo(source: Path, target: Path) -> dict:
|
||||
with wave.open(str(source)) as reader:
|
||||
channels, rate, width = reader.getnchannels(), reader.getframerate(), reader.getsampwidth()
|
||||
frames = reader.readframes(reader.getnframes())
|
||||
if width != 2:
|
||||
raise RuntimeError(f'{source.name}: expected PCM16 from vgmstream')
|
||||
mixed = downmix_pcm16(frames, channels)
|
||||
with wave.open(str(target), 'wb') as writer:
|
||||
writer.setnchannels(min(channels, 2))
|
||||
writer.setsampwidth(2)
|
||||
writer.setframerate(rate)
|
||||
writer.writeframes(mixed)
|
||||
return {**_wav_info(target), 'source_channels': channels}
|
||||
|
||||
|
||||
def _grain_bands(source: Path, folder: Path, prefix: str) -> list[dict]:
|
||||
with wave.open(str(source)) as reader:
|
||||
channels, rate, width = reader.getnchannels(), reader.getframerate(), reader.getsampwidth()
|
||||
frames = reader.readframes(reader.getnframes())
|
||||
if width != 2 or channels != 1:
|
||||
raise RuntimeError(f'{source.name}: expected mono PCM16 from vgmstream')
|
||||
folder.mkdir(parents=True)
|
||||
bands = []
|
||||
for index, pcm in enumerate(loop_bands(frames, GRAIN_BANDS, round(LOOP_CROSSFADE * rate))):
|
||||
target = folder/f'{index}.wav'
|
||||
with wave.open(str(target), 'wb') as writer:
|
||||
writer.setnchannels(1)
|
||||
writer.setsampwidth(2)
|
||||
writer.setframerate(rate)
|
||||
writer.writeframes(pcm)
|
||||
bands.append({'file': f'{prefix}/{index}.wav', **_wav_info(target)})
|
||||
return bands
|
||||
|
||||
|
||||
def _streams(name: str, data: bytes):
|
||||
streams = splc_streams(data) if data[:4] == b'SPLC' else scan_snr(data)
|
||||
if not streams:
|
||||
raise ValueError(f'{name} contains no decodable streams')
|
||||
return streams
|
||||
|
||||
|
||||
def convert(game_root: Path, private: Path, work: Path, vgmstream: Path, report, log) -> dict:
|
||||
"""Write private/audio/** and private/audio/audio_manifest.json; return the manifest."""
|
||||
audio_root = game_root/'data/audio'
|
||||
output = private/'audio'
|
||||
if output.exists():
|
||||
shutil.rmtree(output)
|
||||
work.mkdir(parents=True, exist_ok=True)
|
||||
manifest = {'version': VERSION, 'ambience': {}, 'grains': {}, 'wheels': {}, 'banks': {}, 'patches': {}}
|
||||
|
||||
report('Decoding ambience')
|
||||
beds, headers = BigArchive(audio_root/'ambience.big'), BigArchive(audio_root/'ambienceresident.big')
|
||||
bodies = {Path(e.path).stem: e for e in beds.entries if e.path.endswith('.sns')}
|
||||
heads = {Path(e.path).stem: e for e in headers.entries if e.path.endswith('.snr')}
|
||||
folder = work/'ambience'
|
||||
folder.mkdir()
|
||||
for name in AMBIENCE:
|
||||
if name not in bodies or name not in heads:
|
||||
raise KeyError(f'ambience bed {name} is missing from the disc')
|
||||
(output/'ambience').mkdir(parents=True)
|
||||
for name in AMBIENCE:
|
||||
(folder/(name + '.snr')).write_bytes(headers.read(heads[name]))
|
||||
(folder/(name + '.sns')).write_bytes(beds.read(bodies[name]))
|
||||
_decode(vgmstream, folder, [name + '.snr'], log)
|
||||
decoded = folder/(name + '.snr.wav')
|
||||
info = _stereo(decoded, output/'ambience'/(name + '.wav'))
|
||||
decoded.unlink() # ~65 MB of five-channel PCM each
|
||||
manifest['ambience'][name] = {'file': f'ambience/{name}.wav', **info}
|
||||
|
||||
for kind, archive, suffix in (('grains', 'grains.big', '.grain'), ('wheels', 'wheels.big', '.snr')):
|
||||
report(f'Decoding {kind}')
|
||||
source = BigArchive(audio_root/archive)
|
||||
folder = work/kind
|
||||
folder.mkdir()
|
||||
names = []
|
||||
for entry in source.entries:
|
||||
if not entry.path.endswith(suffix):
|
||||
continue
|
||||
stem, data = Path(entry.path).stem, source.read(entry)
|
||||
stream = grain(data).stream if suffix == '.grain' else scan_snr(data)[0]
|
||||
(folder/(stem + '.snr')).write_bytes(standalone(data, stream))
|
||||
names.append(stem)
|
||||
_decode(vgmstream, folder, [stem + '.snr' for stem in names], log)
|
||||
(output/kind).mkdir(parents=True)
|
||||
for stem in names:
|
||||
decoded = folder/(stem + '.snr.wav')
|
||||
if kind == 'grains':
|
||||
manifest[kind][stem] = {'bands': _grain_bands(decoded, output/kind/stem, f'{kind}/{stem}')}
|
||||
continue
|
||||
target = output/kind/(stem + '.wav')
|
||||
shutil.move(decoded, target)
|
||||
manifest[kind][stem] = {'file': f'{kind}/{stem}.wav', **_wav_info(target)}
|
||||
|
||||
report('Decoding sound effect banks')
|
||||
files = BigArchive(audio_root/'audiofiles.big')
|
||||
by_name = {Path(e.path).name: e for e in files.entries}
|
||||
for bank in BANKS:
|
||||
if bank not in by_name:
|
||||
raise KeyError(f'sound bank {bank} is missing from the disc')
|
||||
data = files.read(by_name[bank])
|
||||
stem = Path(bank).stem
|
||||
folder = work/'banks'/stem
|
||||
folder.mkdir(parents=True)
|
||||
streams = _streams(bank, data)
|
||||
for index, stream in enumerate(streams):
|
||||
(folder/f'{index:04d}.snr').write_bytes(standalone(data, stream))
|
||||
_decode(vgmstream, folder, [f'{index:04d}.snr' for index in range(len(streams))], log)
|
||||
target_folder = output/'banks'/stem
|
||||
target_folder.mkdir(parents=True)
|
||||
samples = []
|
||||
for index in range(len(streams)):
|
||||
target = target_folder/f'{index:04d}.wav'
|
||||
shutil.move(folder/f'{index:04d}.snr.wav', target)
|
||||
samples.append({'file': f'banks/{stem}/{index:04d}.wav', **_wav_info(target)})
|
||||
manifest['banks'][stem] = samples
|
||||
if data[:4] == b'SPLC':
|
||||
# Retail's own layering/randomisation per sound (audio_formats.splc_patches).
|
||||
manifest['patches'][stem] = splc_patches(data)
|
||||
|
||||
(output/'audio_manifest.json').write_text(json.dumps(manifest, indent=1), encoding='utf-8')
|
||||
return manifest
|
||||
@@ -0,0 +1,227 @@
|
||||
"""Containers around Skate 3's EA SNR (EAAC) audio streams.
|
||||
|
||||
Layouts were worked out from the owned disc's archives (docs/hails-additions/11-audio.md):
|
||||
|
||||
- SNR header: u32 version(4) | codec(4) | channels-1(6) | sample rate(18);
|
||||
u32 type(2) | loop(1) | sample count(29); u32 loop start when looping.
|
||||
RAM streams (type 0) are followed by blocks: u8 flag (0x00, or 0x80 = last),
|
||||
u24 block size (incl. this 8-byte header), u32 samples in the block.
|
||||
- SPLC `.bnk` (patch banks): u32 stream count at 0x18, parameter records, then
|
||||
that many SNR streams back to back at unaligned offsets.
|
||||
- `.grain` (rolling surfaces): u32 offset of the SNR stream, f32 duration in
|
||||
seconds, a seek table (the `.sek` layout), then one SNR stream.
|
||||
|
||||
vgmstream reads standalone SNR/SNS and ABKC banks itself; the SPLC and grain
|
||||
streams are cut out with these helpers into standalone `.snr` files for it.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import struct
|
||||
from dataclasses import dataclass
|
||||
|
||||
EA_XMA = 3
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class SnrStream:
|
||||
offset: int
|
||||
end: int
|
||||
codec: int
|
||||
channels: int
|
||||
sample_rate: int
|
||||
samples: int
|
||||
loop_start: int | None
|
||||
|
||||
@property
|
||||
def seconds(self) -> float:
|
||||
return self.samples / self.sample_rate
|
||||
|
||||
|
||||
def _u32(data: bytes, offset: int) -> int:
|
||||
return struct.unpack_from('>I', data, offset)[0]
|
||||
|
||||
|
||||
def snr_at(data: bytes, offset: int, codecs=(EA_XMA,)) -> SnrStream | None:
|
||||
"""The RAM SNR stream starting at `offset`, if its block chain is consistent."""
|
||||
if offset < 0 or offset + 16 > len(data):
|
||||
return None
|
||||
header = _u32(data, offset)
|
||||
version, codec = header >> 28, (header >> 24) & 0xF
|
||||
channels, rate = ((header >> 18) & 0x3F) + 1, header & 0x3FFFF
|
||||
if version > 1 or codec not in codecs or not 8000 <= rate <= 48000 or channels > 8:
|
||||
return None
|
||||
flags = _u32(data, offset + 4)
|
||||
kind, looped, samples = flags >> 30, (flags >> 29) & 1, flags & 0x1FFFFFFF
|
||||
if kind != 0 or samples == 0:
|
||||
return None
|
||||
loop_start = _u32(data, offset + 8) if looped else None
|
||||
cursor = offset + (12 if looped else 8)
|
||||
decoded = 0
|
||||
while decoded < samples:
|
||||
if cursor + 8 > len(data):
|
||||
return None
|
||||
flag, size = data[cursor], _u32(data, cursor) & 0xFFFFFF
|
||||
block_samples = _u32(data, cursor + 4)
|
||||
if flag not in (0x00, 0x80) or size <= 8 or cursor + size > len(data) or block_samples == 0:
|
||||
return None
|
||||
decoded += block_samples
|
||||
cursor += size
|
||||
if flag == 0x80:
|
||||
break
|
||||
if decoded != samples:
|
||||
return None
|
||||
return SnrStream(offset, cursor, codec, channels, rate, samples, loop_start)
|
||||
|
||||
|
||||
def scan_snr(data: bytes, start: int = 0) -> list[SnrStream]:
|
||||
"""Every consistent SNR stream in `data`, in file order."""
|
||||
streams = []
|
||||
offset = start
|
||||
while offset + 16 <= len(data):
|
||||
stream = snr_at(data, offset)
|
||||
if stream is None:
|
||||
offset += 1
|
||||
else:
|
||||
streams.append(stream)
|
||||
offset = stream.end
|
||||
return streams
|
||||
|
||||
|
||||
def splc_streams(data: bytes) -> list[SnrStream]:
|
||||
if data[:4] != b'SPLC':
|
||||
raise ValueError('not an SPLC bank')
|
||||
expected = _u32(data, 0x18)
|
||||
streams = scan_snr(data)
|
||||
if len(streams) != expected:
|
||||
raise ValueError(f'SPLC bank declares {expected} streams; found {len(streams)}')
|
||||
return streams
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class Grain:
|
||||
duration: float
|
||||
stream: SnrStream
|
||||
|
||||
|
||||
def grain(data: bytes) -> Grain:
|
||||
if len(data) < 8:
|
||||
raise ValueError('truncated grain file')
|
||||
offset = _u32(data, 0)
|
||||
duration = struct.unpack_from('>f', data, 4)[0]
|
||||
stream = snr_at(data, offset)
|
||||
if stream is None:
|
||||
raise ValueError(f'grain file has no SNR stream at {offset:#x}')
|
||||
if abs(stream.seconds - duration) > 0.01:
|
||||
raise ValueError(f'grain duration {duration:.3f}s does not match its stream ({stream.seconds:.3f}s)')
|
||||
return Grain(duration, stream)
|
||||
|
||||
|
||||
def standalone(data: bytes, stream: SnrStream) -> bytes:
|
||||
"""The stream as a standalone `.snr` file (header followed by its blocks)."""
|
||||
return data[stream.offset:stream.end]
|
||||
|
||||
|
||||
# Five-channel EA ambience is L, C, R, Ls, Rs (channel 1 correlates with both
|
||||
# 0 and 2; 0 pairs with 3 and 2 with 4). Each output is a weighted average
|
||||
# whose weights sum to 1, so the downmix can never add gain or clip.
|
||||
_FIVE_TO_STEREO = ((1.0, 0.7071, 0.0, 1.0, 0.0), (0.0, 0.7071, 1.0, 0.0, 1.0))
|
||||
|
||||
|
||||
def downmix_weights(channels: int) -> tuple[tuple[float, ...], tuple[float, ...]] | None:
|
||||
"""Per-output weights (normalised to sum 1) for a stereo downmix, or None to keep as is."""
|
||||
if channels <= 2:
|
||||
return None
|
||||
if channels == 5:
|
||||
rows = _FIVE_TO_STEREO
|
||||
else:
|
||||
# Unknown layouts: alternate channels left/right.
|
||||
rows = (tuple(float(i % 2 == 0) for i in range(channels)),
|
||||
tuple(float(i % 2 == 1) for i in range(channels)))
|
||||
return tuple(tuple(w / sum(row) for w in row) for row in rows)
|
||||
|
||||
|
||||
def downmix_pcm16(frames: bytes, channels: int) -> bytes:
|
||||
"""Interleaved little-endian PCM16 frames downmixed to stereo (no gain)."""
|
||||
weights = downmix_weights(channels)
|
||||
if weights is None:
|
||||
return frames
|
||||
import numpy
|
||||
source = numpy.frombuffer(frames, dtype='<i2').reshape(-1, channels).astype(numpy.float32)
|
||||
mixed = source @ numpy.asarray(weights, dtype=numpy.float32).T
|
||||
return numpy.rint(mixed).astype('<i2').tobytes()
|
||||
|
||||
|
||||
def loop_bands(frames: bytes, bands: int, crossfade: int) -> list[bytes]:
|
||||
"""Mono PCM16 cut into `bands` equal parts, each made seamless as a loop.
|
||||
|
||||
A part's last `crossfade` samples are blended into its first ones with
|
||||
linear weights (w and 1 - w), so no sample can exceed the louder of its
|
||||
two sources; the blended tail is then dropped.
|
||||
"""
|
||||
import numpy
|
||||
samples = numpy.frombuffer(frames, dtype='<i2').astype(numpy.float32)
|
||||
size = len(samples) // bands
|
||||
if size <= 2 * crossfade:
|
||||
raise ValueError('grain recording too short for its bands')
|
||||
out = []
|
||||
for index in range(bands):
|
||||
part = samples[index * size:(index + 1) * size].copy()
|
||||
head, tail = part[:crossfade], part[-crossfade:]
|
||||
fade_in = numpy.linspace(0.0, 1.0, crossfade, endpoint=False, dtype=numpy.float32)
|
||||
part[:crossfade] = head * fade_in + tail * (1.0 - fade_in)
|
||||
out.append(numpy.rint(part[:-crossfade]).astype('<i2').tobytes())
|
||||
return out
|
||||
|
||||
|
||||
def splc_patches(data: bytes) -> dict:
|
||||
"""The patch tree of an SPLC bank: what retail plays for each sound id.
|
||||
|
||||
Layout (verified on all 20 banks of the owned disc: the walk ends exactly at
|
||||
the sample table, every member names a valid sample; field meanings follow
|
||||
upstream PR #4's notes, re-derived and checked here):
|
||||
header 60 bytes: +8 sample table offset (from byte 60), +12 record count,
|
||||
+16 container count, +20 24-byte extras (0), +24 sample count;
|
||||
records, 36 bytes: +4 u16 id (= index), +7 u8 group count;
|
||||
containers, 72 bytes: +4 u16 ids (records), +68 u8 count, +69 u8 mode;
|
||||
per record, its groups in order: 12-byte header (+8 u8 member count,
|
||||
+9 u8 mode) then 72-byte members: +0 u16 sample index (= our stream
|
||||
index), +8 f32 gain, +44 f32 pitch, +48 f32 gain range, +64 f32 probability.
|
||||
A record's groups are layers played together, one member each; an id at
|
||||
or above the record count names container (id - records), which picks one
|
||||
of its records. Sample index n is the n-th SNR stream (`splc_streams`).
|
||||
"""
|
||||
if data[:4] != b'SPLC':
|
||||
raise ValueError('not an SPLC bank')
|
||||
table = 60 + _u32(data, 8)
|
||||
records, containers, samples = _u32(data, 12), _u32(data, 16), _u32(data, 24)
|
||||
if _u32(data, 20):
|
||||
raise ValueError('SPLC bank with 24-byte extras is not supported')
|
||||
container_base = 60 + 36 * records
|
||||
out_containers = []
|
||||
for c in range(containers):
|
||||
at = container_base + 72 * c
|
||||
count = data[at + 68]
|
||||
out_containers.append([struct.unpack_from('>H', data, at + 4 + 2 * k)[0] for k in range(count)])
|
||||
cursor = container_base + 72 * containers
|
||||
out_records = []
|
||||
for r in range(records):
|
||||
at = 60 + 36 * r
|
||||
if struct.unpack_from('>H', data, at + 4)[0] != r:
|
||||
raise ValueError(f'SPLC record {r} has id {struct.unpack_from(">H", data, at + 4)[0]}')
|
||||
groups = []
|
||||
for _ in range(data[at + 7]):
|
||||
count, mode = data[cursor + 8], data[cursor + 9]
|
||||
cursor += 12
|
||||
members = []
|
||||
for _ in range(count):
|
||||
sample = struct.unpack_from('>H', data, cursor)[0]
|
||||
if sample >= samples:
|
||||
raise ValueError(f'SPLC member names sample {sample} of {samples}')
|
||||
gain, pitch, gain_range, probability = (struct.unpack_from('>f', data, cursor + o)[0] for o in (8, 44, 48, 64))
|
||||
members.append([sample, round(gain, 4), round(gain_range, 4), round(pitch, 4), round(probability, 4)])
|
||||
cursor += 72
|
||||
groups.append({'mode': mode, 'members': members})
|
||||
out_records.append(groups)
|
||||
if cursor != table:
|
||||
raise ValueError(f'SPLC patch tree ends at {cursor:#x}, sample table at {table:#x}')
|
||||
return {'records': out_records, 'containers': out_containers}
|
||||
@@ -14,6 +14,7 @@ ROOTS = {
|
||||
'assets/private/render-parameters.json', 'assets/private/exposure.json',
|
||||
'assets/private/exposure-profiles.json', 'assets/private/teleports.json', 'assets/private/environment-status'),
|
||||
'maps': ('assets/private/native-props', 'maps.json', 'assets/private/map-status'),
|
||||
'audio': ('assets/private/audio', 'assets/private/audio-availability.json'),
|
||||
}
|
||||
|
||||
|
||||
|
||||
@@ -434,6 +434,8 @@ def _install(iso,base,game_exe,report,game_root=None,refresh=False,finalize=None
|
||||
exports.character(game_root,stage,work,report,log,converted)
|
||||
if 'environment' in groups:
|
||||
exports.environment(game_root,stage,work,report,log,converted,game_exe)
|
||||
if 'audio' in groups:
|
||||
exports.audio(game_root,stage,work,report,log,base/'tools')
|
||||
if 'maps' in groups:
|
||||
report('Preparing authored movable-object models')
|
||||
from .dynamic_props import prepare_catalog
|
||||
|
||||
@@ -0,0 +1,165 @@
|
||||
"""EA SNR container helpers, checked on synthetic streams (never game audio)."""
|
||||
import struct
|
||||
import unittest
|
||||
|
||||
import numpy
|
||||
|
||||
from tools.asset_pipeline import audio_formats as audio
|
||||
|
||||
|
||||
def snr(samples=(300, 200), rate=48000, channels=1, codec=3, loop_start=None, last_flag=True):
|
||||
"""A fake RAM SNR stream: header plus one block per entry in `samples`."""
|
||||
header = struct.pack('>I', (codec << 24) | ((channels - 1) << 18) | rate)
|
||||
flags = (1 << 29 if loop_start is not None else 0) | sum(samples)
|
||||
data = header + struct.pack('>I', flags)
|
||||
if loop_start is not None:
|
||||
data += struct.pack('>I', loop_start)
|
||||
for index, count in enumerate(samples):
|
||||
payload = bytes([0x5A]) * (16 + index)
|
||||
flag = 0x80 if last_flag and index == len(samples) - 1 else 0x00
|
||||
data += struct.pack('>I', (flag << 24) | (8 + len(payload))) + struct.pack('>I', count) + payload
|
||||
return data
|
||||
|
||||
|
||||
class SnrStreams(unittest.TestCase):
|
||||
def test_reads_header_and_block_chain(self):
|
||||
data = b'junk' + snr(rate=44100, channels=2)
|
||||
stream = audio.snr_at(data, 4)
|
||||
self.assertEqual((stream.channels, stream.sample_rate, stream.samples), (2, 44100, 500))
|
||||
self.assertEqual(stream.end, len(data))
|
||||
self.assertIsNone(stream.loop_start)
|
||||
self.assertEqual(audio.standalone(data, stream), data[4:])
|
||||
|
||||
def test_loop_start_and_unflagged_final_block(self):
|
||||
stream = audio.snr_at(snr(loop_start=7, last_flag=False), 0)
|
||||
self.assertEqual((stream.loop_start, stream.samples), (7, 500))
|
||||
|
||||
def test_rejects_inconsistent_chains(self):
|
||||
good = snr()
|
||||
self.assertIsNone(audio.snr_at(good[:-1], 0)) # truncated block
|
||||
self.assertIsNone(audio.snr_at(snr(codec=5), 0)) # not EA-XMA
|
||||
bad_count = bytearray(good)
|
||||
struct.pack_into('>I', bad_count, 4, 499) # header disagrees with blocks
|
||||
self.assertIsNone(audio.snr_at(bytes(bad_count), 0))
|
||||
|
||||
def test_scan_finds_unaligned_back_to_back_streams(self):
|
||||
data = b'\x01\x02\x03' + snr() + b'\xff' + snr(samples=(64,)) + b'tail'
|
||||
streams = audio.scan_snr(data)
|
||||
self.assertEqual([s.offset for s in streams], [3, 3 + len(snr()) + 1])
|
||||
self.assertEqual([s.samples for s in streams], [500, 64])
|
||||
|
||||
def test_splc_count_must_match(self):
|
||||
header = bytearray(0x40)
|
||||
header[:4] = b'SPLC'
|
||||
struct.pack_into('>I', header, 0x18, 2)
|
||||
bank = bytes(header) + snr() + snr()
|
||||
self.assertEqual(len(audio.splc_streams(bank)), 2)
|
||||
struct.pack_into('>I', header, 0x18, 3)
|
||||
with self.assertRaises(ValueError):
|
||||
audio.splc_streams(bytes(header) + snr() + snr())
|
||||
|
||||
def test_grain_offset_and_duration(self):
|
||||
stream = snr(samples=(48000,))
|
||||
data = struct.pack('>If', 0x20, 1.0) + bytes(0x18) + stream
|
||||
parsed = audio.grain(data)
|
||||
self.assertEqual((parsed.stream.offset, parsed.stream.samples), (0x20, 48000))
|
||||
wrong = struct.pack('>If', 0x20, 2.0) + bytes(0x18) + stream
|
||||
with self.assertRaises(ValueError):
|
||||
audio.grain(wrong)
|
||||
|
||||
|
||||
class SplcPatches(unittest.TestCase):
|
||||
@staticmethod
|
||||
def bank():
|
||||
# 2 records, 1 container, 3 samples: record 0 = 1 group of 2 members,
|
||||
# record 1 = 2 groups of 1 member; container 0 picks record 0 or 1.
|
||||
records = []
|
||||
for rid, groups in ((0, 1), (1, 2)):
|
||||
r = bytearray(36)
|
||||
struct.pack_into('>H', r, 4, rid)
|
||||
r[7] = groups
|
||||
records.append(bytes(r))
|
||||
container = bytearray(72)
|
||||
struct.pack_into('>HH', container, 4, 0, 1)
|
||||
container[68] = 2
|
||||
def group(mode, members):
|
||||
g = bytearray(12)
|
||||
g[8], g[9] = len(members), mode
|
||||
out = bytes(g)
|
||||
for sample, gain, pitch, gain_range, probability in members:
|
||||
m = bytearray(72)
|
||||
struct.pack_into('>H', m, 0, sample)
|
||||
for offset, value in ((8, gain), (44, pitch), (48, gain_range), (64, probability)):
|
||||
struct.pack_into('>f', m, offset, value)
|
||||
out += bytes(m)
|
||||
return out
|
||||
tree = b''.join(records) + bytes(container) + group(2, [(0, 1.0, 1.2, 0.125, 1.0), (1, 1.0, 1.0, 0.0, 1.0)]) + group(1, [(2, 0.5, 1.0, 0.0, 0.75)]) + group(0, [(1, 1.0, 1.0, 0.0, 1.0)])
|
||||
header = bytearray(60)
|
||||
header[:4] = b'SPLC'
|
||||
struct.pack_into('>IIIII', header, 8, len(tree), 2, 1, 0, 3)
|
||||
return bytes(header) + tree
|
||||
|
||||
def test_reads_records_layers_and_containers(self):
|
||||
patches = audio.splc_patches(self.bank())
|
||||
self.assertEqual(patches['containers'], [[0, 1]])
|
||||
self.assertEqual(len(patches['records']), 2)
|
||||
first = patches['records'][0]
|
||||
self.assertEqual([g['mode'] for g in first], [2])
|
||||
self.assertEqual(first[0]['members'], [[0, 1.0, 0.125, 1.2, 1.0], [1, 1.0, 0.0, 1.0, 1.0]])
|
||||
self.assertEqual([g['members'][0][0] for g in patches['records'][1]], [2, 1])
|
||||
self.assertEqual(patches['records'][1][0]['members'][0][4], 0.75)
|
||||
|
||||
def test_rejects_a_tree_that_does_not_end_at_the_sample_table(self):
|
||||
data = bytearray(self.bank())
|
||||
struct.pack_into('>I', data, 8, struct.unpack_from('>I', data, 8)[0] + 4)
|
||||
with self.assertRaises(ValueError):
|
||||
audio.splc_patches(bytes(data))
|
||||
|
||||
|
||||
class Downmix(unittest.TestCase):
|
||||
def test_weights_sum_to_one(self):
|
||||
for channels in (3, 4, 5, 6):
|
||||
for row in audio.downmix_weights(channels):
|
||||
self.assertAlmostEqual(sum(row), 1.0, places=6)
|
||||
self.assertIsNone(audio.downmix_weights(2))
|
||||
|
||||
def test_five_channel_downmix_never_exceeds_input_peak(self):
|
||||
frames = numpy.array([[32767, 32767, 32767, 32767, 32767],
|
||||
[-32768, -32768, -32768, -32768, -32768],
|
||||
[1000, 0, -1000, 500, -500]], dtype='<i2')
|
||||
mixed = numpy.frombuffer(audio.downmix_pcm16(frames.tobytes(), 5), dtype='<i2').reshape(-1, 2)
|
||||
self.assertEqual(mixed[0].tolist(), [32767, 32767])
|
||||
self.assertEqual(mixed[1].tolist(), [-32768, -32768])
|
||||
# Left takes L, C and Ls only; right takes C, R and Rs.
|
||||
self.assertEqual(mixed[2].tolist(), [round(1500 / 2.7071), round(-1500 / 2.7071)])
|
||||
|
||||
def test_stereo_passes_through(self):
|
||||
frames = numpy.array([[1, -2], [3, -4]], dtype='<i2').tobytes()
|
||||
self.assertEqual(audio.downmix_pcm16(frames, 2), frames)
|
||||
|
||||
|
||||
|
||||
class LoopBands(unittest.TestCase):
|
||||
def test_bands_wrap_seamlessly_without_gain(self):
|
||||
ramp = numpy.arange(6000, dtype=numpy.float32)
|
||||
signal = (numpy.sin(ramp * 0.05) * (2000 + ramp * 3)).astype('<i2')
|
||||
bands = audio.loop_bands(signal.tobytes(), 3, 100)
|
||||
self.assertEqual(len(bands), 3)
|
||||
part = signal[:2000].astype(numpy.int32)
|
||||
loop = numpy.frombuffer(bands[0], dtype='<i2').astype(numpy.int32)
|
||||
self.assertEqual(len(loop), 1900)
|
||||
# Wrapping from the loop's end to its start continues the original recording.
|
||||
self.assertEqual(loop[-1], part[1899])
|
||||
self.assertEqual(loop[0], part[1900])
|
||||
self.assertLessEqual(numpy.abs(loop).max(), numpy.abs(part).max())
|
||||
# Later bands come from later (louder) parts of the sweep.
|
||||
loudness = [numpy.abs(numpy.frombuffer(b, dtype='<i2').astype(numpy.float32)).mean() for b in bands]
|
||||
self.assertEqual(loudness, sorted(loudness))
|
||||
|
||||
def test_too_short_recordings_are_rejected(self):
|
||||
with self.assertRaises(ValueError):
|
||||
audio.loop_bands(bytes(400), 3, 100)
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
unittest.main()
|
||||
@@ -31,8 +31,9 @@ class AssetVersions(unittest.TestCase):
|
||||
def test_equivalences_only_accept_exact_old_and_new_pair(self):
|
||||
migrations=json.loads(Path(v.__file__).with_name('pipeline-equivalence.json').read_text())
|
||||
# Equivalences describe historical pairs, not every future exporter.
|
||||
current={g:migrations[g][0][1] for g in v.GROUPS}
|
||||
old={g:migrations[g][0][0] for g in v.GROUPS}
|
||||
# Groups without a historical pair (e.g. audio) keep one fingerprint.
|
||||
current={g:migrations[g][0][1] if g in migrations else 'unchanged' for g in v.GROUPS}
|
||||
old={g:migrations[g][0][0] if g in migrations else 'unchanged' for g in v.GROUPS}
|
||||
self.assertEqual(v.changed_groups(old,current),set())
|
||||
changed={**current,'maps':'future-exporter'}
|
||||
self.assertEqual(v.changed_groups(old,changed),{'maps'})
|
||||
|
||||
@@ -6,7 +6,7 @@ import json
|
||||
from pathlib import Path
|
||||
import re
|
||||
|
||||
GROUPS = ('core', 'hud', 'character', 'environment', 'maps')
|
||||
GROUPS = ('core', 'hud', 'character', 'environment', 'maps', 'audio')
|
||||
COMMON = ('owned_game/**/*.py', 'asset_pipeline/fast_refpack.py', 'asset_pipeline/refpack_native.rs',
|
||||
'requirements-setup.txt')
|
||||
PARSERS = ('vendor/utt/**/*.py', 'vendor/university/**/*.py', 'vendor/utt/**/*.json', 'vendor/university/**/*.json')
|
||||
@@ -25,6 +25,7 @@ SOURCES = {
|
||||
'maps': PARSERS + ('asset_pipeline/optional_content.py', 'asset_pipeline/map*.py', 'asset_pipeline/dynamic_props.py',
|
||||
'asset_pipeline/environment.py', 'asset_pipeline/irradiance.py',
|
||||
'asset_pipeline/retail_material.py', 'asset_pipeline/backdrop.py', 'asset_pipeline/sky.py'),
|
||||
'audio': ('asset_pipeline/optional_content.py', 'asset_pipeline/audio_export.py', 'asset_pipeline/audio_formats.py'),
|
||||
}
|
||||
|
||||
|
||||
|
||||
@@ -44,7 +44,7 @@ class HudWorkspace(unittest.TestCase):
|
||||
self.assertEqual(json.loads((assets/'private/hud-availability.json').read_text())['status'],'unavailable')
|
||||
receipt=record(root,'hud')
|
||||
self.assertTrue(receipt)
|
||||
self.assertEqual(damaged(root,{'outputs':{'hud':receipt}},exclude=('core','character','environment','maps')),set())
|
||||
self.assertEqual(damaged(root,{'outputs':{'hud':receipt}},exclude=('core','character','environment','maps','audio')),set())
|
||||
|
||||
def test_failure_keeps_verified_existing_huds(self):
|
||||
with tempfile.TemporaryDirectory() as temp:
|
||||
|
||||
Reference in new issue
Block a user