Extract Fire Hawk

extract_tandy_vgm.py

#!/usr/bin/env python3
"""Extract Fire Hawk's Tandy/PCjr and Sound Blaster soundtracks as VGM files.

The DOS release stores each song as an LSB-first Sierra LZW resource named
SSnnn.  Every unpacked sound resource contains several hardware-specific
tracks.  This converter selects track 0x13 for the three-voice PCjr/Tandy
versions and track 0x08 for the nine-voice Sound Blaster OPL2 versions.  It
uses the matching PATCH.101 and PATCH.003 data and models the respective
Sierra music-driver register writes.

No third-party Python modules are required.
"""

from __future__ import annotations

import argparse
import math
import struct
import wave
from dataclasses import dataclass
from pathlib import Path
from typing import Iterable


VGM_SAMPLE_RATE = 44_100
TICKS_PER_SECOND = 60
SAMPLES_PER_TICK = VGM_SAMPLE_RATE // TICKS_PER_SECOND
TANDY_CLOCK_HZ = 3_579_545
SOUND_BLASTER_CLOCK_HZ = 3_579_545
TANDY_TRACK_TYPE = 0x13
SOUND_BLASTER_TRACK_TYPE = 0x08
ADLIB_TRACK_TYPE = 0x00
PC_SPEAKER_TRACK_TYPE = 0x12
TANDY_DRIVER_PERIOD_TABLE = 0x21
TANDY_DRIVER_VELOCITY_TABLE = 0x273
RESOURCE_PREFIX_SIZE = 2

SONGS = (
    ("SS000", "SS000-opening-theme-tandy.vgm", "Opening Theme"),
    ("SS010", "SS010-mission-1-tandy.vgm", "Mission 1"),
    ("SS011", "SS011-boss-1-tandy.vgm", "Boss 1"),
    ("SS020", "SS020-mission-2-tandy.vgm", "Mission 2"),
    ("SS021", "SS021-boss-2-tandy.vgm", "Boss 2"),
    ("SS030", "SS030-mission-3-tandy.vgm", "Mission 3"),
    ("SS031", "SS031-boss-3-tandy.vgm", "Boss 3"),
    ("SS040", "SS040-mission-4-tandy.vgm", "Mission 4"),
    ("SS041", "SS041-boss-4-tandy.vgm", "Boss 4"),
    ("SS050", "SS050-mission-5-tandy.vgm", "Mission 5"),
    ("SS051", "SS051-boss-5-tandy.vgm", "Boss 5"),
    ("SS060", "SS060-mission-6-tandy.vgm", "Mission 6"),
    ("SS061", "SS061-boss-6-tandy.vgm", "Boss 6"),
    ("SS070", "SS070-mission-7-tandy.vgm", "Mission 7"),
    ("SS071", "SS071-boss-7-tandy.vgm", "Boss 7"),
    ("SS080", "SS080-mission-8-tandy.vgm", "Mission 8"),
    ("SS081", "SS081-boss-8-tandy.vgm", "Boss 8"),
    ("SS090", "SS090-mission-9-tandy.vgm", "Mission 9"),
    ("SS091", "SS091-boss-9-tandy.vgm", "Boss 9"),
    ("SS092", "SS092-ending-tandy.vgm", "Ending"),
    ("SS120", "SS120-moonlight-sonata-tandy.vgm", "Moonlight Sonata"),
)


class ConversionError(Exception):
    """Raised when an input resource does not match the expected format."""


def read_u16(data: bytes, offset: int) -> int:
    return int.from_bytes(data[offset : offset + 2], "little")


def write_u32(data: bytearray, offset: int, value: int) -> None:
    data[offset : offset + 4] = struct.pack("<I", value)


def decompress_sierra_lzw(packed: bytes, source_name: str) -> bytes:
    """Decode the SCI-era LSB-first LZW wrapper used by the loose resources."""
    if len(packed) < 6:
        raise ConversionError(f"{source_name}: truncated LZW resource")

    expected_size = int.from_bytes(packed[:4], "little")
    source = packed[4:]
    source_pos = 0
    bit_buffer = 0
    bit_count = 0
    code_width = 9
    next_code = 258
    code_limit = 512
    dictionary = {value: bytes((value,)) for value in range(256)}
    previous: bytes | None = None
    output = bytearray()

    def get_code() -> int:
        nonlocal source_pos, bit_buffer, bit_count
        while bit_count < code_width:
            if source_pos >= len(source):
                raise ConversionError(f"{source_name}: truncated LZW bitstream")
            bit_buffer |= source[source_pos] << bit_count
            source_pos += 1
            bit_count += 8
        code = bit_buffer & ((1 << code_width) - 1)
        bit_buffer >>= code_width
        bit_count -= code_width
        return code

    while len(output) < expected_size:
        code = get_code()
        if code == 257:
            break
        if code == 256:
            dictionary = {value: bytes((value,)) for value in range(256)}
            code_width = 9
            next_code = 258
            code_limit = 512
            previous = None
            continue

        if code in dictionary:
            current = dictionary[code]
        elif code == next_code and previous is not None:
            current = previous + previous[:1]
        else:
            raise ConversionError(
                f"{source_name}: invalid LZW code {code:#x} "
                f"(next code is {next_code:#x})"
            )

        output.extend(current)
        if previous is not None and next_code < 4096:
            dictionary[next_code] = previous + current[:1]
            next_code += 1
            if next_code == code_limit and code_width < 12:
                code_width += 1
                code_limit <<= 1
        previous = current

    if len(output) != expected_size:
        raise ConversionError(
            f"{source_name}: LZW header says {expected_size} bytes, "
            f"decoded {len(output)}"
        )
    return bytes(output)


@dataclass(frozen=True)
class MidiEvent:
    tick: int
    stream_index: int
    sequence: int
    status: int
    operands: tuple[int, ...]


@dataclass(frozen=True)
class TandyStream:
    part: int
    polyphony: int
    priority: int
    events: tuple[MidiEvent, ...]
    end_tick: int


def find_track_descriptors(
    sound: bytes, source_name: str, track_type_to_find: int, track_name: str
) -> list[tuple[int, int]]:
    """Return the offset/size pairs from a requested SCI sound track.

    Fire Hawk's loose resource still includes its two-byte Sierra resource
    prefix.  Directory offsets are relative to the bytes after that prefix.
    """
    data = sound[RESOURCE_PREFIX_SIZE:]
    pos = 0
    while pos < len(data):
        track_type = data[pos]
        pos += 1
        if track_type == 0xFF:
            break

        descriptors: list[tuple[int, int]] = []
        while pos < len(data) and data[pos] != 0xFF:
            if pos + 6 > len(data):
                raise ConversionError(f"{source_name}: truncated sound-track directory")
            _unknown = read_u16(data, pos)
            offset = read_u16(data, pos + 2)
            size = read_u16(data, pos + 4)
            descriptors.append((offset, size))
            pos += 6
        if pos >= len(data):
            raise ConversionError(f"{source_name}: unterminated sound-track directory")
        pos += 1

        if track_type == track_type_to_find:
            return descriptors

    raise ConversionError(f"{source_name}: no {track_name} track {track_type_to_find:#04x}")


def find_tandy_descriptors(sound: bytes, source_name: str) -> list[tuple[int, int]]:
    descriptors = find_track_descriptors(sound, source_name, TANDY_TRACK_TYPE, "Tandy/PCjr")
    if len(descriptors) != 3:
        raise ConversionError(
            f"{source_name}: Tandy track has {len(descriptors)} channels, expected 3"
        )
    return descriptors


def find_sound_blaster_descriptors(sound: bytes, source_name: str) -> list[tuple[int, int]]:
    descriptors = find_track_descriptors(
        sound, source_name, SOUND_BLASTER_TRACK_TYPE, "Sound Blaster"
    )
    if not 1 <= len(descriptors) <= 9:
        raise ConversionError(
            f"{source_name}: Sound Blaster track has {len(descriptors)} channels"
        )
    return descriptors


def find_adlib_descriptors(sound: bytes, source_name: str) -> list[tuple[int, int]]:
    descriptors = find_track_descriptors(sound, source_name, ADLIB_TRACK_TYPE, "AdLib")
    if not 1 <= len(descriptors) <= 9:
        raise ConversionError(f"{source_name}: AdLib track has {len(descriptors)} channels")
    return descriptors


def find_pc_speaker_descriptors(sound: bytes, source_name: str) -> list[tuple[int, int]]:
    descriptors = find_track_descriptors(
        sound, source_name, PC_SPEAKER_TRACK_TYPE, "PC speaker"
    )
    if not 1 <= len(descriptors) <= 2:
        raise ConversionError(
            f"{source_name}: PC speaker track has {len(descriptors)} channels"
        )
    return descriptors


MIDI_OPERAND_COUNTS = {
    0x80: 2,
    0x90: 2,
    0xA0: 2,
    0xB0: 2,
    0xC0: 1,
    0xD0: 1,
    0xE0: 2,
}


def parse_stream(
    sound: bytes,
    descriptor: tuple[int, int],
    stream_index: int,
    source_name: str,
) -> TandyStream:
    offset, size = descriptor
    start = RESOURCE_PREFIX_SIZE + offset
    end = start + size
    if end > len(sound) or size < 3:
        raise ConversionError(
            f"{source_name}: channel {stream_index + 1} points outside the resource"
        )

    channel = sound[start:end]
    part = channel[0] & 0x0F
    polyphony = channel[1] & 0x0F
    priority = channel[1] >> 4
    pos = 2
    tick = 0
    sequence = 0
    running_status: int | None = None
    events: list[MidiEvent] = []

    while pos < len(channel):
        if channel[pos] == 0xFC:
            return TandyStream(part, polyphony, priority, tuple(events), tick)

        while channel[pos] == 0xF8:
            tick += 240
            pos += 1
            if pos >= len(channel):
                raise ConversionError(f"{source_name}: truncated delay in Tandy channel")

        tick += channel[pos]
        pos += 1
        if pos >= len(channel):
            raise ConversionError(f"{source_name}: missing event after channel delay")

        command = channel[pos]
        pos += 1
        if command == 0xFC:
            return TandyStream(part, polyphony, priority, tuple(events), tick)
        if command == 0xF0:
            while pos < len(channel) and channel[pos] != 0xF7:
                pos += 1
            if pos >= len(channel):
                raise ConversionError(f"{source_name}: unterminated system-exclusive event")
            pos += 1
            sequence += 1
            continue

        first_operand: int | None = None
        if command & 0x80:
            running_status = command
        else:
            if running_status is None:
                raise ConversionError(
                    f"{source_name}: running status before a MIDI status byte"
                )
            first_operand = command
            command = running_status

        operand_count = MIDI_OPERAND_COUNTS.get(command & 0xF0)
        if operand_count is None:
            raise ConversionError(
                f"{source_name}: unsupported MIDI status {command:#04x}"
            )
        operands = [] if first_operand is None else [first_operand]
        needed = operand_count - len(operands)
        if pos + needed > len(channel):
            raise ConversionError(f"{source_name}: truncated MIDI event {command:#04x}")
        operands.extend(channel[pos : pos + needed])
        pos += needed
        events.append(
            MidiEvent(tick, stream_index, sequence, command, tuple(operands))
        )
        sequence += 1

    raise ConversionError(f"{source_name}: Tandy channel has no 0xFC terminator")


def parse_tandy_streams(sound: bytes, source_name: str) -> tuple[TandyStream, ...]:
    descriptors = find_tandy_descriptors(sound, source_name)
    streams = tuple(
        parse_stream(sound, descriptor, index, source_name)
        for index, descriptor in enumerate(descriptors)
    )
    if len({stream.part for stream in streams}) != 3:
        raise ConversionError(f"{source_name}: Tandy channels do not use three distinct parts")
    return streams


def parse_sound_blaster_streams(sound: bytes, source_name: str) -> tuple[TandyStream, ...]:
    descriptors = find_sound_blaster_descriptors(sound, source_name)
    streams = tuple(
        parse_stream(sound, descriptor, index, source_name)
        for index, descriptor in enumerate(descriptors)
    )
    if len({stream.part for stream in streams}) != len(streams):
        raise ConversionError(f"{source_name}: Sound Blaster channels reuse a MIDI part")
    if sum(stream.polyphony for stream in streams) > 9:
        raise ConversionError(f"{source_name}: Sound Blaster track requests over nine OPL voices")
    return streams


def parse_adlib_streams(sound: bytes, source_name: str) -> tuple[TandyStream, ...]:
    descriptors = find_adlib_descriptors(sound, source_name)
    streams = tuple(
        parse_stream(sound, descriptor, index, source_name)
        for index, descriptor in enumerate(descriptors)
    )
    if sum(stream.polyphony for stream in streams) > 9:
        raise ConversionError(f"{source_name}: AdLib track requests over nine OPL voices")
    return streams


def parse_pc_speaker_streams(sound: bytes, source_name: str) -> tuple[TandyStream, ...]:
    descriptors = find_pc_speaker_descriptors(sound, source_name)
    return tuple(
        parse_stream(sound, descriptor, index, source_name)
        for index, descriptor in enumerate(descriptors)
    )


@dataclass
class HardwareChannel:
    part: int
    note: int | None = None
    velocity: int = 0
    base_amplitude: int = 0
    envelope: bytes = b""
    envelope_pos: int = 0
    envelope_count: int = 0
    envelope_volume: int = 0
    release: bool = False
    sustain: bool = False
    duration: int = 0
    attenuation: int = 15


class TandyDriver:
    """Register-level model of Fire Hawk's TANDY3V.DRV music path."""

    def __init__(
        self,
        parts: Iterable[int],
        driver_data: bytes,
        patch_data: bytes,
        output: bytearray,
    ) -> None:
        if len(driver_data) < 0x393 or driver_data[0x20] != 0x73:
            raise ConversionError("TANDY3V.DRV does not match the expected Sierra driver")
        self.periods = tuple(
            read_u16(driver_data, TANDY_DRIVER_PERIOD_TABLE + index * 2)
            for index in range(432)
        )
        self.velocity_table = driver_data[
            TANDY_DRIVER_VELOCITY_TABLE : TANDY_DRIVER_VELOCITY_TABLE + 16
        ]
        self.instrument_offsets, self.instrument_data = parse_patch_101(patch_data)
        self.output = output
        self.channels = [HardwareChannel(part) for part in parts]
        self.programs = [0] * 16
        self.controller_volumes = [15] * 16
        self.pitch_bends = [0] * 16
        self.sustain = [False] * 16
        self.master_volume = 15

        for hardware_channel in range(4):
            self.write_psg(0x9F | (hardware_channel << 5))

    def write_psg(self, value: int) -> None:
        self.output.extend((0x50, value & 0xFF))

    def write_volume(self, index: int, attenuation: int) -> None:
        channel = self.channels[index]
        attenuation = max(0, min(15, attenuation))
        if channel.attenuation != attenuation:
            channel.attenuation = attenuation
            self.write_psg(0x90 | (index << 5) | attenuation)

    def mute(self, index: int) -> None:
        channel = self.channels[index]
        channel.note = None
        channel.base_amplitude = 0
        channel.release = False
        channel.sustain = False
        channel.envelope_count = 0
        self.write_volume(index, 15)

    def period_for(self, part: int, note: int) -> int:
        folded = note
        while folded < 0x2D:
            folded += 12
        while folded > 0x77:
            folded -= 12
        index = (folded - 0x2D) * 4 + self.pitch_bends[part]
        index = max(0, min(431, index))
        return self.periods[index]

    def write_period(self, index: int) -> None:
        channel = self.channels[index]
        if channel.note is None:
            return
        period = self.period_for(channel.part, channel.note)
        self.write_psg(0x80 | (index << 5) | (period & 0x0F))
        self.write_psg((period >> 4) & 0x3F)

    @staticmethod
    def multiply_level(left: int, right: int) -> int:
        # This is the high nibble of TANDY3V.DRV's 17x16 lookup table.
        return ((left + 1) * right) >> 4

    def current_amplitude(self, channel: HardwareChannel) -> int:
        level = self.multiply_level(channel.envelope_volume, channel.base_amplitude)
        return self.multiply_level(level, self.master_volume)

    def update_envelope_volume(self, index: int, value: int) -> None:
        channel = self.channels[index]
        channel.envelope_volume = value
        self.write_volume(index, 15 - self.current_amplitude(channel))

    def note_on(self, part: int, note: int, velocity: int) -> None:
        if note < 0x15 or note > 0x74:
            return
        index = self.channel_for_part(part)
        if index is None:
            return
        if self.channels[index].note is not None:
            self.mute(index)

        channel = self.channels[index]
        channel.note = note
        channel.velocity = velocity
        channel.release = False
        channel.sustain = False
        channel.duration = 0
        channel.envelope_pos = 0
        channel.envelope_count = 0
        instrument_offset = self.instrument_offsets[self.programs[part]]
        channel.envelope = self.instrument_data[instrument_offset:]

        velocity_level = self.velocity_table[min(15, velocity >> 3)]
        channel.base_amplitude = self.multiply_level(
            velocity_level, self.controller_volumes[part]
        )
        self.write_period(index)

    def note_off(self, part: int, note: int) -> None:
        index = self.channel_for_part(part)
        if index is None:
            return
        channel = self.channels[index]
        if channel.note != note:
            return
        if self.sustain[part]:
            channel.sustain = True
        else:
            channel.release = True

    def channel_for_part(self, part: int) -> int | None:
        for index, channel in enumerate(self.channels):
            if channel.part == part:
                return index
        return None

    def control_change(self, part: int, controller: int, value: int) -> None:
        if controller == 7:
            self.controller_volumes[part] = value >> 3
        elif controller == 64:
            enabled = value != 0
            self.sustain[part] = enabled
            if not enabled:
                # TANDY3V.DRV releases every held hardware voice here; its
                # sustain-release loop is not restricted to the MIDI part.
                for channel in self.channels:
                    if channel.sustain:
                        channel.sustain = False
                        channel.release = True
        elif controller in (75,):
            # SCI_MIDI_SET_POLYPHONY.  The three one-voice mappings are already
            # established from the selected track's channel directory.
            return
        elif controller in (120, 123):
            index = self.channel_for_part(part)
            if index is not None:
                self.mute(index)

    def pitch_bend(self, part: int, value: int) -> None:
        if value < 0x2000:
            bend = -((0x2000 - value) // 170)
        else:
            bend = (value - 0x2000) // 170
        self.pitch_bends[part] = bend
        index = self.channel_for_part(part)
        if index is not None:
            self.write_period(index)

    def dispatch(self, event: MidiEvent) -> None:
        command = event.status & 0xF0
        part = event.status & 0x0F
        operands = event.operands
        if command == 0x80:
            self.note_off(part, operands[0])
        elif command == 0x90:
            if operands[1] == 0:
                self.note_off(part, operands[0])
            else:
                self.note_on(part, operands[0], operands[1])
        elif command == 0xB0:
            self.control_change(part, operands[0], operands[1])
        elif command == 0xC0:
            self.programs[part] = operands[0]
        elif command == 0xE0:
            self.pitch_bend(part, operands[0] | (operands[1] << 7))
        # Polyphonic pressure and channel pressure have no Tandy driver action.

    def process_envelopes(self) -> None:
        for index, channel in enumerate(self.channels):
            if channel.note is None:
                continue
            channel.duration += 1
            if channel.envelope_count == 0xFE:
                if not channel.release:
                    continue
                channel.envelope_count = 0
            elif channel.envelope_count:
                channel.envelope_count -= 1
                continue

            if channel.envelope_pos + 1 >= len(channel.envelope):
                raise ConversionError("PATCH.101 instrument envelope is truncated")
            volume = channel.envelope[channel.envelope_pos]
            count = channel.envelope[channel.envelope_pos + 1]
            if volume == 0xFF:
                self.mute(index)
                continue
            channel.envelope_pos += 2
            channel.envelope_count = count
            if volume != channel.envelope_volume:
                self.update_envelope_volume(index, volume)

    def stop(self) -> None:
        for index in range(3):
            self.mute(index)


def parse_patch_101(resource: bytes) -> tuple[tuple[int, ...], bytes]:
    data = resource[RESOURCE_PREFIX_SIZE:]
    if len(data) < 258:
        raise ConversionError("PATCH.101 is truncated")
    offsets = tuple(read_u16(data, index * 2) - 0x100 for index in range(128))
    if min(offsets) < 0:
        raise ConversionError("PATCH.101 has an invalid instrument offset")
    instrument_data = data[256:]
    if max(offsets) >= len(instrument_data):
        raise ConversionError("PATCH.101 instrument offset is outside the patch")
    return offsets, instrument_data


OPL_REGISTER_OFFSETS = (0x00, 0x01, 0x02, 0x08, 0x09, 0x0A, 0x10, 0x11, 0x12)
OPL_VELOCITY_MAP_1 = (
    0, 12, 13, 14, 15, 17, 18, 19, 20, 22, 23, 24, 26, 27, 28, 29,
    31, 32, 33, 34, 35, 36, 37, 38, 39, 40, 41, 42, 43, 45, 45, 46,
    47, 48, 49, 50, 50, 51, 52, 52, 53, 54, 54, 55, 56, 56, 57, 58,
    59, 59, 59, 60, 60, 60, 61, 61, 61, 62, 62, 62, 62, 63, 63, 63,
)
OPL_VELOCITY_MAP_2 = (
    0, 20, 21, 22, 23, 24, 25, 26, 27, 28, 29, 30, 31, 32, 33, 33,
    34, 35, 36, 37, 38, 39, 40, 41, 42, 43, 44, 45, 46, 47, 47, 48,
    49, 50, 50, 51, 52, 52, 53, 54, 54, 55, 56, 57, 57, 58, 58, 58,
    59, 59, 59, 60, 60, 60, 61, 61, 61, 62, 62, 62, 62, 63, 63, 63,
)
OPL_FREQUENCIES = (
    0x157, 0x15C, 0x161, 0x166, 0x16B, 0x171, 0x176, 0x17B,
    0x181, 0x186, 0x18C, 0x192, 0x198, 0x19E, 0x1A4, 0x1AA,
    0x1B0, 0x1B6, 0x1BD, 0x1C3, 0x1CA, 0x1D0, 0x1D7, 0x1DE,
    0x1E5, 0x1EC, 0x1F3, 0x1FA, 0x202, 0x209, 0x211, 0x218,
    0x220, 0x228, 0x230, 0x238, 0x241, 0x249, 0x252, 0x25A,
    0x263, 0x26C, 0x275, 0x27E, 0x287, 0x290, 0x29A, 0x2A4,
)


@dataclass
class OplVoice:
    mapped_part: int = -1
    part: int = -1
    note: int | None = None
    patch: int = -1
    velocity: int = 0
    age: int = 0
    sustained: bool = False


class SoundBlasterDriver:
    """SCI1 nine-voice OPL2 path used by Fire Hawk's SNDBLAST.DRV."""

    def __init__(self, streams: Iterable[TandyStream], patch_resource: bytes, output: bytearray) -> None:
        self.output = output
        self.patches = parse_patch_003(patch_resource)
        self.voices = [OplVoice() for _ in range(9)]
        self.programs = [0] * 16
        self.volumes = [63] * 16
        self.pitch_bends = [0x2000] * 16
        self.sustain = [False] * 16
        self.last_voice = [-1] * 16
        self.master_volume = 15

        voice = 0
        for stream in streams:
            for _ in range(stream.polyphony):
                if voice == len(self.voices):
                    raise ConversionError("Sound Blaster track exceeds OPL2 voice count")
                self.voices[voice].mapped_part = stream.part
                voice += 1
        self.write(0xBD, 0x00)
        self.write(0x08, 0x00)
        self.write(0x01, 0x20)

    def write(self, register: int, value: int) -> None:
        self.output.extend((0x5A, register & 0xFF, value & 0xFF))

    def set_patch(self, voice_index: int, patch_index: int) -> None:
        if not 0 <= patch_index < len(self.patches):
            patch_index = 0
        voice = self.voices[voice_index]
        voice.patch = patch_index
        patch = self.patches[patch_index]
        for operator, offset in ((patch[:13], OPL_REGISTER_OFFSETS[voice_index]), (patch[13:26], OPL_REGISTER_OFFSETS[voice_index] + 3)):
            self.write(0x40 + offset, ((operator[0] & 3) << 6) | (operator[8] & 0x3F))
            self.write(0x60 + offset, ((operator[3] & 0x0F) << 4) | (operator[6] & 0x0F))
            self.write(0x80 + offset, ((operator[4] & 0x0F) << 4) | (operator[7] & 0x0F))
            self.write(0x20 + offset, ((operator[9] & 1) << 7) | ((operator[10] & 1) << 6) | ((operator[5] & 1) << 5) | ((operator[11] & 1) << 4) | (operator[1] & 0x0F))
        self.write(0xE0 + OPL_REGISTER_OFFSETS[voice_index], patch[26] & 3)
        self.write(0xE0 + OPL_REGISTER_OFFSETS[voice_index] + 3, patch[27] & 3)
        self.write(0xC0 + voice_index, ((patch[2] & 7) << 1) | int(not patch[12]))

    def set_velocity(self, voice_index: int) -> None:
        voice = self.voices[voice_index]
        patch = self.patches[voice.patch]
        for op_index, offset in ((1, OPL_REGISTER_OFFSETS[voice_index] + 3), (0, OPL_REGISTER_OFFSETS[voice_index])):
            if op_index == 0 and patch[12]:
                continue
            operator = patch[op_index * 13 : op_index * 13 + 13]
            velocity = self.volumes[voice.part] + 1
            velocity = velocity * (OPL_VELOCITY_MAP_1[voice.velocity] + 1) // 64
            velocity = velocity * (self.master_volume + 1) // 16
            velocity = max(0, velocity - 1)
            velocity = OPL_VELOCITY_MAP_2[velocity] * (63 - (operator[8] & 0x3F)) // 63
            self.write(0x40 + offset, ((operator[0] & 3) << 6) | (63 - velocity))

    def set_note(self, voice_index: int, note: int, key: bool) -> None:
        voice = self.voices[voice_index]
        voice.note = note
        index = note << 2
        bend = abs(self.pitch_bends[voice.part] - 0x2000) // 171
        index += bend if self.pitch_bends[voice.part] > 0x2000 else -bend
        index = max(0, min(0x1FC, index))
        frequency = OPL_FREQUENCIES[index % 48]
        octave = min(7, max(0, index // 48 - 1))
        self.write(0xA0 + voice_index, frequency)
        self.write(0xB0 + voice_index, (int(key) << 5) | (octave << 2) | (frequency >> 8))
        if key:
            self.set_velocity(voice_index)

    def voice_off(self, voice_index: int) -> None:
        voice = self.voices[voice_index]
        if voice.note is None:
            return
        self.set_note(voice_index, voice.note, False)
        voice.note = None
        voice.age = 0
        voice.sustained = False

    def find_voice(self, part: int) -> int | None:
        oldest: int | None = None
        oldest_age = 0
        for offset in range(1, 10):
            index = (self.last_voice[part] + offset) % 9
            voice = self.voices[index]
            if voice.mapped_part != part:
                continue
            if voice.note is None:
                self.last_voice[part] = index
                voice.part = part
                return index
            if voice.age >= oldest_age:
                oldest, oldest_age = index, voice.age
        if oldest is not None and oldest_age:
            self.voice_off(oldest)
            self.last_voice[part] = oldest
            self.voices[oldest].part = part
            return oldest
        return None

    def set_voice_mapping(self, part: int, count: int) -> None:
        mapped = [
            index for index, voice in enumerate(self.voices) if voice.mapped_part == part
        ]
        if count > len(mapped):
            for index, voice in enumerate(self.voices):
                if voice.mapped_part == -1:
                    voice.mapped_part = part
                    mapped.append(index)
                    if len(mapped) == count:
                        return
            raise ConversionError(f"Sound Blaster track cannot allocate {count} voices to part {part}")
        if count < len(mapped):
            for index in reversed(mapped[count:]):
                self.voice_off(index)
                self.voices[index].mapped_part = -1

    def note_on(self, part: int, note: int, velocity: int) -> None:
        if velocity == 0:
            self.note_off(part, note)
            return
        if not 12 <= note <= 107:
            return
        for index, voice in enumerate(self.voices):
            if voice.part == part and voice.note == note:
                self.voice_off(index)
                voice.part = part
                self.voice_on(index, note, velocity >> 1)
                return
        index = self.find_voice(part)
        if index is not None:
            self.voice_on(index, note, velocity >> 1)

    def voice_on(self, voice_index: int, note: int, velocity: int) -> None:
        voice = self.voices[voice_index]
        patch = self.programs[voice.part]
        if patch != voice.patch:
            self.set_patch(voice_index, patch)
        voice.age = 0
        voice.velocity = velocity
        self.set_note(voice_index, note, True)

    def note_off(self, part: int, note: int) -> None:
        for index, voice in enumerate(self.voices):
            if voice.part == part and voice.note == note:
                if self.sustain[part]:
                    voice.sustained = True
                else:
                    self.voice_off(index)
                return

    def dispatch(self, event: MidiEvent) -> None:
        command, part, operands = event.status & 0xF0, event.status & 0x0F, event.operands
        if command == 0x80:
            self.note_off(part, operands[0])
        elif command == 0x90:
            self.note_on(part, operands[0], operands[1])
        elif command == 0xB0:
            controller, value = operands
            if controller == 7:
                self.volumes[part] = value >> 1
                for index, voice in enumerate(self.voices):
                    if voice.part == part and voice.note is not None:
                        self.set_note(index, voice.note, True)
            elif controller == 64:
                self.sustain[part] = value != 0
                if value == 0:
                    for index, voice in enumerate(self.voices):
                        if voice.sustained:
                            self.voice_off(index)
            elif controller == 75:
                self.set_voice_mapping(part, value)
            elif controller in (120, 123):
                for index, voice in enumerate(self.voices):
                    if voice.part == part:
                        self.voice_off(index)
        elif command == 0xC0:
            self.programs[part] = operands[0]
        elif command == 0xE0:
            self.pitch_bends[part] = operands[0] | (operands[1] << 7)
            for index, voice in enumerate(self.voices):
                if voice.part == part and voice.note is not None:
                    self.set_note(index, voice.note, True)

    def tick(self) -> None:
        for voice in self.voices:
            if voice.note is not None:
                voice.age += 1

    def stop(self) -> None:
        for index in range(len(self.voices)):
            self.voice_off(index)


def parse_patch_003(resource: bytes) -> tuple[bytes, ...]:
    data = resource[RESOURCE_PREFIX_SIZE:]
    if not data.startswith(b"AdLib - THEX2 "):
        raise ConversionError("PATCH.003 does not have the expected Fire Hawk AdLib header")
    patches = data[14:]
    if len(patches) != 2690:
        raise ConversionError(f"PATCH.003 has {len(patches)} patch bytes, expected 2690")
    return tuple(patches[index * 28 : index * 28 + 28] for index in range(48)) + tuple(
        patches[2 + index * 28 : 2 + index * 28 + 28] for index in range(48, 96)
    )


def encode_tandy_gd3(title: str, source_resource: str) -> bytes:
    fields = (
        title,
        "",
        "Fire Hawk: Thexder - The Second Contact",
        "",
        "IBM PCjr / Tandy 1000",
        "",
        "",
        "",
        "1990",
        "extract_tandy_vgm.py",
        f"Tandy track 0x13 extracted from {source_resource}; "
        "PATCH.101 envelope and TANDY3V.DRV lookup tables applied.",
    )
    body = "\0".join(fields).encode("utf-16le") + b"\0\0"
    return b"Gd3 " + struct.pack("<II", 0x00000100, len(body)) + body


def encode_opl_gd3(title: str, source_resource: str, hardware: str, track_type: int) -> bytes:
    fields = (
        title,
        "",
        "Fire Hawk: Thexder - The Second Contact",
        "",
        f"IBM PC / {hardware}",
        "",
        "",
        "",
        "1990",
        "extract_tandy_vgm.py",
        f"{hardware} track {track_type:#04x} extracted from {source_resource}; "
        f"PATCH.003 and the {hardware} OPL2 music path applied.",
    )
    body = "\0".join(fields).encode("utf-16le") + b"\0\0"
    return b"Gd3 " + struct.pack("<II", 0x00000100, len(body)) + body


def build_vgm(
    streams: tuple[TandyStream, ...],
    driver_data: bytes,
    patch_data: bytes,
    title: str,
    source_resource: str,
) -> tuple[bytes, int, int]:
    header_size = 0x100
    output = bytearray(header_size)
    output[:4] = b"Vgm "
    write_u32(output, 0x08, 0x00000171)
    write_u32(output, 0x0C, TANDY_CLOCK_HZ)
    output[0x28:0x2A] = struct.pack("<H", 0x000C)
    output[0x2A] = 17
    write_u32(output, 0x34, header_size - 0x34)

    emulated_driver = TandyDriver(
        (stream.part for stream in streams), driver_data, patch_data, output
    )
    events = sorted(
        (event for stream in streams for event in stream.events),
        key=lambda event: (event.tick, event.stream_index, event.sequence),
    )
    events_by_tick: dict[int, list[MidiEvent]] = {}
    for event in events:
        events_by_tick.setdefault(event.tick, []).append(event)

    final_tick = max(stream.end_tick for stream in streams)
    event_count = 0
    # The real driver's 60 Hz callback first advances the sequence and then
    # advances each active instrument envelope.
    for tick in range(final_tick + 1):
        for event in events_by_tick.get(tick, ()):
            emulated_driver.dispatch(event)
            event_count += 1
        emulated_driver.process_envelopes()
        if tick != final_tick:
            output.append(0x62)  # exactly 735 samples at the VGM 44.1 kHz rate

    emulated_driver.stop()
    output.append(0x66)

    total_samples = final_tick * SAMPLES_PER_TICK
    write_u32(output, 0x18, total_samples)
    gd3_offset = len(output)
    output.extend(encode_tandy_gd3(title, source_resource))
    write_u32(output, 0x14, gd3_offset - 0x14)
    write_u32(output, 0x04, len(output) - 0x04)
    return bytes(output), final_tick, event_count


def build_sound_blaster_vgm(
    streams: tuple[TandyStream, ...],
    patch_data: bytes,
    title: str,
    source_resource: str,
    hardware: str = "Sound Blaster",
    track_type: int = SOUND_BLASTER_TRACK_TYPE,
) -> tuple[bytes, int, int]:
    header_size = 0x100
    output = bytearray(header_size)
    output[:4] = b"Vgm "
    write_u32(output, 0x08, 0x00000171)
    write_u32(output, 0x50, SOUND_BLASTER_CLOCK_HZ)
    write_u32(output, 0x34, header_size - 0x34)

    emulated_driver = SoundBlasterDriver(streams, patch_data, output)
    events = sorted(
        (event for stream in streams for event in stream.events),
        key=lambda event: (event.tick, event.stream_index, event.sequence),
    )
    events_by_tick: dict[int, list[MidiEvent]] = {}
    for event in events:
        events_by_tick.setdefault(event.tick, []).append(event)

    final_tick = max(stream.end_tick for stream in streams)
    event_count = 0
    for tick in range(final_tick + 1):
        for event in events_by_tick.get(tick, ()):
            emulated_driver.dispatch(event)
            event_count += 1
        emulated_driver.tick()
        if tick != final_tick:
            output.append(0x62)

    emulated_driver.stop()
    output.append(0x66)
    total_samples = final_tick * SAMPLES_PER_TICK
    write_u32(output, 0x18, total_samples)
    gd3_offset = len(output)
    output.extend(encode_opl_gd3(title, source_resource, hardware, track_type))
    write_u32(output, 0x14, gd3_offset - 0x14)
    write_u32(output, 0x04, len(output) - 0x04)
    return bytes(output), final_tick, event_count


def build_pc_speaker_wav(
    streams: tuple[TandyStream, ...], title: str, destination: Path
) -> tuple[int, int]:
    """Render the one-bit PC-speaker arrangement as mono 16-bit PCM."""
    events = sorted(
        (event for stream in streams for event in stream.events),
        key=lambda event: (event.tick, event.stream_index, event.sequence),
    )
    events_by_tick: dict[int, list[MidiEvent]] = {}
    for event in events:
        events_by_tick.setdefault(event.tick, []).append(event)

    current_note: int | None = None
    pitch_bend = 0
    final_tick = max(stream.end_tick for stream in streams)
    frames = bytearray()
    for tick in range(final_tick):
        for event in events_by_tick.get(tick, ()):
            command = event.status & 0xF0
            if command == 0x90 and event.operands[1]:
                # STD.DRV has one hardware channel and cuts off the previous
                # tone whenever a new note arrives.
                current_note = event.operands[0]
            elif command == 0x80 or (command == 0x90 and not event.operands[1]):
                if current_note == event.operands[0]:
                    current_note = None
            elif command == 0xE0:
                pitch_bend = ((event.operands[0] | (event.operands[1] << 7)) - 0x2000) // 171

        if current_note is None or not 24 <= current_note <= 119:
            frames.extend(b"\x00\x00" * SAMPLES_PER_TICK)
            continue
        halftone_delta = (current_note - 129) * 4 + pitch_bend
        octave_delta = ((halftone_delta + 10 * 48) // 48) - 10
        halftone_index = (halftone_delta + 48 * 100) % 48
        table_frequency = int(2.0 ** ((288 + halftone_index) / 48.0) * 440.0)
        frequency = table_frequency / (1 << max(0, -octave_delta))
        for sample in range(SAMPLES_PER_TICK):
            phase = ((tick * SAMPLES_PER_TICK + sample) * frequency / VGM_SAMPLE_RATE) % 1.0
            frames.extend(struct.pack("<h", 12000 if phase < 0.5 else -12000))

    with wave.open(str(destination), "wb") as wav:
        wav.setnchannels(1)
        wav.setsampwidth(2)
        wav.setframerate(VGM_SAMPLE_RATE)
        wav.writeframes(frames)
    return final_tick, len(events)


def validate_vgm(data: bytes, expected_samples: int, output_name: str) -> None:
    if data[:4] != b"Vgm ":
        raise ConversionError(f"{output_name}: missing VGM signature")
    if int.from_bytes(data[0x18:0x1C], "little") != expected_samples:
        raise ConversionError(f"{output_name}: incorrect total-sample field")
    data_start = 0x34 + int.from_bytes(data[0x34:0x38], "little")
    pos = data_start
    samples = 0
    found_end = False
    while pos < len(data):
        command = data[pos]
        pos += 1
        if command == 0x50:
            pos += 1
        elif command == 0x5A:
            pos += 2
        elif command == 0x61:
            samples += read_u16(data, pos)
            pos += 2
        elif command == 0x62:
            samples += 735
        elif command == 0x63:
            samples += 882
        elif 0x70 <= command <= 0x7F:
            samples += (command & 0x0F) + 1
        elif command == 0x66:
            found_end = True
            break
        else:
            raise ConversionError(f"{output_name}: unexpected VGM command {command:#04x}")
    if not found_end or samples != expected_samples:
        raise ConversionError(
            f"{output_name}: command stream has {samples} samples, "
            f"expected {expected_samples}"
        )


def locate_resource(name: str, disk1: Path, disk2: Path) -> Path:
    for directory in (disk1, disk2):
        candidate = directory / name
        if candidate.is_file():
            return candidate
    raise ConversionError(f"could not find {name} in {disk1} or {disk2}")


def convert_all(disk1: Path, disk2: Path, output_dir: Path) -> None:
    tandy_patch_path = locate_resource("PATCH.101", disk1, disk2)
    sound_blaster_patch_path = locate_resource("PATCH.003", disk1, disk2)
    driver_path = locate_resource("TANDY3V.DRV", disk1, disk2)
    tandy_patch_data = decompress_sierra_lzw(
        tandy_patch_path.read_bytes(), tandy_patch_path.name
    )
    sound_blaster_patch_data = decompress_sierra_lzw(
        sound_blaster_patch_path.read_bytes(), sound_blaster_patch_path.name
    )
    driver_data = driver_path.read_bytes()
    output_dir.mkdir(parents=True, exist_ok=True)

    for resource_name, output_name, title in SONGS:
        source_path = locate_resource(resource_name, disk1, disk2)
        sound = decompress_sierra_lzw(source_path.read_bytes(), resource_name)
        tandy_streams = parse_tandy_streams(sound, resource_name)
        tandy_vgm, tandy_ticks, tandy_event_count = build_vgm(
            tandy_streams, driver_data, tandy_patch_data, title, resource_name
        )
        validate_vgm(tandy_vgm, tandy_ticks * SAMPLES_PER_TICK, output_name)
        tandy_destination = output_dir / output_name
        tandy_destination.write_bytes(tandy_vgm)
        print(
            f"{tandy_destination.name}: {tandy_ticks / TICKS_PER_SECOND:.2f}s, "
            f"{tandy_event_count} MIDI events, {len(tandy_vgm)} bytes"
        )
        sound_blaster_streams = parse_sound_blaster_streams(sound, resource_name)
        sound_blaster_vgm, sound_blaster_ticks, sound_blaster_event_count = (
            build_sound_blaster_vgm(
                sound_blaster_streams,
                sound_blaster_patch_data,
                title,
                resource_name,
            )
        )
        sound_blaster_name = output_name.removesuffix("-tandy.vgm") + "-sound-blaster.vgm"
        validate_vgm(
            sound_blaster_vgm,
            sound_blaster_ticks * SAMPLES_PER_TICK,
            sound_blaster_name,
        )
        sound_blaster_destination = output_dir / sound_blaster_name
        sound_blaster_destination.write_bytes(sound_blaster_vgm)
        print(
            f"{sound_blaster_destination.name}: "
            f"{sound_blaster_ticks / TICKS_PER_SECOND:.2f}s, "
            f"{sound_blaster_event_count} MIDI events, "
            f"{len(sound_blaster_vgm)} bytes"
        )
        adlib_streams = parse_adlib_streams(sound, resource_name)
        adlib_vgm, adlib_ticks, adlib_event_count = build_sound_blaster_vgm(
            adlib_streams,
            sound_blaster_patch_data,
            title,
            resource_name,
            hardware="AdLib",
            track_type=ADLIB_TRACK_TYPE,
        )
        adlib_name = output_name.removesuffix("-tandy.vgm") + "-adlib.vgm"
        validate_vgm(adlib_vgm, adlib_ticks * SAMPLES_PER_TICK, adlib_name)
        adlib_destination = output_dir / adlib_name
        adlib_destination.write_bytes(adlib_vgm)
        print(
            f"{adlib_destination.name}: {adlib_ticks / TICKS_PER_SECOND:.2f}s, "
            f"{adlib_event_count} MIDI events, {len(adlib_vgm)} bytes"
        )
        pc_speaker_streams = parse_pc_speaker_streams(sound, resource_name)
        pc_speaker_name = output_name.removesuffix("-tandy.vgm") + "-pc-speaker.wav"
        pc_speaker_destination = output_dir / pc_speaker_name
        pc_ticks, pc_event_count = build_pc_speaker_wav(
            pc_speaker_streams, title, pc_speaker_destination
        )
        print(
            f"{pc_speaker_destination.name}: {pc_ticks / TICKS_PER_SECOND:.2f}s, "
            f"{pc_event_count} MIDI events, {pc_speaker_destination.stat().st_size} bytes"
        )


def main() -> int:
    script_dir = Path(__file__).resolve().parent
    firehawk_dir = script_dir.parent
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--disk1", type=Path, default=firehawk_dir / "disk1")
    parser.add_argument("--disk2", type=Path, default=firehawk_dir / "disk2")
    parser.add_argument("--output-dir", type=Path, default=script_dir)
    args = parser.parse_args()

    try:
        convert_all(args.disk1, args.disk2, args.output_dir)
    except (ConversionError, OSError) as error:
        parser.exit(1, f"error: {error}\n")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())