diff --git a/.gitignore b/.gitignore index 2dbf912d..a2aa28a7 100644 --- a/.gitignore +++ b/.gitignore @@ -36,6 +36,20 @@ __pycache__/ *.stategraph *.rwcmset !/crates/skate-data/tests/fixtures/retail-collision.rwcmset +# Retail audio and anything decoded from it (tools/asset_pipeline/audio_export.py). +*.snr +*.sns +*.sek +*.abk +*.bnk +*.csi +*.ems +*.grain +*.mus +*.mpf +*.xma +*.wav +!/mods/Skyline_Drive_Mod/audio/*.wav .env .env.* *.pem diff --git a/tools/asset_pipeline/asset_exports.py b/tools/asset_pipeline/asset_exports.py index 4443631c..60108a51 100644 --- a/tools/asset_pipeline/asset_exports.py +++ b/tools/asset_pipeline/asset_exports.py @@ -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) diff --git a/tools/asset_pipeline/audio_export.py b/tools/asset_pipeline/audio_export.py new file mode 100644 index 00000000..c7b7e355 --- /dev/null +++ b/tools/asset_pipeline/audio_export.py @@ -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/). +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 `` files in `directory` to `.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 diff --git a/tools/asset_pipeline/audio_formats.py b/tools/asset_pipeline/audio_formats.py new file mode 100644 index 00000000..4f1e0a85 --- /dev/null +++ b/tools/asset_pipeline/audio_formats.py @@ -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=' 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=' 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} diff --git a/tools/asset_pipeline/group_receipts.py b/tools/asset_pipeline/group_receipts.py index 3f7a05c1..9b1eb5a7 100644 --- a/tools/asset_pipeline/group_receipts.py +++ b/tools/asset_pipeline/group_receipts.py @@ -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'), } diff --git a/tools/asset_pipeline/install.py b/tools/asset_pipeline/install.py index b497bb4d..43abc7e8 100644 --- a/tools/asset_pipeline/install.py +++ b/tools/asset_pipeline/install.py @@ -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 diff --git a/tools/asset_pipeline/test_audio_formats.py b/tools/asset_pipeline/test_audio_formats.py new file mode 100644 index 00000000..b4f2b8e9 --- /dev/null +++ b/tools/asset_pipeline/test_audio_formats.py @@ -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='