diff --git a/CLAUDE.md b/CLAUDE.md index 7e219ce..9791439 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -67,13 +67,19 @@ persistence) → GUI (Qt widgets that read the manager and connect to its signal without recursing. Widgets react to its signals (`playlists_changed`, `playlist_content_changed(pid)`, `track_updated(id)`, `track_fields_edited`). -- **`lintunes/player.py`** — `Player(QObject)` wraps `QMediaPlayer`/`QAudioOutput`. - It is deliberately **context-agnostic** (knows a queue + shuffle walk order, not - which view started playback). It exposes `track_changed`/`playing_changed`/ - `position_changed`/`duration_changed` signals consumed by the transport, MPRIS, - and the visualizer (a `QAudioBufferOutput` tee feeds PCM to the spectrum bars). - Playback *context* ("library" / "playlist:") is tracked in `MainWindow`, - not the player. +- **`lintunes/player.py`** — `Player(QObject)` walks a queue through a swappable + **`PlaybackSink`**. Player owns the queue, shuffle walk, per-track start/stop + times and the play-count/scrobble bookkeeping; a sink owns only "make this file + come out of something, and report where it's up to". `LocalSink` (the + `QMediaPlayer`/`QAudioOutput` pipeline, with the `QAudioBufferOutput` PCM tee + that feeds the visualizer) **stays in `player.py`** — the Player tests stub Qt + Multimedia with `patch.multiple(player_module, QMediaPlayer=..., …)`, so those + names must resolve in this module. `cast/sink.py::CastSink` is the other + implementation. `set_sink()` carries the current track, position and + playing-state across a swap and deliberately does *not* re-emit + `track_changed` (that would double-scrobble the same song). + Player is also deliberately **context-agnostic**: playback *context* + ("library" / "playlist:") is tracked in `MainWindow`, not the player. - **`lintunes/gui/`** — `main_window.py` assembles a top `TransportBar` over a horizontal `QSplitter` (`SidebarPanel` | stacked `LibraryView`/`PlaylistView`). @@ -105,6 +111,21 @@ persistence) → GUI (Qt widgets that read the manager and connect to its signal plus an Auxio-importable `.m3u`); it never deletes outside that folder and only ever *reads* local library files. +- **`lintunes/cast/`** — Chromecast playback (Connections menu), using the + **media-receiver model**: `server.py` runs a `ThreadingHTTPServer` on an + ephemeral port for the life of a session and the device fetches the *original* + file itself (bit-exact, no transcode). URLs carry an opaque random token, never + a path, so traversal is structurally impossible; Range + HEAD are mandatory + (the device seeks by re-requesting ranges and won't report a duration without + them). `support.py` is the format gate — ALAC, AIFF and protected AAC are + refused and Player skips them with a status-bar message. `discovery.py` wraps + `CastBrowser`; `sink.py` is the `PlaybackSink`; `controller.py` owns the + session and its own `SleepInhibitor`. **pychromecast is imported lazily**, never + at module scope, so the app still launches where the dep isn't installed yet. + While casting there is no local PCM, so the visualizer shows the cast glyph + instead of bars, and position comes from a 500 ms poll of + `adjusted_current_time` (only trusted while PLAYING — it creeps while paused). + ## Conventions & gotchas - **Tests are organized as `tests/test_roundN.py`** — each development round adds diff --git a/lintunes/__init__.py b/lintunes/__init__.py index fab3560..0322b42 100644 --- a/lintunes/__init__.py +++ b/lintunes/__init__.py @@ -1,3 +1,3 @@ """LinTunes — iTunes-style music library manager and player for Linux.""" -__version__ = "0.3.0" +__version__ = "0.4.0" diff --git a/lintunes/cast/__init__.py b/lintunes/cast/__init__.py new file mode 100644 index 0000000..0e78248 --- /dev/null +++ b/lintunes/cast/__init__.py @@ -0,0 +1,14 @@ +"""Chromecast playback: discovery, the local file server, and the session. + +Casting uses the *media receiver* model — lintunes hands the Chromecast a URL +and the device fetches and decodes the original file itself. That keeps the +audio bit-exact (no transcode) and lets the device do its own buffering, at the +cost of lintunes becoming a remote control while connected: no local PCM, so +the visualizer goes flat, and position is polled from the device rather than +read from Qt. + +Nothing here imports pychromecast at module scope — the import is deferred into +the functions that need it, so lintunes still launches on a machine that has +pulled this code but not yet run ``pip install -e .`` (see CLAUDE.md on the +self-updater shipping code but not dependencies). +""" diff --git a/lintunes/cast/controller.py b/lintunes/cast/controller.py new file mode 100644 index 0000000..a503edc --- /dev/null +++ b/lintunes/cast/controller.py @@ -0,0 +1,189 @@ +"""Owns a cast session on MainWindow's behalf. + +Widget-free by design: it reports state as signals and MainWindow decides what +to show. Held as an attribute on MainWindow — a worker object that goes out of +scope stops delivering its signals. +""" + +import threading + +from PyQt6.QtCore import QObject, QTimer, pyqtSignal + +from lintunes.cast.discovery import CastDiscovery, is_available +from lintunes.cast.server import TrackServer +from lintunes.cast.sink import CastSink +from lintunes.eventlog import log_control +from lintunes.inhibit import INHIBIT_LOGOUT, INHIBIT_SUSPEND, SleepInhibitor + +CONNECT_TIMEOUT_S = 15.0 + + +class CastController(QObject): + """Connects to a Chromecast, hands Player a sink, and cleans up after.""" + + connecting = pyqtSignal(object) # CastDevice + connected = pyqtSignal(str) # device name + disconnected = pyqtSignal(str) # reason; "" when the user asked + failed = pyqtSignal(str) + + def __init__(self, player, parent=None): + super().__init__(parent) + self._player = player + self._sink: CastSink | None = None + self._server: TrackServer | None = None + self._discovery: CastDiscovery | None = None + self._attempt = 0 + self._connect_done = _ConnectResult(self) + self._connect_done.ready.connect(self._on_connect_result) + # Its own instance, like the device-sync one: while casting, this + # machine is the file server for the whole room, so a suspend or a + # logout cuts the music off mid-song. + self._inhibitor = SleepInhibitor( + reason="Casting to a Chromecast", + flags=INHIBIT_LOGOUT | INHIBIT_SUSPEND) + + # ---- state ---- + + def is_available(self) -> tuple[bool, str]: + return is_available() + + def is_connected(self) -> bool: + return self._sink is not None + + def device_name(self) -> str: + return self._sink.name if self._sink is not None else "" + + def discovery(self) -> CastDiscovery: + """Created lazily and shared with the picker dialog, which needs the + live browser to turn a chosen device back into a connection.""" + if self._discovery is None: + self._discovery = CastDiscovery(self) + return self._discovery + + # ---- connecting ---- + + def connect_to(self, device): + """Connect in the background; the outcome arrives as a signal.""" + if self._sink is not None: + self.disconnect() + self._attempt += 1 + attempt = self._attempt + self.connecting.emit(device) + log_control("cast", "connect", f"{device.name} at {device.host}") + + discovery = self.discovery() + browser, zconf = discovery.browser(), discovery.zconf() + + def work(): + try: + import pychromecast + info = None + if browser is not None: + info = browser.devices.get(_uuid(device.uuid)) + if info is not None: + cast = pychromecast.get_chromecast_from_cast_info(info, zconf) + else: + # The browser was already torn down (the dialog closed) — + # reach the device directly by address instead. + cast = pychromecast.get_chromecast_from_host( + (device.host, device.port, _uuid(device.uuid), + device.model, device.name)) + cast.wait(timeout=CONNECT_TIMEOUT_S) + self._connect_done.ready.emit({"cast": cast}, attempt) + except Exception as e: # noqa: BLE001 — report, never crash + self._connect_done.ready.emit({"error": str(e)}, attempt) + + threading.Thread(target=work, daemon=True).start() + + def _on_connect_result(self, result: dict, attempt: int): + # The user can cancel and re-pick while a connect is in flight; a + # result from a superseded attempt must not install a sink. + if attempt != self._attempt: + cast = result.get("cast") + if cast is not None: + try: + cast.disconnect() + except Exception: # noqa: BLE001 + pass + return + if "error" in result: + self.failed.emit(f"Couldn't connect: {result['error']}") + return + + server = TrackServer() + try: + server.start() + except OSError as e: + self.failed.emit(f"Couldn't start the local media server: {e}") + return + + # Built here, on the GUI thread, so its timers belong to a thread that + # actually runs an event loop. + self._server = server + self._sink = CastSink(result["cast"], server, self) + # Deferred to the next event-loop turn on purpose. Player also listens + # for `unavailable` and swaps back to the local sink, which disconnects + # this signal mid-emission; whichever slot ran second would silently be + # dropped. Letting Player do the fallback first and tidying up after is + # order-independent. + self._sink.unavailable.connect( + lambda reason: QTimer.singleShot( + 0, lambda: self._on_unavailable(reason))) + self._player.set_sink(self._sink) + self._inhibitor.inhibit() + self.connected.emit(self._sink.name) + + # ---- disconnecting ---- + + def disconnect(self, reason: str = ""): + """Drop the session and hand playback back to the local sink.""" + sink, self._sink = self._sink, None + self._server = None + if sink is None: + return + log_control("cast", "disconnect", f"{sink.name} reason={reason or 'user'}") + # Player carries the position across, so playback continues locally + # from the same spot rather than restarting the track. A no-op when + # Player already fell back on its own (the device-vanished path). + self._player.set_sink(None) + self._teardown(sink, reason) + + def _on_unavailable(self, reason: str): + """The device went away on its own. Player has already fallen back by + the time this lands, so this only tidies up and reports.""" + sink, self._sink = self._sink, None + self._server = None + if sink is None: + return + log_control("cast", "lost", reason) + self._teardown(sink, reason) + + def _teardown(self, sink, reason: str): + sink.shutdown() + self._inhibitor.release() + self.disconnected.emit(reason) + + def shutdown(self): + """Quitting. Unlike a user disconnect, playback is not handed back: + stopping first clears the current track, so the swap to the local sink + can't fire a burst of audio out of the speakers on the way out.""" + if self._discovery is not None: + self._discovery.stop() + if self._sink is not None: + self._player.stop() + self.disconnect() + self._inhibitor.release() + + +class _ConnectResult(QObject): + """Marshals a background connect result onto the GUI thread.""" + + ready = pyqtSignal(dict, int) + + +def _uuid(text: str): + import uuid + try: + return uuid.UUID(text) + except (ValueError, AttributeError, TypeError): + return text diff --git a/lintunes/cast/discovery.py b/lintunes/cast/discovery.py new file mode 100644 index 0000000..d3158fc --- /dev/null +++ b/lintunes/cast/discovery.py @@ -0,0 +1,160 @@ +"""Finding Chromecasts on the local network. + +pychromecast's CastBrowser does the mDNS work on its own thread; everything +leaves here as a Qt signal, so the picker dialog never touches zeroconf state +directly and nothing touches a widget off the GUI thread. + +Discovery runs only while the picker is open — an idle mDNS browser is chatter +on the network for no benefit. +""" + +from dataclasses import dataclass + +from PyQt6.QtCore import QObject, pyqtSignal + + +@dataclass(frozen=True) +class CastDevice: + uuid: str + name: str + model: str + host: str + port: int + + +def is_available() -> tuple[bool, str]: + """(True, "") when pychromecast can be imported, else (False, why not). + + Uses find_spec rather than a real import: the menu asks this every time it + opens, and there's no reason to drag protobuf and zeroconf into the process + just to decide whether to grey an item out. + """ + import importlib.util + try: + if importlib.util.find_spec("pychromecast") is None: + return False, "pychromecast isn't installed" + except (ImportError, ValueError): + return False, "pychromecast isn't installed" + return True, "" + + +def _to_device(info) -> CastDevice: + """Every pychromecast field access in one place. + + CastInfo's shape has drifted across major versions (host/hostname, services + vs a bare host), so a version bump should be a fix here rather than a hunt + through the dialog. + """ + host, port = "", 8009 + services = getattr(info, "services", None) + if services: + first = next(iter(services)) + host = getattr(first, "host", "") or "" + port = getattr(first, "port", 8009) or 8009 + host = host or getattr(info, "host", "") or "" + return CastDevice( + uuid=str(info.uuid), + name=info.friendly_name or "Chromecast", + model=getattr(info, "model_name", "") or "", + host=host, + port=port, + ) + + +class CastDiscovery(QObject): + """Browses for Chromecasts, reporting them as they appear and disappear.""" + + device_found = pyqtSignal(object) # CastDevice + device_lost = pyqtSignal(object) # CastDevice + failed = pyqtSignal(str) + + def __init__(self, parent=None): + super().__init__(parent) + self._browser = None + self._zconf = None + self._devices: dict[str, CastDevice] = {} + + def is_running(self) -> bool: + return self._browser is not None + + def devices(self) -> list[CastDevice]: + return list(self._devices.values()) + + def browser(self): + """The live CastBrowser, needed to build a Chromecast from a pick.""" + return self._browser + + def zconf(self): + return self._zconf + + def start(self): + if self._browser is not None: + return + # Imported here, not at module scope: lintunes must still launch on a + # machine that has pulled this code but not yet run `pip install -e .`. + try: + import pychromecast + import zeroconf + except ImportError as e: + self.failed.emit(f"Chromecast support isn't installed ({e}).") + return + + outer = self + + class _Listener(pychromecast.discovery.AbstractCastListener): + """Callbacks arrive on the browser's thread; they only ever emit.""" + + def add_cast(self, uuid, service): + outer._changed(uuid, added=True) + + def remove_cast(self, uuid, service, cast_info): + outer._removed(uuid) + + def update_cast(self, uuid, service): + outer._changed(uuid, added=False) + + try: + self._zconf = zeroconf.Zeroconf() + self._browser = pychromecast.CastBrowser(_Listener(), self._zconf) + self._browser.start_discovery() + except Exception as e: # noqa: BLE001 — report, never crash the dialog + self.stop() + self.failed.emit(f"Couldn't search for Chromecasts: {e}") + + def stop(self): + """Idempotent: safe to call from every dialog exit path.""" + browser, self._browser = self._browser, None + zconf, self._zconf = self._zconf, None + if browser is not None: + try: + browser.stop_discovery() + except Exception: # noqa: BLE001 — teardown is best-effort + pass + if zconf is not None: + try: + zconf.close() + except Exception: # noqa: BLE001 + pass + self._devices.clear() + + # ---- browser thread ---- + + def _changed(self, uuid, added: bool): + browser = self._browser + if browser is None: + return + info = browser.devices.get(uuid) + if info is None: + return + device = _to_device(info) + known = self._devices.get(device.uuid) + self._devices[device.uuid] = device + # An update for a device already listed would otherwise add a duplicate + # row every time zeroconf refreshes its record. + if known is None or added: + self.device_found.emit(device) + + def _removed(self, uuid): + device = self._devices.pop(str(uuid), None) + if device is not None: + self.device_lost.emit(device) diff --git a/lintunes/cast/server.py b/lintunes/cast/server.py new file mode 100644 index 0000000..6001172 --- /dev/null +++ b/lintunes/cast/server.py @@ -0,0 +1,182 @@ +"""A minimal local HTTP server that hands single files to the Chromecast. + +The device fetches audio over the LAN, so it needs a URL rather than a path. +This is the only server code in lintunes, so the exposure is kept as small as +it can be: + +* It binds an ephemeral port and lives only as long as the cast session. +* The URL never carries a path. ``publish()`` mints an opaque random token and + keeps the ``token -> absolute path`` mapping in memory; an unknown token is a + 404. Path traversal isn't defended against, it's structurally impossible — + there is no path in the request to traverse with. +* Only tracks actually being cast are published, and only the few most recent + stay resolvable. + +Range requests are mandatory, not an optimization: the Chromecast seeks by +re-requesting byte ranges, and won't report a duration without ``Accept-Ranges``. +""" + +import secrets +import threading +from http import HTTPStatus +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from pathlib import Path + +from lintunes.cast.support import parse_range + +# How many previously published tracks stay resolvable. More than one because +# the device re-requests the current file when it seeks and can briefly reach +# back for the previous one across a track change; small because every live +# token is a file this machine will hand to anything on the LAN that asks. +KEEP_TOKENS = 3 + +CHUNK = 64 * 1024 + + +class _Handler(BaseHTTPRequestHandler): + protocol_version = "HTTP/1.1" # Chromecast wants keep-alive + Content-Length + + # BaseHTTPRequestHandler logs every request to stderr; the app has a status + # bar for anything worth saying. + def log_message(self, format, *args): + pass + + @property + def _tracks(self): + return self.server.track_server + + def _resolve(self) -> Path | None: + prefix = "/t/" + if not self.path.startswith(prefix): + return None + return self._tracks.lookup(self.path[len(prefix):]) + + def do_HEAD(self): + self._serve(body=False) + + def do_GET(self): + self._serve(body=True) + + def _serve(self, body: bool): + path = self._resolve() + if path is None: + self.send_error(HTTPStatus.NOT_FOUND) + return + try: + size = path.stat().st_size + handle = open(path, "rb") + except OSError: + # The file moved or the drive went away since it was published. + self.send_error(HTTPStatus.NOT_FOUND) + return + + with handle: + span = parse_range(self.headers.get("Range"), size) + if span is None: + start, end = 0, max(size - 1, 0) + status = HTTPStatus.OK + else: + start, end = span + status = HTTPStatus.PARTIAL_CONTENT + length = end - start + 1 if size else 0 + + self.send_response(status) + self.send_header("Content-Type", self._tracks.content_type(path)) + self.send_header("Accept-Ranges", "bytes") + self.send_header("Content-Length", str(length)) + if status == HTTPStatus.PARTIAL_CONTENT: + self.send_header("Content-Range", f"bytes {start}-{end}/{size}") + self.end_headers() + if not body or not length: + return + + handle.seek(start) + remaining = length + try: + while remaining > 0: + chunk = handle.read(min(CHUNK, remaining)) + if not chunk: + break + self.wfile.write(chunk) + remaining -= len(chunk) + except (BrokenPipeError, ConnectionResetError): + # Normal: the device drops the connection the moment it seeks + # or the track is replaced. Nothing to report. + pass + + +class TrackServer: + """Serves published files over the LAN for the duration of a cast session. + + Thread-safe: ``publish`` is called from the GUI thread while the server's + own threads read the token map. + """ + + def __init__(self): + self._httpd: ThreadingHTTPServer | None = None + self._thread: threading.Thread | None = None + self._lock = threading.Lock() + self._tokens: dict[str, Path] = {} + self._types: dict[Path, str] = {} + self._order: list[str] = [] + + # ---- lifecycle ---- + + def start(self) -> int: + """Bind an ephemeral port on every interface and serve. Returns the port.""" + if self._httpd is not None: + return self.port + # 0.0.0.0, not localhost: the Chromecast fetches from another machine. + httpd = ThreadingHTTPServer(("0.0.0.0", 0), _Handler) + httpd.daemon_threads = True + httpd.track_server = self + self._httpd = httpd + self._thread = threading.Thread(target=httpd.serve_forever, daemon=True) + self._thread.start() + return self.port + + @property + def port(self) -> int: + return self._httpd.server_address[1] if self._httpd else 0 + + def stop(self): + """Shut the listener down. Idempotent.""" + httpd, self._httpd = self._httpd, None + if httpd is not None: + httpd.shutdown() + httpd.server_close() + self._thread = None + with self._lock: + self._tokens.clear() + self._types.clear() + self._order.clear() + + # ---- publishing ---- + + def publish(self, path, content_type: str) -> str: + """Make `path` fetchable and return its token.""" + path = Path(path) + with self._lock: + token = secrets.token_urlsafe(16) + self._tokens[token] = path + self._types[path] = content_type + self._order.append(token) + # Drop the oldest so a long listening session doesn't leave the + # whole played history reachable. + while len(self._order) > KEEP_TOKENS: + stale = self._order.pop(0) + dropped = self._tokens.pop(stale, None) + if dropped is not None and dropped not in self._tokens.values(): + self._types.pop(dropped, None) + return token + + def lookup(self, token: str) -> Path | None: + with self._lock: + return self._tokens.get(token) + + def content_type(self, path: Path) -> str: + with self._lock: + return self._types.get(path, "application/octet-stream") + + def url_for(self, host: str, token: str) -> str: + return f"http://{host}:{self.port}/t/{token}" diff --git a/lintunes/cast/sink.py b/lintunes/cast/sink.py new file mode 100644 index 0000000..5ab7fff --- /dev/null +++ b/lintunes/cast/sink.py @@ -0,0 +1,408 @@ +"""The Player sink that plays through a Chromecast. + +lintunes hands the device a URL served by the local TrackServer and then acts +as a remote control: there is no local decode, so position and state are polled +from the device rather than read from Qt. + +Two threading rules hold this together: + +* pychromecast delivers its listener callbacks on its own socket worker + thread. Every callback here does nothing but ``emit`` a private signal, + which Qt delivers queued onto the GUI thread. No timer, no Player state and + no widget is ever touched from that thread. +* Every outbound control call is a blocking socket write. They go through a + single serialized worker thread, so a device that has vanished stalls its own + queue instead of freezing the UI — which is exactly the case the feature has + to survive. +""" + +import queue +import threading +import time + +from PyQt6.QtCore import QTimer, pyqtSignal + +from lintunes.cast.support import content_type_for, local_ip_for, uncastable_reason +from lintunes.player import PlaybackSink + +# The stock receiver lintunes launches, and the idle "backdrop" app. Anything +# else running means someone cast to the device from another app. +DEFAULT_RECEIVER_APP_ID = "CC1AD845" +BACKDROP_APP_ID = "E8C28D3C" + +PLAYING_STATES = ("PLAYING", "BUFFERING") + + +# ---- pure status reading (no pychromecast needed; unit-tested) ---- + +def is_natural_end(status) -> bool: + """True when the device finished the track by itself. + + Distinguished from every other way of going idle: ERROR means the load + failed, INTERRUPTED/CANCELLED mean something replaced our media, and none + of those should count a play or advance the queue. + """ + return (getattr(status, "player_state", None) == "IDLE" + and getattr(status, "idle_reason", None) == "FINISHED") + + +def is_load_error(status) -> bool: + return (getattr(status, "player_state", None) == "IDLE" + and getattr(status, "idle_reason", None) == "ERROR") + + +def is_playing_state(status) -> bool: + """BUFFERING counts as playing so the play button doesn't flicker to + "paused" every time the device re-fills its buffer.""" + return getattr(status, "player_state", None) in PLAYING_STATES + + +def position_ms_from(status, fallback_ms: int) -> int: + """Where the device is, in ms. + + ``adjusted_current_time`` extrapolates from the last status using the wall + clock, which is what keeps the seek slider smooth between the device's + infrequent pushes — but it keeps creeping while paused, so it is only + trusted while actually playing. + """ + if status is None: + return fallback_ms + if is_playing_state(status): + value = getattr(status, "adjusted_current_time", None) + else: + value = getattr(status, "current_time", None) + if value is None: + value = getattr(status, "current_time", None) + return int(value * 1000) if value is not None else fallback_ms + + +def duration_ms_from(status, fallback_ms: int) -> int: + value = getattr(status, "duration", None) if status is not None else None + return int(value * 1000) if value else fallback_ms + + +def app_stolen(app_id) -> bool: + """True when a foreign receiver is running — another app took the device.""" + return bool(app_id) and app_id not in (DEFAULT_RECEIVER_APP_ID, + BACKDROP_APP_ID) + + +def connection_verdict(status_name: str) -> str: + """"ok" | "grace" | "lost" for a pychromecast connection status. + + LOST is only "grace": pychromecast retries a dropped socket on its own + every few seconds, so a brief Wi-Fi blip heals itself and dropping straight + back to the laptop speakers would be far more disruptive than waiting. + """ + if status_name == "CONNECTED": + return "ok" + if status_name in ("LOST", "CONNECTING"): + return "grace" + if status_name in ("DISCONNECTED", "FAILED", "FAILED_RESOLVE"): + return "lost" + return "ok" + + +class CastSink(PlaybackSink): + """Plays through a Chromecast by serving it the original file.""" + + provides_pcm = False + + POLL_MS = 500 # how often we ask the device where it is + LOST_GRACE_MS = 15_000 # how long a dropped socket may heal itself + SEEK_SETTLE_MS = 2_000 # trust our own seek over stale status this long + VOLUME_COALESCE_MS = 120 # a slider drag is ~60 events/s; one write is enough + + # Private relays: emitted on pychromecast's thread, handled on the GUI's. + _status_arrived = pyqtSignal(object) + _connection_arrived = pyqtSignal(str) + _app_arrived = pyqtSignal(object) + + def __init__(self, cast, server, parent=None): + super().__init__(parent) + self._cast = cast + self._mc = cast.media_controller + self._server = server + self.name = getattr(cast.cast_info, "friendly_name", "") or "Chromecast" + self._host = getattr(cast.cast_info, "host", "") or "" + if not self._host: + services = getattr(cast.cast_info, "services", None) + if services: + self._host = getattr(next(iter(services)), "host", "") or "" + + self._position_ms = 0 + self._duration_ms = 0 + self._playing = False + # What lintunes last *asked* for, as opposed to what the device last + # reported. They diverge exactly when the connection dies — see _fail. + self._intent_playing = False + self._dying = False + self._optimistic_until = 0.0 + self._optimistic_ms = 0 + self._last_finished_session = None + self._pending_volume = 1.0 + self._shutdown_done = False + # Restored on disconnect: cast.set_volume changes the device's own + # system volume, which would otherwise persist for whatever plays next. + self._device_volume_at_connect = getattr( + getattr(cast, "status", None), "volume_level", None) + + # Serialized outbound control calls (see the module docstring). + self._outbox: queue.Queue = queue.Queue() + self._sender = threading.Thread(target=self._send_loop, daemon=True) + self._sender.start() + + self._status_arrived.connect(self._on_status) + self._connection_arrived.connect(self._on_connection) + self._app_arrived.connect(self._on_app) + + self._poll = QTimer(self) + self._poll.setInterval(self.POLL_MS) + self._poll.timeout.connect(self._on_poll) + self._poll.start() + + self._lost_timer = QTimer(self) + self._lost_timer.setSingleShot(True) + self._lost_timer.setInterval(self.LOST_GRACE_MS) + self._lost_timer.timeout.connect(self._on_grace_expired) + + self._volume_timer = QTimer(self) + self._volume_timer.setSingleShot(True) + self._volume_timer.setInterval(self.VOLUME_COALESCE_MS) + self._volume_timer.timeout.connect(self._flush_volume) + + self._mc.register_status_listener(self) + cast.register_status_listener(self) + cast.register_connection_listener(self) + + # ---- PlaybackSink ---- + + def can_play(self, track) -> bool: + return content_type_for(track.location, track.kind) is not None + + def unplayable_reason(self, track) -> str: + return uncastable_reason(track.location, track.kind) + + def load(self, track, autoplay: bool, start_ms: int): + content_type = content_type_for(track.location, track.kind) + if content_type is None: # Player gates on can_play; belt and braces + self.error.emit(self.unplayable_reason(track)) + return + # Re-resolved per track so a network change (docking, VPN, new lease) + # is picked up rather than baking in a stale address for the session. + try: + host = local_ip_for(self._host) if self._host else "" + except OSError as e: + self._fail(f"Lost the network route to {self.name} ({e}).") + return + token = self._server.publish(track.location, content_type) + url = self._server.url_for(host, token) + + # Report length and state straight away; the device won't say anything + # for a second or two and the transport shouldn't sit blank until then. + self._set_optimistic(start_ms) + self._duration_ms = track.total_time or 0 + self.duration_changed.emit(self._duration_ms) + self._intent_playing = autoplay + self._set_playing(autoplay) + + metadata = { + "metadataType": 3, # MusicTrackMediaMetadata + "title": track.name or "", + "artist": track.artist or "", + "albumName": track.album or "", + } + self._submit( + lambda: self._mc.play_media( + url, content_type, + title=track.name or "", + # NOT pychromecast's STREAM_TYPE_LIVE default: LIVE tells the + # receiver the stream is unbounded, which kills both seeking + # and the duration readout. + stream_type="BUFFERED", + autoplay=autoplay, + current_time=max(0, start_ms) / 1000.0, + metadata=metadata, + )) + + def play(self): + self._intent_playing = True + self._set_playing(True) + self._submit(self._mc.play) + + def pause(self): + self._intent_playing = False + self._set_playing(False) + self._submit(self._mc.pause) + + def stop(self): + self._intent_playing = False + self._set_playing(False) + self._submit(self._mc.stop) + + def seek(self, position_ms: int): + target = max(0, position_ms) + self._set_optimistic(target) + self._submit(lambda: self._mc.seek(target / 1000.0)) + + def position_ms(self) -> int: + return self._position_ms + + def duration_ms(self) -> int: + return self._duration_ms + + def is_playing(self) -> bool: + return self._playing + + def set_volume(self, level: float): + self._pending_volume = max(0.0, min(1.0, float(level))) + self._volume_timer.start() + + def shutdown(self): + """Stop the device, hand its volume back, and drop the session. + + Every step is best-effort and independent: this runs while quitting and + while the device is already gone, and a failure in one step must not + leave the rest undone. Skipping it entirely would leave the Chromecast + buffering from an HTTP server that is about to die. + """ + if self._shutdown_done: + return + self._shutdown_done = True + self._poll.stop() + self._lost_timer.stop() + self._volume_timer.stop() + for step in ( + lambda: (self._device_volume_at_connect is not None + and self._cast.set_volume(self._device_volume_at_connect)), + self._mc.stop, + self._cast.quit_app, + self._cast.disconnect, + ): + try: + step() + except Exception: # noqa: BLE001 — teardown is best-effort + pass + self._outbox.put(None) # retire the sender thread + self._server.stop() + + # ---- outbound calls, serialized off the GUI thread ---- + + def _submit(self, call): + if not self._shutdown_done: + self._outbox.put(call) + + def _send_loop(self): + while True: + call = self._outbox.get() + if call is None: + return + try: + call() + except Exception as e: # noqa: BLE001 + if not self._shutdown_done: + self._fail(f"Lost {self.name} ({e}).") + return + + # ---- pychromecast listeners (foreign thread: emit only) ---- + + def new_media_status(self, status): + self._status_arrived.emit(status) + + def load_media_failed(self, item, error_code): + self._status_arrived.emit(None) + + def new_cast_status(self, status): + self._app_arrived.emit(getattr(status, "app_id", None)) + + def new_connection_status(self, status): + self._connection_arrived.emit(getattr(status, "status", "") or "") + + # ---- GUI thread ---- + + def _on_status(self, status): + if status is None or self._shutdown_done or self._dying: + return + if is_load_error(status): + self.error.emit("the Chromecast couldn't play it") + return + if is_natural_end(status): + session = getattr(status, "media_session_id", None) + # The device repeats its final status; without this the track would + # be counted, scrobbled and advanced more than once. + if session is not None and session == self._last_finished_session: + return + self._last_finished_session = session + self._set_playing(False) + self.ended.emit() + return + self._set_playing(is_playing_state(status)) + + def _on_poll(self): + if self._shutdown_done or self._dying: + return + status = getattr(self._mc, "status", None) + if time.monotonic() * 1000 < self._optimistic_until: + # Hold our own seek target briefly: the device keeps reporting the + # old position for a beat, which snaps the slider back mid-scrub. + position = self._optimistic_ms + else: + position = position_ms_from(status, self._position_ms) + self._position_ms = position + self.position_changed.emit(position) + + duration = duration_ms_from(status, self._duration_ms) + if duration != self._duration_ms: + self._duration_ms = duration + self.duration_changed.emit(duration) + + def _on_connection(self, status_name: str): + verdict = connection_verdict(status_name) + if verdict == "ok": + self._lost_timer.stop() + elif verdict == "grace": + if not self._lost_timer.isActive(): + self._lost_timer.start() + else: + self._lost_timer.stop() + self._fail(f"Lost the connection to {self.name}.") + + def _on_grace_expired(self): + self._fail(f"{self.name} stopped responding.") + + def _on_app(self, app_id): + if app_stolen(app_id): + self._fail(f"Something else started casting to {self.name}.") + + def _flush_volume(self): + level = self._pending_volume + self._submit(lambda: self._cast.set_volume(level)) + + # ---- internals ---- + + def _fail(self, reason: str): + """Give up on the device and let Player fall back to local playback. + + Reports the state lintunes last *asked* for rather than whatever the + socket last said. A dying connection pushes an IDLE media status just + before the disconnect, so reading the device's own view here would make + the music stop on the fallback instead of continuing on the speakers. + """ + if self._dying: + return + self._dying = True + self._poll.stop() + self._lost_timer.stop() + self._playing = self._intent_playing + self.unavailable.emit(reason) + + def _set_playing(self, playing: bool): + if playing != self._playing: + self._playing = playing + self.state_changed.emit(playing) + + def _set_optimistic(self, position_ms: int): + self._optimistic_ms = max(0, position_ms) + self._position_ms = self._optimistic_ms + self._optimistic_until = time.monotonic() * 1000 + self.SEEK_SETTLE_MS + self.position_changed.emit(self._position_ms) diff --git a/lintunes/cast/support.py b/lintunes/cast/support.py new file mode 100644 index 0000000..0003f6c --- /dev/null +++ b/lintunes/cast/support.py @@ -0,0 +1,122 @@ +"""Pure helpers for casting: format support, range parsing, source address. + +Deliberately free of pychromecast and Qt so the fiddly parts (what the device +will actually accept, byte-range arithmetic) are unit-testable offline. +""" + +import socket +from pathlib import Path + +# The Chromecast's audio codec support, keyed by container. Anything absent is +# refused rather than handed over to fail on the device: the receiver stalls +# silently on an unsupported stream, which reads as a bug. +# +# Notably missing: ALAC and AIFF (never supported), and DRM'd .m4p purchases. +_BY_SUFFIX = { + ".mp3": "audio/mpeg", + ".mpga": "audio/mpeg", + ".m4a": "audio/mp4", + ".aac": "audio/mp4", + ".wav": "audio/wav", + ".flac": "audio/flac", + ".ogg": "audio/ogg", + ".opus": "audio/ogg", +} + +# iTunes ``kind`` strings that live in an .m4a container but hold something the +# device can't decode, so the suffix alone isn't enough to say yes. Note that +# "Purchased AAC audio file" is *not* here: those are DRM-free iTunes Plus +# downloads and cast fine. Only "Protected AAC" is FairPlay-encrypted. +_REFUSED_KINDS = { + "Apple Lossless audio file": "Apple Lossless", + "Protected AAC audio file": "copy-protected AAC", +} + +# Human names for the containers the device can't decode, for the skip message. +_REFUSED_SUFFIXES = { + ".aif": "AIFF", + ".aiff": "AIFF", + ".m4p": "copy-protected AAC", +} + + +def content_type_for(location: str, kind: str = "") -> str | None: + """The MIME type to advertise for `location`, or None if it can't be cast. + + `kind` is the iTunes description carried on the track; it disambiguates the + .m4a container, which holds both plain AAC (fine) and Apple Lossless or + FairPlay-protected audio (not fine). + """ + if not location: + return None + if kind in _REFUSED_KINDS: + return None + return _BY_SUFFIX.get(Path(location).suffix.lower()) + + +def uncastable_reason(location: str, kind: str = "") -> str: + """Why `location` can't be cast, phrased for the status bar. + + Only meaningful when `content_type_for` returned None. + """ + if not location: + return "it has no file" + if kind in _REFUSED_KINDS: + return f"a Chromecast can't play {_REFUSED_KINDS[kind]}" + suffix = Path(location).suffix.lower() + if suffix in _REFUSED_SUFFIXES: + return f"a Chromecast can't play {_REFUSED_SUFFIXES[suffix]}" + return f"a Chromecast can't play {suffix.lstrip('.').upper() or 'this format'}" + + +def local_ip_for(peer_ip: str, port: int = 8009) -> str: + """This machine's address *as the Chromecast will see it*. + + Resolving the local hostname is wrong here: trav's machines run Tailscale, + so the first address found is often a 100.x tailnet address the Chromecast + can't reach. Opening a UDP socket toward the device asks the routing table + which source address actually reaches it — no packet is sent, since UDP + connect() only fixes the peer. Re-run for every load so a network change + (docking, VPN up/down, new DHCP lease) is picked up on the next track. + """ + sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + try: + sock.connect((peer_ip, port)) + return sock.getsockname()[0] + finally: + sock.close() + + +def parse_range(header: str | None, size: int) -> tuple[int, int] | None: + """Interpret a Range header against a file of `size` bytes. + + Returns an inclusive (start, end) pair, or None to serve the whole file — + which is also the answer for anything malformed or unsatisfiable, since a + 200 with the full body is always a valid response to a Range request and + keeps playback working rather than erroring the device out. + + Only single byte ranges are handled; that is all the Chromecast sends. + """ + if not header or size <= 0: + return None + header = header.strip() + if not header.startswith("bytes=") or "," in header: + return None + spec = header[len("bytes="):].strip() + first, _, last = spec.partition("-") + if not _: + return None + try: + if not first: + # "-500" = the final 500 bytes. + length = int(last) + if length <= 0: + return None + return max(0, size - length), size - 1 + start = int(first) + end = int(last) if last else size - 1 + except ValueError: + return None + if start < 0 or start >= size or end < start: + return None + return start, min(end, size - 1) diff --git a/lintunes/gui/cast_dialog.py b/lintunes/gui/cast_dialog.py new file mode 100644 index 0000000..c58c94d --- /dev/null +++ b/lintunes/gui/cast_dialog.py @@ -0,0 +1,129 @@ +"""The "Connect to Chromecast…" picker. + +Searching starts when the dialog opens and stops when it closes, whichever way +it closes. Connecting is not awaited here — the dialog's job ends at "which +device", and CastController reports the outcome on the status bar. +""" + +from PyQt6.QtCore import Qt, QTimer +from PyQt6.QtWidgets import ( + QDialog, QDialogButtonBox, QHBoxLayout, QLabel, QListWidget, + QListWidgetItem, QVBoxLayout) + +from lintunes.gui.spinner import Spinner + +# How long to look before admitting there's nothing there. Searching carries on +# regardless — a speaker that wakes up later still shows up. +EMPTY_AFTER_MS = 12_000 + + +class ChromecastDialog(QDialog): + """Lists Chromecasts as they're discovered; returns the one picked.""" + + def __init__(self, discovery, parent=None): + super().__init__(parent) + self._discovery = discovery + self._selected = None + + self.setWindowTitle("Connect to Chromecast") + self.setMinimumWidth(340) + layout = QVBoxLayout(self) + + header = QHBoxLayout() + self._spinner = Spinner() + header.addWidget(self._spinner) + self._status = QLabel("Searching for Chromecasts…") + header.addWidget(self._status, stretch=1) + layout.addLayout(header) + + self._list = QListWidget() + self._list.setAlternatingRowColors(True) + self._list.itemSelectionChanged.connect(self._on_selection_changed) + self._list.itemDoubleClicked.connect(lambda _: self._on_connect()) + layout.addWidget(self._list, stretch=1) + + self._buttons = QDialogButtonBox(QDialogButtonBox.StandardButton.Cancel) + self._connect_button = self._buttons.addButton( + "Connect", QDialogButtonBox.ButtonRole.AcceptRole) + self._connect_button.setEnabled(False) + self._connect_button.setDefault(True) + self._buttons.rejected.connect(self.reject) + self._connect_button.clicked.connect(self._on_connect) + layout.addWidget(self._buttons) + + discovery.device_found.connect(self._on_found) + discovery.device_lost.connect(self._on_lost) + discovery.failed.connect(self._on_failed) + + self._empty_timer = QTimer(self) + self._empty_timer.setSingleShot(True) + self._empty_timer.setInterval(EMPTY_AFTER_MS) + self._empty_timer.timeout.connect(self._on_empty_timeout) + self._empty_timer.start() + + # Anything discovery already knows about (it may still be running from + # a previous open) should be on screen before the first new callback. + for device in discovery.devices(): + self._on_found(device) + discovery.start() + + @property + def selected_device(self): + return self._selected + + # ---- discovery ---- + + def _on_found(self, device): + for row in range(self._list.count()): + existing = self._list.item(row).data(Qt.ItemDataRole.UserRole) + if existing.uuid == device.uuid: + return + label = device.name + if device.model: + label = f"{device.name} · {device.model}" + item = QListWidgetItem(label) + item.setData(Qt.ItemDataRole.UserRole, device) + self._list.addItem(item) + self._status.setText("Select a Chromecast to connect to.") + if self._list.count() == 1: + self._list.setCurrentRow(0) + + def _on_lost(self, device): + for row in range(self._list.count()): + if self._list.item(row).data( + Qt.ItemDataRole.UserRole).uuid == device.uuid: + self._list.takeItem(row) + break + if self._list.count() == 0: + self._status.setText("Searching for Chromecasts…") + + def _on_failed(self, message: str): + self._spinner.hide() + self._status.setText(message) + self._empty_timer.stop() + + def _on_empty_timeout(self): + if self._list.count() == 0: + # The spinner keeps turning: discovery is still running, and a + # device that comes online in a minute will still appear. + self._status.setText("No Chromecasts found on this network yet.") + + # ---- selection ---- + + def _on_selection_changed(self): + self._connect_button.setEnabled(self._list.currentItem() is not None) + + def _on_connect(self): + item = self._list.currentItem() + if item is None: + return + self._selected = item.data(Qt.ItemDataRole.UserRole) + self.accept() + + def done(self, result): + # The single exit funnel — accept, reject, Esc and the window's close + # button all land here, so discovery can't be left running. + self._empty_timer.stop() + self._spinner.stop() + self._discovery.stop() + super().done(result) diff --git a/lintunes/gui/cast_indicator.py b/lintunes/gui/cast_indicator.py new file mode 100644 index 0000000..76b3389 --- /dev/null +++ b/lintunes/gui/cast_indicator.py @@ -0,0 +1,55 @@ +"""The small cast glyph under the volume slider. Click to disconnect.""" + +from PyQt6.QtCore import QSize, Qt +from PyQt6.QtGui import QIcon +from PyQt6.QtWidgets import QToolButton + +from lintunes.gui.icons import transport_icon + +HEIGHT = 18 + + +class CastIndicator(QToolButton): + """Shows what playback is coming out of, when it isn't this machine. + + Never hidden, even when nothing is connected. `setVisible(False)` would + take it out of the layout's height budget, which lifts the volume slider + the instant you connect and drops it again when you disconnect — so idle + is a visible widget with an empty icon, holding its space. + """ + + def __init__(self, parent=None): + super().__init__(parent) + self._connected = False + self._device = "" + self.setAutoRaise(True) + self.setFocusPolicy(Qt.FocusPolicy.NoFocus) + self.setFixedHeight(HEIGHT) + self.setIconSize(QSize(HEIGHT, HEIGHT)) + self._apply() + + def set_connected(self, connected: bool, device_name: str = ""): + self._connected = bool(connected) + self._device = device_name or "" + self._apply() + + def refresh_theme(self): + self._apply() + + def sizeHint(self) -> QSize: + return QSize(HEIGHT + 6, HEIGHT) + + def _apply(self): + if self._connected: + self.setIcon(transport_icon( + "cast_connected", self.palette().highlight().color())) + name = self._device or "a Chromecast" + self.setToolTip(f"Casting to {name} — click to stop") + self.setCursor(Qt.CursorShape.PointingHandCursor) + self.setEnabled(True) + else: + # Empty icon rather than a hidden widget: holds the slot open. + self.setIcon(QIcon()) + self.setToolTip("") + self.setCursor(Qt.CursorShape.ArrowCursor) + self.setEnabled(False) diff --git a/lintunes/gui/icons.py b/lintunes/gui/icons.py index 9dfe717..47016fe 100644 --- a/lintunes/gui/icons.py +++ b/lintunes/gui/icons.py @@ -69,10 +69,48 @@ def _shuffle(p, color): p.drawPolygon(QPolygonF([QPointF(15, 11), QPointF(19, 14), QPointF(15, 17)])) +def _cast_outline(p, color, filled: bool): + """The standard cast glyph: a screen with signal arcs at its lower left.""" + pen = QPen(color, 1.6) + pen.setCapStyle(Qt.PenCapStyle.RoundCap) + p.setPen(pen) + p.setBrush(Qt.BrushStyle.NoBrush) + # Screen, its lower-left corner left open for the arcs. + p.drawPolyline(QPolygonF([ + QPointF(6.5, 15.5), QPointF(17.5, 15.5), QPointF(17.5, 4.5), + QPointF(3.5, 4.5), QPointF(3.5, 7.0), + ])) + if filled: + # "Receiving": the screen is lit rather than empty. + p.setPen(Qt.PenStyle.NoPen) + p.setBrush(color) + p.drawRect(QRectF(8.5, 8.0, 7.2, 5.8)) + p.setPen(pen) + p.setBrush(Qt.BrushStyle.NoBrush) + # Two arcs radiating from the corner (16ths of a degree, 0..90). + for radius in (4.5, 8.0): + p.drawArc(QRectF(3.5 - radius, 15.5 - radius, radius * 2, radius * 2), + 0, 90 * 16) + # The corner dot the arcs radiate from. + p.setPen(Qt.PenStyle.NoPen) + p.setBrush(color) + p.drawEllipse(QPointF(3.8, 15.2), 1.5, 1.5) + + +def _cast(p, color): + _cast_outline(p, color, filled=False) + + +def _cast_connected(p, color): + _cast_outline(p, color, filled=True) + + _DRAWERS = { "play": _play, "pause": _pause, "next": _next, "previous": _previous, "shuffle": _shuffle, + "cast": _cast, + "cast_connected": _cast_connected, } diff --git a/lintunes/gui/main_window.py b/lintunes/gui/main_window.py index 1814fba..7ecf231 100644 --- a/lintunes/gui/main_window.py +++ b/lintunes/gui/main_window.py @@ -11,11 +11,13 @@ from PyQt6.QtGui import QAction, QKeySequence from lintunes import device_sync, mpris, tagging, theme from lintunes.art_search import AlbumArtFetcher +from lintunes.cast.controller import CastController from lintunes.eventlog import log_control from lintunes.inhibit import INHIBIT_LOGOUT, INHIBIT_SUSPEND, SleepInhibitor from lintunes.player import Player from lintunes.importers import file_importer from lintunes.gui.album_art_dialog import AlbumArtDialog +from lintunes.gui.cast_dialog import ChromecastDialog from lintunes.gui.sidebar import SidebarPanel from lintunes.gui.library_view import LibraryView from lintunes.gui.playlist_view import PlaylistView @@ -71,6 +73,9 @@ class MainWindow(QMainWindow): self._lastfm = lastfm self._prefs_dialog = None self.player = Player(manager, self) + # Owns the cast session, if any. Built before the transport bar, which + # takes it to drive the little cast indicator under the volume slider. + self._cast = CastController(self.player, self) # Keep the machine awake while audio is actually playing. self._inhibitor = SleepInhibitor() # Context playback started from: "library" or "playlist:". Scopes @@ -90,7 +95,8 @@ class MainWindow(QMainWindow): layout.setSpacing(0) self.setCentralWidget(central) - self._transport = TransportBar(self.player, manager, self._prefs) + self._transport = TransportBar(self.player, manager, self._prefs, + cast=self._cast) layout.addWidget(self._transport) splitter = QSplitter(Qt.Orientation.Horizontal) @@ -179,6 +185,12 @@ class MainWindow(QMainWindow): manager.file_move_failed.connect( lambda name, err: self.statusBar().showMessage( f"Couldn't move the file for “{name}”: {err}", 8000)) + self.player.track_unplayable.connect(self._on_track_unplayable) + self._cast.connecting.connect(self._on_cast_connecting) + self._cast.connected.connect(self._on_cast_connected) + self._cast.disconnected.connect(self._on_cast_disconnected) + self._cast.failed.connect( + lambda msg: QMessageBox.warning(self, "Chromecast", msg)) manager.library_reloaded.connect(self._on_library_reloaded) manager.conflict_resolved.connect(self._on_conflict_resolved) prefs.changed.connect(self._on_prefs_changed) @@ -286,14 +298,18 @@ class MainWindow(QMainWindow): self._add_action(track_menu, "Go to Current Song", "Ctrl+L", self._go_to_current_song) - device_menu = bar.addMenu("&Device") + # Mnemonic on the "n": Alt+C already belongs to &Controls. + connections_menu = bar.addMenu("Co&nnections") self._sync_action = self._add_action( - device_menu, "Sync Playlist to Rabbit", "", + connections_menu, "Sync Playlist to Rabbit", "", self._sync_playlist_to_device) self._sync_action.setEnabled(False) + connections_menu.addSeparator() + self._cast_action = self._add_action( + connections_menu, "Connect to Chromecast…", "", self._toggle_cast) # Re-checked every time the menu opens: cheap (one gvfs listdir), and - # always reflects plug/unplug and the current view. - device_menu.aboutToShow.connect(self._refresh_device_actions) + # always reflects plug/unplug, the cast session and the current view. + connections_menu.aboutToShow.connect(self._refresh_connection_actions) def _add_action(self, menu, text, shortcut, slot): action = QAction(text, self) @@ -314,9 +330,9 @@ class MainWindow(QMainWindow): self._redo_action.setText( f"Redo {stack.redo_label()}" if can_redo else "Redo") - # ---- device sync ---- + # ---- connections: device sync ---- - def _refresh_device_actions(self): + def _refresh_connection_actions(self): syncing = self._sync_worker is not None and self._sync_worker.busy() on_playlist = (self._content.currentWidget() is self._playlist_view and bool(self._playlist_view.playlist_id)) @@ -324,6 +340,51 @@ class MainWindow(QMainWindow): not syncing and on_playlist and device_sync.find_device() is not None) + if self._cast.is_connected(): + self._cast_action.setText( + f"Disconnect from {self._cast.device_name()}") + self._cast_action.setEnabled(True) + self._cast_action.setToolTip("") + return + self._cast_action.setText("Connect to Chromecast…") + available, why = self._cast.is_available() + self._cast_action.setEnabled(available) + # The self-updater ships code but not dependencies, so a machine that + # has pulled this round hasn't necessarily installed pychromecast yet. + self._cast_action.setToolTip( + "" if available else f"{why} — run: pip install -e .") + + # ---- connections: casting ---- + + def _toggle_cast(self): + if self._cast.is_connected(): + self._cast.disconnect() + return + available, why = self._cast.is_available() + if not available: + QMessageBox.information( + self, "Chromecast support not installed", + f"{why}.\n\nRun “pip install -e .” in the lintunes checkout, " + "then restart LinTunes.") + return + dialog = ChromecastDialog(self._cast.discovery(), self) + if dialog.exec() and dialog.selected_device is not None: + self._cast.connect_to(dialog.selected_device) + + def _on_cast_connecting(self, device): + self.statusBar().showMessage(f"Connecting to {device.name}…", 8000) + + def _on_cast_connected(self, name: str): + self.statusBar().showMessage(f"Casting to {name}", 6000) + + def _on_cast_disconnected(self, reason: str): + self.statusBar().showMessage( + reason or "Stopped casting — playing here again", 8000) + + def _on_track_unplayable(self, track, reason: str): + self.statusBar().showMessage( + f"Skipped “{track.name}” — {reason}", 6000) + def _sync_playlist_to_device(self): # Everything may have changed since the menu opened — re-verify. if self._sync_worker is not None and self._sync_worker.busy(): @@ -779,6 +840,9 @@ class MainWindow(QMainWindow): return self._inhibitor.release() self._sync_inhibitor.release() + # Before the player: this stops the device and kills the HTTP server it + # is fetching from. Skipped, the Chromecast sits buffering a dead URL. + self._cast.shutdown() self.player.shutdown() self._manager.flush() super().closeEvent(event) diff --git a/lintunes/gui/spinner.py b/lintunes/gui/spinner.py new file mode 100644 index 0000000..e9d5475 --- /dev/null +++ b/lintunes/gui/spinner.py @@ -0,0 +1,67 @@ +"""A small indeterminate spinner. + +Everything visual in lintunes is hand-painted (see gui/icons.py) — there is no +SVG, no .qrc and no QMovie asset anywhere — so this is drawn too. The timer +runs only while the widget is visible, so it costs nothing sitting in a closed +dialog. +""" + +from PyQt6.QtCore import QRectF, Qt, QTimer +from PyQt6.QtGui import QColor, QPainter, QPen +from PyQt6.QtWidgets import QWidget + +PERIOD_MS = 900 # one full turn +FRAME_MS = 40 +ARC_SPAN = 280 # degrees of the circle actually drawn + + +class Spinner(QWidget): + """A rotating arc, in the theme highlight color unless told otherwise.""" + + def __init__(self, diameter: int = 18, parent=None): + super().__init__(parent) + self._angle = 0 + self._color: QColor | None = None + self.setFixedSize(diameter, diameter) + self._timer = QTimer(self) + self._timer.setInterval(FRAME_MS) + self._timer.timeout.connect(self._advance) + + def set_color(self, color: QColor): + self._color = color + self.update() + + def start(self): + if not self._timer.isActive(): + self._timer.start() + + def stop(self): + self._timer.stop() + + def showEvent(self, event): + super().showEvent(event) + self.start() + + def hideEvent(self, event): + super().hideEvent(event) + self.stop() + + def _advance(self): + self._angle = (self._angle + int(360 * FRAME_MS / PERIOD_MS)) % 360 + self.update() + + def paintEvent(self, event): + painter = QPainter(self) + painter.setRenderHint(QPainter.RenderHint.Antialiasing) + color = self._color or self.palette().highlight().color() + pen = QPen(color, 2.0) + pen.setCapStyle(Qt.PenCapStyle.RoundCap) + painter.setPen(pen) + painter.setBrush(Qt.BrushStyle.NoBrush) + inset = 2.0 + box = QRectF(inset, inset, + self.width() - inset * 2, self.height() - inset * 2) + # Qt measures arcs in 16ths of a degree, counter-clockwise from 3 + # o'clock; negate to spin the way every other spinner does. + painter.drawArc(box, -self._angle * 16, -ARC_SPAN * 16) + painter.end() diff --git a/lintunes/gui/transport.py b/lintunes/gui/transport.py index db69883..efdf845 100644 --- a/lintunes/gui/transport.py +++ b/lintunes/gui/transport.py @@ -8,6 +8,7 @@ from PyQt6.QtCore import Qt, QTimer, pyqtSignal from PyQt6.QtGui import QColor, QFont from lintunes.tap_tempo import TapTempo +from lintunes.gui.cast_indicator import CastIndicator from lintunes.gui.icons import transport_icon from lintunes.gui.track_table import format_time from lintunes.gui.visualizer import VisualizerWidget @@ -234,10 +235,11 @@ class TransportBar(QWidget): play_clicked = pyqtSignal() # MainWindow decides what "play" means - def __init__(self, player, manager, prefs, parent=None): + def __init__(self, player, manager, prefs, cast=None, parent=None): super().__init__(parent) self._player = player self._prefs = prefs + self._cast = cast self._scrubbing = False self._current_track = None @@ -280,8 +282,9 @@ class TransportBar(QWidget): layout.addSpacing(10) # Master output volume: a slim horizontal slider in its own slot - # between the visualizer and the timeline. The HBox centers it - # vertically, so nothing stacks above or below it (iTunes-style). + # between the visualizer and the timeline, with the cast indicator + # parked directly beneath it. The column is shorter than the boxes + # beside it, so the HBox still centers it and BAR_HEIGHT is untouched. self._volume_slider = ClickJumpSlider() self._volume_slider.setRange(0, 100) self._volume_slider.setFixedWidth(96) @@ -297,7 +300,15 @@ class TransportBar(QWidget): self._volume_save_timer.setInterval(400) self._volume_save_timer.timeout.connect(self._save_volume) self._volume_slider.valueChanged.connect(self._on_volume_changed) - layout.addWidget(self._volume_slider) + volume_col = QVBoxLayout() + volume_col.setSpacing(2) + volume_col.setContentsMargins(0, 0, 0, 0) + volume_col.addWidget(self._volume_slider, + alignment=Qt.AlignmentFlag.AlignHCenter) + self._cast_indicator = CastIndicator() + volume_col.addWidget(self._cast_indicator, + alignment=Qt.AlignmentFlag.AlignHCenter) + layout.addLayout(volume_col) layout.addSpacing(10) center = QVBoxLayout() @@ -365,6 +376,12 @@ class TransportBar(QWidget): player.playing_changed.connect(self._on_playing_changed) player.position_changed.connect(self._on_position_changed) player.duration_changed.connect(self._on_duration_changed) + if cast is not None: + cast.connected.connect( + lambda name: self._cast_indicator.set_connected(True, name)) + cast.disconnected.connect( + lambda _reason: self._cast_indicator.set_connected(False)) + self._cast_indicator.clicked.connect(lambda: cast.disconnect()) def refresh_theme(self): """Re-apply everything driven by Preferences: button glyphs (highlight @@ -378,6 +395,7 @@ class TransportBar(QWidget): box.setStyleSheet(_box_style(box_bg)) self._apply_now_playing_bg() self._apply_now_playing_font() + self._cast_indicator.refresh_theme() self._visualizer.update() # dim-mode bar color may have changed def _apply_now_playing_bg(self): diff --git a/lintunes/gui/visualizer.py b/lintunes/gui/visualizer.py index 5858cfc..1132b5b 100644 --- a/lintunes/gui/visualizer.py +++ b/lintunes/gui/visualizer.py @@ -12,6 +12,7 @@ from PyQt6.QtGui import QPainter, QColor, QPainterPath, QPen from PyQt6.QtMultimedia import QAudioFormat from lintunes import theme +from lintunes.gui.icons import transport_icon BANDS = 20 @@ -21,6 +22,7 @@ FRAME_MS = 16 # ~60 fps DECAY = 0.82 # slow decay; attack is instant FLOOR_DB = -55.0 RADIUS = 8 # rounded corners, matching the transport button boxes +ICON_SIZE = 20 # the cast glyph shown in place of bars while casting # Visualizer click cycles through these brightness modes in order. MODE_ON = "on" @@ -54,6 +56,9 @@ class VisualizerWidget(QWidget): self._samples = np.zeros(FFT_SIZE, dtype=np.float32) self._sample_rate = 44100 self._window = np.hanning(FFT_SIZE).astype(np.float32) + # True while audio is coming out of something other than this machine + # (a Chromecast), so there is no decoded PCM to draw. + self._no_pcm = False self.setFixedWidth(140) self.setMinimumHeight(36) @@ -66,6 +71,23 @@ class VisualizerWidget(QWidget): player.audio_buffer.connect(self._on_buffer) player.playing_changed.connect(self._on_playing_changed) player.track_changed.connect(self._on_track_changed) + if hasattr(player, "sink_changed"): + player.sink_changed.connect(self._on_sink_changed) + + def _on_sink_changed(self, sink): + """Casting means no local decode, so there is nothing to draw. Say so + with the cast glyph rather than animating a flat line off a buffer of + zeros 30 times a second.""" + self._no_pcm = not getattr(sink, "provides_pcm", True) + if self._no_pcm: + self._timer.stop() + self._clear() + self.setToolTip("No spectrum while casting") + else: + self.setToolTip("Click to cycle: on / dim / off") + if self._mode != MODE_OFF and self._player.is_playing(): + self._timer.start() + self.update() # ---- audio intake ---- @@ -94,6 +116,8 @@ class VisualizerWidget(QWidget): # ---- state ---- def _on_playing_changed(self, playing): + if self._no_pcm: + return # nothing to animate; don't let play/pause restart the timer if playing: if self._mode != MODE_OFF: self._timer.start() @@ -179,6 +203,16 @@ class VisualizerWidget(QWidget): painter.setPen(QPen(self.palette().mid().color(), 1)) painter.drawPath(panel) painter.setClipPath(panel) # keep bars inside the rounded shape + if self._no_pcm: + # Casting: the audio never passes through this machine, so show + # where it went instead of pretending to analyze silence. + glyph = transport_icon("cast_connected", self._bar_color()).pixmap( + ICON_SIZE, ICON_SIZE) + painter.drawPixmap( + (self.width() - ICON_SIZE) // 2, + (self.height() - ICON_SIZE) // 2, glyph) + painter.end() + return inset = 3 # bar field sits 3px in from every edge avail_w = self.width() - 2 * inset avail_h = self.height() - 2 * inset diff --git a/lintunes/player.py b/lintunes/player.py index 484e032..c32f9f5 100644 --- a/lintunes/player.py +++ b/lintunes/player.py @@ -29,40 +29,89 @@ def make_shuffle_order(count: int, start_index: int) -> list[int]: return order -class Player(QObject): - """Playback engine: wraps QMediaPlayer and walks a queue of track ids. +class PlaybackSink(QObject): + """One audio output engine sitting behind Player. - The queue is the displayed order of whatever view playback started from. - Play counts are recorded (via LibraryManager) when a track finishes - playing naturally. + Player owns the queue, the shuffle walk, the play-count/scrobble + bookkeeping and the custom start/stop times. A sink owns only "make this + file come out of something, and tell me where it's up to". + + Everything a sink reports leaves as a Qt signal, so a sink whose events + originate on a foreign thread (the cast sink's pychromecast socket worker) + delivers them queued onto the GUI thread — the art_search/lastfm pattern. """ - track_changed = pyqtSignal(object) # Track or None - playing_changed = pyqtSignal(bool) position_changed = pyqtSignal('qint64') # ms duration_changed = pyqtSignal('qint64') # ms - error_occurred = pyqtSignal(str) - track_missing = pyqtSignal(object) # Track whose file is gone/moved - audio_buffer = pyqtSignal(object) # QAudioBuffer (decoded PCM) - track_finished = pyqtSignal(object) # Track played to the very end + state_changed = pyqtSignal(bool) # is it playing? + ended = pyqtSignal() # reached its natural end + error = pyqtSignal(str) # this track failed; Player skips + audio_buffer = pyqtSignal(object) # decoded PCM tee (local only) + unavailable = pyqtSignal(str) # the sink itself died - def __init__(self, manager, parent=None): + provides_pcm = False # does audio_buffer ever fire? (gates the visualizer) + name = "" # "" for local; the device name while casting + + def load(self, track, autoplay: bool, start_ms: int): + raise NotImplementedError + + def play(self): + raise NotImplementedError + + def pause(self): + raise NotImplementedError + + def stop(self): + raise NotImplementedError + + def seek(self, position_ms: int): + raise NotImplementedError + + def position_ms(self) -> int: + raise NotImplementedError + + def duration_ms(self) -> int: + raise NotImplementedError + + def is_playing(self) -> bool: + raise NotImplementedError + + def set_volume(self, level: float): + """Set output volume from a logical 0..1 value (0=silent, 1=full).""" + raise NotImplementedError + + def can_play(self, track) -> bool: + """False when this sink can't decode the track at all, so Player skips + past it rather than stalling on something that will never start.""" + return True + + def unplayable_reason(self, track) -> str: + return "" + + def shutdown(self): + pass + + +class LocalSink(PlaybackSink): + """The built-in sink: QMediaPlayer + QAudioOutput, with the PCM tee. + + Stays in player.py on purpose — the Player tests stub the Qt Multimedia + classes with ``patch.multiple(player_module, QMediaPlayer=..., ...)``, so + they have to be looked up in this module's namespace. + """ + + provides_pcm = True + + def __init__(self, parent=None): super().__init__(parent) - self._manager = manager - self._queue: list[int] = [] - self._index = -1 - self._current_track: Track | None = None - self._shuffle = False - self._shuffle_order: list[int] = [] - # Custom start/stop times (ms) for the loaded track: seek to start once - # the media is seekable, and end the track early at stop. + self._volume = 1.0 # logical 0..1; what the volume slider shows + # Custom start time (ms) for the loaded track: applied once the media + # reports itself loaded, which is when it is reliably seekable. self._pending_start_ms = 0 - self._stop_at_ms = 0 - self._counted_finish_id: int | None = None self._paused_at: float | None = None # monotonic time of last pause + self._shutdown_done = False self._audio = QAudioOutput(self) - self._volume = 1.0 # logical 0..1; what the volume slider shows self._media = QMediaPlayer(self) self._media.setAudioOutput(self._audio) # Tee of the decoded PCM, feeding the visualizer @@ -77,13 +126,77 @@ class Player(QObject): self._media_devices.audioOutputsChanged.connect( self._on_audio_outputs_changed) self._media.positionChanged.connect(self.position_changed) - self._media.positionChanged.connect(self._on_position) self._media.durationChanged.connect(self.duration_changed) self._media.playbackStateChanged.connect(self._on_state_changed) self._media.mediaStatusChanged.connect(self._on_media_status) self._media.errorOccurred.connect(self._on_error) - self._shutdown_done = False + def load(self, track, autoplay: bool, start_ms: int): + self._pending_start_ms = max(0, start_ms) + url = QUrl.fromLocalFile(track.location) + if self._media.source() == url: + # setSource() no-ops on an unchanged URL, so LoadedMedia never + # re-fires and the armed start time would be skipped (and forcing + # a clear+reload races the FFmpeg backend, which snaps the seek + # back to 0). The media is already loaded: rewind and seek now. + self._media.stop() + if self._pending_start_ms: + self._media.setPosition(self._pending_start_ms) + self._pending_start_ms = 0 + else: + self._media.setSource(url) + if autoplay: + self._media.play() + + def play(self): + self._apply_volume() + if (self._paused_at is not None + and time.monotonic() - self._paused_at + >= RESUME_NUDGE_THRESHOLD_S): + # Re-prime a possibly idle-suspended sink (the manual workaround + # was "rewind slightly"); seek in place so the listener doesn't + # lose their spot. + self._media.setPosition(self._media.position()) + self._paused_at = None + self._media.play() + + def pause(self): + self._paused_at = time.monotonic() + self._media.pause() + + def stop(self): + self._media.stop() + + def seek(self, position_ms: int): + self._media.setPosition(max(0, position_ms)) + + def position_ms(self) -> int: + return self._media.position() + + def duration_ms(self) -> int: + return self._media.duration() + + def is_playing(self) -> bool: + return self._media.playbackState() == QMediaPlayer.PlaybackState.PlayingState + + def set_volume(self, level: float): + """Set output volume from a logical 0..1 value (0=silent, 1=full). + The slider scale is perceptual, so convert to the linear gain + QAudioOutput expects — that makes the knob feel iTunes-like rather + than jumping to "loud" in the first few percent.""" + self._volume = max(0.0, min(1.0, float(level))) + self._apply_volume() + + def _apply_volume(self): + """Push the stored logical volume to the sink. Called again on resume + and after device swaps: some backends (seen with Bluetooth sinks) + re-create the sink with a stale/zero gain, which played silently until + a seek re-primed it.""" + self._audio.setVolume(QAudio.convertVolume( + self._volume, + QAudio.VolumeScale.LogarithmicVolumeScale, + QAudio.VolumeScale.LinearVolumeScale, + )) def shutdown(self): """Tear down the Qt Multimedia pipeline in a safe order before Qt @@ -104,6 +217,157 @@ class Player(QObject): self._media.setAudioBufferOutput(None) self._media.setAudioOutput(None) # detach the sink while both live + def _on_audio_outputs_changed(self): + """Re-point the output at the current default device when the set of + available outputs changes (a device was added or removed).""" + new_default = QMediaDevices.defaultAudioOutput() + if new_default.isNull() or new_default.id() == self._audio.device().id(): + return + was_playing = self.is_playing() + self._audio.setDevice(new_default) + self._apply_volume() # the new sink may come up at a stale/zero gain + # Some backends stall the sink across a device swap; nudge it back. + # play() is a no-op if playback actually continued. + if was_playing: + self._media.play() + + def _on_state_changed(self, state): + self.state_changed.emit(state == QMediaPlayer.PlaybackState.PlayingState) + + def _on_media_status(self, status): + if (status == QMediaPlayer.MediaStatus.LoadedMedia + and self._pending_start_ms): + # The media is only reliably seekable once loaded; jump to the + # custom start time now (works whether or not we're autoplaying). + start = self._pending_start_ms + self._pending_start_ms = 0 + self._media.setPosition(start) + return + if status == QMediaPlayer.MediaStatus.BufferedMedia: + self._apply_volume() # sink can be re-created per source + return + if status == QMediaPlayer.MediaStatus.EndOfMedia: + self.ended.emit() + + def _on_error(self, error, error_string): + self.error.emit(error_string) + + +class Player(QObject): + """Playback engine: walks a queue of track ids through a swappable sink. + + The queue is the displayed order of whatever view playback started from. + Play counts are recorded (via LibraryManager) when a track finishes + playing naturally — on whichever sink is active, so casting scrobbles and + counts plays exactly like local playback does. + """ + + track_changed = pyqtSignal(object) # Track or None + playing_changed = pyqtSignal(bool) + position_changed = pyqtSignal('qint64') # ms + duration_changed = pyqtSignal('qint64') # ms + error_occurred = pyqtSignal(str) + track_missing = pyqtSignal(object) # Track whose file is gone/moved + audio_buffer = pyqtSignal(object) # QAudioBuffer (decoded PCM) + track_finished = pyqtSignal(object) # Track played to the very end + track_unplayable = pyqtSignal(object, str) # skipped: sink can't decode it + sink_changed = pyqtSignal(object) # the new PlaybackSink + + # Signals relayed from whichever sink is active. + _SINK_SIGNALS = ("position_changed", "duration_changed", "state_changed", + "ended", "error", "audio_buffer", "unavailable") + + def __init__(self, manager, parent=None): + super().__init__(parent) + self._manager = manager + self._queue: list[int] = [] + self._index = -1 + self._current_track: Track | None = None + self._shuffle = False + self._shuffle_order: list[int] = [] + # Custom stop time (ms) for the loaded track: end it early. Enforced + # here rather than in a sink so both playback paths honor it. + self._stop_at_ms = 0 + self._counted_finish_id: int | None = None + self._volume = 1.0 # logical 0..1; what the volume slider shows + + self._local = LocalSink(self) + self._sink: PlaybackSink = self._local + self._connect_sink(self._sink) + + self._shutdown_done = False + + # ---- sinks ---- + + def _connect_sink(self, sink: PlaybackSink): + sink.position_changed.connect(self.position_changed) + sink.position_changed.connect(self._on_position) + sink.duration_changed.connect(self.duration_changed) + sink.state_changed.connect(self.playing_changed) + sink.ended.connect(self._on_sink_ended) + sink.error.connect(self._on_sink_error) + sink.audio_buffer.connect(self.audio_buffer) + sink.unavailable.connect(self._on_sink_unavailable) + + def _disconnect_sink(self, sink: PlaybackSink): + for name in self._SINK_SIGNALS: + try: + getattr(sink, name).disconnect() + except TypeError: + pass + + def active_sink(self) -> PlaybackSink: + return self._sink + + def provides_pcm(self) -> bool: + return self._sink.provides_pcm + + def set_sink(self, sink: PlaybackSink | None): + """Swap the output engine; None restores the built-in local one. + + The current track and position carry across and playback resumes in + whatever state it was in, so connecting or losing a Chromecast picks up + mid-song rather than restarting it. + + `track_changed` is deliberately *not* re-emitted: to the listener this + is the same song still playing, and re-emitting would fire a second + Last.fm now-playing update and reset an in-progress bpm tap. + """ + new_sink = sink if sink is not None else self._local + if new_sink is self._sink: + return + track = self._current_track + position = self._sink.position_ms() + was_playing = self._sink.is_playing() + + self._sink.stop() + self._disconnect_sink(self._sink) + self._sink = new_sink + self._connect_sink(self._sink) + # Whatever the slider last showed applies to the new output too. + self._sink.set_volume(self._volume) + + if track is not None: + if self._sink.can_play(track): + self._sink.load(track, autoplay=was_playing, start_ms=position) + else: + self.track_unplayable.emit( + track, self._sink.unplayable_reason(track)) + self._skip_unplayable(autoplay=was_playing) + self.sink_changed.emit(self._sink) + + def shutdown(self): + """Tear down every sink. Idempotent; wired to closeEvent/aboutToQuit.""" + if self._shutdown_done: + return + self._shutdown_done = True + self._sink.shutdown() + if self._sink is not self._local: + # The local sink is kept alive across a swap so switching back is + # instant — which means its Qt Multimedia pipeline is still around + # at quit and still needs its ordered teardown. + self._local.shutdown() + # ---- queue ---- def play_queue(self, track_ids: list[int], start_index: int): @@ -161,35 +425,23 @@ class Player(QObject): return self._current_track def is_playing(self) -> bool: - return self._media.playbackState() == QMediaPlayer.PlaybackState.PlayingState + return self._sink.is_playing() def position_ms(self) -> int: - return self._media.position() + return self._sink.position_ms() def duration_ms(self) -> int: - return self._media.duration() + return self._sink.duration_ms() # ---- volume ---- def set_volume(self, level: float): """Set output volume from a logical 0..1 value (0=silent, 1=full). - The slider scale is perceptual, so convert to the linear gain - QAudioOutput expects — that makes the knob feel iTunes-like rather - than jumping to "loud" in the first few percent.""" - level = max(0.0, min(1.0, float(level))) - self._volume = level - self._apply_volume() - - def _apply_volume(self): - """Push the stored logical volume to the sink. Called again on resume - and after device swaps: some backends (seen with Bluetooth sinks) - re-create the sink with a stale/zero gain, which played silently until - a seek re-primed it.""" - self._audio.setVolume(QAudio.convertVolume( - self._volume, - QAudio.VolumeScale.LogarithmicVolumeScale, - QAudio.VolumeScale.LinearVolumeScale, - )) + The value is kept here so it survives a sink swap — the slider keeps + meaning the same thing whether audio is coming out of this machine or + a Chromecast.""" + self._volume = max(0.0, min(1.0, float(level))) + self._sink.set_volume(self._volume) def volume(self) -> float: """The logical 0..1 volume last set (what the slider shows).""" @@ -202,19 +454,9 @@ class Player(QObject): f"playing={self.is_playing()} pos={self.position_ms()}ms " f"track={self._current_track.name if self._current_track else None!r}") if self.is_playing(): - self._paused_at = time.monotonic() - self._media.pause() + self._sink.pause() elif self._current_track is not None: - self._apply_volume() - if (self._paused_at is not None - and time.monotonic() - self._paused_at - >= RESUME_NUDGE_THRESHOLD_S): - # Re-prime a possibly idle-suspended sink (the manual - # workaround was "rewind slightly"); seek in place so the - # listener doesn't lose their spot. - self._media.setPosition(self._media.position()) - self._paused_at = None - self._media.play() + self._sink.play() elif self._queue: self._index = max(self._index, 0) self._load_current(autoplay=True) @@ -226,11 +468,10 @@ class Player(QObject): def pause(self): if self.is_playing(): log_control("player", "pause", f"pos={self.position_ms()}ms") - self._paused_at = time.monotonic() - self._media.pause() + self._sink.pause() def stop(self): - self._media.stop() + self._sink.stop() self._current_track = None self.track_changed.emit(None) @@ -243,11 +484,11 @@ class Player(QObject): if not self._queue: return prev_index = self._step_index(-1) - if self._media.position() > RESTART_THRESHOLD_MS or prev_index is None: + if self.position_ms() > RESTART_THRESHOLD_MS or prev_index is None: # "Restart" means the track's custom start time, not 0:00. start = (self._current_track.start_time if self._current_track else 0) - self._media.setPosition(max(0, start)) + self._sink.seek(max(0, start)) else: self._index = prev_index self._load_current(autoplay=self.is_playing()) @@ -258,10 +499,10 @@ class Player(QObject): self._index = next_index self._load_current(autoplay=autoplay) elif self._queue: - self._media.stop() + self._sink.stop() def seek(self, position_ms: int): - self._media.setPosition(max(0, position_ms)) + self._sink.seek(max(0, position_ms)) def retry_current(self): """Re-attempt the track at the current queue position (e.g. after its @@ -271,51 +512,56 @@ class Player(QObject): # ---- internals ---- def _load_current(self, autoplay: bool): - if not (0 <= self._index < len(self._queue)): + """Load the queue entry at _index, stepping past tracks the active + sink can't decode (a Chromecast can't play ALAC, AIFF or protected + AAC). The skip is a loop, not recursion: shuffling into a run of + unsupported tracks must not walk the stack down.""" + while 0 <= self._index < len(self._queue): + track = self._manager.library.tracks.get(self._queue[self._index]) + if track is None: + # The queue references a track that's no longer in the library; + # there's nothing to locate, so just report and stop. + self.error_occurred.emit( + f"Can't play “track {self._queue[self._index]}”.") + self.stop() + return + if not track.location or not Path(track.location).exists(): + # Stop here rather than auto-advancing: if the whole drive is + # unmounted every queued track is missing, and skipping to the + # next would walk the entire queue for nothing. + # Stop *before* emitting: the UI shows a modal "Locate File…" + # dialog synchronously from this signal and may relocate + + # retry, which a later stop() would otherwise undo. + self.stop() + self.track_missing.emit(track) + return + if not self._sink.can_play(track): + # Reported on its own signal, not error_occurred: that one + # opens a modal dialog, and shuffling past 100 lossless files + # must not mean 100 dialogs. + self.track_unplayable.emit( + track, self._sink.unplayable_reason(track)) + next_index = self._step_index(1) + if next_index is None: + self.stop() + return + self._index = next_index + continue + + self._current_track = track + # Arm the custom stop time for this track. It's honored only when + # it falls before the end (otherwise the natural finish handles it). + if track.stop_time > 0 and (track.total_time == 0 + or track.stop_time < track.total_time): + self._stop_at_ms = track.stop_time + else: + self._stop_at_ms = 0 + self._counted_finish_id = None + self._sink.load( + track, autoplay, + track.start_time if track.start_time > 0 else 0) + self.track_changed.emit(track) return - track = self._manager.library.tracks.get(self._queue[self._index]) - if track is None: - # The queue references a track that's no longer in the library; - # there's nothing to locate, so just report and stop. - self.error_occurred.emit( - f"Can't play “track {self._queue[self._index]}”.") - self.stop() - return - if not track.location or not Path(track.location).exists(): - # Stop here rather than auto-advancing: if the whole drive is - # unmounted every queued track is missing, and skipping to the - # next would recurse through the entire queue (RecursionError). - # Stop *before* emitting: the UI shows a modal "Locate File…" - # dialog synchronously from this signal and may relocate + retry, - # which a later stop() would otherwise undo. - self.stop() - self.track_missing.emit(track) - return - self._current_track = track - # Arm the custom start/stop times for this track. Stop is honored only - # when it falls before the end (otherwise EndOfMedia handles the finish). - self._pending_start_ms = track.start_time if track.start_time > 0 else 0 - if track.stop_time > 0 and (track.total_time == 0 - or track.stop_time < track.total_time): - self._stop_at_ms = track.stop_time - else: - self._stop_at_ms = 0 - self._counted_finish_id = None - url = QUrl.fromLocalFile(track.location) - if self._media.source() == url: - # setSource() no-ops on an unchanged URL, so LoadedMedia never - # re-fires and the armed start time would be skipped (and forcing - # a clear+reload races the FFmpeg backend, which snaps the seek - # back to 0). The media is already loaded: rewind and seek now. - self._media.stop() - if self._pending_start_ms: - self._media.setPosition(self._pending_start_ms) - self._pending_start_ms = 0 - else: - self._media.setSource(url) - if autoplay: - self._media.play() - self.track_changed.emit(track) def _skip_unplayable(self, autoplay: bool = True): next_index = self._step_index(1) @@ -325,39 +571,25 @@ class Player(QObject): else: self.stop() - def _on_audio_outputs_changed(self): - """Re-point the output at the current default device when the set of - available outputs changes (a device was added or removed).""" - new_default = QMediaDevices.defaultAudioOutput() - if new_default.isNull() or new_default.id() == self._audio.device().id(): - return - was_playing = self.is_playing() - self._audio.setDevice(new_default) - self._apply_volume() # the new sink may come up at a stale/zero gain - # Some backends stall the sink across a device swap; nudge it back. - # play() is a no-op if playback actually continued. - if was_playing: - self._media.play() + # ---- sink events ---- - def _on_state_changed(self, state): - self.playing_changed.emit(state == QMediaPlayer.PlaybackState.PlayingState) + def _on_sink_ended(self): + """The active sink played the track to its natural end.""" + self._note_finished(self._current_track) + # Playback has already stopped at the end, so force autoplay. + self._advance(autoplay=True) - def _on_media_status(self, status): - if (status == QMediaPlayer.MediaStatus.LoadedMedia - and self._pending_start_ms): - # The media is only reliably seekable once loaded; jump to the - # custom start time now (works whether or not we're autoplaying). - start = self._pending_start_ms - self._pending_start_ms = 0 - self._media.setPosition(start) - return - if status == QMediaPlayer.MediaStatus.BufferedMedia: - self._apply_volume() # sink can be re-created per source - return - if status == QMediaPlayer.MediaStatus.EndOfMedia: - self._note_finished(self._current_track) - # The player has already stopped at EndOfMedia, so force autoplay - self._advance(autoplay=True) + def _on_sink_error(self, message: str): + track_name = self._current_track.name if self._current_track else "?" + self.error_occurred.emit(f"Playback error on '{track_name}': {message}") + self._skip_unplayable() + + def _on_sink_unavailable(self, reason: str): + """The sink itself died (the Chromecast went away). Drop back to local + playback, which picks up at the same position, and say why.""" + self.set_sink(None) + if reason: + self.error_occurred.emit(reason) def _on_position(self, position): """Stop a track at its custom stop time and move on, mirroring a @@ -376,8 +608,3 @@ class Player(QObject): self._counted_finish_id = track.track_id self._manager.record_play(track.track_id) self.track_finished.emit(track) - - def _on_error(self, error, error_string): - track_name = self._current_track.name if self._current_track else "?" - self.error_occurred.emit(f"Playback error on '{track_name}': {error_string}") - self._skip_unplayable() diff --git a/requirements.txt b/requirements.txt index 792dea1..7d070a5 100644 --- a/requirements.txt +++ b/requirements.txt @@ -2,3 +2,4 @@ PyQt6>=6.8.0 mutagen>=1.46 numpy>=1.24 requests>=2.28 +pychromecast>=14.0.10 diff --git a/setup.py b/setup.py index 9b53730..4084abd 100644 --- a/setup.py +++ b/setup.py @@ -17,6 +17,7 @@ setup( "mutagen>=1.46", "numpy>=1.24", "requests>=2.28", + "pychromecast>=14.0.10", ], data_files=[ ("share/applications", ["packaging/lintunes.desktop"]), @@ -26,5 +27,5 @@ setup( "lintunes=lintunes.main:main", ], }, - python_requires=">=3.9", + python_requires=">=3.11", ) diff --git a/tasks-done.md b/tasks-done.md index 16fe553..9c02078 100644 --- a/tasks-done.md +++ b/tasks-done.md @@ -1,5 +1,40 @@ ## Done +### Round 29 (2026-08-13) — Cast to Chromecast (v0.4.0) + +The Device menu is now **Connections**, with "Connect to Chromecast…" alongside +the Rabbit sync. Picking a device from the search dialog hands playback to it; +a small cast glyph appears under the volume slider and clicking it disconnects. +See `tests/test_round29.py` (67 tests). + +Uses the **media-receiver model**: lintunes serves the original file over the +LAN and the Chromecast decodes it. Bit-exact (no transcode, no double-lossy on +already-lossy files) and the device buffers for itself; the cost is that +lintunes becomes a remote control while connected — no local PCM, so the +visualizer shows the cast glyph instead of bars, and transport actions land +with ~1 s of round trip. + +- [x] **`lintunes/cast/`** — `support.py` (format gate, source-address + selection, range parsing), `server.py` (token-addressed + `ThreadingHTTPServer` with Range/HEAD), `discovery.py` (`CastBrowser` behind + Qt signals), `sink.py` (`CastSink`), `controller.py` (session lifecycle + + its own `SleepInhibitor`). +- [x] **`Player` sink split** — new `PlaybackSink` base with `LocalSink` and + `CastSink`; `set_sink()` carries track/position/playing-state both ways, so + connecting and disconnecting pick up mid-song. Play counts and Last.fm + scrobbles fire identically on both paths. +- [x] **Format handling** — 12,610 MP3 + ~8,750 AAC cast natively; the ~119 + Apple Lossless / AIFF / protected-AAC tracks are skipped with a status-bar + message rather than stalling the device. +- [x] **Failure handling** — a dropped socket gets a 15 s grace period + (pychromecast retries on its own, so a Wi-Fi blip heals itself); a real loss, + a foreign app taking the device, or a network change falls back to local + playback **still playing**, at the same position. Quitting stops the device + rather than leaving it fetching from a dead server. +- [x] **New dependency: `pychromecast>=14.0.10`**, imported lazily so the app + still launches where it isn't installed; `python_requires` raised to >=3.11 + (its floor). The menu item explains the `pip install -e .` when missing. + ### Misc (recorded 2026-07-02, moved from TASKS.md during the Round 17 rewrite) - [x] Confirm Last.fm scrobbling works diff --git a/tests/test_player.py b/tests/test_player.py index 13c99c4..3bd4e20 100644 --- a/tests/test_player.py +++ b/tests/test_player.py @@ -56,7 +56,7 @@ def test_missing_file_emits_track_missing_and_stops(qapp, tmp_path): assert missing == [track] # Stopped, not advanced into a track; setSource was never called. assert player.current_track is None - player._media.setSource.assert_not_called() + player._local._media.setSource.assert_not_called() def test_all_missing_queue_does_not_recurse(qapp, tmp_path): @@ -104,4 +104,4 @@ def test_retry_current_plays_after_relocate(qapp, tmp_path): assert errors == [] and missing == [] assert player.current_track is track - player._media.setSource.assert_called_once() + player._local._media.setSource.assert_called_once() diff --git a/tests/test_round10.py b/tests/test_round10.py index 9d42e30..bdb6a25 100644 --- a/tests/test_round10.py +++ b/tests/test_round10.py @@ -42,7 +42,7 @@ class TestPlayerVolume: player.set_volume(0.5) assert player.volume() == 0.5 # The linear gain handed to QAudioOutput is set (perceptual->linear). - player._audio.setVolume.assert_called() + player._local._audio.setVolume.assert_called() def test_clamps_out_of_range(self, qapp): player = _mock_player() diff --git a/tests/test_round17.py b/tests/test_round17.py index 1b8101f..992685f 100644 --- a/tests/test_round17.py +++ b/tests/test_round17.py @@ -223,12 +223,12 @@ class TestPlayerShutdown: def test_shutdown_detaches_in_order_and_is_idempotent(self, qapp, tmp_path): player = _mock_player(qapp, tmp_path) player.shutdown() - player._media.stop.assert_called_once() - player._media.setSource.assert_called_once_with(QUrl()) - player._media.setAudioBufferOutput.assert_called_with(None) - player._media.setAudioOutput.assert_called_with(None) + player._local._media.stop.assert_called_once() + player._local._media.setSource.assert_called_once_with(QUrl()) + player._local._media.setAudioBufferOutput.assert_called_with(None) + player._local._media.setAudioOutput.assert_called_with(None) player.shutdown() # second call must be a no-op - player._media.stop.assert_called_once() + player._local._media.stop.assert_called_once() class TestVolumeReapply: @@ -239,18 +239,18 @@ class TestVolumeReapply: expected = QAudio.convertVolume( 0.5, QAudio.VolumeScale.LogarithmicVolumeScale, QAudio.VolumeScale.LinearVolumeScale) - player._audio.setVolume.assert_called_with(expected) + player._local._audio.setVolume.assert_called_with(expected) def test_resume_reapplies_volume(self, qapp, tmp_path, monkeypatch): player = _mock_player(qapp, tmp_path) player._current_track = Track(track_id=1, name="A") applied = [] - monkeypatch.setattr(player, "_apply_volume", + monkeypatch.setattr(player._local, "_apply_volume", lambda: applied.append(True)) monkeypatch.setattr(player, "is_playing", lambda: False) player.toggle_play() # paused with a cued track → resume assert applied == [True] - player._media.play.assert_called_once() + player._local._media.play.assert_called_once() def test_long_pause_resume_seeks_in_place(self, qapp, tmp_path, monkeypatch): @@ -259,15 +259,15 @@ class TestVolumeReapply: player._current_track = Track(track_id=1, name="A") monkeypatch.setattr(player, "is_playing", lambda: False) # Short pause: no nudge. - player._paused_at = time_module.monotonic() - 5 + player._local._paused_at = time_module.monotonic() - 5 player.toggle_play() - player._media.setPosition.assert_not_called() + player._local._media.setPosition.assert_not_called() # Walk-away pause: seek-in-place re-primes the suspended sink. - player._media.position.return_value = 12345 - player._paused_at = time_module.monotonic() - 600 + player._local._media.position.return_value = 12345 + player._local._paused_at = time_module.monotonic() - 600 player.toggle_play() - player._media.setPosition.assert_called_once_with(12345) - assert player._paused_at is None + player._local._media.setPosition.assert_called_once_with(12345) + assert player._local._paused_at is None # ---- Task B: search bar lives in the header strip ---- diff --git a/tests/test_round22.py b/tests/test_round22.py index e69fbf4..b44a3ca 100644 --- a/tests/test_round22.py +++ b/tests/test_round22.py @@ -48,21 +48,21 @@ class TestSameSourceReplay: url = QUrl.fromLocalFile(loc) player.play_queue([1], 0) # first load: source was empty, plain set - assert player._media.setSource.call_args_list[-1].args == (url,) + assert player._local._media.setSource.call_args_list[-1].args == (url,) # The track is now the loaded source; the user edits its start time # and plays it again. The media is already loaded, so the player must # seek directly instead of waiting for a LoadedMedia that never comes. - player._media.source.return_value = url + player._local._media.source.return_value = url track.start_time = 30_000 - player._media.setSource.reset_mock() + player._local._media.setSource.reset_mock() player.play_queue([1], 0) - player._media.setSource.assert_not_called() - player._media.stop.assert_called() - player._media.setPosition.assert_called_with(30_000) - assert player._pending_start_ms == 0 # consumed - player._media.play.assert_called() + player._local._media.setSource.assert_not_called() + player._local._media.stop.assert_called() + player._local._media.setPosition.assert_called_with(30_000) + assert player._local._pending_start_ms == 0 # consumed + player._local._media.play.assert_called() def test_unchanged_url_without_start_time_restarts_at_zero( self, qapp, tmp_path): @@ -72,13 +72,13 @@ class TestSameSourceReplay: player, _ = _player(qapp, [track]) player.play_queue([1], 0) - player._media.source.return_value = QUrl.fromLocalFile(loc) - player._media.setPosition.reset_mock() + player._local._media.source.return_value = QUrl.fromLocalFile(loc) + player._local._media.setPosition.reset_mock() player.play_queue([1], 0) # stop() rewinds to 0; no seek should be issued. - player._media.stop.assert_called() - player._media.setPosition.assert_not_called() + player._local._media.stop.assert_called() + player._local._media.setPosition.assert_not_called() def test_different_url_loads_via_setsource(self, qapp, tmp_path): for name in ("a.mp3", "b.mp3"): @@ -87,16 +87,16 @@ class TestSameSourceReplay: t2 = _track(2, location=str(tmp_path / "b.mp3"), total_time=200_000, start_time=30_000) player, _ = _player(qapp, [t1, t2]) - player._media.source.return_value = QUrl.fromLocalFile(t1.location) + player._local._media.source.return_value = QUrl.fromLocalFile(t1.location) player.play_queue([1, 2], 1) # loading b.mp3 while a.mp3 is the source - assert [c.args for c in player._media.setSource.call_args_list] \ + assert [c.args for c in player._local._media.setSource.call_args_list] \ == [(QUrl.fromLocalFile(t2.location),)] # New source: the seek waits for LoadedMedia as before. - assert player._pending_start_ms == 30_000 - player._on_media_status(player_module.QMediaPlayer.MediaStatus.LoadedMedia) - player._media.setPosition.assert_called_with(30_000) - assert player._pending_start_ms == 0 + assert player._local._pending_start_ms == 30_000 + player._local._on_media_status(player_module.QMediaPlayer.MediaStatus.LoadedMedia) + player._local._media.setPosition.assert_called_with(30_000) + assert player._local._pending_start_ms == 0 class TestPreviousRestart: @@ -107,9 +107,9 @@ class TestPreviousRestart: player, _ = _player(qapp, [track]) player.play_queue([1], 0) - player._media.position.return_value = RESTART_THRESHOLD_MS + 1 + player._local._media.position.return_value = RESTART_THRESHOLD_MS + 1 player.previous() - player._media.setPosition.assert_called_with(30_000) + player._local._media.setPosition.assert_called_with(30_000) def test_restart_without_start_time_seeks_to_zero(self, qapp, tmp_path): (tmp_path / "a.mp3").write_bytes(b"x") @@ -117,6 +117,6 @@ class TestPreviousRestart: player, _ = _player(qapp, [track]) player.play_queue([1], 0) - player._media.position.return_value = RESTART_THRESHOLD_MS + 1 + player._local._media.position.return_value = RESTART_THRESHOLD_MS + 1 player.previous() - player._media.setPosition.assert_called_with(0) + player._local._media.setPosition.assert_called_with(0) diff --git a/tests/test_round23.py b/tests/test_round23.py index b0b2a0f..09132ab 100644 --- a/tests/test_round23.py +++ b/tests/test_round23.py @@ -96,8 +96,8 @@ class TestPlayerLogging: track = Track(track_id=1, name="Song A", location=str(tmp_path / "a.mp3"), total_time=100_000) player = self._player(qapp, [track]) - player._media.position.return_value = 4321 - player._media.playbackState.return_value = None # != PlayingState + player._local._media.position.return_value = 4321 + player._local._media.playbackState.return_value = None # != PlayingState player.play_queue([1], 0) player.toggle_play() text = path.read_text() diff --git a/tests/test_round29.py b/tests/test_round29.py new file mode 100644 index 0000000..ea7f02b --- /dev/null +++ b/tests/test_round29.py @@ -0,0 +1,911 @@ +"""Round 29: casting to a Chromecast. + +Covers the parts that are verifiable without a device on the network — the +format gate, byte-range arithmetic, and the local media server that hands the +Chromecast its files. The server tests drive a real socket on an ephemeral +port, so range requests, HEAD and token containment are exercised end to end. + +Not covered here (hand-verified against a real Chromecast): the pychromecast +handshake and discovery, the receiver's own decode of any given file, seek +behavior over Wi-Fi, and the MainWindow menu/dialog wiring. +""" + +import http.client +import socket +from types import SimpleNamespace +from unittest.mock import MagicMock, patch + +import pytest +from PyQt6.QtCore import QObject, pyqtSignal + +from lintunes import player as player_module +from lintunes.cast.discovery import CastDevice +from lintunes.cast.server import KEEP_TOKENS, TrackServer +from lintunes.cast.sink import ( + BACKDROP_APP_ID, DEFAULT_RECEIVER_APP_ID, CastSink, app_stolen, + connection_verdict, duration_ms_from, is_load_error, is_natural_end, + is_playing_state, position_ms_from) +from lintunes.cast.support import ( + content_type_for, local_ip_for, parse_range, uncastable_reason) +from lintunes.gui.cast_dialog import ChromecastDialog +from lintunes.gui.cast_indicator import CastIndicator +from lintunes.gui.icons import transport_icon +from lintunes.gui.spinner import Spinner +from lintunes.models import Track +from lintunes.models.library import Library +from lintunes.player import PlaybackSink, Player + + +# ---- the format gate ---- + +class TestContentType: + def test_mp3_and_aac_cast_natively(self): + assert content_type_for("/m/a.mp3", "MPEG audio file") == "audio/mpeg" + assert content_type_for("/m/a.mpga") == "audio/mpeg" + assert content_type_for("/m/a.m4a", "Matched AAC audio file") == "audio/mp4" + assert content_type_for("/m/a.m4a", "AAC audio file") == "audio/mp4" + assert content_type_for("/m/a.wav", "WAV audio file") == "audio/wav" + + def test_purchased_aac_is_drm_free_and_casts(self): + # iTunes Plus downloads are unencrypted; only "Protected AAC" isn't. + assert content_type_for("/m/a.m4a", "Purchased AAC audio file") == "audio/mp4" + + def test_lossless_and_protected_are_refused(self): + assert content_type_for("/m/a.m4a", "Apple Lossless audio file") is None + assert content_type_for("/m/a.m4a", "Protected AAC audio file") is None + assert content_type_for("/m/a.m4p") is None + assert content_type_for("/m/a.aiff", "AIFF audio file") is None + assert content_type_for("/m/a.aif") is None + + def test_unknown_container_is_refused(self): + assert content_type_for("/m/a.wma") is None + assert content_type_for("") is None + + def test_suffix_case_is_ignored(self): + assert content_type_for("/m/A.MP3") == "audio/mpeg" + + def test_reason_names_the_format(self): + assert "Apple Lossless" in uncastable_reason( + "/m/a.m4a", "Apple Lossless audio file") + assert "copy-protected AAC" in uncastable_reason("/m/a.m4p") + assert "AIFF" in uncastable_reason("/m/a.aiff") + assert "WMA" in uncastable_reason("/m/a.wma") + + +# ---- byte ranges ---- + +class TestParseRange: + def test_no_header_serves_whole_file(self): + assert parse_range(None, 100) is None + assert parse_range("", 100) is None + + def test_closed_range(self): + assert parse_range("bytes=10-19", 100) == (10, 19) + + def test_open_ended_range_runs_to_the_end(self): + assert parse_range("bytes=10-", 100) == (10, 99) + + def test_suffix_range_takes_the_tail(self): + assert parse_range("bytes=-20", 100) == (80, 99) + assert parse_range("bytes=-500", 100) == (0, 99) # longer than the file + + def test_end_past_eof_is_clamped(self): + assert parse_range("bytes=90-999", 100) == (90, 99) + + def test_unsatisfiable_and_malformed_fall_back_to_whole_file(self): + # A 200 with the full body is a legal answer to any Range request and + # keeps the device playing instead of erroring it out. + assert parse_range("bytes=100-", 100) is None # start at/past EOF + assert parse_range("bytes=50-10", 100) is None # inverted + assert parse_range("bytes=abc-def", 100) is None + assert parse_range("items=0-10", 100) is None + assert parse_range("bytes=0-10,20-30", 100) is None # multi-range + assert parse_range("bytes=0-10", 0) is None # empty file + + +# ---- source address selection ---- + +class TestLocalIp: + def test_returns_the_address_that_routes_to_the_peer(self): + # Route toward a public address; any answer is fine as long as it is a + # real local v4 address and not loopback-by-accident. + ip = local_ip_for("8.8.8.8") + assert ip.count(".") == 3 + socket.inet_aton(ip) # raises if it isn't a v4 address + + def test_unreachable_peer_still_resolves(self): + # connect() on UDP only picks a route, so an unroutable-but-valid + # address still yields the interface the kernel would use. + assert local_ip_for("192.0.2.1").count(".") == 3 + + +# ---- the media server ---- + +@pytest.fixture +def served(tmp_path): + """A running TrackServer plus a helper to talk to it over loopback.""" + server = TrackServer() + server.start() + yield server + server.stop() + + +def _request(server, path, method="GET", headers=None): + conn = http.client.HTTPConnection("127.0.0.1", server.port, timeout=5) + try: + conn.request(method, path, headers=headers or {}) + response = conn.getresponse() + return response.status, dict(response.getheaders()), response.read() + finally: + conn.close() + + +def _audio(tmp_path, name="song.mp3", data=None): + path = tmp_path / name + path.write_bytes(data if data is not None else bytes(range(256)) * 8) + return path + + +class TestTrackServer: + def test_serves_the_published_file_verbatim(self, served, tmp_path): + path = _audio(tmp_path) + token = served.publish(path, "audio/mpeg") + + status, headers, body = _request(served, f"/t/{token}") + + assert status == 200 + assert body == path.read_bytes() + assert headers["Content-Type"] == "audio/mpeg" + assert headers["Content-Length"] == str(path.stat().st_size) + # Without this the device won't seek and won't report a duration. + assert headers["Accept-Ranges"] == "bytes" + + def test_head_reports_size_without_a_body(self, served, tmp_path): + path = _audio(tmp_path) + token = served.publish(path, "audio/mpeg") + + status, headers, body = _request(served, f"/t/{token}", method="HEAD") + + assert status == 200 + assert body == b"" + assert headers["Content-Length"] == str(path.stat().st_size) + + def test_range_request_returns_the_slice(self, served, tmp_path): + path = _audio(tmp_path) + size = path.stat().st_size + token = served.publish(path, "audio/mpeg") + + status, headers, body = _request( + served, f"/t/{token}", headers={"Range": "bytes=10-19"}) + + assert status == 206 + assert body == path.read_bytes()[10:20] + assert headers["Content-Range"] == f"bytes 10-19/{size}" + assert headers["Content-Length"] == "10" + + def test_open_ended_range_returns_the_tail(self, served, tmp_path): + path = _audio(tmp_path) + size = path.stat().st_size + token = served.publish(path, "audio/mpeg") + + status, headers, body = _request( + served, f"/t/{token}", headers={"Range": "bytes=100-"}) + + assert status == 206 + assert body == path.read_bytes()[100:] + assert headers["Content-Range"] == f"bytes 100-{size - 1}/{size}" + + def test_unknown_token_is_not_found(self, served): + status, _, _ = _request(served, "/t/nonexistent") + assert status == 404 + + def test_no_path_can_be_smuggled_through_the_url(self, served, tmp_path): + # There is no path in the request to traverse with — only a token + # lookup — so these are 404s by construction, not by sanitizing. + _audio(tmp_path) + for probe in ("/t/../../../etc/passwd", "/etc/passwd", + "/t/%2e%2e%2fetc%2fpasswd", "/", f"/t/{tmp_path}"): + status, _, _ = _request(served, probe) + assert status == 404, probe + + def test_only_the_most_recent_tokens_stay_live(self, served, tmp_path): + tokens = [served.publish(_audio(tmp_path, f"{i}.mp3"), "audio/mpeg") + for i in range(KEEP_TOKENS + 2)] + + for stale in tokens[:2]: + status, _, _ = _request(served, f"/t/{stale}") + assert status == 404 + for live in tokens[2:]: + status, _, _ = _request(served, f"/t/{live}") + assert status == 200 + + def test_previous_track_stays_fetchable(self, served, tmp_path): + # The device re-requests the outgoing file while draining its buffer. + first = served.publish(_audio(tmp_path, "a.mp3"), "audio/mpeg") + served.publish(_audio(tmp_path, "b.mp3"), "audio/mpeg") + + status, _, _ = _request(served, f"/t/{first}") + assert status == 200 + + def test_file_deleted_after_publishing_is_not_found(self, served, tmp_path): + path = _audio(tmp_path) + token = served.publish(path, "audio/mpeg") + path.unlink() + + status, _, _ = _request(served, f"/t/{token}") + assert status == 404 + + def test_a_client_hanging_up_mid_stream_does_not_kill_the_server( + self, served, tmp_path): + # Exactly what the Chromecast does on every seek, pause and stop. + big = _audio(tmp_path, "big.mp3", b"x" * (4 * 1024 * 1024)) + token = served.publish(big, "audio/mpeg") + + conn = http.client.HTTPConnection("127.0.0.1", served.port, timeout=5) + conn.request("GET", f"/t/{token}") + conn.getresponse().read(1024) + conn.close() # hang up mid-body + + # The server thread survived and still answers. + status, _, body = _request(served, f"/t/{token}") + assert status == 200 + assert len(body) == big.stat().st_size + + def test_url_for_uses_the_bound_port(self, served, tmp_path): + token = served.publish(_audio(tmp_path), "audio/mpeg") + assert served.url_for("192.168.1.5", token) == ( + f"http://192.168.1.5:{served.port}/t/{token}") + + def test_stop_is_idempotent_and_closes_the_port(self, tmp_path): + server = TrackServer() + port = server.start() + server.publish(_audio(tmp_path), "audio/mpeg") + server.stop() + server.stop() # must not raise + + with pytest.raises(OSError): + conn = http.client.HTTPConnection("127.0.0.1", port, timeout=2) + conn.request("GET", "/t/anything") + conn.getresponse() + + def test_tokens_do_not_survive_a_restart(self, tmp_path): + server = TrackServer() + server.start() + token = server.publish(_audio(tmp_path), "audio/mpeg") + server.stop() + server.start() + try: + assert server.lookup(token) is None + finally: + server.stop() + + +# ---- reading the device's status ---- + +def _status(**fields): + fields.setdefault("player_state", "PLAYING") + fields.setdefault("idle_reason", None) + fields.setdefault("media_session_id", 1) + return SimpleNamespace(**fields) + + +class TestStatusReading: + def test_finished_is_a_natural_end(self): + assert is_natural_end(_status(player_state="IDLE", idle_reason="FINISHED")) + + def test_other_idle_reasons_are_not_a_finish(self): + # Counting a play or advancing on these would be wrong: ERROR means it + # never played, INTERRUPTED means something else took the device. + for reason in ("ERROR", "INTERRUPTED", "CANCELLED", None): + assert not is_natural_end( + _status(player_state="IDLE", idle_reason=reason)) + assert is_load_error(_status(player_state="IDLE", idle_reason="ERROR")) + + def test_buffering_still_counts_as_playing(self): + # Otherwise the play button flickers every time the device re-buffers. + assert is_playing_state(_status(player_state="BUFFERING")) + assert is_playing_state(_status(player_state="PLAYING")) + assert not is_playing_state(_status(player_state="PAUSED")) + + def test_position_extrapolates_only_while_playing(self): + playing = _status(player_state="PLAYING", + adjusted_current_time=12.5, current_time=10.0) + assert position_ms_from(playing, 0) == 12_500 + + # adjusted_current_time keeps creeping off the wall clock while + # paused, which would drift the seek slider during a pause. + paused = _status(player_state="PAUSED", + adjusted_current_time=99.0, current_time=10.0) + assert position_ms_from(paused, 0) == 10_000 + + def test_position_falls_back_when_the_device_says_nothing(self): + assert position_ms_from(None, 4321) == 4321 + assert position_ms_from( + _status(adjusted_current_time=None, current_time=None), 4321) == 4321 + + def test_duration_falls_back_to_the_library_value(self): + assert duration_ms_from(_status(duration=180.0), 0) == 180_000 + assert duration_ms_from(_status(duration=None), 200_000) == 200_000 + assert duration_ms_from(None, 200_000) == 200_000 + + def test_foreign_receiver_reads_as_stolen(self): + assert app_stolen("YouTube") + assert not app_stolen(DEFAULT_RECEIVER_APP_ID) + assert not app_stolen(BACKDROP_APP_ID) + assert not app_stolen(None) + assert not app_stolen("") + + def test_lost_is_a_grace_period_not_a_disconnect(self): + # pychromecast retries a dropped socket on its own, so a Wi-Fi blip + # must not dump playback back to the laptop speakers. + assert connection_verdict("LOST") == "grace" + assert connection_verdict("CONNECTING") == "grace" + assert connection_verdict("CONNECTED") == "ok" + assert connection_verdict("DISCONNECTED") == "lost" + assert connection_verdict("FAILED") == "lost" + assert connection_verdict("FAILED_RESOLVE") == "lost" + + +# ---- Player's sink contract ---- + +class _FakeSink(PlaybackSink): + """A sink that records what Player asked of it, standing in for a cast.""" + + def __init__(self, playable=True, parent=None): + super().__init__(parent) + self.name = "Kitchen" + self.loads = [] + self.volume = None + self.stopped = 0 + self.shutdowns = 0 + self._playing = False + self._position = 0 + self._playable = playable + + def load(self, track, autoplay, start_ms): + self.loads.append((track.track_id, autoplay, start_ms)) + self._playing = autoplay + self._position = start_ms + + def play(self): + self._playing = True + + def pause(self): + self._playing = False + + def stop(self): + self.stopped += 1 + self._playing = False + + def seek(self, position_ms): + self._position = position_ms + + def position_ms(self): + return self._position + + def duration_ms(self): + return 0 + + def is_playing(self): + return self._playing + + def set_volume(self, level): + self.volume = level + + def can_play(self, track): + return self._playable + + def unplayable_reason(self, track): + return "a Chromecast can't play Apple Lossless" + + def shutdown(self): + self.shutdowns += 1 + + +def _player(qapp, tracks): + library = Library() + for track in tracks: + library.tracks[track.track_id] = track + with patch.multiple( + player_module, + QMediaPlayer=MagicMock(), + QAudioOutput=MagicMock(), + QAudioBufferOutput=MagicMock(), + QMediaDevices=MagicMock(), + ): + return Player(MagicMock(library=library)) + + +def _local_track(tmp_path, track_id, name="Song", **fields): + path = tmp_path / f"{track_id}.mp3" + path.write_bytes(b"x") + return Track(track_id=track_id, name=name, location=str(path), **fields) + + +class TestPlayerSinks: + def test_default_sink_is_local(self, qapp): + player = _player(qapp, []) + assert player.active_sink() is player._local + assert player.provides_pcm() is True + + def test_swap_carries_track_position_and_playing_state(self, qapp, tmp_path): + track = _local_track(tmp_path, 1, total_time=200_000) + player = _player(qapp, [track]) + player.play_queue([1], 0) + player._local._media.position.return_value = 42_000 + player._local._media.playbackState.return_value = \ + player_module.QMediaPlayer.PlaybackState.PlayingState + + sink = _FakeSink() + player.set_sink(sink) + + assert player.active_sink() is sink + assert sink.loads == [(1, True, 42_000)] # same song, same spot + assert player.provides_pcm() is False + + def test_swap_does_not_re_announce_the_track(self, qapp, tmp_path): + # A second track_changed would fire a duplicate Last.fm now-playing + # and reset an in-progress bpm tap, for what is the same song. + track = _local_track(tmp_path, 1) + player = _player(qapp, [track]) + player.play_queue([1], 0) + seen = [] + player.track_changed.connect(seen.append) + + player.set_sink(_FakeSink()) + + assert seen == [] + + def test_volume_follows_the_swap_both_ways(self, qapp): + player = _player(qapp, []) + player.set_volume(0.42) + + sink = _FakeSink() + player.set_sink(sink) + assert sink.volume == 0.42 # the slider keeps its meaning + + player.set_volume(0.7) + assert sink.volume == 0.7 + + player.set_sink(None) + assert player.volume() == 0.7 + assert player.active_sink() is player._local + + def test_swapping_to_the_same_sink_is_a_no_op(self, qapp): + player = _player(qapp, []) + sink = _FakeSink() + player.set_sink(sink) + player.set_sink(sink) + assert sink.stopped == 0 + + def test_natural_end_on_a_cast_records_the_play_and_advances( + self, qapp, tmp_path): + t1 = _local_track(tmp_path, 1, total_time=200_000) + t2 = _local_track(tmp_path, 2, total_time=200_000) + player = _player(qapp, [t1, t2]) + finished = [] + player.track_finished.connect(lambda t: finished.append(t.track_id)) + player.play_queue([1, 2], 0) + sink = _FakeSink() + player.set_sink(sink) + + sink.ended.emit() + + player._manager.record_play.assert_called_once_with(1) + assert finished == [1] + assert player.current_track.track_id == 2 + + def test_repeated_end_counts_the_play_once(self, qapp, tmp_path): + track = _local_track(tmp_path, 1, total_time=200_000) + player = _player(qapp, [track]) + player.play_queue([1], 0) + sink = _FakeSink() + player.set_sink(sink) + + sink.ended.emit() + sink.ended.emit() + + player._manager.record_play.assert_called_once_with(1) + + def test_custom_stop_time_still_ends_early_on_a_cast(self, qapp, tmp_path): + t1 = _local_track(tmp_path, 1, total_time=200_000, stop_time=60_000) + t2 = _local_track(tmp_path, 2, total_time=200_000) + player = _player(qapp, [t1, t2]) + player.play_queue([1, 2], 0) + sink = _FakeSink() + player.set_sink(sink) + + sink.position_changed.emit(59_000) + assert player._manager.record_play.call_count == 0 + sink.position_changed.emit(60_500) + + player._manager.record_play.assert_called_once_with(1) + assert player.current_track.track_id == 2 + + def test_custom_start_time_reaches_the_sink(self, qapp, tmp_path): + track = _local_track(tmp_path, 1, total_time=200_000, start_time=30_000) + player = _player(qapp, [track]) + sink = _FakeSink() + player.set_sink(sink) + + player.play_queue([1], 0) + + assert sink.loads[-1] == (1, True, 30_000) + + def test_unplayable_track_is_skipped_not_dialogged(self, qapp, tmp_path): + t1 = _local_track(tmp_path, 1, name="Lossless") + t2 = _local_track(tmp_path, 2, name="Fine") + player = _player(qapp, [t1, t2]) + skipped, errors = [], [] + player.track_unplayable.connect( + lambda track, reason: skipped.append((track.name, reason))) + # error_occurred opens a modal dialog in the GUI — the wrong channel + # for "this one track's format isn't supported". + player.error_occurred.connect(errors.append) + + sink = _FakeSink() + # Only the second track is castable. + sink.can_play = lambda track: track.track_id == 2 + player.set_sink(sink) + player.play_queue([1, 2], 0) + + assert [name for name, _ in skipped] == ["Lossless"] + assert "Apple Lossless" in skipped[0][1] + assert errors == [] + assert player.current_track.track_id == 2 + + def test_a_queue_of_unplayable_tracks_stops_without_recursing( + self, qapp, tmp_path): + tracks = [_local_track(tmp_path, i) for i in range(1, 600)] + player = _player(qapp, tracks) + skipped = [] + player.track_unplayable.connect(lambda t, r: skipped.append(t)) + + player.set_sink(_FakeSink(playable=False)) + player.play_queue([t.track_id for t in tracks], 0) + + assert len(skipped) == len(tracks) + assert player.current_track is None # stopped cleanly, no crash + + def test_a_dead_sink_falls_back_to_local_playback(self, qapp, tmp_path): + track = _local_track(tmp_path, 1, total_time=200_000) + player = _player(qapp, [track]) + player.play_queue([1], 0) + sink = _FakeSink() + player.set_sink(sink) + sink._position = 55_000 + sink._playing = True + errors = [] + player.error_occurred.connect(errors.append) + + sink.unavailable.emit("Lost the connection to Kitchen.") + + assert player.active_sink() is player._local + assert errors == ["Lost the connection to Kitchen."] + + def test_missing_file_still_reports_on_a_cast_sink(self, qapp, tmp_path): + track = Track(track_id=1, name="Ghost", + location=str(tmp_path / "gone" / "a.mp3")) + player = _player(qapp, [track]) + missing = [] + player.track_missing.connect(missing.append) + + player.set_sink(_FakeSink()) + player.play_queue([1], 0) + + assert missing == [track] + assert player.current_track is None + + def test_shutdown_tears_down_both_sinks(self, qapp): + player = _player(qapp, []) + sink = _FakeSink() + player.set_sink(sink) + + player.shutdown() + + assert sink.shutdowns == 1 + # The local sink is kept alive across a swap, so its Qt Multimedia + # pipeline still needs the ordered teardown that avoids a segfault. + player._local._media.setAudioOutput.assert_called_with(None) + + +# ---- the cast sink's failure handling ---- + +class _FakeCast: + """Enough of a pychromecast Chromecast to build a CastSink against.""" + + def __init__(self): + self.cast_info = SimpleNamespace(friendly_name="Kitchen", host="10.0.0.5") + self.status = SimpleNamespace(volume_level=0.8) + self.media_controller = SimpleNamespace( + status=None, register_status_listener=lambda _l: None, + play_media=lambda *a, **k: None, play=lambda: None, + pause=lambda: None, stop=lambda: None, seek=lambda _s: None) + self.volumes = [] + self.quit_calls = 0 + self.disconnects = 0 + + def register_status_listener(self, listener): + pass + + def register_connection_listener(self, listener): + pass + + def set_volume(self, level): + self.volumes.append(level) + + def quit_app(self): + self.quit_calls += 1 + + def disconnect(self): + self.disconnects += 1 + + +@pytest.fixture +def cast_sink(qapp): + cast = _FakeCast() + server = TrackServer() + sink = CastSink(cast, server) + yield sink, cast + sink.shutdown() + + +class TestCastSinkFailure: + def test_failure_reports_the_state_lintunes_asked_for(self, cast_sink): + sink, _cast = cast_sink + sink._intent_playing = True + sink._playing = True + # A dying connection pushes an IDLE media status just before the + # disconnect. Believing it here would stop the music on the fallback. + sink._on_status(_status(player_state="IDLE", idle_reason=None)) + assert sink.is_playing() is False + + reasons = [] + sink.unavailable.connect(reasons.append) + sink._on_connection("DISCONNECTED") + + assert reasons == ["Lost the connection to Kitchen."] + assert sink.is_playing() is True # so Player resumes locally playing + + def test_a_paused_session_does_not_start_playing_on_failure(self, cast_sink): + sink, _cast = cast_sink + sink._intent_playing = False + sink._on_connection("DISCONNECTED") + assert sink.is_playing() is False + + def test_failure_fires_once(self, cast_sink): + sink, _cast = cast_sink + reasons = [] + sink.unavailable.connect(reasons.append) + + sink._on_connection("DISCONNECTED") + sink._on_connection("FAILED") + sink._on_grace_expired() + + assert len(reasons) == 1 + + def test_a_dropped_socket_gets_a_grace_period(self, cast_sink): + sink, _cast = cast_sink + reasons = [] + sink.unavailable.connect(reasons.append) + + sink._on_connection("LOST") + assert sink._lost_timer.isActive() + assert reasons == [] # pychromecast is still retrying + + sink._on_connection("CONNECTED") + assert not sink._lost_timer.isActive() + assert reasons == [] # the blip healed itself + + def test_grace_expiring_gives_up(self, cast_sink): + sink, _cast = cast_sink + reasons = [] + sink.unavailable.connect(reasons.append) + + sink._on_connection("LOST") + sink._on_grace_expired() + + assert reasons == ["Kitchen stopped responding."] + + def test_another_app_taking_the_device_ends_the_session(self, cast_sink): + sink, _cast = cast_sink + reasons = [] + sink.unavailable.connect(reasons.append) + + sink._on_app(DEFAULT_RECEIVER_APP_ID) + sink._on_app(BACKDROP_APP_ID) + assert reasons == [] + + sink._on_app("YouTube") + assert reasons == ["Something else started casting to Kitchen."] + + def test_natural_end_is_reported_once_per_session(self, cast_sink): + sink, _cast = cast_sink + ends = [] + sink.ended.connect(lambda: ends.append(True)) + + done = _status(player_state="IDLE", idle_reason="FINISHED", + media_session_id=7) + sink._on_status(done) + sink._on_status(done) # the device repeats its final status + + assert ends == [True] + + def test_shutdown_hands_the_device_volume_back(self, qapp): + cast = _FakeCast() + sink = CastSink(cast, TrackServer()) + sink.set_volume(0.2) + + sink.shutdown() + sink.shutdown() # idempotent + + # The device's own volume persists for whatever plays next, so the + # session's setting must not be left behind. + assert cast.volumes[-1] == 0.8 + assert cast.quit_calls == 1 + assert cast.disconnects == 1 + + def test_uncastable_track_is_refused_by_the_sink(self, cast_sink): + sink, _cast = cast_sink + alac = Track(track_id=1, name="Lossless", location="/m/a.m4a", + kind="Apple Lossless audio file") + mp3 = Track(track_id=2, name="Fine", location="/m/a.mp3") + + assert not sink.can_play(alac) + assert "Apple Lossless" in sink.unplayable_reason(alac) + assert sink.can_play(mp3) + + +# ---- widgets ---- + +class _FakeDiscovery(QObject): + device_found = pyqtSignal(object) + device_lost = pyqtSignal(object) + failed = pyqtSignal(str) + + def __init__(self): + super().__init__() + self.started = 0 + self.stopped = 0 + + def devices(self): + return [] + + def start(self): + self.started += 1 + + def stop(self): + self.stopped += 1 + + +def _device(uuid="u1", name="Kitchen", model="Chromecast Audio"): + return CastDevice(uuid=uuid, name=name, model=model, + host="10.0.0.5", port=8009) + + +class TestCastIndicator: + def test_holds_its_space_when_idle(self, qapp): + indicator = CastIndicator() + indicator.show() + # Hiding it would collapse the layout and shift the volume slider the + # moment you connect — idle is a visible widget with no icon. + assert indicator.icon().isNull() + assert not indicator.isEnabled() + idle_height = indicator.sizeHint().height() + + indicator.set_connected(False) + assert not indicator.isHidden() # never hides itself + + indicator.set_connected(True, "Kitchen") + assert not indicator.icon().isNull() + assert indicator.isEnabled() + assert "Kitchen" in indicator.toolTip() + assert indicator.sizeHint().height() == idle_height + + indicator.set_connected(False) + assert indicator.icon().isNull() + + +class TestTransportLayout: + def test_cast_indicator_does_not_move_the_volume_slider( + self, qapp, tmp_path): + from lintunes.gui.transport import BAR_HEIGHT, TransportBar + from lintunes.preferences import Preferences + + player = _player(qapp, []) + bar = TransportBar(player, MagicMock(), Preferences(tmp_path)) + bar.resize(1200, BAR_HEIGHT) + bar.show() + idle_y = bar._volume_slider.pos().y() + + bar._cast_indicator.set_connected(True, "Kitchen") + bar.layout().activate() + + assert bar._volume_slider.pos().y() == idle_y + # The whole column has to stay inside the bar's documented height. + assert bar.sizeHint().height() <= BAR_HEIGHT + bar.hide() + + +class TestCastDialog: + def test_devices_appear_and_disappear(self, qapp): + discovery = _FakeDiscovery() + dialog = ChromecastDialog(discovery) + assert discovery.started == 1 + + discovery.device_found.emit(_device()) + assert dialog._list.count() == 1 + assert "Kitchen" in dialog._list.item(0).text() + + # zeroconf re-announces constantly; the same device must not stack up. + discovery.device_found.emit(_device()) + assert dialog._list.count() == 1 + + discovery.device_lost.emit(_device()) + assert dialog._list.count() == 0 + + def test_connect_needs_a_selection(self, qapp): + discovery = _FakeDiscovery() + dialog = ChromecastDialog(discovery) + assert not dialog._connect_button.isEnabled() + + discovery.device_found.emit(_device()) + assert dialog._connect_button.isEnabled() + + dialog._on_connect() + assert dialog.selected_device.name == "Kitchen" + + def test_closing_always_stops_the_search(self, qapp): + for close in (lambda d: d.reject(), lambda d: d.accept(), + lambda d: d.done(0)): + discovery = _FakeDiscovery() + dialog = ChromecastDialog(discovery) + close(dialog) + assert discovery.stopped == 1 + + def test_missing_dependency_is_reported_not_spun_forever(self, qapp): + discovery = _FakeDiscovery() + dialog = ChromecastDialog(discovery) + + discovery.failed.emit("Chromecast support isn't installed.") + + assert "isn't installed" in dialog._status.text() + assert dialog._spinner.isHidden() + + +class TestSpinner: + def test_animates_only_while_visible(self, qapp): + spinner = Spinner() + spinner.show() + assert spinner._timer.isActive() + spinner.hide() + assert not spinner._timer.isActive() + + +class TestCastIcons: + def test_cast_glyphs_render_and_differ(self, qapp): + from PyQt6.QtGui import QColor + idle = transport_icon("cast", QColor("#4A4A4A")) + live = transport_icon("cast_connected", QColor("#4A4A4A")) + assert not idle.isNull() and not live.isNull() + assert (idle.pixmap(20, 20).toImage() + != live.pixmap(20, 20).toImage()) + + +class TestVisualizerWhileCasting: + def test_stops_animating_when_the_sink_has_no_pcm(self, qapp, tmp_path): + from lintunes.gui.visualizer import VisualizerWidget + from lintunes.preferences import Preferences + + player = _player(qapp, []) + vis = VisualizerWidget(player, Preferences(tmp_path)) + + player.set_sink(_FakeSink()) + assert vis._no_pcm is True + assert not vis._timer.isActive() + # A play/pause must not restart an FFT over a buffer of zeros. + player.playing_changed.emit(True) + assert not vis._timer.isActive() + + player.set_sink(None) + assert vis._no_pcm is False diff --git a/tests/test_round8.py b/tests/test_round8.py index ed87b11..cf1e1e4 100644 --- a/tests/test_round8.py +++ b/tests/test_round8.py @@ -248,7 +248,7 @@ class TestPlayerStartStop: total_time=200_000, start_time=30_000) player, _ = self._player(qapp, [track]) player.play_queue([1], 0) - assert player._pending_start_ms == 30_000 - player._on_media_status(player_module.QMediaPlayer.MediaStatus.LoadedMedia) - player._media.setPosition.assert_called_with(30_000) - assert player._pending_start_ms == 0 # consumed (one-shot) + assert player._local._pending_start_ms == 30_000 + player._local._on_media_status(player_module.QMediaPlayer.MediaStatus.LoadedMedia) + player._local._media.setPosition.assert_called_with(30_000) + assert player._local._pending_start_ms == 0 # consumed (one-shot)