1
#!/usr/bin/env python3
2
"""Prepare private version 1 music for the archived C radio and FFT benchmark.
3
Requires ffmpeg, ffprobe and NumPy. Leaves the maintained radio's music pack alone.
4
"""
5
import argparse
6
import json
7
import os
8
from pathlib import Path
9
import struct
10
import subprocess
11
import unicodedata
12
import zlib
13
 
14
 
15
ROOT = Path(__file__).resolve().parents[1]
16
MAX_TAG_BYTES = 256
17
 
18
 
19
def music_tag(value, name, required=False):
20
    value = value.strip()
21
    if (required and not value) or len(value.encode("utf-8")) > MAX_TAG_BYTES:
22
        raise ValueError(f"{name} must be {'1' if required else '0'}..{MAX_TAG_BYTES} UTF-8 bytes; use --{name.lower()} to override it")
23
    if any(unicodedata.category(character) == "Cc" for character in value):
24
        raise ValueError(f"{name} contains control characters; use --{name.lower()} to override it")
25
    return value
26
 
27
 
28
def ogg_packets(data):
29
    cursor = 0
30
    packet = bytearray()
31
    while cursor < len(data):
32
        if data[cursor:cursor + 4] != b"OggS":
33
            raise ValueError("Invalid Ogg page")
34
        segments = data[cursor + 26]
35
        sizes = data[cursor + 27:cursor + 27 + segments]
36
        cursor += 27 + segments
37
        for size in sizes:
38
            packet.extend(data[cursor:cursor + size])
39
            cursor += size
40
            if size < 255:
41
                yield bytes(packet)
42
                packet.clear()
43
    if packet:
44
        raise ValueError("Truncated Ogg packet")
45
 
46
 
47
def legacy_spectrum(file, packets, preskip):
48
    import numpy as np
49
    pcm = np.frombuffer(subprocess.check_output([
50
        "ffmpeg", "-v", "error", "-i", str(file), "-map", "0:a:0", "-vn",
51
        "-ar", "48000", "-ac", "1", "-f", "f32le", "pipe:1"]), dtype="<f4")
52
    window_size = 2048
53
    window = np.hanning(window_size)
54
    padded = np.pad(pcm, (window_size, window_size))
55
    frequencies = np.fft.rfftfreq(window_size, 1 / 48000)
56
    edges = np.geomspace(30, 20000, 33)
57
    bins = [np.flatnonzero((frequencies >= low) & (frequencies < high))
58
            for low, high in zip(edges[:-1], edges[1:])]
59
    bins = [group if len(group) else np.array([np.argmin(abs(frequencies - (edges[i] + edges[i + 1]) / 2))])
60
            for i, group in enumerate(bins)]
61
    for index in range(len(packets)):
62
        center = index * 960 + 480 - preskip + window_size
63
        sample = padded[center - window_size // 2:center + window_size // 2]
64
        sample = np.pad(sample, (0, max(0, window_size - len(sample))))
65
        power = abs(np.fft.rfft(sample * window)) * 2 / window.sum()
66
        db = np.array([20 * np.log10(max(1e-6, power[group].max())) for group in bins])
67
        bands = np.clip((db + 72) / 66 * 255, 0, 255).astype(np.uint8).tobytes()
68
        yield bands
69
 
70
 
71
def main():
72
    parser = argparse.ArgumentParser(description=__doc__)
73
    parser.add_argument("file", type=Path, nargs="?", default=os.environ.get("TRACK") or None)
74
    parser.set_defaults(legacy_spectrum=True)
75
    parser.add_argument("--title", help="Override the audio file's title tag")
76
    parser.add_argument("--artist", help="Override the audio file's artist tag")
77
    args = parser.parse_args()
78
    if args.file is None:
79
        parser.error("provide an audio file or set TRACK for make music")
80
    out = ROOT / "artifacts/archive/music"
81
    out.mkdir(parents=True, exist_ok=True, mode=0o700)
82
    metadata = json.loads(subprocess.check_output([
83
        "ffprobe", "-v", "error", "-show_format", "-of", "json", str(args.file)]))["format"]
84
    tags = {key.lower(): value for key, value in metadata.get("tags", {}).items()}
85
    title = music_tag(args.title if args.title is not None else tags.get("title", args.file.stem), "Title", required=True)
86
    artist = music_tag(args.artist if args.artist is not None else tags.get("artist", ""), "Artist")
87
    encoded = subprocess.check_output([
88
        "ffmpeg", "-v", "error", "-i", str(args.file), "-map", "0:a:0", "-vn",
89
        "-ar", "48000", "-ac", "2", "-c:a", "libopus", "-b:a", "96k",
90
        "-vbr", "on", "-frame_duration", "20", "-application", "audio", "-f", "opus", "pipe:1"])
91
    packets = list(ogg_packets(encoded))
92
    if not packets[0].startswith(b"OpusHead") or not packets[1].startswith(b"OpusTags"):
93
        raise ValueError("Expected Opus headers")
94
    preskip = struct.unpack_from("<H", packets[0], 10)[0]
95
    packets = packets[2:]
96
    legacy = legacy_spectrum(args.file, packets, preskip) if args.legacy_spectrum else None
97
    frames = bytearray()
98
    for index, packet in enumerate(packets):
99
        if not 0 < len(packet) <= 1275:
100
            raise ValueError("Unexpected Opus packet size")
101
        bands = next(legacy) if legacy is not None else b""
102
        frames.extend(struct.pack("<H", len(packet)) + bands + packet)
103
    if not 1 <= len(packets) <= 30000:
104
        raise ValueError("Track must contain 1..30000 audio frames")
105
    version = 1 if args.legacy_spectrum else 3
106
    tag_bytes = b"" if args.legacy_spectrum else (
107
        struct.pack("<HH", len(title.encode("utf-8")), len(artist.encode("utf-8")))
108
        + title.encode("utf-8") + artist.encode("utf-8"))
109
    payload = tag_bytes + frames
110
    header = struct.pack("<8sIIIIII32x", b"S3MUSIC\0", version, len(packets), 48000, 2, 20, zlib.crc32(payload))
111
    binary = header + payload
112
    if len(binary) > 4 * 1024 * 1024:
113
        raise ValueError("Music exceeds the configured 4 MiB partition")
114
    (out / "music.bin").write_bytes(binary)
115
    (out / "music.bin").chmod(0o600)
116
    manifest = {"title": title, "artist": artist,
117
                "durationMs": len(packets) * 20, "sourceDurationMs": round(float(metadata["duration"]) * 1000),
118
                "codec": "Opus", "sampleRate": 48000, "channels": 2, "frameMs": 20,
119
                "bands": 32, "frames": len(packets), "bytes": len(binary), "packVersion": version,
120
                "spectrum": "precomputed" if args.legacy_spectrum else "live-fft"}
121
    (out / "music.json").write_text(json.dumps(manifest, indent=2) + "\n")
122
    print(json.dumps(manifest, indent=2))
123
 
124
 
125
if __name__ == "__main__":
126
    main()