Skip to content

Instantly share code, notes, and snippets.

@pathawks

pathawks/snd_to_wav.py

Last active August 1, 2026 23:28
Show Gist options

  • Select an option

    Learn more about clone URLs
  • Save pathawks/6b01a2a97941b31abaf1cc346ac019c6 to your computer and use it in GitHub Desktop.

Select an option

Learn more about clone URLs
Save pathawks/6b01a2a97941b31abaf1cc346ac019c6 to your computer and use it in GitHub Desktop.
Download ZIP
Desk Mate Song to MIDI
Raw
snd_to_wav.py
#!/usr/bin/env python3
"""Convert DeskMate 3 Sound Editor .SND files to 8-bit mono WAV files.
The bundled DeskMate 3 files use the 2500-series .SND container: a 114-byte
header followed by linked 46-byte sample descriptors. Uncompressed samples
are unsigned 8-bit PCM. Music-compressed samples are decoded from the native
COMPRESS.RES codebook and nibble stream.
"""
from __future__ import annotations
import argparse
import json
import re
import struct
from dataclasses import asdict, dataclass
from pathlib import Path
FIXED_HEADER_SIZE = 0x72
DESCRIPTOR_SIZE = 46
MAX_SAMPLES = 16
NEW_FORMAT_ID = b"\x1a\x80"
COMPRESSION_NAMES = {
0: "none",
1: "music",
2: "speech",
}
@dataclass
class SampleDescriptor:
index: int
descriptor_offset: int
pitch: int
range_low: int
range_high: int
data_offset: int
stored_bytes: int
sample_count: int
sustain_start: int
sustain_end: int
@dataclass
class SoundFile:
path: Path
name: str
instrument_id: int
sample_rate: int
compression: int
descriptors: list[SampleDescriptor]
data: bytes
class SndFormatError(ValueError):
"""The input does not match the DeskMate 3 .SND format."""
def read_u16(data: bytes, offset: int) -> int:
return struct.unpack_from("<H", data, offset)[0]
def read_u32(data: bytes, offset: int) -> int:
return struct.unpack_from("<I", data, offset)[0]
def parse_snd(path: Path) -> SoundFile:
data = path.read_bytes()
if len(data) < FIXED_HEADER_SIZE:
raise SndFormatError(f"{path}: too short for a DeskMate 3 .SND header")
if data[0x2C:0x2E] != NEW_FORMAT_ID:
raise SndFormatError(
f"{path}: not a DeskMate 3 / Tandy 2500-series .SND file "
f"(expected ID 1a80 at 0x2c)"
)
count = read_u16(data, 0x2E)
if not 1 <= count <= MAX_SAMPLES:
raise SndFormatError(f"{path}: invalid sample count {count}")
compression = read_u16(data, 0x42)
if compression not in COMPRESSION_NAMES:
raise SndFormatError(f"{path}: unsupported compression code {compression}")
sample_rate = read_u16(data, 0x58)
if sample_rate == 0:
raise SndFormatError(f"{path}: invalid zero sample rate")
raw_name = data[:10].split(b"\0", 1)[0]
name = raw_name.decode("cp437", errors="replace")
instrument_id = read_u16(data, 0x30)
descriptors: list[SampleDescriptor] = []
descriptor_offset = FIXED_HEADER_SIZE
seen_offsets: set[int] = set()
for index in range(1, count + 1):
if descriptor_offset in seen_offsets:
raise SndFormatError(f"{path}: cyclic sample-descriptor chain")
seen_offsets.add(descriptor_offset)
if descriptor_offset + DESCRIPTOR_SIZE > len(data):
raise SndFormatError(f"{path}: sample descriptor {index} is truncated")
link = read_u32(data, descriptor_offset)
descriptor = SampleDescriptor(
index=index,
descriptor_offset=descriptor_offset,
pitch=data[descriptor_offset + 6],
range_low=data[descriptor_offset + 8],
range_high=data[descriptor_offset + 9],
data_offset=read_u32(data, descriptor_offset + 10),
stored_bytes=read_u32(data, descriptor_offset + 14),
sample_count=read_u32(data, descriptor_offset + 18),
sustain_start=read_u32(data, descriptor_offset + 22),
sustain_end=read_u32(data, descriptor_offset + 26),
)
validate_descriptor(path, descriptor, len(data), compression)
descriptors.append(descriptor)
if index < count:
if link == 0:
raise SndFormatError(f"{path}: sample-descriptor chain ends at entry {index}")
descriptor_offset = link
elif link != 0:
raise SndFormatError(f"{path}: final sample descriptor has a nonzero link")
return SoundFile(
path=path,
name=name,
instrument_id=instrument_id,
sample_rate=sample_rate,
compression=compression,
descriptors=descriptors,
data=data,
)
def validate_descriptor(
path: Path, descriptor: SampleDescriptor, file_size: int, compression: int
) -> None:
if descriptor.sample_count == 0:
return
if descriptor.data_offset >= file_size:
raise SndFormatError(
f"{path}: sample {descriptor.index} starts outside the file "
f"at 0x{descriptor.data_offset:x}"
)
stored_bytes = descriptor.stored_bytes
if compression == 0 and stored_bytes == 0:
stored_bytes = descriptor.sample_count
if stored_bytes == 0 or descriptor.data_offset + stored_bytes > file_size:
raise SndFormatError(f"{path}: sample {descriptor.index} data is truncated")
def decode_music_compression(payload: bytes, expected_samples: int) -> tuple[bytes, int]:
"""Decode DeskMate's music-compression stream from COMPRESS.RES.
Each stream starts with fourteen signed big-endian delta values and a
big-endian extended-run base. Codes are four-bit values, packed as the
first low nibble twice, then high/low nibble pairs. The final byte can
complete a command after the requested sample count; those samples are
padding and are discarded by the caller.
"""
if len(payload) < 30:
raise SndFormatError("music-compressed sample is shorter than its 30-byte codebook")
deltas = struct.unpack_from(">14h", payload)
extended_run_base = struct.unpack_from(">H", payload, 28)[0]
codes: list[int] = []
stream = payload[30:]
if stream:
codes.extend((stream[0] & 0x0F, stream[0] & 0x0F))
for byte in stream[1:]:
codes.extend((byte >> 4, byte & 0x0F))
output = bytearray()
sample = 0x80
state = 0
value = 0
def emit_run(length: int) -> None:
output.extend(bytes([sample]) * length)
for code in codes:
if state == 0:
if code < 14:
sample = max(0, min(255, sample + deltas[code]))
output.append(sample)
elif code == 14:
state = 4
else:
state = 1
elif state == 1:
if code < 14:
emit_run(code + 1)
elif code == 14:
sample = 0x80
state = 6
else:
state = 2
elif state == 2:
value = code << 4
state = 3
elif state == 3:
emit_run((value | code) + 14)
state = 0
elif state == 4:
value = code << 4
state = 5
elif state == 5:
sample = value | code
output.append(sample)
state = 0
elif state == 6:
value = code << 4
state = 7
elif state == 7:
emit_run((value | code) + extended_run_base + 1)
state = 0
else: # pragma: no cover - protects against accidental future edits.
raise AssertionError(f"invalid decoder state {state}")
if state != 0:
raise SndFormatError(f"music-compressed sample ends in incomplete decoder state {state}")
if len(output) < expected_samples:
raise SndFormatError(
f"music-compressed sample decodes to {len(output)} samples, "
f"shorter than its declared {expected_samples} samples"
)
return bytes(output[:expected_samples]), len(output) - expected_samples
def decode_sample(sound: SoundFile, descriptor: SampleDescriptor) -> tuple[bytes, int]:
if descriptor.sample_count == 0:
return b"", 0
stored_bytes = descriptor.stored_bytes
if sound.compression == 0 and stored_bytes == 0:
stored_bytes = descriptor.sample_count
payload = sound.data[descriptor.data_offset : descriptor.data_offset + stored_bytes]
if sound.compression == 0:
if len(payload) != descriptor.sample_count:
raise SndFormatError(
f"{sound.path}: uncompressed sample {descriptor.index} has "
f"{len(payload)} stored bytes but declares {descriptor.sample_count} samples"
)
return payload, 0
if sound.compression == 1:
return decode_music_compression(payload, descriptor.sample_count)
raise SndFormatError(
f"{sound.path}: speech compression is not implemented; no verified "
"speech-compressed sample is present in this corpus"
)
def midi_note_name(pitch: int) -> str | None:
if not 1 <= pitch <= 0x3F:
return None
midi_note = 32 + pitch
names = ("C", "C#", "D", "D#", "E", "F", "F#", "G", "G#", "A", "A#", "B")
return f"{names[midi_note % 12]}{midi_note // 12 - 1}"
def safe_name(value: str) -> str:
value = re.sub(r"[^A-Za-z0-9._-]+", "-", value).strip(".-")
return value or "sound"
def output_name(sound: SoundFile, descriptor: SampleDescriptor) -> str:
base = safe_name(sound.path.stem)
if len(sound.descriptors) == 1:
return f"{base}.wav"
pitch = midi_note_name(descriptor.pitch)
suffix = f"sample-{descriptor.index:02d}"
if pitch:
suffix += f"-{pitch}"
return f"{base}-{suffix}.wav"
def write_wav(path: Path, samples: bytes, sample_rate: int, descriptor: SampleDescriptor) -> None:
loop_enabled = (
descriptor.sustain_end > descriptor.sustain_start
and descriptor.sustain_end < len(samples)
)
chunks = [
(b"fmt ", struct.pack("<HHIIHH", 1, 1, sample_rate, sample_rate, 1, 8)),
]
if loop_enabled:
unity_note = 60
if 1 <= descriptor.pitch <= 0x3F:
unity_note = 32 + descriptor.pitch
sample_period = int(1_000_000_000 / sample_rate)
smpl = struct.pack(
"<9I6I",
0,
0,
sample_period,
unity_note,
0,
0,
0,
1,
0,
0,
0,
descriptor.sustain_start,
descriptor.sustain_end,
0,
0,
)
chunks.append((b"smpl", smpl))
chunks.append((b"data", samples))
body = bytearray(b"WAVE")
for chunk_id, chunk_data in chunks:
body += chunk_id + struct.pack("<I", len(chunk_data)) + chunk_data
if len(chunk_data) & 1:
body += b"\0"
path.write_bytes(b"RIFF" + struct.pack("<I", len(body)) + body)
def prepare_conversion(path: Path) -> tuple[SoundFile, list[tuple[SampleDescriptor, bytes, int]]]:
sound = parse_snd(path)
decoded = [(descriptor, *decode_sample(sound, descriptor)) for descriptor in sound.descriptors]
return sound, decoded
def write_conversion(
sound: SoundFile, decoded: list[tuple[SampleDescriptor, bytes, int]], out_dir: Path
) -> dict[str, object]:
out_dir.mkdir(parents=True, exist_ok=True)
sample_reports: list[dict[str, object]] = []
for descriptor, samples, discarded_padding in decoded:
report = asdict(descriptor)
report["pitch_name"] = midi_note_name(descriptor.pitch)
report["decoded_samples"] = len(samples)
report["discarded_padding_samples"] = discarded_padding
if samples:
wav_path = out_dir / output_name(sound, descriptor)
write_wav(wav_path, samples, sound.sample_rate, descriptor)
report["wav"] = str(wav_path)
else:
report["wav"] = None
sample_reports.append(report)
report = {
"source": str(sound.path),
"format": "DeskMate 3 / Tandy 2500-series SND",
"name": sound.name,
"instrument_id": sound.instrument_id,
"sample_rate": sound.sample_rate,
"compression": sound.compression,
"compression_name": COMPRESSION_NAMES[sound.compression],
"samples": sample_reports,
}
report_path = out_dir / f"{safe_name(sound.path.stem)}.json"
report_path.write_text(json.dumps(report, indent=2) + "\n", encoding="utf-8")
return {"report": str(report_path), **report}
def main() -> int:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("inputs", nargs="+", type=Path, help="DeskMate 3 .SND files")
parser.add_argument("--out-dir", type=Path, default=Path("recovered/snd-wav"))
args = parser.parse_args()
prepared = []
for path in args.inputs:
try:
prepared.append(prepare_conversion(path))
except (OSError, SndFormatError) as exc:
parser.error(str(exc))
results = [write_conversion(sound, decoded, args.out_dir) for sound, decoded in prepared]
print(json.dumps(results, indent=2))
return 0
if __name__ == "__main__":
raise SystemExit(main())
Raw
sng_to_midi.py
#!/usr/bin/env python3
"""Convert Tandy DeskMate Music .SNG files to Standard MIDI files.
This is intentionally a small, dependency-free decoder. The container,
duration table, and pitch indexing are taken from the DeskMate 3 Music player.
"""
from __future__ import annotations
import argparse
import json
import struct
from dataclasses import dataclass, field
from pathlib import Path
PPQ = 96
DEFAULT_TEMPO_BPM = 100
# Native table at MUSIC.PDM DS:2268. A quarter note is 24 DeskMate units;
# scaling each unit to four MIDI ticks makes PPQ=96 exact.
DURATION_UNITS = (
168, 144, 96, 84, 72, 64, 48, 42,
36, 32, 24, 21, 18, 16, 12, -10,
9, 8, 6, -5, -4, 4, 3, 2,
)
# DeskMate's instrument numbers are the numbers found in the .SND headers.
# MIDI programs are zero-based General MIDI program numbers.
MIDI_PROGRAMS = {
1: 71, # clarinet
2: 0, # acoustic grand piano
3: 9, # glockenspiel, a useful bells approximation
4: 42, # cello
}
@dataclass
class Song:
path: Path
title: str
composer: str
voices: list[list[int]]
control: list[int]
diagnostics: dict[str, object] = field(default_factory=dict)
def read_c_string(data: bytes, offset: int) -> tuple[str, int]:
end = data.find(b"\0", offset)
if end < 0:
raise ValueError(f"unterminated metadata string at 0x{offset:x}")
return data[offset:end].decode("cp437", errors="replace"), end + 1
def parse_sng(path: Path) -> Song:
data = path.read_bytes()
if len(data) < 0x18 or data[:4] != b"\x0cSNG":
raise ValueError(f"{path}: not a DeskMate .SNG file")
title, cursor = read_c_string(data, 0x18)
composer, cursor = read_c_string(data, cursor)
streams: list[list[int]] = []
search = cursor
while True:
start = data.find(b"\xfe\x7f", search)
if start < 0:
break
end = data.find(b"\xff\x7f", start + 2)
if end < 0:
raise ValueError(f"{path}: unterminated stream at 0x{start:x}")
payload = data[start + 2 : end]
if len(payload) % 2:
raise ValueError(f"{path}: odd-sized stream at 0x{start:x}")
streams.append(list(struct.unpack(f"<{len(payload) // 2}H", payload)))
search = end + 2
if len(streams) < 3:
raise ValueError(f"{path}: expected three voice streams, found {len(streams)}")
return Song(
path=path,
title=title,
composer=composer,
voices=streams[:3],
control=streams[3] if len(streams) > 3 else [],
)
def vlq(value: int) -> bytes:
value = max(0, value)
out = bytearray([value & 0x7F])
value >>= 7
while value:
out.insert(0, (value & 0x7F) | 0x80)
value >>= 7
return bytes(out)
def text_meta(kind: int, value: str) -> bytes:
encoded = value.encode("utf-8", errors="replace")
return bytes([0xFF, kind]) + vlq(len(encoded)) + encoded
def make_track(events: list[tuple[int, bytes]], name: str) -> bytes:
body = bytearray()
body += b"\0" + text_meta(0x03, name)
for delta, event in events:
body += vlq(delta)
body += event
body += b"\0\xff\x2f\0"
return b"MTrk" + struct.pack(">I", len(body)) + body
def absolute_to_delta(events: list[tuple[int, bytes]]) -> list[tuple[int, bytes]]:
ordered = sorted(enumerate(events), key=lambda item: (item[1][0], item[0]))
result: list[tuple[int, bytes]] = []
previous = 0
for _, (position, event) in ordered:
result.append((position - previous, event))
previous = position
return result
def dynamic_volume(value: int) -> int:
# DeskMate stores a small dynamic-level index in 0x72xx. These values are
# deliberately moderate because the original player applies instrument
# sample scaling as well as the score dynamic.
return min(127, 32 + value * 16)
def decode_voice(
words: list[int], channel: int, pitch_offset: int, all_piano: bool
) -> tuple[bytes, dict[str, object]]:
events: list[tuple[int, bytes]] = []
if all_piano:
events.append((0, bytes([0xC0 | channel, 0])))
diagnostics = {
"initial_program": 0 if all_piano else None,
"instrument_changes": [],
"volume_changes": [],
"rests": 0,
"notes": 0,
"note_flag_words": [],
"unknown_words": [],
}
pending = 0
program = 0
volume = 100
for index, word in enumerate(words):
high = word >> 8
low = word & 0xFF
if high == 0x73:
program = 0 if all_piano else MIDI_PROGRAMS.get(low, 0)
diagnostics["instrument_changes"].append(
{"word": f"{word:04x}", "instrument_id": low, "program": program}
)
events.append((pending, bytes([0xC0 | channel, program])))
pending = 0
continue
if high == 0x72:
volume = dynamic_volume(low)
diagnostics["volume_changes"].append({"word": f"{word:04x}", "volume": volume})
events.append((pending, bytes([0xB0 | channel, 7, volume])))
pending = 0
continue
if high < 0x80:
diagnostics["unknown_words"].append({"index": index, "word": f"{word:04x}"})
continue
duration_index = high & 0x1F
if duration_index >= len(DURATION_UNITS) or DURATION_UNITS[duration_index] <= 0:
diagnostics["unknown_words"].append(
{"index": index, "word": f"{word:04x}", "reason": "unsupported duration index"}
)
continue
duration = DURATION_UNITS[duration_index] * 4
pending += duration
pitch_index = low & 0x3F
if pitch_index == 0:
diagnostics["rests"] += 1
continue
if low & 0xC0 or high & 0xE0 != 0x80:
diagnostics["note_flag_words"].append(
{
"index": index,
"word": f"{word:04x}",
"pitch_flags": low & 0xC0,
"event_flags": high & 0xE0,
}
)
# The native tone table has entries 1..63 for A1..B6.
pitch = 32 + pitch_index + pitch_offset
pitch = max(0, min(127, pitch))
# Consume the duration as a note length. A later event begins after
# this event, so its delta is accumulated in `pending`.
events.append((pending - duration, bytes([0x90 | channel, pitch, volume])))
events.append((duration, bytes([0x80 | channel, pitch, 0])))
pending = 0
diagnostics["notes"] += 1
if pending:
# MIDI delta times must precede a real event. Use an empty text meta
# event to carry a final rest before End of Track.
events.append((pending, bytes([0xFF, 0x01, 0x00])))
return make_track(events, f"DeskMate voice {channel + 1}"), diagnostics
def decode_control(words: list[int]) -> tuple[list[tuple[int, bytes]], dict[str, object]]:
position_units = 0
events: list[tuple[int, bytes]] = []
diagnostics: dict[str, object] = {
"tempo_changes": [],
"time_signature_changes": [],
"repeat_markers": [],
"other_controls": [],
}
for index, word in enumerate(words):
kind = word >> 12
value = word & 0x0FFF
position = position_units * 4
if kind == 0x2:
position_units += value
elif kind == 0x1:
bpm = word & 0xFF
if bpm:
tempo = int(60_000_000 / bpm)
events.append((position, bytes([0xFF, 0x51, 0x03]) + tempo.to_bytes(3, "big")))
diagnostics["tempo_changes"].append({"position_units": position_units, "bpm": bpm})
elif kind == 0x6:
numerator = ((word >> 4) & 0xFF) // 4
denominator = word & 0x0F
if numerator > 0 and denominator in (2, 4, 8, 16, 32):
denominator_power = denominator.bit_length() - 1
events.append(
(position, bytes([0xFF, 0x58, 0x04, numerator, denominator_power, 24, 8]))
)
diagnostics["time_signature_changes"].append(
{"position_units": position_units, "numerator": numerator, "denominator": denominator}
)
elif kind in (0x3, 0x4, 0x5):
diagnostics["repeat_markers"].append(
{"index": index, "position_units": position_units, "word": f"{word:04x}"}
)
elif kind != 0x7:
diagnostics["other_controls"].append(
{"index": index, "position_units": position_units, "word": f"{word:04x}"}
)
return events, diagnostics
def make_midi(song: Song, pitch_offset: int, all_piano: bool) -> tuple[bytes, dict[str, object]]:
tracks: list[bytes] = []
conductor_absolute = [
(0, text_meta(0x03, song.title)),
(0, text_meta(0x02, song.composer)),
(0, bytes([0xFF, 0x51, 0x03]) + int(60_000_000 / DEFAULT_TEMPO_BPM).to_bytes(3, "big")),
(0, bytes([0xFF, 0x58, 0x04, 4, 2, 24, 8])),
]
control_events, control_diagnostics = decode_control(song.control)
conductor_absolute.extend(control_events)
tracks.append(make_track(absolute_to_delta(conductor_absolute), "DeskMate conductor"))
diagnostics: dict[str, object] = {
"source": str(song.path),
"title": song.title,
"composer": song.composer,
"voice_count": len(song.voices),
"control_word_count": len(song.control),
"control": control_diagnostics,
"voices": [],
"pitch_offset": pitch_offset,
"all_piano": all_piano,
}
for channel, words in enumerate(song.voices):
track, voice_diag = decode_voice(words, channel, pitch_offset, all_piano)
tracks.append(track)
diagnostics["voices"].append(voice_diag)
header = b"MThd" + struct.pack(">IHHH", 6, 1, len(tracks), PPQ)
return header + b"".join(tracks), diagnostics
def convert(path: Path, out_dir: Path, pitch_offset: int, all_piano: bool) -> dict[str, object]:
song = parse_sng(path)
midi, diagnostics = make_midi(song, pitch_offset, all_piano)
out_dir.mkdir(parents=True, exist_ok=True)
midi_path = out_dir / f"{path.stem}.mid"
report_path = out_dir / f"{path.stem}.json"
midi_path.write_bytes(midi)
report_path.write_text(json.dumps(diagnostics, indent=2) + "\n", encoding="utf-8")
return {"midi": str(midi_path), "report": str(report_path), **diagnostics}
def main() -> int:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("inputs", nargs="+", type=Path, help="DeskMate .SNG files")
parser.add_argument("--out-dir", type=Path, default=Path("recovered/sng-midi"))
parser.add_argument(
"--pitch-offset",
type=int,
default=0,
help="offset applied to the stored pitch byte (default: 0)",
)
parser.add_argument(
"--all-piano",
action="store_true",
help="use Acoustic Grand Piano for every instrument change",
)
args = parser.parse_args()
results = []
for path in args.inputs:
try:
results.append(convert(path, args.out_dir, args.pitch_offset, args.all_piano))
except (OSError, ValueError) as exc:
parser.error(str(exc))
print(json.dumps(results, indent=2))
return 0
if __name__ == "__main__":
raise SystemExit(main())
Raw
sng_to_wav.py
#!/usr/bin/env python3
"""Render DeskMate Music .SNG scores with their original .SND instruments.
The score decoder uses the native DeskMate duration and pitch indexing already
recovered by sng_to_midi.py. Instrument sample selection, pitch transposition,
and Clarinet sustain loops come directly from the DeskMate .SND descriptors.
Repeat blocks and alternate endings are unfolded into playback order.
"""
from __future__ import annotations
import argparse
import bisect
import json
import math
import struct
import sys
import wave
from array import array
from dataclasses import dataclass, field
from pathlib import Path
from snd_to_wav import SampleDescriptor, SndFormatError, decode_sample, parse_snd
from sng_to_midi import DEFAULT_TEMPO_BPM, DURATION_UNITS, Song, parse_sng
DEFAULT_OUTPUT_RATE = 44_000
DEFAULT_INSTRUMENT_ID = 2
DEFAULT_VOLUME_LEVEL = 3
MAX_VOLUME_LEVEL = 5
VOICE_MIX_GAIN = 1.0 / 3.0
RELEASE_SECONDS = 0.004
@dataclass(frozen=True)
class NoteEvent:
voice: int
start: int
duration: int
pitch: int
instrument_id: int
volume_level: int
connected_from_previous: bool
connected_to_next: bool
@property
def end(self) -> int:
return self.start + self.duration
@dataclass(frozen=True)
class InstrumentSample:
descriptor: SampleDescriptor
pcm: bytes
@dataclass
class Instrument:
instrument_id: int
name: str
source: Path
sample_rate: int
samples: list[InstrumentSample]
def sample_for_pitch(self, pitch: int) -> InstrumentSample:
for sample in self.samples:
descriptor = sample.descriptor
if descriptor.range_low <= pitch <= descriptor.range_high:
return sample
return min(self.samples, key=lambda item: abs(item.descriptor.pitch - pitch))
@dataclass
class RepeatNode:
start: int
count: int
marker_index: int
end: int | None = None
endings: list[tuple[int, int, int]] = field(default_factory=list)
children: list["RepeatNode"] = field(default_factory=list)
@dataclass(frozen=True)
class PlaybackInterval:
source_start: int
source_end: int
output_start_seconds: float
@dataclass(frozen=True)
class NoteOccurrence:
event: NoteEvent
start_frame: int
end_frame: int
class TempoMap:
def __init__(self, changes: list[tuple[int, int]]) -> None:
by_position: dict[int, int] = {0: DEFAULT_TEMPO_BPM}
for position, bpm in changes:
by_position[position] = bpm
self.changes = sorted(by_position.items())
self.positions = [position for position, _ in self.changes]
def tempo_at(self, position: int) -> int:
index = bisect.bisect_right(self.positions, position) - 1
return self.changes[max(0, index)][1]
def seconds_between(self, start: int, end: int) -> float:
if end <= start:
return 0.0
seconds = 0.0
cursor = start
bpm = self.tempo_at(start)
first = bisect.bisect_right(self.positions, start)
for position, new_bpm in self.changes[first:]:
if position >= end:
break
seconds += (position - cursor) * 60.0 / (bpm * 24.0)
cursor = position
bpm = new_bpm
return seconds + (end - cursor) * 60.0 / (bpm * 24.0)
def decode_tempo_changes(song: Song) -> list[tuple[int, int]]:
position = 0
changes: list[tuple[int, int]] = []
for word in song.control:
kind = word >> 12
if kind == 0x2:
position += word & 0x0FFF
elif kind == 0x1:
bpm = word & 0xFF
if bpm:
changes.append((position, bpm))
return changes
def decode_voice(words: list[int], voice: int) -> tuple[list[NoteEvent], int, dict[str, int]]:
position = 0
instrument_id = 0
volume_level = DEFAULT_VOLUME_LEVEL
notes: list[NoteEvent] = []
diagnostics = {
"notes": 0,
"rests": 0,
"instrument_changes": 0,
"volume_changes": 0,
"connected_notes": 0,
"unknown_words": 0,
}
for word in words:
high = word >> 8
low = word & 0xFF
if high == 0x73:
instrument_id = low
diagnostics["instrument_changes"] += 1
continue
if high == 0x72:
volume_level = low
diagnostics["volume_changes"] += 1
continue
if high < 0x80:
diagnostics["unknown_words"] += 1
continue
duration_index = high & 0x1F
if duration_index >= len(DURATION_UNITS):
diagnostics["unknown_words"] += 1
continue
duration = DURATION_UNITS[duration_index]
if duration <= 0:
diagnostics["unknown_words"] += 1
continue
pitch = low & 0x3F
if pitch:
# Bits 5 and 6 are inverted connection boundaries. 0xe0 is an
# ordinary detached note, 0xc0 begins a tie/slur, 0xa0 ends one,
# and 0x80 is an interior connected note.
connected_from = (high & 0x40) == 0
connected_to = (high & 0x20) == 0
notes.append(
NoteEvent(
voice=voice,
start=position,
duration=duration,
pitch=pitch,
instrument_id=instrument_id,
volume_level=volume_level,
connected_from_previous=connected_from,
connected_to_next=connected_to,
)
)
diagnostics["notes"] += 1
if connected_from or connected_to:
diagnostics["connected_notes"] += 1
else:
diagnostics["rests"] += 1
position += duration
return notes, position, diagnostics
def parse_repeat_tree(song: Song) -> tuple[list[RepeatNode], list[dict[str, int]]]:
roots: list[RepeatNode] = []
stack: list[RepeatNode] = []
records: list[dict[str, int]] = []
position = 0
for index, word in enumerate(song.control):
kind = word >> 12
value = word & 0x0FFF
if kind == 0x2:
position += value
continue
if kind == 0x3:
if value < 2:
raise ValueError(f"{song.path}: invalid repeat count {value} at control word {index}")
node = RepeatNode(start=position, count=value, marker_index=index)
if stack:
stack[-1].children.append(node)
else:
roots.append(node)
stack.append(node)
records.append({"kind": 3, "position_units": position, "value": value})
elif kind == 0x4:
if not stack:
raise ValueError(f"{song.path}: alternate ending outside a repeat at word {index}")
stack[-1].endings.append((position, value, index))
records.append({"kind": 4, "position_units": position, "value": value})
elif kind == 0x5:
if not stack:
raise ValueError(f"{song.path}: repeat end without a start at word {index}")
node = stack.pop()
if position <= node.start:
raise ValueError(f"{song.path}: empty repeat at control word {index}")
node.end = position
records.append({"kind": 5, "position_units": position, "value": value})
if stack:
raise ValueError(f"{song.path}: {len(stack)} unterminated repeat block(s)")
return roots, records
def unfold_repeats(roots: list[RepeatNode], total_units: int) -> list[tuple[int, int]]:
intervals: list[tuple[int, int]] = []
def append_interval(start: int, end: int) -> None:
if end <= start:
return
if intervals and intervals[-1][1] == start:
intervals[-1] = (intervals[-1][0], end)
else:
intervals.append((start, end))
def expand_range(start: int, end: int, children: list[RepeatNode]) -> None:
cursor = start
for child in sorted(children, key=lambda item: (item.start, item.marker_index)):
if child.end is None or child.start < start or child.end > end:
continue
if child.start < cursor:
raise ValueError("overlapping repeat blocks are not supported")
append_interval(cursor, child.start)
expand_repeat(child)
cursor = child.end
append_interval(cursor, end)
def expand_repeat(node: RepeatNode) -> None:
assert node.end is not None
endings = sorted(node.endings, key=lambda item: (item[1], item[0], item[2]))
first_ending_position = min((item[0] for item in endings), default=node.end)
if not node.start <= first_ending_position <= node.end:
raise ValueError("alternate ending lies outside its repeat")
for pass_number in range(1, node.count + 1):
expand_range(
node.start,
first_ending_position,
[child for child in node.children if child.end is not None and child.end <= first_ending_position],
)
if not endings:
continue
selected_index: int | None = None
for ending_index, (_, ending_number, _) in enumerate(endings):
next_number = endings[ending_index + 1][1] if ending_index + 1 < len(endings) else ending_number + 1
if ending_number <= pass_number < next_number:
selected_index = ending_index
break
if selected_index is None:
continue
ending_start = endings[selected_index][0]
ending_end = endings[selected_index + 1][0] if selected_index + 1 < len(endings) else node.end
expand_range(
ending_start,
ending_end,
[
child
for child in node.children
if child.end is not None and child.start >= ending_start and child.end <= ending_end
],
)
expand_range(0, total_units, roots)
return intervals
def make_playback_intervals(
source_intervals: list[tuple[int, int]], tempo_map: TempoMap
) -> tuple[list[PlaybackInterval], float]:
result: list[PlaybackInterval] = []
output_seconds = 0.0
for start, end in source_intervals:
result.append(PlaybackInterval(start, end, output_seconds))
output_seconds += tempo_map.seconds_between(start, end)
return result, output_seconds
def load_instruments(snd_dir: Path) -> dict[int, Instrument]:
instruments: dict[int, Instrument] = {}
candidates = sorted(
(path for path in snd_dir.iterdir() if path.is_file() and path.suffix.lower() == ".snd"),
key=lambda path: path.name.lower(),
)
for path in candidates:
sound = parse_snd(path)
# ID zero is used by ordinary Sound Editor recordings such as
# DESKTOP.SND. It is not a Music instrument file.
if sound.instrument_id == 0 or len(sound.descriptors) < 2:
continue
if sound.instrument_id in instruments:
other = instruments[sound.instrument_id].source
raise ValueError(f"duplicate instrument ID {sound.instrument_id}: {other} and {path}")
samples = [
InstrumentSample(descriptor, decode_sample(sound, descriptor)[0])
for descriptor in sound.descriptors
if descriptor.sample_count
]
if samples:
instruments[sound.instrument_id] = Instrument(
instrument_id=sound.instrument_id,
name=sound.name,
source=path,
sample_rate=sound.sample_rate,
samples=samples,
)
return instruments
def effective_instrument_id(instrument_id: int) -> int:
return DEFAULT_INSTRUMENT_ID if instrument_id == 0 else instrument_id
def note_occurrences(
notes: list[NoteEvent],
intervals: list[PlaybackInterval],
tempo_map: TempoMap,
sample_rate: int,
) -> list[NoteOccurrence]:
occurrences: list[NoteOccurrence] = []
note_starts = [note.start for note in notes]
for interval in intervals:
index = bisect.bisect_left(note_starts, interval.source_start)
while index < len(notes):
note = notes[index]
if note.start >= interval.source_end:
break
end = min(note.end, interval.source_end)
start_seconds = interval.output_start_seconds + tempo_map.seconds_between(
interval.source_start, note.start
)
end_seconds = interval.output_start_seconds + tempo_map.seconds_between(
interval.source_start, end
)
start_frame = round(start_seconds * sample_rate)
end_frame = max(start_frame + 1, round(end_seconds * sample_rate))
occurrences.append(NoteOccurrence(note, start_frame, end_frame))
index += 1
return occurrences
def connected_phrases(occurrences: list[NoteOccurrence]) -> list[list[NoteOccurrence]]:
phrases: list[list[NoteOccurrence]] = []
for occurrence in occurrences:
if not phrases:
phrases.append([occurrence])
continue
previous = phrases[-1][-1]
connected = (
previous.event.connected_to_next
and occurrence.event.connected_from_previous
and abs(previous.end_frame - occurrence.start_frame) <= 1
and previous.event.end == occurrence.event.start
and effective_instrument_id(previous.event.instrument_id)
== effective_instrument_id(occurrence.event.instrument_id)
)
if connected:
phrases[-1].append(occurrence)
else:
phrases.append([occurrence])
return phrases
def render_phrase(
mix: array,
phrase: list[NoteOccurrence],
instruments: dict[int, Instrument],
output_rate: int,
) -> None:
source_position = 0.0
active_sample: InstrumentSample | None = None
phrase_end = phrase[-1].end_frame
release_frames = max(1, round(RELEASE_SECONDS * output_rate))
for occurrence in phrase:
note = occurrence.event
instrument_id = effective_instrument_id(note.instrument_id)
instrument = instruments[instrument_id]
sample = instrument.sample_for_pitch(note.pitch)
if sample is not active_sample:
active_sample = sample
source_position = 0.0
descriptor = sample.descriptor
pcm = sample.pcm
step = math.pow(2.0, (note.pitch - descriptor.pitch) / 12.0)
step *= instrument.sample_rate / output_rate
loop_enabled = (
descriptor.sustain_end > descriptor.sustain_start
and descriptor.sustain_end < len(pcm)
)
gain = max(0.0, min(MAX_VOLUME_LEVEL, note.volume_level)) / MAX_VOLUME_LEVEL
gain *= VOICE_MIX_GAIN
for frame in range(occurrence.start_frame, occurrence.end_frame):
if loop_enabled and source_position >= descriptor.sustain_end:
loop_length = descriptor.sustain_end - descriptor.sustain_start
source_position = descriptor.sustain_start + (
(source_position - descriptor.sustain_start) % loop_length
)
source_index = int(source_position)
if source_index >= len(pcm):
break
envelope = 1.0
if frame >= phrase_end - release_frames:
envelope = max(0.0, (phrase_end - frame) / release_frames)
mix[frame] += ((pcm[source_index] - 128) / 128.0) * gain * envelope
source_position += step
def render_song(
song: Song,
instruments: dict[int, Instrument],
output_rate: int,
) -> tuple[array, dict[str, object]]:
voices: list[list[NoteEvent]] = []
voice_diagnostics: list[dict[str, int]] = []
total_units = 0
for voice_index, words in enumerate(song.voices):
notes, voice_units, diagnostics = decode_voice(words, voice_index)
voices.append(notes)
voice_diagnostics.append(diagnostics)
total_units = max(total_units, voice_units)
control_position = sum((word & 0x0FFF) for word in song.control if word >> 12 == 0x2)
total_units = max(total_units, control_position)
roots, repeat_records = parse_repeat_tree(song)
source_intervals = unfold_repeats(roots, total_units)
tempo_changes = decode_tempo_changes(song)
tempo_map = TempoMap(tempo_changes)
playback, duration_seconds = make_playback_intervals(source_intervals, tempo_map)
total_frames = max(1, round(duration_seconds * output_rate))
required_ids = sorted(
{
effective_instrument_id(note.instrument_id)
for voice in voices
for note in voice
}
)
missing = [instrument_id for instrument_id in required_ids if instrument_id not in instruments]
if missing:
raise ValueError(
f"{song.path}: missing DeskMate instrument ID(s) {', '.join(map(str, missing))}"
)
mix = array("f", [0.0]) * total_frames
occurrence_counts: list[int] = []
phrase_counts: list[int] = []
for notes in voices:
occurrences = note_occurrences(notes, playback, tempo_map, output_rate)
phrases = connected_phrases(occurrences)
occurrence_counts.append(len(occurrences))
phrase_counts.append(len(phrases))
for phrase in phrases:
render_phrase(mix, phrase, instruments, output_rate)
peak = max((abs(value) for value in mix), default=0.0)
clipped_frames = sum(1 for value in mix if abs(value) > 1.0)
rms = math.sqrt(sum(value * value for value in mix) / len(mix))
report: dict[str, object] = {
"source": str(song.path),
"title": song.title,
"composer": song.composer,
"sample_rate": output_rate,
"channels": 1,
"sample_width_bits": 16,
"duration_seconds": duration_seconds,
"source_duration_units": total_units,
"playback_duration_units": sum(end - start for start, end in source_intervals),
"tempo_changes": [
{"position_units": position, "bpm": bpm} for position, bpm in tempo_changes
],
"repeat_records": repeat_records,
"playback_intervals": [
{"source_start": start, "source_end": end} for start, end in source_intervals
],
"default_instrument": {
"score_id": 0,
"resolved_instrument_id": DEFAULT_INSTRUMENT_ID,
"name": instruments[DEFAULT_INSTRUMENT_ID].name,
},
"instruments": [
{
"instrument_id": instrument_id,
"name": instruments[instrument_id].name,
"source": str(instruments[instrument_id].source),
}
for instrument_id in required_ids
],
"voices": [
{
**diagnostics,
"rendered_note_occurrences": occurrence_counts[index],
"rendered_phrases": phrase_counts[index],
}
for index, diagnostics in enumerate(voice_diagnostics)
],
"peak": peak,
"rms": rms,
"clipped_frames": clipped_frames,
}
return mix, report
def write_pcm16_wav(path: Path, mix: array, sample_rate: int) -> None:
path.parent.mkdir(parents=True, exist_ok=True)
with wave.open(str(path), "wb") as output:
output.setnchannels(1)
output.setsampwidth(2)
output.setframerate(sample_rate)
block_size = 65_536
for start in range(0, len(mix), block_size):
pcm = array(
"h",
(
round(max(-1.0, min(1.0, value)) * 32767.0)
for value in mix[start : start + block_size]
),
)
if sys.byteorder != "little":
pcm.byteswap()
output.writeframesraw(pcm.tobytes())
def convert(
path: Path,
out_dir: Path,
instruments: dict[int, Instrument],
sample_rate: int,
) -> dict[str, object]:
song = parse_sng(path)
mix, report = render_song(song, instruments, sample_rate)
out_dir.mkdir(parents=True, exist_ok=True)
wav_path = out_dir / f"{path.stem}.wav"
report_path = out_dir / f"{path.stem}.json"
write_pcm16_wav(wav_path, mix, sample_rate)
report["wav"] = str(wav_path)
report_path.write_text(json.dumps(report, indent=2) + "\n", encoding="utf-8")
return {"wav": str(wav_path), "report": str(report_path), **report}
def main() -> int:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("inputs", nargs="+", type=Path, help="DeskMate Music .SNG files")
parser.add_argument(
"--snd-dir",
type=Path,
help="directory containing the original DeskMate instrument .SND files "
"(default: directory of the first input)",
)
parser.add_argument("--out-dir", type=Path, default=Path("recovered/sng-wav"))
parser.add_argument(
"--sample-rate",
type=int,
default=DEFAULT_OUTPUT_RATE,
help=f"output sample rate (default: {DEFAULT_OUTPUT_RATE})",
)
args = parser.parse_args()
if args.sample_rate < 8_000 or args.sample_rate > 192_000:
parser.error("--sample-rate must be between 8000 and 192000")
snd_dir = args.snd_dir if args.snd_dir is not None else args.inputs[0].parent
try:
instruments = load_instruments(snd_dir)
if DEFAULT_INSTRUMENT_ID not in instruments:
raise ValueError(f"{snd_dir}: the default Piano instrument (ID 2) was not found")
results = [convert(path, args.out_dir, instruments, args.sample_rate) for path in args.inputs]
except (OSError, SndFormatError, ValueError) as exc:
parser.error(str(exc))
print(json.dumps(results, indent=2))
return 0
if __name__ == "__main__":
raise SystemExit(main())
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment