1
#!/usr/bin/env python3
2
"""Prepare locally supplied audio for the S3. Requires ffmpeg and ffprobe.
3
4
The v4 catalog contains ordered v3 songs with metadata and 20 ms Opus packets.
5
Writes the flash image music.bin and local summary music.json to artifacts/radio.
6
"""
7
import argparse
8
import json
9
import os
10
from pathlib import Path
11
import struct
12
import subprocess
13
import unicodedata
14
import zlib
15
 
16
 
17
from project import ROOT
18
from music_partition import MUSIC_BYTES
19
MAX_TAG_BYTES = 256
20
MAX_TRACKS = 32
21
MAX_TOTAL_FRAMES = 180_000
22
AUDIO_SUFFIXES = {".flac", ".wav", ".mp3", ".m4a", ".ogg", ".opus"}
23
DEFAULT_BITRATE_KBPS = 96
24
 
25
 
26
def music_bitrate(value):
27
    if type(value) is not int or not 1 <= value <= 512:
28
        raise ValueError("bitrateKbps must be an integer from 1 to 512")
29
    return value
30
 
31
 
32
def music_tag(value, name, required=False):
33
    value = value.strip()
34
    if (required and not value) or len(value.encode("utf-8")) > MAX_TAG_BYTES:
35
        raise ValueError(f"{name} must be {'1' if required else '0'}..{MAX_TAG_BYTES} UTF-8 bytes; use --{name.lower()} to override it")
36
    if any(unicodedata.category(character) == "Cc" for character in value):
37
        raise ValueError(f"{name} contains control characters; use --{name.lower()} to override it")
38
    return value
39
 
40
 
41
def ogg_packets(data):
42
    cursor = 0
43
    packet = bytearray()
44
    while cursor < len(data):
45
        if data[cursor:cursor + 4] != b"OggS":
46
            raise ValueError("Invalid Ogg page")
47
        segments = data[cursor + 26]
48
        sizes = data[cursor + 27:cursor + 27 + segments]
49
        cursor += 27 + segments
50
        for size in sizes:
51
            packet.extend(data[cursor:cursor + size])
52
            cursor += size
53
            if size < 255:
54
                yield bytes(packet)
55
                packet.clear()
56
    if packet:
57
        raise ValueError("Truncated Ogg packet")
58
 
59
 
60
def pack_song(packets, title, artist):
61
    title = music_tag(title, "Title", required=True)
62
    artist = music_tag(artist, "Artist")
63
    if not 1 <= len(packets) <= 30000:
64
        raise ValueError("Each song must contain 1..30000 audio frames (up to 10 minutes)")
65
    title_bytes, artist_bytes = title.encode(), artist.encode()
66
    payload = bytearray(struct.pack("<HH", len(title_bytes), len(artist_bytes)) + title_bytes + artist_bytes)
67
    for packet in packets:
68
        if not 0 < len(packet) <= 1275:
69
            raise ValueError("Unexpected Opus packet size")
70
        payload.extend(struct.pack("<H", len(packet)) + packet)
71
    header = struct.pack("<8sIIIIII32x", b"S3MUSIC\0", 3, len(packets), 48000, 2, 20, zlib.crc32(payload))
72
    return header + payload
73
 
74
 
75
def pack_catalog(songs):
76
    if not 1 <= len(songs) <= MAX_TRACKS:
77
        raise ValueError(f"A playlist needs 1..{MAX_TRACKS} songs")
78
    if sum(struct.unpack_from("<I", song, 12)[0] for song in songs) > MAX_TOTAL_FRAMES:
79
        raise ValueError("Playlist exceeds the bounded 60-minute frame index")
80
    payload = b"".join(struct.pack("<I", len(song)) + song for song in songs)
81
    used = 64 + len(payload)
82
    if used > MUSIC_BYTES:
83
        raise ValueError("Encoded playlist exceeds the 16 MiB music partition")
84
    header = bytearray(struct.pack("<8sIIIIII32x", b"S3MUSIC\0", 4, len(songs), 48000, 2, 20, zlib.crc32(payload)))
85
    struct.pack_into("<I", header, 32, used)
86
    return header + payload
87
 
88
 
89
def inputs(args):
90
    if args.playlist and args.files:
91
        raise ValueError("Use either audio files or --playlist")
92
    playlist = args.playlist or (os.environ.get("PLAYLIST") if not args.files else None)
93
    if playlist:
94
        path = Path(playlist)
95
        entries = json.loads(path.read_text())
96
        if not isinstance(entries, list):
97
            raise ValueError("A playlist JSON file must contain an ordered array")
98
        result = []
99
        for entry in entries:
100
            if isinstance(entry, str): entry = {"file": entry}
101
            if not isinstance(entry, dict) or set(entry) - {"file", "title", "artist", "bitrateKbps"} or not isinstance(entry.get("file"), str):
102
                raise ValueError("Each playlist entry needs a file and optional title/artist/bitrateKbps")
103
            if any(key in entry and not isinstance(entry[key], str) for key in ["title", "artist"]):
104
                raise ValueError("Title and artist overrides must be strings")
105
            result.append({**entry, "file": path.parent / entry["file"]})
106
    else:
107
        paths = args.files or ([Path(os.environ["TRACK"])] if os.environ.get("TRACK") else [])
108
        result = []
109
        for path in paths:
110
            if path.is_dir():
111
                result.extend({"file": child} for child in sorted(path.rglob("*")) if child.suffix.lower() in AUDIO_SUFFIXES)
112
            else: result.append({"file": path})
113
    if not 1 <= len(result) <= MAX_TRACKS:
114
        raise ValueError(f"Provide 1..{MAX_TRACKS} songs, a folder, or --playlist")
115
    for entry in result:
116
        entry["bitrateKbps"] = music_bitrate(entry.get("bitrateKbps", DEFAULT_BITRATE_KBPS))
117
    if args.title is not None or args.artist is not None:
118
        if len(result) != 1: raise ValueError("Use playlist entries for per-song metadata overrides")
119
        if args.title is not None: result[0]["title"] = args.title
120
        if args.artist is not None: result[0]["artist"] = args.artist
121
    return result
122
 
123
 
124
def encode_song(entry):
125
    file = entry["file"]
126
    bitrate = music_bitrate(entry.get("bitrateKbps", DEFAULT_BITRATE_KBPS))
127
    metadata = json.loads(subprocess.check_output([
128
        "ffprobe", "-v", "error", "-show_format", "-of", "json", str(file)]))["format"]
129
    tags = {key.lower(): value for key, value in metadata.get("tags", {}).items()}
130
    title = music_tag(entry.get("title", tags.get("title", file.stem)), "Title", required=True)
131
    artist = music_tag(entry.get("artist", tags.get("artist", "")), "Artist")
132
    encoded = subprocess.check_output([
133
        "ffmpeg", "-v", "error", "-i", str(file), "-map", "0:a:0", "-vn",
134
        "-ar", "48000", "-ac", "2", "-c:a", "libopus", "-b:a", f"{bitrate}k",
135
        "-vbr", "on", "-frame_duration", "20", "-application", "audio", "-f", "opus", "pipe:1"])
136
    packets = list(ogg_packets(encoded))
137
    if len(packets) < 3 or not packets[0].startswith(b"OpusHead") or not packets[1].startswith(b"OpusTags"):
138
        raise ValueError("Expected Opus headers")
139
    packets = packets[2:]
140
    packed = pack_song(packets, title, artist)
141
    return packed, {"title": title, "artist": artist, "durationMs": len(packets) * 20,
142
                    "sourceDurationMs": round(float(metadata["duration"]) * 1000),
143
                    "frames": len(packets), "bytes": len(packed), "bitrateKbps": bitrate}
144
 
145
 
146
def main():
147
    parser = argparse.ArgumentParser(description=__doc__)
148
    parser.add_argument("files", type=Path, nargs="*", help="Audio files or folders in playlist order")
149
    parser.add_argument("--playlist", type=Path, help="Ordered JSON list of files and optional metadata/bitrate overrides")
150
    parser.add_argument("--title", help="Override the title for a single input")
151
    parser.add_argument("--artist", help="Override the artist for a single input")
152
    args = parser.parse_args()
153
    try:
154
        encoded = [encode_song(entry) for entry in inputs(args)]
155
        binary = pack_catalog([song for song, _ in encoded])
156
    except ValueError as error:
157
        parser.error(str(error))
158
    manifest = {"packVersion": 4, "trackCount": len(encoded), "tracks": [track for _, track in encoded],
159
                "durationMs": sum(track["durationMs"] for _, track in encoded), "bytes": len(binary),
160
                "codec": "Opus", "sampleRate": 48000, "channels": 2, "frameMs": 20, "spectrum": "live-fft"}
161
    out = ROOT / "artifacts/radio"
162
    out.mkdir(parents=True, exist_ok=True, mode=0o700)
163
    for name, value in [("music.bin", binary), ("music.json", (json.dumps(manifest, indent=2, ensure_ascii=False) + "\n").encode())]:
164
        temporary = out / (name + ".tmp")
165
        temporary.touch(mode=0o600, exist_ok=True)
166
        temporary.chmod(0o600)
167
        temporary.write_bytes(value)
168
        temporary.replace(out / name)
169
    print(json.dumps(manifest, indent=2, ensure_ascii=False))
170
 
171
 
172
if __name__ == "__main__":
173
    main()