v0.5.0: send album art to the Chromecast

The device is on a TV, so it should show the cover. play_media now carries
thumb=, which pychromecast folds into metadata["images"] — the field the
receiver paints full-screen. Verified against the real device: it fetches
both the audio and the artwork URL from us on every track change.

Album art lives in the audio file's tags rather than as a file of its own,
so TrackServer tokens now resolve to an _Asset that is either a path or a
blob held in memory. Audio and art get separate eviction rings so a cover
can't push out the previous track's audio while the device is still
fetching it; Range and HEAD work on both.

The image type is sniffed from the cover's magic bytes rather than trusted
from the tag — ID3 APIC mimes are routinely wrong or blank, and the
receiver silently drops an image whose declared type doesn't match its
content. Anything unrecognized is treated as "no cover".

Best-effort throughout: no art, junk where the art should be, or an
unreadable file all just play without a cover rather than failing the
load. Also sends albumArtist and trackNumber in the metadata.

tests/test_round29.py: 74 tests; 447 pass overall.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
2026-08-13 22:06:27 -04:00
co-authored by Claude Opus 5
parent 087c4bf103
commit 6c746c3330
7 changed files with 217 additions and 33 deletions
+6 -1
View File
@@ -117,7 +117,12 @@ persistence) → GUI (Qt widgets that read the manager and connect to its signal
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
them). A token resolves to an `_Asset` that is either a file on disk (audio)
or an in-memory blob (album art, which lives in tags rather than as its own
file); the two have **separate eviction rings** so a cover can't push out the
previous track's audio. Art is passed as `play_media(thumb=…)`, which
pychromecast folds into `metadata["images"]` — that's what a TV paints
full-screen. `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
+1 -1
View File
@@ -1,3 +1,3 @@
"""LinTunes — iTunes-style music library manager and player for Linux."""
__version__ = "0.4.0"
__version__ = "0.5.0"
+55 -28
View File
@@ -16,8 +16,10 @@ 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 io
import secrets
import threading
from dataclasses import dataclass
from http import HTTPStatus
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path
@@ -28,11 +30,34 @@ from lintunes.cast.support import parse_range
# 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.
# Audio and artwork are counted separately so a track's cover can't evict the
# audio of the track before it.
KEEP_TOKENS = 3
CHUNK = 64 * 1024
@dataclass(frozen=True)
class _Asset:
"""Something published for the device to fetch: a library file on disk, or
a blob held in memory (album art, which is extracted from tags rather than
being a file of its own)."""
content_type: str
path: Path | None = None
data: bytes | None = None
def size(self) -> int:
if self.data is not None:
return len(self.data)
return self.path.stat().st_size
def open(self):
if self.data is not None:
return io.BytesIO(self.data)
return open(self.path, "rb")
class _Handler(BaseHTTPRequestHandler):
protocol_version = "HTTP/1.1" # Chromecast wants keep-alive + Content-Length
@@ -45,7 +70,7 @@ class _Handler(BaseHTTPRequestHandler):
def _tracks(self):
return self.server.track_server
def _resolve(self) -> Path | None:
def _resolve(self) -> "_Asset | None":
prefix = "/t/"
if not self.path.startswith(prefix):
return None
@@ -58,13 +83,13 @@ class _Handler(BaseHTTPRequestHandler):
self._serve(body=True)
def _serve(self, body: bool):
path = self._resolve()
if path is None:
asset = self._resolve()
if asset is None:
self.send_error(HTTPStatus.NOT_FOUND)
return
try:
size = path.stat().st_size
handle = open(path, "rb")
size = asset.size()
handle = asset.open()
except OSError:
# The file moved or the drive went away since it was published.
self.send_error(HTTPStatus.NOT_FOUND)
@@ -81,7 +106,7 @@ class _Handler(BaseHTTPRequestHandler):
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("Content-Type", asset.content_type)
self.send_header("Accept-Ranges", "bytes")
self.send_header("Content-Length", str(length))
if status == HTTPStatus.PARTIAL_CONTENT:
@@ -116,9 +141,10 @@ class TrackServer:
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] = []
self._assets: dict[str, _Asset] = {}
# One eviction ring per kind, so publishing a track's cover can't push
# the previous track's audio out from under a device still fetching it.
self._rings: dict[str, list[str]] = {"audio": [], "art": []}
# ---- lifecycle ----
@@ -147,36 +173,37 @@ class TrackServer:
httpd.server_close()
self._thread = None
with self._lock:
self._tokens.clear()
self._types.clear()
self._order.clear()
self._assets.clear()
for ring in self._rings.values():
ring.clear()
# ---- publishing ----
def publish(self, path, content_type: str) -> str:
"""Make `path` fetchable and return its token."""
path = Path(path)
"""Make a library file fetchable and return its token."""
return self._publish(
_Asset(content_type=content_type, path=Path(path)), "audio")
def publish_bytes(self, data: bytes, content_type: str) -> str:
"""Make an in-memory blob (album art) fetchable and return its token."""
return self._publish(
_Asset(content_type=content_type, data=data), "art")
def _publish(self, asset: _Asset, kind: str) -> str:
with self._lock:
token = secrets.token_urlsafe(16)
self._tokens[token] = path
self._types[path] = content_type
self._order.append(token)
self._assets[token] = asset
ring = self._rings[kind]
ring.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)
while len(ring) > KEEP_TOKENS:
self._assets.pop(ring.pop(0), None)
return token
def lookup(self, token: str) -> Path | None:
def lookup(self, token: str) -> _Asset | 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")
return self._assets.get(token)
def url_for(self, host: str, token: str) -> str:
return f"http://{host}:{self.port}/t/{token}"
+31 -1
View File
@@ -22,7 +22,9 @@ import time
from PyQt6.QtCore import QTimer, pyqtSignal
from lintunes.cast.support import content_type_for, local_ip_for, uncastable_reason
from lintunes import tagging
from lintunes.cast.support import (
content_type_for, image_type_for, local_ip_for, uncastable_reason)
from lintunes.player import PlaybackSink
# The stock receiver lintunes launches, and the idle "backdrop" app. Anything
@@ -212,10 +214,21 @@ class CastSink(PlaybackSink):
"artist": track.artist or "",
"albumName": track.album or "",
}
if track.album_artist:
metadata["albumArtist"] = track.album_artist
if track.track_number:
metadata["trackNumber"] = track.track_number
thumb = self._publish_artwork(track)
if thumb:
# pychromecast folds `thumb` into metadata["images"], which is what
# the receiver paints full-screen on a TV.
thumb = self._server.url_for(host, thumb)
self._submit(
lambda: self._mc.play_media(
url, content_type,
title=track.name or "",
thumb=thumb or None,
# NOT pychromecast's STREAM_TYPE_LIVE default: LIVE tells the
# receiver the stream is unbounded, which kills both seeking
# and the duration readout.
@@ -225,6 +238,23 @@ class CastSink(PlaybackSink):
metadata=metadata,
))
def _publish_artwork(self, track) -> str | None:
"""Serve the track's embedded cover, returning its token.
Art lives in the audio file's tags, not as a file of its own, so it is
published as an in-memory blob. Best-effort throughout: a track with no
cover (or an unreadable one) just plays without one rather than
failing the load.
"""
try:
data = tagging.read_embedded_artwork(track.location)
except Exception: # noqa: BLE001 — art is never worth a failure
return None
image_type = image_type_for(data or b"")
if image_type is None:
return None
return self._server.publish_bytes(data, image_type)
def play(self):
self._intent_playing = True
self._set_playing(True)
+20
View File
@@ -69,6 +69,26 @@ def uncastable_reason(location: str, kind: str = "") -> str:
return f"a Chromecast can't play {suffix.lstrip('.').upper() or 'this format'}"
def image_type_for(data: bytes) -> str | None:
"""The MIME type of an embedded cover image, or None if we can't tell.
Sniffed from the magic bytes rather than trusted from the tag: ID3 APIC
frames carry a declared mime that is routinely wrong (or empty), and the
Chromecast just drops an image whose type doesn't match its content.
"""
if not data:
return None
if data[:3] == b"\xff\xd8\xff":
return "image/jpeg"
if data[:8] == b"\x89PNG\r\n\x1a\n":
return "image/png"
if data[:6] in (b"GIF87a", b"GIF89a"):
return "image/gif"
if data[:4] == b"RIFF" and data[8:12] == b"WEBP":
return "image/webp"
return None
def local_ip_for(peer_ip: str, port: int = 8009) -> str:
"""This machine's address *as the Chromecast will see it*.
+19
View File
@@ -1,5 +1,24 @@
## Done
### Round 30 (2026-08-13) — Album art on the cast device (v0.5.0)
The Chromecast is on a TV, so it should show the cover. `play_media` now
carries `thumb=`, which pychromecast folds into `metadata["images"]` — the
field the receiver paints full-screen. Verified against the real device: it
fetches both the audio and the artwork URL from us on every track change.
- [x] **`TrackServer` serves in-memory blobs** — album art lives in the audio
file's tags, not as a file of its own, so a token now resolves to an
`_Asset` that is either a path or bytes. Audio and art get **separate
eviction rings**, so a cover can't push out the previous track's audio while
a device is still fetching it. Range and HEAD work on both.
- [x] **`support.image_type_for`** sniffs the cover's type from its magic bytes
rather than trusting the tag — ID3 APIC mimes are routinely wrong or blank,
and the receiver silently drops an image whose type doesn't match.
- [x] **Best-effort throughout** — no cover, junk where the cover should be, or
an unreadable file all just play without art rather than failing the load.
- [x] Also sends `albumArtist` and `trackNumber` in the metadata.
### Round 29 (2026-08-13) — Cast to Chromecast (v0.4.0)
The Device menu is now **Connections**, with "Connect to Chromecast…" alongside
+85 -2
View File
@@ -18,7 +18,7 @@ from unittest.mock import MagicMock, patch
import pytest
from PyQt6.QtCore import QObject, pyqtSignal
from lintunes import player as player_module
from lintunes import player as player_module, tagging
from lintunes.cast.discovery import CastDevice
from lintunes.cast.server import KEEP_TOKENS, TrackServer
from lintunes.cast.sink import (
@@ -26,7 +26,8 @@ from lintunes.cast.sink import (
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)
content_type_for, image_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
@@ -72,6 +73,22 @@ class TestContentType:
assert "WMA" in uncastable_reason("/m/a.wma")
class TestImageType:
def test_sniffs_the_common_cover_formats(self):
assert image_type_for(b"\xff\xd8\xff\xe0rest") == "image/jpeg"
assert image_type_for(b"\x89PNG\r\n\x1a\nrest") == "image/png"
assert image_type_for(b"GIF89a...") == "image/gif"
assert image_type_for(b"RIFF\x00\x00\x00\x00WEBPVP8 ") == "image/webp"
def test_unrecognized_bytes_are_refused(self):
# Better no cover than a cover the receiver silently drops: the type
# is sniffed rather than trusted, because ID3 APIC mimes are often
# wrong or blank.
assert image_type_for(b"") is None
assert image_type_for(b"not an image at all") is None
assert image_type_for(b"\xff\xd8") is None # truncated JPEG magic
# ---- byte ranges ----
class TestParseRange:
@@ -268,6 +285,41 @@ class TestTrackServer:
conn.request("GET", "/t/anything")
conn.getresponse()
def test_serves_album_art_from_memory(self, served):
# Cover art lives in the audio file's tags, not as a file of its own.
art = b"\x89PNG\r\n\x1a\n" + bytes(range(256)) * 4
token = served.publish_bytes(art, "image/png")
status, headers, body = _request(served, f"/t/{token}")
assert status == 200
assert body == art
assert headers["Content-Type"] == "image/png"
assert headers["Content-Length"] == str(len(art))
def test_range_and_head_work_on_art_too(self, served):
art = b"\x89PNG\r\n\x1a\n" + bytes(range(256))
token = served.publish_bytes(art, "image/png")
status, _, body = _request(
served, f"/t/{token}", headers={"Range": "bytes=8-15"})
assert status == 206
assert body == art[8:16]
status, headers, body = _request(served, f"/t/{token}", method="HEAD")
assert status == 200 and body == b""
assert headers["Content-Length"] == str(len(art))
def test_art_and_audio_evict_separately(self, served, tmp_path):
# A track's cover must not push the previous track's audio out from
# under a device that is still fetching it.
audio = served.publish(_audio(tmp_path), "audio/mpeg")
for i in range(KEEP_TOKENS + 2):
served.publish_bytes(b"\x89PNG\r\n\x1a\n" + bytes([i]), "image/png")
status, _, _ = _request(served, f"/t/{audio}")
assert status == 200
def test_tokens_do_not_survive_a_restart(self, tmp_path):
server = TrackServer()
server.start()
@@ -745,6 +797,37 @@ class TestCastSinkFailure:
assert cast.quit_calls == 1
assert cast.disconnects == 1
def test_embedded_cover_is_published_for_the_device(
self, cast_sink, monkeypatch):
sink, _cast = cast_sink
art = b"\xff\xd8\xff\xe0" + b"jpegbody"
monkeypatch.setattr(tagging, "read_embedded_artwork", lambda _p: art)
token = sink._publish_artwork(
Track(track_id=1, name="A", location="/m/a.mp3"))
assert token is not None
asset = sink._server.lookup(token)
assert asset.data == art
assert asset.content_type == "image/jpeg"
def test_a_track_without_a_cover_still_plays(self, cast_sink, monkeypatch):
sink, _cast = cast_sink
track = Track(track_id=1, name="A", location="/m/a.mp3")
monkeypatch.setattr(tagging, "read_embedded_artwork", lambda _p: None)
assert sink._publish_artwork(track) is None
# Unreadable tags, or junk where the cover should be, are equally
# never worth failing the load over.
monkeypatch.setattr(tagging, "read_embedded_artwork", lambda _p: b"junk")
assert sink._publish_artwork(track) is None
def boom(_p):
raise OSError("drive went away")
monkeypatch.setattr(tagging, "read_embedded_artwork", boom)
assert sink._publish_artwork(track) is None
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",