rsf2wav converts Relentless Software RSF streamed audio to at3 or wav.

rsf2wav.py

#!/usr/bin/env python3
"""
rsf2wav converts Relentless Software RSF streamed audio to at3 or wav. RSF likely stands for "Relentless Streaming Format"

RSF is the streamed music container used by Buzz! Quiz TV and related titles. It is a 32 byte big endian header repeated at the start of every 64 KiB block, wrapping raw Sony ATRAC3 (this is NOT ATRAC3plus, and NOT scrambled)

header layout, big endian:
    0x00  u8   format tag, always 6 so far
    0x02  u8   channel count
    0x04  u16  sample rate
    0x06  u16  number of 64 KiB blocks in the file
    0x08  u32  bytes of ATRAC3 data in this block
    0x0C  u32  size of a later block, used for prefetch
    0x10  u32  block align, 192 per channel per frame
    0x14  u32  total ATRAC3 bytes in the file

the bytes between the end of a block's audio data and the next 64 KiB boundary are uninitialised memory from the machine that built the file, not audio. skip them, this tested consistently on three input files
"""

import argparse
import os
import shutil
import struct
import subprocess
import sys

BLOCK_SIZE = 0x10000
HEADER_SIZE = 0x20
FRAME_SIZE = 192
WAVE_FORMAT_SONY_SCX = 0x0270


class RsfError(Exception):
    pass


def parse_header(buf, offset=0):
    if len(buf) < offset + HEADER_SIZE:
        raise RsfError("truncated header at 0x%x" % offset)
    head = buf[offset:offset + HEADER_SIZE]
    tag, channels = head[0], head[2]
    rate, block_count = struct.unpack(">HH", head[4:8])
    data_size, next_size, block_align, total_size = struct.unpack(">IIII", head[8:24])
    return {
        "tag": tag,
        "channels": channels,
        "rate": rate,
        "block_count": block_count,
        "data_size": data_size,
        "next_size": next_size,
        "block_align": block_align,
        "total_size": total_size,
    }


def read_rsf(path):
    with open(path, "rb") as handle:
        buf = handle.read()
    if len(buf) < HEADER_SIZE:
        raise RsfError("file is too small to be an RSF")

    info = parse_header(buf)
    if info["tag"] != 6:
        raise RsfError("unexpected format tag %d, this may not be an RSF" % info["tag"])
    if info["channels"] < 1 or info["channels"] > 8:
        raise RsfError("implausible channel count %d" % info["channels"])
    if info["block_align"] != FRAME_SIZE * info["channels"]:
        raise RsfError("unexpected block align %d for %d channels" % (info["block_align"], info["channels"]))

    chunks = []
    for offset in range(0, len(buf), BLOCK_SIZE):
        block = parse_header(buf, offset)
        size = block["data_size"]
        start = offset + HEADER_SIZE
        available = min(BLOCK_SIZE - HEADER_SIZE, len(buf) - start)
        if size > available:
            info.setdefault("warnings", []).append(
                "block at 0x%x claims %d bytes but only %d are present" % (offset, size, available))
            size = available
        chunks.append(buf[start:start + size])

    info["blocks_found"] = len(chunks)
    info["data"] = b"".join(chunks)
    return info

def check_frames(data):
    total = len(data) // FRAME_SIZE
    bad = sum(1 for i in range(total) if (data[i * FRAME_SIZE] >> 2) != 0x28) // atrac3 start unit begin with 0x28 in its 6 top bits
    return total, bad

def build_at3(info):
    channels = info["channels"]
    rate = info["rate"]
    block_align = info["block_align"]
    data = info["data"]

    # WAVEFORMATEX plus the 14 byte ATRAC3 extension. Coding mode 0 means the channels are coded independently, which is what RSF seems to use.
    extra = struct.pack("<HIHHHH", 1, 1024, 0, 0, 1, 0)
    fmt = struct.pack("<HHIIHH", WAVE_FORMAT_SONY_SCX, channels, rate, rate * block_align // 1024, block_align, 16)
    fmt += struct.pack("<H", len(extra)) + extra

    chunks = b"fmt " + struct.pack("<I", len(fmt)) + fmt
    chunks += b"data" + struct.pack("<I", len(data)) + data
    return b"RIFF" + struct.pack("<I", 4 + len(chunks)) + b"WAVE" + chunks


def describe(path, info):
    total, bad = check_frames(info["data"])
    seconds = total / info["channels"] * 1024 / info["rate"]
    print("%s" % os.path.basename(path))
    print("  %d ch, %d Hz, block align %d, %d blocks" % (info["channels"], info["rate"], info["block_align"], info["blocks_found"]))
    print("  %d bytes of ATRAC3, header says %d%s" % (len(info["data"]), info["total_size"], "" if len(info["data"]) == info["total_size"] else " (mismatch)"))
    print("  %d frames, %d failed the sound unit check, %.2f seconds" % (total, bad, seconds))
    if info["block_count"] != info["blocks_found"]:
        print("  header says %d blocks, found %d" % (info["block_count"], info["blocks_found"]))
    for warning in info.get("warnings", []):
        print("  warning: %s" % warning)

def convert(path, out_dir, want_wav, ffmpeg, keep_at3, quiet):
    info = read_rsf(path)
    if not quiet:
        describe(path, info)

    stem = os.path.splitext(os.path.basename(path))[0]
    at3_path = os.path.join(out_dir, stem + ".at3")
    with open(at3_path, "wb") as handle:
        handle.write(build_at3(info))

    if not want_wav:
        print("wrote %s" % at3_path)
        return

    wav_path = os.path.join(out_dir, stem + ".wav")
    result = subprocess.run([ffmpeg, "-v", "error", "-y", "-i", at3_path, "-c:a", "pcm_s16le", wav_path], capture_output=True, text=True)
    if not keep_at3:
        os.remove(at3_path)
    if result.returncode != 0:
        raise RsfError("ffmpeg failed: %s" % result.stderr.strip())
    print("wrote %s" % wav_path)


def main():
    parser = argparse.ArgumentParser(description="Convert Relentless RSF streamed audio to at3 or wav.")
    parser.add_argument("files", nargs="+", help="one or more .rsf files")
    parser.add_argument("-o", "--out-dir", default=".", help="output directory")
    parser.add_argument("-f", "--format", choices=["wav", "at3"], default="wav", help="output format, wav needs ffmpeg (default: wav)")
    parser.add_argument("-k", "--keep-at3", action="store_true", help="keep the intermediate at3 when writing wav")
    parser.add_argument("-i", "--info", action="store_true", help="print header details and exit without converting")
    parser.add_argument("-q", "--quiet", action="store_true", help="do not print header details")
    parser.add_argument("--ffmpeg", default="ffmpeg", help="path to the ffmpeg binary")
    args = parser.parse_args()

    want_wav = args.format == "wav" and not args.info
    if want_wav and shutil.which(args.ffmpeg) is None:
        sys.exit("ffmpeg not found, install it or use --format at3")
    if not args.info:
        os.makedirs(args.out_dir, exist_ok=True)

    failed = 0
    for path in args.files:
        try:
            if args.info:
                describe(path, read_rsf(path))
            else:
                convert(path, args.out_dir, want_wav, args.ffmpeg, args.keep_at3, args.quiet)
        except (RsfError, OSError) as error:
            print("%s: %s" % (os.path.basename(path), error), file=sys.stderr)
            failed += 1
    return 1 if failed else 0


if __name__ == "__main__":
    sys.exit(main())