Created
August 1, 2026 18:09
-
-
Save pathawks/182a9a02c8b0417b54c8ba0cd8c73c95 to your computer and use it in GitHub Desktop.
Lakers versus Celtics and the NBA Playoffs - VGM
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| #!/usr/bin/env python3 | |
| """Extract the "Lakers versus Celtics and the NBA Playoffs" title music and sound effects as Tandy VGM. | |
| The selected Tandy sound resource is the game's ``T`` file. BBALL.EXE | |
| decodes it in place and then uses it for both the title-screen score in ``S`` | |
| and ten embedded effect-command entries. This script models both sides of | |
| that driver and records its SN76489-compatible port writes in VGM 1.71 files. | |
| The game does not attach names to the effect entries, so those outputs keep | |
| their stable command indices (00 through 09). | |
| No third-party Python modules are required. | |
| """ | |
| from __future__ import annotations | |
| import argparse | |
| import struct | |
| from dataclasses import dataclass | |
| from pathlib import Path | |
| VGM_SAMPLE_RATE = 44_100 | |
| TANDY_CLOCK_HZ = 3_579_545 | |
| PIT_INPUT_HZ = 1_193_182 | |
| PIT_DIVISOR = 0x4000 | |
| HEADER_SIZE = 0x100 | |
| # These offsets are in the decoded ``T`` resource, whose code is loaded at | |
| # segment 1d05 by BBALL.EXE. | |
| COMMAND_TABLE_OFFSET = 0x3B4 | |
| COMMAND_COUNT = 10 | |
| FIRST_SEQUENCER = 0x447 | |
| SECOND_SEQUENCER = 0x45D | |
| class ConversionError(Exception): | |
| """Raised when an input resource is not the expected Tandy driver.""" | |
| def read_u16(data: bytes | bytearray, offset: int) -> int: | |
| if offset < 0 or offset + 2 > len(data): | |
| raise ConversionError(f"read outside T resource at {offset:#x}") | |
| return int.from_bytes(data[offset : offset + 2], "little") | |
| def read_i16(data: bytes | bytearray, offset: int) -> int: | |
| value = read_u16(data, offset) | |
| return value - 0x1_0000 if value & 0x8000 else value | |
| def write_u32(data: bytearray, offset: int, value: int) -> None: | |
| data[offset : offset + 4] = struct.pack("<I", value) | |
| def decode_tandy_resource(packed: bytes) -> bytes: | |
| """Apply BBALL.EXE's three-byte in-place decode to the Tandy resource.""" | |
| if len(packed) < 0xA6E: | |
| raise ConversionError(f"T: {len(packed)} bytes; resource is truncated") | |
| decoded = bytearray(packed) | |
| # BBALL.EXE always walks a 0xffc-byte buffer, but the shipped file is | |
| # smaller (the unused tail is whatever remained in its load segment). | |
| # All code and command streams reside inside T itself. | |
| decoded_length = min(0xFFC, len(decoded)) | |
| for offset in range(0, decoded_length - decoded_length % 3, 3): | |
| first, middle, third = decoded[offset : offset + 3] | |
| decoded[offset] = third ^ middle | |
| decoded[offset + 2] = first ^ middle | |
| return bytes(decoded) | |
| @dataclass(frozen=True) | |
| class EffectCommand: | |
| """One playable entry from the driver's AL command table.""" | |
| index: int | |
| sequencer: str | |
| master_stream: int | |
| @property | |
| def filename(self) -> str: | |
| return f"tandy-command-{self.index:02d}.vgm" | |
| @property | |
| def title(self) -> str: | |
| return f"Tandy command {self.index:02d}" | |
| def identify_effect_commands(driver: bytes) -> tuple[EffectCommand, ...]: | |
| """Decode the ten table entries BBALL.EXE can request with AH=3.""" | |
| if driver[:3] != b"\xe9\xd1\x03" or read_u16(driver, 0x3CC) != 0x4FA: | |
| raise ConversionError("T: decoded bytes do not match the Lakers vs Celtics driver") | |
| commands: list[EffectCommand] = [] | |
| for index in range(COMMAND_COUNT): | |
| handler = read_u16(driver, COMMAND_TABLE_OFFSET + index * 2) | |
| if handler + 6 > len(driver) or driver[handler] != 0xB8 or driver[handler + 3] != 0xE8: | |
| raise ConversionError(f"T: command {index} has an unexpected handler at {handler:#x}") | |
| master_stream = read_u16(driver, handler + 1) | |
| sequencer_target = handler + 6 + read_i16(driver, handler + 4) | |
| if sequencer_target == FIRST_SEQUENCER: | |
| sequencer = "first" | |
| elif sequencer_target == SECOND_SEQUENCER: | |
| sequencer = "second" | |
| else: | |
| raise ConversionError( | |
| f"T: command {index} calls unknown sequencer {sequencer_target:#x}" | |
| ) | |
| if not 0 <= master_stream < len(driver): | |
| raise ConversionError(f"T: command {index} stream is outside the resource") | |
| commands.append(EffectCommand(index, sequencer, master_stream)) | |
| return tuple(commands) | |
| class VgmWriter: | |
| """Minimal VGM command writer for one SN76489-compatible PSG.""" | |
| def __init__(self) -> None: | |
| self.commands = bytearray() | |
| self.samples = 0 | |
| def psg(self, value: int) -> None: | |
| self.commands.extend((0x50, value & 0xFF)) | |
| def wait(self, samples: int) -> None: | |
| while samples: | |
| chunk = min(samples, 0xFFFF) | |
| self.commands.extend((0x61, chunk & 0xFF, chunk >> 8)) | |
| self.samples += chunk | |
| samples -= chunk | |
| @dataclass | |
| class SequencerState: | |
| active: bool = False | |
| master_start: int = 0 | |
| master_pos: int = 0 | |
| phase_count: int = 0 | |
| phase_initial: int = 0 | |
| repeat_count: int = 0 | |
| repeat_resets_pattern: bool = False | |
| pattern_start: int = 0 | |
| pattern_pos: int = 0 | |
| note_count: int = 0 | |
| note_initial: int = 0 | |
| pitch_start: int = 0 | |
| pitch: int = 0 | |
| pitch_delta: int = 0 | |
| tone_mode: bool = False | |
| class TandySequencer: | |
| """Faithful model of the two simple streams in the decoded T driver.""" | |
| def __init__(self, driver: bytes, writer: VgmWriter) -> None: | |
| self.driver = driver | |
| self.writer = writer | |
| self.first = SequencerState() | |
| self.second = SequencerState() | |
| def start(self, command: EffectCommand) -> None: | |
| state = self.first if command.sequencer == "first" else self.second | |
| state.active = True | |
| state.master_start = command.master_stream | |
| state.master_pos = command.master_stream | |
| state.phase_count = 0 | |
| state.repeat_count = 0 | |
| def running(self) -> bool: | |
| return self.first.active or self.second.active | |
| def tick(self) -> None: | |
| self._tick_first() | |
| self._tick_second() | |
| def _take_byte(self, state: SequencerState) -> int: | |
| if state.master_pos >= len(self.driver): | |
| raise ConversionError("T: master stream extends outside the resource") | |
| value = self.driver[state.master_pos] | |
| state.master_pos += 1 | |
| return value | |
| def _take_u16(self, state: SequencerState) -> int: | |
| if state.master_pos + 2 > len(self.driver): | |
| raise ConversionError("T: master stream has a truncated word") | |
| value = read_u16(self.driver, state.master_pos) | |
| state.master_pos += 2 | |
| return value | |
| def _set_pattern(self, state: SequencerState, pointer: int) -> None: | |
| if pointer == 0 or pointer >= len(self.driver): | |
| raise ConversionError(f"T: invalid pattern pointer {pointer:#x}") | |
| state.pattern_start = pointer | |
| state.pattern_pos = pointer | |
| state.note_initial = self.driver[pointer - 1] | |
| state.note_count = state.note_initial | |
| @staticmethod | |
| def _split_period(period: int) -> tuple[int, int]: | |
| period &= 0x3FF | |
| return period & 0x0F, (period >> 4) & 0x3F | |
| def _advance_master(self, state: SequencerState, first: bool) -> bool: | |
| """Advance a stream to one 0x83 unit, or stop it at 0x84.""" | |
| while True: | |
| command = self._take_byte(state) | |
| if command == 0x80: | |
| self._set_pattern(state, self._take_u16(state)) | |
| elif command == 0x84: | |
| self.writer.psg(0xBF if first else 0xDF) | |
| if not first: | |
| self.writer.psg(0xFF) | |
| state.active = False | |
| return False | |
| elif command == 0x83: | |
| state.phase_count = self._take_byte(state) | |
| state.phase_initial = state.phase_count | |
| state.pitch_start = self._take_u16(state) | |
| state.pitch = state.pitch_start | |
| state.pitch_delta = self._take_u16(state) | |
| return True | |
| elif command == 0x81: | |
| state.repeat_count = self._take_byte(state) | |
| state.repeat_resets_pattern = False | |
| elif command == 0x82: | |
| state.repeat_count = self._take_byte(state) | |
| state.repeat_resets_pattern = True | |
| elif command == 0x87: | |
| state.master_pos = state.master_start | |
| # The first sequencer's dispatcher has no 0x85/0x86 cases and | |
| # simply skips them. Two shipped first-voice streams contain | |
| # 0x85, inherited from the second sequencer's vocabulary. | |
| elif first and command in (0x85, 0x86): | |
| pass | |
| elif not first and command == 0x85: | |
| state.tone_mode = True | |
| elif not first and command == 0x86: | |
| state.tone_mode = False | |
| else: | |
| raise ConversionError(f"T: unexpected stream command {command:#04x}") | |
| def _prepare_tick(self, state: SequencerState, first: bool) -> bool: | |
| if not state.active: | |
| return False | |
| if state.phase_count: | |
| state.pitch = (state.pitch + state.pitch_delta) & 0xFFFF | |
| state.phase_count = (state.phase_count - 1) & 0xFF | |
| return True | |
| if state.repeat_count: | |
| state.repeat_count = (state.repeat_count - 1) & 0xFF | |
| state.pitch = state.pitch_start | |
| state.phase_count = state.phase_initial | |
| if state.repeat_resets_pattern: | |
| state.pattern_pos = state.pattern_start | |
| state.note_count = state.note_initial | |
| return True | |
| return self._advance_master(state, first) | |
| def _next_note(self, state: SequencerState) -> int | None: | |
| state.note_count = (state.note_count - 1) & 0xFF | |
| if state.note_count: | |
| return None | |
| state.note_count = state.note_initial | |
| if state.pattern_pos >= len(self.driver): | |
| raise ConversionError("T: pattern extends outside the resource") | |
| note = self.driver[state.pattern_pos] | |
| if note == 0xFF: | |
| return None | |
| state.pattern_pos += 1 | |
| return note | |
| def _tick_first(self) -> None: | |
| state = self.first | |
| if not self._prepare_tick(state, first=True): | |
| return | |
| note = self._next_note(state) | |
| if note is not None: | |
| self.writer.psg(0xB0 | note) | |
| low, high = self._split_period(state.pitch) | |
| self.writer.psg(0xA0 | low) | |
| self.writer.psg(high) | |
| def _tick_second(self) -> None: | |
| state = self.second | |
| if not self._prepare_tick(state, first=False): | |
| return | |
| note = self._next_note(state) | |
| if note is not None: | |
| if state.tone_mode: | |
| self.writer.psg(0xFF) | |
| self.writer.psg(0xD0 | note) | |
| else: | |
| self.writer.psg(0xE7 | note) | |
| self.writer.psg(0xF0 | note) | |
| self.writer.psg(0xDF) | |
| low, high = self._split_period(state.pitch) | |
| self.writer.psg(0xC0 | low) | |
| self.writer.psg(high) | |
| @dataclass | |
| class MusicVoice: | |
| """Runtime state for one of the title score's four mapped PSG voices.""" | |
| index: int | |
| midi_channel: int | |
| transpose: int | |
| active: bool = False | |
| note: int = 0 | |
| age: int = 0 | |
| envelope_delay_initial: int = 0 | |
| envelope_delay: int = 0 | |
| attenuation: int = 0x0F | |
| envelope_active: bool = False | |
| volume_start: int = 0 | |
| volume_pos: int = 0 | |
| pitch_start: int = 0 | |
| pitch_pos: int = 0 | |
| pitch_divisor: int = 0 | |
| base_period: int = 0 | |
| pitch_bend_event: bool = False | |
| class TandyMusicDriver: | |
| """Model the decoded T driver's external score path used for resource S.""" | |
| def __init__(self, driver: bytes, score: bytes, writer: VgmWriter) -> None: | |
| if len(score) <= 0x4A: | |
| raise ConversionError("S: title score is truncated") | |
| self.driver = driver | |
| self.score = score | |
| self.writer = writer | |
| self.start = read_u16(score, 0) | |
| if not 0 < self.start < len(score): | |
| raise ConversionError(f"S: invalid event-stream offset {self.start:#x}") | |
| self.pos = self.start | |
| self.delay = 0 | |
| self.running = True | |
| self.running_status: int | None = None | |
| self.event_count = 0 | |
| self.loop_count = 0 | |
| self.terminator: int | None = None | |
| self.voices = [ | |
| MusicVoice( | |
| index=index, | |
| midi_channel=score[0x43 + index], | |
| transpose=self._signed_byte(score[0x47 + index]), | |
| ) | |
| for index in range(4) | |
| ] | |
| @staticmethod | |
| def _signed_byte(value: int) -> int: | |
| return value - 0x100 if value & 0x80 else value | |
| def _take_score_byte(self) -> int: | |
| if self.pos >= len(self.score): | |
| raise ConversionError("S: event stream reaches the end without FC 80/81") | |
| value = self.score[self.pos] | |
| self.pos += 1 | |
| return value | |
| def _write_period(self, voice: MusicVoice, period: int) -> None: | |
| if voice.index >= 3: | |
| return | |
| period &= 0x3FF | |
| latch = self.driver[0x1CE + voice.index] | |
| self.writer.psg(latch | (period & 0x0F)) | |
| self.writer.psg((period >> 4) & 0x3F) | |
| def _write_attenuation(self, voice: MusicVoice, attenuation: int) -> None: | |
| voice.attenuation = attenuation & 0xFF | |
| latch = self.driver[0x1D2 + voice.index] | |
| self.writer.psg(latch | voice.attenuation) | |
| def _set_program(self, voice: MusicVoice, program: int) -> None: | |
| map_offset = 0x7A + program | |
| if map_offset >= len(self.driver): | |
| raise ConversionError(f"T: program {program} is outside the patch map") | |
| patch_index = self.driver[map_offset] | |
| descriptor = read_u16(self.driver, 0xFA + patch_index * 2) | |
| voice.volume_start = read_u16(self.driver, descriptor) | |
| voice.pitch_start = read_u16(self.driver, descriptor + 2) | |
| voice.pitch_divisor = self.driver[descriptor + 4] | |
| if voice.volume_start == 0 or voice.volume_start >= len(self.driver): | |
| raise ConversionError(f"T: program {program} has an invalid volume envelope") | |
| if voice.pitch_start >= len(self.driver): | |
| raise ConversionError(f"T: program {program} has an invalid pitch envelope") | |
| voice.envelope_delay_initial = self.driver[voice.volume_start - 1] | |
| def _start_envelope(self, voice: MusicVoice, base_period: int) -> None: | |
| if voice.volume_start == 0: | |
| raise ConversionError( | |
| f"S: channel {voice.midi_channel} plays before selecting a program" | |
| ) | |
| voice.envelope_delay = voice.envelope_delay_initial | |
| voice.envelope_active = True | |
| self._write_attenuation(voice, self.driver[voice.volume_start]) | |
| voice.volume_pos = voice.volume_start | |
| voice.base_period = base_period & 0xFFFF | |
| voice.pitch_pos = voice.pitch_start | |
| def _normal_period(self, voice: MusicVoice, note: int) -> int: | |
| adjusted = (note + voice.transpose) & 0xFF | |
| adjusted = self._signed_byte(adjusted) | |
| while adjusted < 0: | |
| adjusted += 12 | |
| while adjusted > 0x32: | |
| adjusted -= 12 | |
| return read_u16(self.driver, 0x14 + adjusted * 2) | |
| def _note_on(self, channel: int, note: int, velocity: int) -> None: | |
| if velocity == 0: | |
| self._note_off(channel, note) | |
| return | |
| if channel == 9: | |
| if note not in (0x24, 0x26): | |
| return | |
| voice = self.voices[3] | |
| program = 4 if note == 0x24 else 5 | |
| self._set_program(voice, program) | |
| self.writer.psg(0xE6 if note == 0x24 else 0xE4) | |
| self._start_envelope(voice, program) | |
| return | |
| voice = next( | |
| ( | |
| candidate | |
| for candidate in self.voices | |
| if candidate.midi_channel == channel and not candidate.active | |
| ), | |
| None, | |
| ) | |
| if voice is None: | |
| candidates = [ | |
| candidate | |
| for candidate in self.voices | |
| if candidate.midi_channel == channel and candidate.active | |
| ] | |
| voice = max(candidates, key=lambda candidate: candidate.age, default=None) | |
| if voice is None or voice.age == 0: | |
| return | |
| voice.age = 0 | |
| voice.active = True | |
| voice.note = note | |
| period = self._normal_period(voice, note) | |
| self._write_period(voice, period) | |
| self._start_envelope(voice, period) | |
| def _note_off(self, channel: int, note: int) -> None: | |
| for voice in self.voices: | |
| if voice.midi_channel != channel or voice.note != note: | |
| continue | |
| voice.active = False | |
| voice.note = 0 | |
| self._write_attenuation(voice, 0x0F) | |
| voice.envelope_active = False | |
| voice.age = 0 | |
| return | |
| def _program_change(self, channel: int, program: int) -> None: | |
| if channel == 9: | |
| return | |
| for voice in self.voices: | |
| if voice.midi_channel == channel: | |
| self._set_program(voice, program) | |
| def _pitch_bend(self, channel: int, low: int, high: int) -> None: | |
| bend = (high << 7) | low | |
| if bend == 0: | |
| raise ConversionError("S: zero pitch-bend divisor") | |
| for voice in self.voices[:3]: | |
| if voice.midi_channel != channel or not voice.active: | |
| continue | |
| voice.pitch_bend_event = True | |
| period = (voice.base_period * 0x2000) // bend | |
| self._write_period(voice, period) | |
| def _tick_voice(self, voice: MusicVoice) -> None: | |
| if voice.envelope_active: | |
| voice.envelope_delay = (voice.envelope_delay - 1) & 0xFF | |
| if voice.envelope_delay == 0: | |
| voice.envelope_delay = voice.envelope_delay_initial | |
| if voice.volume_pos >= len(self.driver): | |
| raise ConversionError("T: volume envelope extends outside the driver") | |
| attenuation = self.driver[voice.volume_pos] | |
| if attenuation != 0xFF: | |
| voice.volume_pos += 1 | |
| self._write_attenuation(voice, attenuation) | |
| if voice.pitch_divisor == 0 or voice.pitch_bend_event: | |
| return | |
| quotient = voice.base_period // voice.pitch_divisor | |
| if quotient > 0xFF: | |
| raise ConversionError("T: pitch envelope division overflows the driver") | |
| if voice.pitch_pos >= len(self.driver): | |
| raise ConversionError("T: pitch envelope extends outside the driver") | |
| pitch_step = self.driver[voice.pitch_pos] | |
| if pitch_step == 0x80: | |
| voice.pitch_pos = voice.pitch_start | |
| pitch_step = self.driver[voice.pitch_pos] | |
| signed_quotient = self._signed_byte(quotient) | |
| signed_step = self._signed_byte(pitch_step) | |
| period = (voice.base_period + signed_quotient * signed_step) & 0xFFFF | |
| voice.pitch_pos += 1 | |
| if period >= 0x3FF: | |
| period = 0x3FF | |
| self._write_period(voice, period) | |
| def _tick_voices(self) -> None: | |
| for voice in self.voices: | |
| self._tick_voice(voice) | |
| voice.pitch_bend_event = False | |
| if voice.active: | |
| voice.age = (voice.age + 1) & 0xFFFF | |
| def tick(self) -> None: | |
| if not self.running: | |
| return | |
| if self.delay: | |
| self.delay = (self.delay - 1) & 0xFF | |
| self._tick_voices() | |
| return | |
| while True: | |
| event = self._take_score_byte() | |
| if event == 0xFC: | |
| terminator = self._take_score_byte() | |
| if terminator == 0x80: | |
| self.pos = self.start | |
| self.loop_count += 1 | |
| continue | |
| self.terminator = terminator | |
| self.running = False | |
| return | |
| if event & 0x80: | |
| self.running_status = event | |
| else: | |
| if self.running_status is None: | |
| raise ConversionError("S: data byte appears before a status byte") | |
| self.pos -= 1 | |
| status = self.running_status | |
| assert status is not None | |
| command = status & 0xF0 | |
| channel = status & 0x0F | |
| if command == 0xE0: | |
| low = self._take_score_byte() | |
| high = self._take_score_byte() | |
| self._pitch_bend(channel, low, high) | |
| elif command == 0xC0: | |
| self._program_change(channel, self._take_score_byte()) | |
| elif command == 0x90: | |
| note = self._take_score_byte() | |
| velocity = self._take_score_byte() | |
| self._note_on(channel, note, velocity) | |
| elif command == 0x80: | |
| note = self._take_score_byte() | |
| self._take_score_byte() | |
| self._note_off(channel, note) | |
| else: | |
| # This matches the original driver's fallback for statuses it | |
| # does not synthesize: consume two data bytes and continue. | |
| self._take_score_byte() | |
| self._take_score_byte() | |
| self.event_count += 1 | |
| delta = self._take_score_byte() | |
| if delta == 0: | |
| continue | |
| self.delay = delta - 1 | |
| self._tick_voices() | |
| return | |
| def encode_gd3(title: str, source_resource: str, notes: str) -> bytes: | |
| fields = ( | |
| title, | |
| "", | |
| "Lakers vs Celtics", | |
| "", | |
| "IBM PC / Tandy 1000", | |
| "", | |
| "", | |
| "", | |
| "", | |
| "extract_music.py", | |
| f"{notes} Source resource: {source_resource}. The game advances the " | |
| "Tandy driver from IRQ 0 with PIT divisor 0x4000.", | |
| ) | |
| body = "\0".join(fields).encode("utf-16le") + b"\0\0" | |
| return b"Gd3 " + struct.pack("<II", 0x00000100, len(body)) + body | |
| def wait_one_irq(writer: VgmWriter, completed_intervals: int) -> None: | |
| before = completed_intervals * VGM_SAMPLE_RATE * PIT_DIVISOR // PIT_INPUT_HZ | |
| after = (completed_intervals + 1) * VGM_SAMPLE_RATE * PIT_DIVISOR // PIT_INPUT_HZ | |
| writer.wait(after - before) | |
| def finish_vgm( | |
| writer: VgmWriter, title: str, source_resource: str, notes: str | |
| ) -> bytes: | |
| writer.commands.append(0x66) | |
| output = bytearray(HEADER_SIZE) | |
| output[:4] = b"Vgm " | |
| write_u32(output, 0x08, 0x00000171) | |
| write_u32(output, 0x0C, TANDY_CLOCK_HZ) | |
| # Match the Tandy settings used by the existing Fire Hawk extractor. | |
| output[0x28:0x2A] = struct.pack("<H", 0x000C) | |
| output[0x2A] = 17 | |
| write_u32(output, 0x34, HEADER_SIZE - 0x34) | |
| write_u32(output, 0x18, writer.samples) | |
| output.extend(writer.commands) | |
| gd3_offset = len(output) | |
| output.extend(encode_gd3(title, source_resource, notes)) | |
| write_u32(output, 0x14, gd3_offset - 0x14) | |
| write_u32(output, 0x04, len(output) - 0x04) | |
| return bytes(output) | |
| def build_effect_vgm(driver: bytes, command: EffectCommand) -> tuple[bytes, int]: | |
| writer = VgmWriter() | |
| # This is the driver's AH=1 initialization: mute all three tones and noise. | |
| for value in (0x9F, 0xBF, 0xDF, 0xFF): | |
| writer.psg(value) | |
| sequencer = TandySequencer(driver, writer) | |
| sequencer.start(command) | |
| ticks = 0 | |
| max_ticks = 20_000 | |
| while sequencer.running(): | |
| sequencer.tick() | |
| ticks += 1 | |
| if ticks > max_ticks: | |
| raise ConversionError(f"T: command {command.index} did not finish") | |
| if sequencer.running(): | |
| wait_one_irq(writer, ticks - 1) | |
| return ( | |
| finish_vgm( | |
| writer, | |
| command.title, | |
| "T", | |
| "Tandy effect command-table entry rendered from the decoded driver.", | |
| ), | |
| ticks, | |
| ) | |
| def build_title_vgm(driver: bytes, score: bytes) -> tuple[bytes, int, int]: | |
| writer = VgmWriter() | |
| # AH=1 initializes the score state and mutes all PSG voices. | |
| for value in (0x9F, 0xBF, 0xDF, 0xFF): | |
| writer.psg(value) | |
| music = TandyMusicDriver(driver, score, writer) | |
| ticks = 0 | |
| max_ticks = 1_000_000 | |
| while music.running: | |
| music.tick() | |
| ticks += 1 | |
| if ticks > max_ticks: | |
| raise ConversionError("S: title score did not finish") | |
| if music.running: | |
| wait_one_irq(writer, ticks - 1) | |
| if music.terminator != 0x81 or music.pos != len(score): | |
| raise ConversionError( | |
| "S: title score did not end at its expected FC 81 terminator" | |
| ) | |
| # The game stops the driver when it leaves this presentation state. | |
| for value in (0x9F, 0xBF, 0xDF, 0xFF): | |
| writer.psg(value) | |
| return ( | |
| finish_vgm( | |
| writer, | |
| "Title Screen", | |
| "S and T", | |
| "Title score S rendered through the decoded Tandy driver T.", | |
| ), | |
| ticks, | |
| music.event_count, | |
| ) | |
| def validate_vgm(data: bytes, expected_samples: int, name: str) -> None: | |
| if data[:4] != b"Vgm ": | |
| raise ConversionError(f"{name}: missing VGM signature") | |
| if read_u16(data, 0x08) != 0x0171: | |
| raise ConversionError(f"{name}: unexpected VGM version") | |
| if int.from_bytes(data[0x18:0x1C], "little") != expected_samples: | |
| raise ConversionError(f"{name}: incorrect total-sample field") | |
| position = 0x34 + int.from_bytes(data[0x34:0x38], "little") | |
| samples = 0 | |
| ended = False | |
| while position < len(data): | |
| opcode = data[position] | |
| position += 1 | |
| if opcode == 0x50: | |
| position += 1 | |
| elif opcode == 0x61: | |
| samples += read_u16(data, position) | |
| position += 2 | |
| elif opcode == 0x66: | |
| ended = True | |
| break | |
| else: | |
| raise ConversionError(f"{name}: unexpected VGM command {opcode:#04x}") | |
| if not ended or position > len(data) or samples != expected_samples: | |
| raise ConversionError( | |
| f"{name}: command stream has {samples} samples; expected {expected_samples}" | |
| ) | |
| def extract(data_dir: Path, output_dir: Path, list_only: bool = False) -> None: | |
| driver_path = data_dir / "T" | |
| score_path = data_dir / "S" | |
| if not driver_path.is_file(): | |
| raise ConversionError(f"could not find the Tandy driver resource: {driver_path}") | |
| if not score_path.is_file(): | |
| raise ConversionError(f"could not find the title-score resource: {score_path}") | |
| driver = decode_tandy_resource(driver_path.read_bytes()) | |
| score = score_path.read_bytes() | |
| commands = identify_effect_commands(driver) | |
| if list_only: | |
| print( | |
| f"title external stream {read_u16(score, 0):#05x} " | |
| "S-title-screen-tandy.vgm" | |
| ) | |
| for command in commands: | |
| print( | |
| f"{command.index:02d} {command.sequencer:6s} " | |
| f"stream {command.master_stream:#05x} {command.filename}" | |
| ) | |
| return | |
| output_dir.mkdir(parents=True, exist_ok=True) | |
| title_name = "S-title-screen-tandy.vgm" | |
| title_vgm, title_ticks, title_events = build_title_vgm(driver, score) | |
| validate_vgm( | |
| title_vgm, | |
| int.from_bytes(title_vgm[0x18:0x1C], "little"), | |
| title_name, | |
| ) | |
| (output_dir / title_name).write_bytes(title_vgm) | |
| title_samples = int.from_bytes(title_vgm[0x18:0x1C], "little") | |
| print( | |
| f"{title_name}: {title_ticks} IRQ ticks, {title_events} events, " | |
| f"{title_samples} samples" | |
| ) | |
| for command in commands: | |
| vgm, ticks = build_effect_vgm(driver, command) | |
| validate_vgm(vgm, int.from_bytes(vgm[0x18:0x1C], "little"), command.filename) | |
| destination = output_dir / command.filename | |
| destination.write_bytes(vgm) | |
| samples = int.from_bytes(vgm[0x18:0x1C], "little") | |
| print(f"{destination.name}: {ticks} IRQ ticks, {samples} samples") | |
| def main() -> None: | |
| script_dir = Path(__file__).resolve().parent | |
| parser = argparse.ArgumentParser(description=__doc__) | |
| parser.add_argument( | |
| "--data-dir", | |
| type=Path, | |
| default=script_dir / "Lakers vs Celtics Data", | |
| help="directory containing the title score S and Tandy driver T", | |
| ) | |
| parser.add_argument( | |
| "--output-dir", | |
| type=Path, | |
| default=script_dir / "Music", | |
| help="directory in which to write the Tandy VGM files", | |
| ) | |
| parser.add_argument( | |
| "--list", | |
| action="store_true", | |
| help="list the title score and Tandy effect resources without writing files", | |
| ) | |
| args = parser.parse_args() | |
| try: | |
| extract(args.data_dir, args.output_dir, args.list) | |
| except ConversionError as error: | |
| parser.error(str(error)) | |
| if __name__ == "__main__": | |
| main() |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment