Something went wrong. Try again.
My personal dotfiles and NixOS configurations.
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629--- a/src/playback_backend.py+++ b/src/playback_backend.py@@ -1,4 +1,6 @@ import gi+import threading+from collections import deque gi.require_version('Gst', '1.0') @@ -12,7 +14,8 @@ class PlaybackBackend: def __init__(self, state_callback=None, finish_callback=None, error_callback=None, position_callback=None, codec_callback=None, buffering_callback=None, eos_callback=None):- self.pipeline = Gst.ElementFactory.make("playbin3")+ self.pipeline = Gst.ElementFactory.make("playbin")+ self.pipeline.connect("source-setup", self._on_source_setup) # qtdemux cannot seek fragmented MP4 (what /decrypt serves) while # streaming over HTTP. Download buffering keeps the track in a # seekable temp file, which needs the server's Range support.@@ -24,11 +27,23 @@ class PlaybackBackend: self._codec_callback = codec_callback self._buffering_callback = buffering_callback self._eos_callback = eos_callback- self._next_uri = ""+ self._next_uri = None+ self._lock = threading.RLock()+ self._epoch = 0+ self._streams = deque()+ self._starts = {}+ self._assigned = None+ self._assigned_uri = self._current_uri = None+ self._pending_seek = None+ self._want_playing = False+ self._live = False+ self._buffering = {}+ self._target_state = Gst.State.NULL bus = self.pipeline.get_bus() self._closed = False self._bus = bus+ bus.set_sync_handler(self._on_sync_message) bus.add_signal_watch() self._bus_handlers = [bus.connect("message", self._on_bus_message), bus.connect("message::stream-collection", self._on_stream_collection)]@@ -37,6 +52,8 @@ class PlaybackBackend: self._start_position_poll() def set_live(self, live: bool) -> None:+ self._live = live+ self._buffering.clear() # An endless live stream must never fill a download-to-disk buffer. flags = self.pipeline.get_property("flags") self.pipeline.set_property("flags", flags & ~_PLAY_FLAG_DOWNLOAD if live else flags | _PLAY_FLAG_DOWNLOAD)@@ -95,7 +112,7 @@ class PlaybackBackend: def poll(): if self._closed: return GLib.SOURCE_REMOVE- if self._poll_position:+ if self._poll_position and self._pending_seek is None: success, pos = self.pipeline.query_position(Gst.Format.TIME) if success and self._position_callback: self._position_callback(pos)@@ -111,6 +128,7 @@ class PlaybackBackend: for handler in self._bus_handlers: self._bus.disconnect(handler) self._bus.remove_signal_watch()+ self._bus.set_sync_handler(None) self.pipeline.disconnect(self._finish_handler) self.stop() self._state_callback = self._finish_callback = self._error_callback = None@@ -119,6 +137,46 @@ class PlaybackBackend: def _on_bus_message(self, bus, message): message_type = message.type + if message_type == Gst.MessageType.TAG and self._codec_callback:+ index = self.pipeline.get_property("current-audio")+ pad = self.pipeline.emit("get-audio-pad", index) if index >= 0 else None+ caps = pad.get_current_caps() if pad is not None else None+ if caps is not None and not caps.is_empty():+ label = self._codec_label_from_structure(caps.get_structure(0))+ if label:+ self._codec_callback(label)++ if message_type == Gst.MessageType.STREAM_START and message.src == self.pipeline:+ with self._lock:+ item = self._starts.pop(message.get_seqnum(), None)+ if item is None or item[0] != self._epoch:+ return True+ token = item[1]+ if token is not None and token == self._assigned:+ self._current_uri = self._assigned_uri+ self._assigned = None+ else:+ token = None+ if token is not None:+ self._apply_state()+ if self._finish_callback:+ self._finish_callback(token)+ return True++ if message_type == Gst.MessageType.ASYNC_DONE and message.src == self.pipeline:+ pending = self._pending_seek+ if pending is not None:+ timestamp, callback, phase = pending+ if phase == "preroll":+ self._pending_seek = (timestamp, callback, "seek")+ if not self.pipeline.seek_simple(+ Gst.Format.TIME, Gst.SeekFlags.FLUSH | Gst.SeekFlags.ACCURATE,+ timestamp * Gst.MSECOND):+ self._finish_seek(False)+ else:+ self._finish_seek(True)+ return True+ if message_type == Gst.MessageType.EOS: if self._eos_callback: self._eos_callback()@@ -126,59 +184,144 @@ class PlaybackBackend: if message_type == Gst.MessageType.ERROR: err, debug = message.parse_error()+ self._finish_seek(False) if self._error_callback: self._error_callback(err, debug) return True if message_type == Gst.MessageType.BUFFERING:- if self._buffering_callback:- self._buffering_callback(message.parse_buffering())+ self._handle_buffering(message) return True if message_type != Gst.MessageType.STATE_CHANGED or message.src != self.pipeline: return True _, new_state, _ = message.parse_state_changed()- if self._state_callback:+ if self._state_callback and not (new_state == Gst.State.PAUSED and self._want_playing): self._state_callback(new_state) return True + def _buffer_uri(self, source):+ while source is not None and source != self.pipeline:+ if source.find_property("uri"):+ return source.get_property("uri")+ source = source.get_parent()+ return None++ def _handle_buffering(self, message):+ if self._live:+ return+ uri = self._buffer_uri(message.src)+ percent = message.parse_buffering()+ if percent < 100:+ self._buffering[message.src] = uri+ else:+ self._buffering.pop(message.src, None)+ if uri is None or uri == self._current_uri:+ if self._buffering_callback:+ self._buffering_callback(percent)+ self._apply_state()++ def _apply_state(self):+ playing = self._want_playing and self._pending_seek is None+ blocked = any(uri is None or uri == self._current_uri for uri in self._buffering.values())+ state = Gst.State.PLAYING if playing and (self._live or not blocked) else Gst.State.PAUSED+ if state == self._target_state:+ return+ self._target_state = state+ result = self.pipeline.set_state(state)+ if result == Gst.StateChangeReturn.NO_PREROLL:+ self._live = True+ self._buffering.clear()+ if self._want_playing:+ self.pipeline.set_state(Gst.State.PLAYING)++ def _on_sync_message(self, bus, message, *args):+ if message.type == Gst.MessageType.STREAM_START and message.src == self.pipeline:+ try:+ item = self._streams.popleft()+ except IndexError:+ pass+ else:+ self._starts[message.get_seqnum()] = item+ return Gst.BusSyncReply.PASS+ def _on_about_to_finish(self, *args):- # about-to-finish comes from the pipeline; marshal onto the main loop- # so the switch mutates GObject state safely.- if self._next_uri and self._finish_callback:- uri = self._next_uri- def finish():- if not self._closed and uri == self._next_uri and self._finish_callback:- self._finish_callback(uri)- return GLib.SOURCE_REMOVE- GLib.idle_add(finish)+ with self._lock:+ if self._closed or self._next_uri is None or self._assigned is not None:+ return+ uri, token = self._next_uri+ self._next_uri = None+ self._assigned = token+ self._assigned_uri = uri+ self._streams.append((self._epoch, token))+ self.pipeline.set_property("uri", uri)++ def queue_uri(self, uri: str, token) -> None:+ with self._lock:+ self._next_uri = (uri, token) - def queue_uri(self, uri: str) -> None:- self._next_uri = uri+ def clear_next_uri(self) -> bool:+ with self._lock:+ self._next_uri = None+ return self._assigned is not None - def clear_next_uri(self) -> None:- self._next_uri = ""+ def _on_source_setup(self, pipeline, source):+ uri = source.get_property("location") if source.find_property("location") else ""+ if uri and uri.startswith("http://127.0.0.1:8129/decrypt/") and source.find_property("timeout"):+ source.set_property("timeout", 180) def load_uri(self, uri: str, user_initiated: bool = True) -> None:+ with self._lock:+ self._streams.append((self._epoch, None))+ self._current_uri = uri+ self._want_playing = user_initiated self.pipeline.set_property("uri", uri)- if user_initiated:- self.pipeline.set_state(Gst.State.PLAYING)- else:- self.pipeline.set_state(Gst.State.PAUSED)+ self._apply_state() def play(self) -> None:- self.pipeline.set_state(Gst.State.PLAYING)+ self._want_playing = True+ self._apply_state()+ if self._state_callback:+ self._state_callback(Gst.State.PLAYING) def pause(self) -> None:- self.pipeline.set_state(Gst.State.PAUSED)+ self._want_playing = False+ self._apply_state()+ if self._state_callback:+ self._state_callback(Gst.State.PAUSED) def stop(self) -> None:+ self._pending_seek = None+ self._want_playing = False+ self._buffering.clear()+ with self._lock:+ self._epoch += 1+ self._next_uri = self._assigned = None+ self._streams.clear()+ self._starts.clear()+ self._target_state = Gst.State.NULL self.pipeline.set_state(Gst.State.NULL)+ self._bus.set_flushing(True)+ self._bus.set_flushing(False)++ def seek(self, timestamp_milli: int, callback=None) -> None:+ uri, playing = self._current_uri, self._want_playing+ self.stop()+ if not uri:+ if callback:+ callback(False)+ return+ self._pending_seek = (timestamp_milli, callback, "preroll")+ self.load_uri(uri, False)+ self._want_playing = playing - def seek(self, timestamp_milli: int) -> None:- self.pipeline.seek_simple(- Gst.Format.TIME,- Gst.SeekFlags.FLUSH | Gst.SeekFlags.KEY_UNIT,- timestamp_milli * Gst.MSECOND,- )+ def _finish_seek(self, success):+ pending = self._pending_seek+ if pending is None:+ return+ self._pending_seek = None+ self._apply_state()+ callback = pending[1]+ if callback:+ callback(success)--- a/src/preloader.py+++ b/src/preloader.py@@ -1,78 +1,122 @@-"""Get the queue's next song ready before the current one ends."""-from typing import Any, TYPE_CHECKING+import logging+import threading+import time+import urllib.request -from .playback_backend import PlaybackBackend-from .playback_preparer import PlaybackPreparer-from .resources import SongQuality+from gi.repository import GLib+from .playback_preparer import DECRYPT_URL, PlaybackPreparer from .song import Song -if TYPE_CHECKING:- from .api import AppleMusicAPI -# How long before the current song ends the next one is prepared-PRELOAD_AHEAD_MS = 15_000+WARM_ATTEMPTS = 4+WARM_TIMEOUT = 30+WARM_BUDGET = 180+WARM_BACKOFF = 1 class NextSongPreloader:- """Prepares the next queue entry and hands it to the pipeline in advance.-- playbin3 then switches to it at about-to-finish without a gap. The entry- is remembered by identity, so a queue edit that changes what comes next- can drop a preload that would otherwise play under the wrong song.- """-- def __init__(self, backend: PlaybackBackend) -> None:+ def __init__(self, backend): self._backend = backend- # Separate from the current song's preparer, so neither cancels the other self._preparer = PlaybackPreparer()- # The queue entry being prepared or ready, and its URI once ready- self._entry: dict[str, Any] | None = None+ self._entry = None+ self._token = None self._uri = ""- # An entry that failed to prepare; not retried on every position update- self._failed: dict[str, Any] | None = None+ self._failed = None+ self._retry_after = 0+ self._cancel_warm = threading.Event() - @property- def ready_uri(self) -> str:- return self._uri-- def consider(self, remaining_milli: int, entry: dict[str, Any] | None,- api: "AppleMusicAPI | None", quality: SongQuality) -> None:- """Start preparing `entry` once the current song is close enough to its end."""- if (entry is None or self._entry is not None or entry is self._failed- or remaining_milli > PRELOAD_AHEAD_MS):+ def consider(self, remaining_milli, entry, api, quality):+ if (entry is None or self._entry is not None+ or (entry is self._failed and time.monotonic() < self._retry_after)): return self._entry = entry+ token = self._token = object()+ cancel = self._cancel_warm = threading.Event() song = Song.new(entry)- self._preparer.prepare(song, api, quality,- lambda uri, error: self._prepared(entry, song, uri, error)) - def _prepared(self, entry: dict[str, Any], song: Song, uri: str | None, error: BaseException | None) -> None:- if entry is not self._entry:- return- if error or not uri:- print(f"Cannot preload '{song.name}': {error}")- self._entry = None- self._failed = entry- return- self._uri = uri- self._backend.queue_uri(uri)+ def ready(uri, error):+ if token is not self._token:+ return+ if error or not uri:+ print(f"Cannot preload: {error or 'no URI'}; retry in 10 seconds")+ self._failed = entry+ self._retry_after = time.monotonic() + 10+ self._entry = self._token = None+ self._uri = ""+ return+ self._failed = None+ if uri.startswith(DECRYPT_URL):+ threading.Thread(target=warm, args=(uri,), daemon=True,+ name="gapless-prefetch").start()+ else:+ arm(uri) - def queue_changed(self, next_entry: dict[str, Any] | None) -> None:- """Drop the preload if the queue no longer has its entry next."""+ def arm(uri):+ if token is self._token:+ self._uri = uri+ self._backend.queue_uri(uri, token)+ return GLib.SOURCE_REMOVE++ def warm(uri):+ deadline = time.monotonic() + WARM_BUDGET+ for attempt in range(WARM_ATTEMPTS):+ if cancel.is_set():+ return+ try:+ remaining = deadline - time.monotonic()+ if remaining <= 0:+ raise TimeoutError("prefetch budget exhausted")+ with urllib.request.urlopen(uri, timeout=min(WARM_TIMEOUT, remaining)) as response:+ expected = int(response.headers.get("Content-Length", "0"))+ if response.status != 200 or expected <= 0:+ raise ValueError("prefetch needs HTTP 200 and a positive Content-Length")+ received = 0+ while True:+ if cancel.is_set():+ return+ if time.monotonic() >= deadline:+ raise TimeoutError("prefetch budget exhausted")+ chunk = response.read1(64 * 1024)+ if not chunk:+ break+ received += len(chunk)+ if received != expected:+ raise ValueError(f"short prefetch: {received}/{expected} bytes")+ if not cancel.is_set():+ GLib.idle_add(arm, uri)+ return+ except Exception as error:+ if cancel.is_set():+ return+ logging.warning("Gapless prefetch attempt %d/%d failed (%s)",+ attempt + 1, WARM_ATTEMPTS, type(error).__name__)+ if attempt + 1 == WARM_ATTEMPTS or time.monotonic() >= deadline:+ GLib.idle_add(ready, None, RuntimeError("prefetch exhausted; using EOS fallback"))+ return+ if cancel.wait(min(WARM_BACKOFF * 2 ** attempt,+ max(0, deadline - time.monotonic()))):+ return++ self._preparer.prepare(song, api, quality, ready)++ def queue_changed(self, next_entry): if self._entry is not None and next_entry is not self._entry:- self.drop()+ return self.drop()+ return False - def take(self) -> str:- """The ready URI, handed over once: the preload is done after this."""- uri = self._uri- if uri:- self._entry = None- self._uri = ""- self._backend.clear_next_uri()- return uri+ def take(self, token):+ if token is not self._token or not self._uri:+ return False+ self._cancel_warm.set()+ self._entry = self._token = None+ self._uri = ""+ return True - def drop(self) -> None:+ def drop(self):+ self._cancel_warm.set()+ self._failed = None+ self._retry_after = 0 self._preparer.cancel()- self._entry = None+ self._entry = self._token = None self._uri = ""- self._backend.clear_next_uri()+ return self._backend.clear_next_uri()--- a/src/player.py+++ b/src/player.py@@ -4,6 +4,7 @@ from pathlib import Path from typing import Any import gi+import os gi.require_version('Gst', '1.0') @@ -27,8 +28,9 @@ from .song import Song from .stations import StationLoader Gst.init(None)-Gst.debug_set_active(False)-Gst.debug_set_default_threshold(0)+if not os.environ.get("GST_DEBUG"):+ Gst.debug_set_active(False)+ Gst.debug_set_default_threshold(0) MILLISECOND: int = 1_000@@ -244,9 +246,7 @@ class Player(GObject.GObject): if self.episode_active: self.radio.update_position() return- if self.song and self.song.duration_milli:- self.preloader.consider(self.song.duration_milli - self.timestamp_milli,- self._queue_entry(self.track_pos + 1), self.api, self.quality)+ self._prepare_successor() def _on_eos(self) -> None: """The stream ended without a gapless switch to the next song.@@ -262,9 +262,6 @@ class Player(GObject.GObject): return if self.track_pos < 0: return- if self.preloader.ready_uri:- self._switch_to_preloaded()- return self.preloader.drop() self.activity.ended(END_NATURAL_END_OF_TRACK, self.song.duration_milli if self.song else 0) self.track_pos += 1@@ -281,20 +278,24 @@ class Player(GObject.GObject): else: self.emit("playback-error", _("Couldn’t play {name}").format(name=name)) - def _on_about_to_finish(self, uri: str) -> None:- self._switch_to_preloaded()-- def _switch_to_preloaded(self) -> None:- uri = self.preloader.take()- if not uri:+ def _on_about_to_finish(self, token) -> None:+ if not self.preloader.take(token): return- self.backend.load_uri(uri, True) self.activity.ended(END_NATURAL_END_OF_TRACK, self.song.duration_milli if self.song else 0) self._gapless_switch = True try: self.track_pos += 1 finally: self._gapless_switch = False+ self._prepare_successor()++ def _prepare_successor(self) -> None:+ if self.radio_active or self.episode_active or self.looping == LoopStatus.TRACK:+ return+ entry = self._queue_entry(self.track_pos + 1)+ if entry is not None and Song.new(entry).has_drm and not self.previews_only and not self.wrapper_ready:+ return+ self.preloader.consider(0, entry, self.api, self.quality) # The queue @@ -306,14 +307,18 @@ class Player(GObject.GObject): def _on_queue_changed(self, *args: Any) -> None: # Inserting, removing, moving or reordering songs can change what # comes next; a song preloaded before that must not play instead.- self.preloader.queue_changed(self._queue_entry(self.track_pos + 1))+ if self.preloader.queue_changed(self._queue_entry(self.track_pos + 1)):+ self._on_pos_changed()+ elif self.state != PlaybackState.STOPPED:+ self._prepare_successor() def _on_pos_changed(self, *args: Any) -> None: """Play the song at the new queue position.""" if self._reindexing: return user_initiated = not self._restoring- self.activity.ended(END_SELECTED_OTHER_ITEM)+ if not self._gapless_switch:+ self.activity.ended(END_SELECTED_OTHER_ITEM) if self.track_pos < 0: return if self.track_pos >= self.tracklist.n_items:@@ -372,6 +377,7 @@ class Player(GObject.GObject): self.loading = False self.emit("playback-error", toast_text(failure)) + self._prepare_successor() self.preparer.prepare(song, self.api, self.quality, prepared) def _start_playback(self, uri: str, user_initiated: bool) -> None:@@ -380,12 +386,14 @@ class Player(GObject.GObject): self._pending_playback = (uri, user_initiated) return self._pending_playback = None+ self._prepare_successor() self.backend.load_uri(uri, user_initiated) def _on_wrapper_ready(self, *args: Any) -> None: if not self.wrapper_ready: return self.radio.wrapper_ready()+ self._prepare_successor() if self._pending_playback: uri, user_initiated = self._pending_playback self._pending_playback = None@@ -442,7 +450,7 @@ class Player(GObject.GObject): # Live radio cannot be paused; the button stops it instead. Pressing # play again restarts the same station. self.radio.stop_for_resume()- elif self.state == PlaybackState.PLAYING:+ elif self.backend._want_playing: self.pause() else: self.play()@@ -456,7 +464,7 @@ class Player(GObject.GObject): def pause(self) -> None: if self.radio_active: self.play_pause()- elif self.state == PlaybackState.PLAYING:+ elif self.backend._want_playing: self.backend.pause() def stop(self) -> None:@@ -494,9 +502,15 @@ class Player(GObject.GObject): def seek(self, timestamp_milli: int) -> None: if not self.can_seek: return- self.timestamp_milli = timestamp_milli- self.backend.seek(timestamp_milli)- self.emit("seeked")+ self.preloader.drop()++ def applied(success):+ if success:+ self.timestamp_milli = timestamp_milli+ self._prepare_successor()+ self.emit("seeked")++ self.backend.seek(timestamp_milli, applied) def toggle_mute(self) -> None: self.mute = not self.mute