scriptosandClaude Opus 5.5
67a687f4e6
Build & Push Docker Image / Build & push image (push) Successful in 6m49s
LOSSLESS_ONLY (default true): follow-up tracks are only converted when nothing is lost (MP3 into a FLAC stream). FLAC after MP3 or higher resolution FLAC plays in original quality with a normal track change. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
521 lines
20 KiB
Python
521 lines
20 KiB
Python
"""Gapless playback for bluos-upnp-bridge.
|
|
|
|
BluOS restarts its pipeline for every URL, which leaves a 1-2 s gap between tracks. In gapless
|
|
mode the bridge serves one continuous stream and appends the next track's audio data directly
|
|
after the current one:
|
|
|
|
- MP3: ID3v2/ID3v1 tags and the Xing/Info frame of follow-up tracks are dropped, the MPEG frames
|
|
are simply concatenated (a few ms of encoder padding remain).
|
|
- FLAC: the first track's STREAMINFO gets "total samples unknown" and a zero MD5 so decoders keep
|
|
going; follow-up tracks are appended without their metadata blocks. Tracks must have the same
|
|
sample rate, channel count and bit depth.
|
|
|
|
Tracks that can't be stitched (different codec or FLAC parameters) end the stream; the bridge then
|
|
falls back to a normal track change.
|
|
"""
|
|
import logging
|
|
import shutil
|
|
import subprocess
|
|
import threading
|
|
import time
|
|
import urllib.request
|
|
|
|
log = logging.getLogger("gapless")
|
|
|
|
CHUNK = 65536
|
|
|
|
# ---------------------------------------------------------------------------- MP3
|
|
|
|
_MP3_BITRATES = {
|
|
# (version_is_mpeg1, layer3) -> kbit/s table (index 0..15)
|
|
True: [0, 32, 40, 48, 56, 64, 80, 96, 112, 128, 160, 192, 224, 256, 320, 0],
|
|
False: [0, 8, 16, 24, 32, 40, 48, 56, 64, 80, 96, 112, 128, 144, 160, 0],
|
|
}
|
|
_MP3_RATES = {3: [44100, 48000, 32000], 2: [22050, 24000, 16000], 0: [11025, 12000, 8000]}
|
|
|
|
|
|
def id3v2_size(buf):
|
|
"""Length of a leading ID3v2 tag (0 if none)."""
|
|
if len(buf) >= 10 and buf[:3] == b"ID3":
|
|
size = (buf[6] & 0x7F) << 21 | (buf[7] & 0x7F) << 14 | (buf[8] & 0x7F) << 7 | (buf[9] & 0x7F)
|
|
return 10 + size + (10 if buf[5] & 0x10 else 0)
|
|
return 0
|
|
|
|
|
|
def mp3_frame_len(buf, pos):
|
|
"""Length of the MPEG layer III frame at pos, or 0 if no valid header."""
|
|
if pos + 4 > len(buf) or buf[pos] != 0xFF or (buf[pos + 1] & 0xE0) != 0xE0:
|
|
return 0
|
|
version = (buf[pos + 1] >> 3) & 0x03 # 3 = MPEG1, 2 = MPEG2, 0 = MPEG2.5
|
|
layer = (buf[pos + 1] >> 1) & 0x03 # 1 = layer III
|
|
br_idx = buf[pos + 2] >> 4
|
|
sr_idx = (buf[pos + 2] >> 2) & 0x03
|
|
padding = (buf[pos + 2] >> 1) & 0x01
|
|
if version == 1 or layer != 1 or br_idx in (0, 15) or sr_idx == 3:
|
|
return 0
|
|
bitrate = _MP3_BITRATES[version == 3][br_idx] * 1000
|
|
rate = _MP3_RATES[version][sr_idx]
|
|
factor = 144 if version == 3 else 72
|
|
return factor * bitrate // rate + padding
|
|
|
|
|
|
def mp3_sync(buf, start=0):
|
|
"""Offset of the first frame at/after start that is followed by another valid frame."""
|
|
pos = start
|
|
while pos < len(buf) - 4:
|
|
n = mp3_frame_len(buf, pos)
|
|
if n and (pos + n + 4 > len(buf) or mp3_frame_len(buf, pos + n)):
|
|
return pos
|
|
pos += 1
|
|
return -1
|
|
|
|
|
|
def mp3_skip_info_frame(buf, pos):
|
|
"""If the frame at pos is a Xing/Info/VBRI header frame, return the offset after it."""
|
|
n = mp3_frame_len(buf, pos)
|
|
if n and (b"Xing" in buf[pos:pos + 64] or b"Info" in buf[pos:pos + 64] or b"VBRI" in buf[pos:pos + 64]):
|
|
return pos + n
|
|
return pos
|
|
|
|
|
|
# ---------------------------------------------------------------------------- FLAC
|
|
|
|
def flac_header_len(buf):
|
|
"""Length of 'fLaC' + all metadata blocks, or -1 if buf is too short, or 0 if not FLAC."""
|
|
if buf[:4] != b"fLaC":
|
|
return 0
|
|
pos = 4
|
|
while True:
|
|
if pos + 4 > len(buf):
|
|
return -1
|
|
last = buf[pos] & 0x80
|
|
length = int.from_bytes(buf[pos + 1:pos + 4], "big")
|
|
pos += 4 + length
|
|
if last:
|
|
return pos if pos <= len(buf) else -1
|
|
|
|
|
|
def flac_params(header):
|
|
"""Sample rate / channels / bits per sample bytes of STREAMINFO for compatibility checks."""
|
|
si = header[8:8 + 34]
|
|
return bytes(si[10:13]) + bytes([si[13] & 0xF0])
|
|
|
|
|
|
def flac_format(params):
|
|
"""(sample rate, channels, bits per sample) from flac_params()."""
|
|
b = params
|
|
rate = (b[0] << 12) | (b[1] << 4) | (b[2] >> 4)
|
|
channels = ((b[2] >> 1) & 0x07) + 1
|
|
bits = (((b[2] & 0x01) << 4) | (b[3] >> 4)) + 1
|
|
return rate, channels, bits
|
|
|
|
|
|
def mp3_rate(buf, pos):
|
|
version = (buf[pos + 1] >> 3) & 0x03
|
|
return _MP3_RATES[version][(buf[pos + 2] >> 2) & 0x03]
|
|
|
|
|
|
FFMPEG = shutil.which("ffmpeg")
|
|
LOSSLESS_ONLY = True # only convert follow-up tracks if no quality is lost (set by the bridge)
|
|
|
|
|
|
def _lossless_conversion(src_kind, dst_kind, src_format, dst_params):
|
|
"""True if converting src into the stream format keeps the full quality.
|
|
|
|
MP3 -> FLAC only re-packs the decoded audio. FLAC -> FLAC is fine if sample rate and bit depth don't
|
|
go down (and the channels match). FLAC -> MP3 or downsampling would lose quality.
|
|
"""
|
|
if src_kind == "mp3" and dst_kind == "flac":
|
|
return True
|
|
if src_kind == "flac" and dst_kind == "flac" and src_format and dst_params:
|
|
rate, channels, bits = src_format
|
|
d_rate, d_channels, d_bits = flac_format(dst_params)
|
|
return channels == d_channels and d_rate >= rate and d_bits >= bits
|
|
return False
|
|
|
|
|
|
class Transcoder:
|
|
"""ffmpeg converting a follow-up track into the session's codec (file-like: read/close)."""
|
|
|
|
def __init__(self, url, kind, rate, channels=2, bits=16):
|
|
cmd = [FFMPEG, "-hide_banner", "-loglevel", "error", "-nostdin", "-i", url, "-map", "0:a:0",
|
|
"-ar", str(rate), "-ac", str(channels)]
|
|
if kind == "flac":
|
|
fmt = {16: "s16", 24: "s32", 32: "s32"}.get(bits, "s16")
|
|
cmd += ["-c:a", "flac", "-sample_fmt", fmt]
|
|
if bits == 24:
|
|
cmd += ["-bits_per_raw_sample", "24"]
|
|
cmd += ["-f", "flac", "-"]
|
|
else:
|
|
cmd += ["-c:a", "libmp3lame", "-b:a", "320k", "-write_xing", "0", "-id3v2_version", "0",
|
|
"-f", "mp3", "-"]
|
|
self.proc = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.DEVNULL)
|
|
|
|
def read(self, n=CHUNK):
|
|
return self.proc.stdout.read(n)
|
|
|
|
def close(self):
|
|
if self.proc.poll() is None:
|
|
self.proc.kill()
|
|
self.proc.stdout.close()
|
|
self.proc.wait()
|
|
|
|
|
|
def flac_open_ended(header):
|
|
"""STREAMINFO with total samples = 0 (unknown) and MD5 = 0, so decoders don't stop early."""
|
|
h = bytearray(header)
|
|
si = 8 # 'fLaC' + block header
|
|
# block/frame sizes "any": follow-up tracks (or ffmpeg conversions) may use other block sizes
|
|
h[si:si + 4] = (16).to_bytes(2, "big") + (65535).to_bytes(2, "big")
|
|
h[si + 4:si + 10] = b"\x00" * 6
|
|
h[si + 13] &= 0xF0
|
|
h[si + 14:si + 18] = b"\x00\x00\x00\x00"
|
|
h[si + 18:si + 34] = b"\x00" * 16
|
|
return bytes(h)
|
|
|
|
|
|
def flac_sync(buf, start=0):
|
|
"""Offset of the next plausible FLAC frame header (sync 0xFFF8/0xFFF9)."""
|
|
pos = start
|
|
while pos < len(buf) - 4:
|
|
if buf[pos] == 0xFF and buf[pos + 1] in (0xF8, 0xF9):
|
|
bs, sr = buf[pos + 2] >> 4, buf[pos + 2] & 0x0F
|
|
ch, ss = buf[pos + 3] >> 4, (buf[pos + 3] >> 1) & 0x07
|
|
if bs != 0 and sr != 15 and ch < 11 and ss not in (3, 7):
|
|
return pos
|
|
pos += 1
|
|
return -1
|
|
|
|
|
|
# ---------------------------------------------------------------------------- session
|
|
|
|
class Segment:
|
|
def __init__(self, upstream, uri, meta, info, label, cover=""):
|
|
self.upstream = upstream
|
|
self.cover = cover
|
|
self.uri = uri
|
|
self.meta = meta
|
|
self.info = info
|
|
self.label = label
|
|
self.streamed = False # proxy started sending its audio
|
|
|
|
@property
|
|
def duration(self):
|
|
return self.info.get("duration") or 0
|
|
|
|
|
|
class Session:
|
|
"""One continuous stream made of several tracks."""
|
|
|
|
def __init__(self, first, start_secs=0, lead_secs=20):
|
|
self.segments = [first]
|
|
self.start_secs = start_secs # stream begins this far into the first track
|
|
self.lead_secs = lead_secs # follow-up tracks are appended this long before they're due
|
|
self.cond = threading.Condition()
|
|
self.closed = False
|
|
self.ended = False # proxy finished writing (no more segments)
|
|
self.served = 0 # number of player connections served
|
|
self.generation = 0 # bumped by rebase(); writers of older generations stop
|
|
self.mp3_rate = 44100 # sample rate of the first MP3 track (target for conversions)
|
|
self.playback_secs = 0 # updated by the renderer from BluOS' counter
|
|
|
|
def starts(self):
|
|
"""Playback time (stream seconds) at which each segment starts."""
|
|
out, t = [], -self.start_secs
|
|
for seg in self.segments:
|
|
out.append(max(0, t))
|
|
t += seg.duration
|
|
return out
|
|
|
|
def end_of_last(self):
|
|
return sum(s.duration for s in self.segments) - self.start_secs
|
|
|
|
def end_of_streamed(self):
|
|
"""Playback time at which the audio sent so far runs out."""
|
|
sent = [s for i, s in enumerate(self.segments) if s.streamed or i == 0]
|
|
return sum(s.duration for s in sent) - self.start_secs
|
|
|
|
def hold_until_played(self, generation, margin=1):
|
|
"""Keep the connection open until playback reaches the end of the sent audio.
|
|
|
|
BluOS treats an EOF like a dropped radio stream and reconnects at once, so ending the
|
|
stream early would make it replay the session from the start.
|
|
"""
|
|
with self.cond:
|
|
while not self.closed and self.generation == generation and self.playback_secs < self.end_of_streamed() - margin:
|
|
self.cond.wait(1)
|
|
|
|
def _queued(self):
|
|
"""The follow-up segment that hasn't been sent to the player yet (or None)."""
|
|
if len(self.segments) >= 2 and not self.segments[-1].streamed:
|
|
return self.segments[-1]
|
|
return None
|
|
|
|
def offer_next(self, seg, current_index):
|
|
"""Queue seg after the playing track, replacing a queued one that wasn't sent yet.
|
|
|
|
Returns False if a different track was already sent to the player (too late to change).
|
|
"""
|
|
with self.cond:
|
|
queued = self._queued()
|
|
if queued is not None and len(self.segments) - 1 > current_index:
|
|
if queued.uri != seg.uri:
|
|
self.segments[-1] = seg
|
|
log.info("gapless: queued track changed to %r", seg.label)
|
|
elif len(self.segments) - 1 == current_index:
|
|
self.segments.append(seg)
|
|
elif self.segments[-1].uri != seg.uri:
|
|
return False
|
|
self.cond.notify_all()
|
|
return True
|
|
|
|
def clear_next(self, current_index):
|
|
"""Drop the queued follow-up track (queue emptied in the app). False if already sent."""
|
|
with self.cond:
|
|
if len(self.segments) - 1 <= current_index:
|
|
return True
|
|
if self._queued() is not None:
|
|
self.segments.pop()
|
|
log.info("gapless: queued track removed")
|
|
self.cond.notify_all()
|
|
return True
|
|
return False
|
|
|
|
def wait_segment(self, i, timeout_after_end=8):
|
|
"""Block until segment i exists and is due. None if the session ends.
|
|
|
|
Follow-up tracks are only sent lead_secs before they start, so queue changes in the app
|
|
until then still take effect.
|
|
"""
|
|
with self.cond:
|
|
while not self.closed:
|
|
if len(self.segments) > i:
|
|
if i == 0 or self.playback_secs >= self.starts()[i] - self.lead_secs:
|
|
return self.segments[i]
|
|
elif self.playback_secs >= self.end_of_last() + timeout_after_end:
|
|
# nothing queued: give up shortly after playback reaches the end of the stream
|
|
return None
|
|
self.cond.wait(1)
|
|
return None
|
|
|
|
def rebase(self, index, position):
|
|
"""Continue the session from segment `index` at `position` seconds (player reconnected).
|
|
|
|
BluOS drops paused "radio" streams after about two minutes and reconnects on resume with
|
|
its counter reset to 0 - without this it would hear the session from the very start.
|
|
"""
|
|
with self.cond:
|
|
self.segments = self.segments[index:]
|
|
for seg in self.segments:
|
|
seg.streamed = False
|
|
self.start_secs = max(0, int(position))
|
|
self.playback_secs = 0
|
|
self.ended = False
|
|
self.generation += 1
|
|
self.cond.notify_all()
|
|
|
|
def close(self):
|
|
with self.cond:
|
|
self.closed = True
|
|
self.cond.notify_all()
|
|
|
|
|
|
class IcyWriter:
|
|
"""Writes audio, inserting ICY metadata (StreamTitle) every metaint bytes if requested."""
|
|
|
|
def __init__(self, out, metaint=0):
|
|
self.out = out
|
|
self.metaint = metaint
|
|
self.count = 0
|
|
self.title = ""
|
|
self.cover = "" # sent as StreamUrl - some players show it as artwork
|
|
self.sent_title = None
|
|
|
|
def write(self, data):
|
|
if not self.metaint:
|
|
self.out.write(data)
|
|
return
|
|
while data:
|
|
room = self.metaint - self.count
|
|
part, data = data[:room], data[room:]
|
|
self.out.write(part)
|
|
self.count += len(part)
|
|
if self.count == self.metaint:
|
|
self.count = 0
|
|
if (self.title, self.cover) != self.sent_title:
|
|
text = f"StreamTitle='{self.title.replace(chr(39), '`')}';"
|
|
if self.cover:
|
|
text += f"StreamUrl='{self.cover}';"
|
|
text = text.encode("utf-8")[:4000]
|
|
blocks = (len(text) + 15) // 16
|
|
self.out.write(bytes([blocks]) + text.ljust(blocks * 16, b"\x00"))
|
|
self.sent_title = (self.title, self.cover)
|
|
else:
|
|
self.out.write(b"\x00")
|
|
|
|
|
|
def _open(url, start=0):
|
|
req = urllib.request.Request(url, headers={"Range": f"bytes={start}-"})
|
|
return urllib.request.urlopen(req, timeout=20)
|
|
|
|
|
|
def _total_size(resp):
|
|
cr = resp.headers.get("Content-Range", "")
|
|
tail = cr.rsplit("/", 1)[-1] if cr else resp.headers.get("Content-Length", "")
|
|
return int(tail) if tail.isdigit() else 0
|
|
|
|
|
|
def _read_at_least(resp, buf, n):
|
|
while len(buf) < n:
|
|
chunk = resp.read(CHUNK)
|
|
if not chunk:
|
|
break
|
|
buf += chunk
|
|
return buf
|
|
|
|
|
|
def stream_session(session, writer, kind_hint=None):
|
|
"""Write all segments of the session into writer. Returns the number of segments sent."""
|
|
kind = None
|
|
params = None
|
|
sent = 0
|
|
i = 0
|
|
gen = session.generation
|
|
while True:
|
|
seg = session.wait_segment(i)
|
|
if seg is None or session.generation != gen:
|
|
break
|
|
start_secs = session.start_secs if i == 0 else 0
|
|
try:
|
|
resp = _open(seg.upstream)
|
|
except Exception as exc:
|
|
log.warning("segment %r: upstream failed: %s", seg.label, exc)
|
|
break
|
|
ctype = resp.headers.get("Content-Type", "")
|
|
seg_kind = "flac" if "flac" in ctype else "mp3" if ("mpeg" in ctype or "mp3" in ctype) else None
|
|
total = _total_size(resp)
|
|
buf = _read_at_least(resp, b"", 4096)
|
|
convert = False
|
|
src_format = None
|
|
if i > 0 and seg_kind == "flac" and kind == "flac":
|
|
while flac_header_len(buf) == -1:
|
|
more = resp.read(CHUNK)
|
|
if not more:
|
|
break
|
|
buf += more
|
|
hl = flac_header_len(buf)
|
|
convert = hl > 0 and flac_params(buf[:hl]) != params # different sample rate / bit depth
|
|
src_format = flac_format(flac_params(buf[:hl])) if hl > 0 else None
|
|
if i > 0 and (seg_kind != kind or convert) and LOSSLESS_ONLY and not _lossless_conversion(seg_kind, kind, src_format, params):
|
|
log.info("segment %r (%s) would lose quality as %s, ending gapless stream (normal track change)",
|
|
seg.label, ctype or "?", kind)
|
|
resp.close()
|
|
break
|
|
if i > 0 and (seg_kind != kind or convert):
|
|
if not FFMPEG:
|
|
log.info("segment %r (%s) can't be stitched to %s (no ffmpeg), ending gapless stream",
|
|
seg.label, ctype, kind)
|
|
resp.close()
|
|
break
|
|
resp.close()
|
|
rate, channels, bits = flac_format(params) if kind == "flac" else (session.mp3_rate, 2, 16)
|
|
log.info("gapless: converting %r (%s) to %s %s Hz", seg.label, ctype or "?", kind, rate)
|
|
resp = Transcoder(seg.upstream, kind, rate, channels, bits)
|
|
seg_kind, total = kind, 0
|
|
buf = _read_at_least(resp, b"", 4096)
|
|
|
|
if seg_kind == "mp3":
|
|
head = id3v2_size(buf)
|
|
buf = _read_at_least(resp, buf, head + 4096)
|
|
audio = mp3_sync(buf, head)
|
|
if audio < 0:
|
|
log.warning("segment %r: no MP3 frames found", seg.label)
|
|
resp.close()
|
|
break
|
|
if i == 0:
|
|
session.mp3_rate = mp3_rate(buf, audio)
|
|
if i > 0 or start_secs:
|
|
audio = mp3_skip_info_frame(buf, audio)
|
|
if start_secs and seg.duration and total:
|
|
# jump into the file: approximate byte position, then resync to a frame
|
|
target = audio + int((total - audio) * start_secs / seg.duration)
|
|
resp.close()
|
|
resp = _open(seg.upstream, target)
|
|
buf = _read_at_least(resp, b"", 8192)
|
|
pos = mp3_sync(buf, 0)
|
|
buf = buf[pos:] if pos >= 0 else b""
|
|
else:
|
|
buf = buf[audio:] if i > 0 else buf # first track: keep its tags for the decoder
|
|
else: # flac
|
|
hlen = flac_header_len(buf)
|
|
while hlen == -1:
|
|
more = resp.read(CHUNK)
|
|
if not more:
|
|
break
|
|
buf += more
|
|
hlen = flac_header_len(buf)
|
|
if hlen <= 0:
|
|
log.warning("segment %r: invalid FLAC header", seg.label)
|
|
resp.close()
|
|
break
|
|
header = buf[:hlen]
|
|
p = flac_params(header)
|
|
if params and p != params:
|
|
log.info("segment %r: different FLAC format, ending gapless stream", seg.label)
|
|
resp.close()
|
|
break
|
|
params = params or p
|
|
if i == 0:
|
|
writer.write(flac_open_ended(header))
|
|
if start_secs and seg.duration and total:
|
|
target = hlen + int((total - hlen) * start_secs / seg.duration)
|
|
resp.close()
|
|
resp = _open(seg.upstream, target)
|
|
buf = _read_at_least(resp, b"", 16384)
|
|
pos = flac_sync(buf, 0)
|
|
buf = buf[pos:] if pos >= 0 else b""
|
|
else:
|
|
buf = buf[hlen:]
|
|
|
|
kind = kind or seg_kind
|
|
with session.cond:
|
|
if session.generation != gen:
|
|
resp.close() # player reconnected, a newer writer takes over
|
|
return sent
|
|
if session.closed or i >= len(session.segments) or session.segments[i] is not seg:
|
|
resp.close() # queue changed while the upstream was opening
|
|
continue
|
|
seg.streamed = True
|
|
writer.title = seg.label
|
|
writer.cover = seg.cover
|
|
log.info("gapless: streaming segment %d %r (%s)", i, seg.label, seg_kind)
|
|
# hold back the last 128 bytes so a trailing ID3v1 tag can be dropped
|
|
tail = b""
|
|
data = buf
|
|
while True:
|
|
if session.closed or session.generation != gen:
|
|
resp.close()
|
|
return sent
|
|
data = tail + data
|
|
tail = data[-128:]
|
|
if len(data) > 128:
|
|
try:
|
|
writer.write(data[:-128])
|
|
except Exception:
|
|
resp.close() # player went away - don't leave the server stream hanging
|
|
raise
|
|
more = resp.read(CHUNK)
|
|
if not more:
|
|
break
|
|
data = more
|
|
resp.close()
|
|
if not (seg_kind == "mp3" and tail[:3] == b"TAG" and len(tail) == 128):
|
|
writer.write(tail)
|
|
sent += 1
|
|
i += 1
|
|
if sent:
|
|
session.hold_until_played(gen)
|
|
if session.generation == gen:
|
|
session.ended = True
|
|
return sent
|