music-assistant-server

1"""
2Color palette extraction from artwork.
3
4Derives a 6-field MediaItemPalette per the Sendspin color@v1 spec from an
5image: a `modern_colorthief` MMCQ quantizer produces image candidates, then `primary`,
6`accent`, `on_dark`, `on_light`, `background_dark`, and `background_light` are
7chosen (and adjusted where needed) so that every spec-mandated contrast pair
8clears the WCAG AA 4.5:1 threshold.
9
10Extracted palettes are stored in the cache controller (sqlite), keyed on a
11hash of provider and image path, so results persist across restarts and are
12shared process-wide.
13"""
14
15from __future__ import annotations
16
17import asyncio
18from typing import TYPE_CHECKING
19
20from modern_colorthief import get_palette as _mmcq_palette
21from music_assistant_models.errors import MusicAssistantError
22from music_assistant_models.media_items import MediaItemPalette
23
24from music_assistant.helpers.images import (
25    _extract_imageproxy_id,
26    create_thumb_hash,
27    get_image_data,
28)
29from music_assistant.helpers.util import join_task
30
31if TYPE_CHECKING:
32    from music_assistant.mass import MusicAssistant
33
34
35_PALETTE_QUANTIZE_COLORS = 5
36_COLORTHIEF_QUALITY = 10
37# Minimum contrast ratio between colors (Sendspin color@v1 requires WCAG AA ≥ 4.5:1).
38_MIN_CONTRAST = 4.5
39# Preferred contrast (we try this first for richer, more vivid picks)
40_PREFERRED_CONTRAST = 7.0
41# Upper cap when picking the dark on-light color. Without this, near-black
42# image regions (e.g. text outlines) win and the picked color looks like pure
43# black instead of a vibrant darker shade from the artwork.
44_MAX_DARK_PICK_CONTRAST = 17.35
45_BACKGROUND_ADJUST_STEPS = 20
46_SIMILARITY_THRESHOLD = 60  # squared euclidean RGB distance for accent picking
47
48# Cache controller namespace for extracted palettes. Palettes are
49# content-addressed (hash of provider+path) and deterministic, so a long
50# expiration is safe: a palette only changes if the underlying image changes.
51_CACHE_PROVIDER = "palette"
52_CACHE_EXPIRATION = 90 * 24 * 3600  # 90 days
53
54_RGB = tuple[int, int, int]
55
56
57def _relative_luminance(rgb: _RGB) -> float:
58    """Compute WCAG relative luminance for an sRGB color."""
59
60    def channel(c: int) -> float:
61        s = c / 255.0
62        return s / 12.92 if s <= 0.03928 else ((s + 0.055) / 1.055) ** 2.4
63
64    r, g, b = rgb
65    return 0.2126 * channel(r) + 0.7152 * channel(g) + 0.0722 * channel(b)
66
67
68def _contrast_ratio(a: _RGB, b: _RGB) -> float:
69    """Compute WCAG contrast ratio between two RGB colors."""
70    la = _relative_luminance(a)
71    lb = _relative_luminance(b)
72    lighter, darker = (la, lb) if la >= lb else (lb, la)
73    return (lighter + 0.05) / (darker + 0.05)
74
75
76def _color_distance_sq(a: _RGB, b: _RGB) -> int:
77    return (a[0] - b[0]) ** 2 + (a[1] - b[1]) ** 2 + (a[2] - b[2]) ** 2
78
79
80def _mix(rgb: _RGB, target: _RGB, factor: float) -> _RGB:
81    """Blend rgb toward target. factor=0 returns rgb, factor=1 returns target."""
82    return (
83        round(rgb[0] + (target[0] - rgb[0]) * factor),
84        round(rgb[1] + (target[1] - rgb[1]) * factor),
85        round(rgb[2] + (target[2] - rgb[2]) * factor),
86    )
87
88
89def _adjust_until_contrast(
90    color: _RGB,
91    mix_toward: _RGB,
92    refs: tuple[_RGB, ...],
93    min_contrast: float = _MIN_CONTRAST,
94) -> _RGB | None:
95    """
96    Mix color toward mix_toward until contrast >= min_contrast vs all refs.
97
98    :param color: Starting color.
99    :param mix_toward: Direction to blend (e.g. black to darken, white to lighten).
100    :param refs: Colors that the result must contrast against.
101    :param min_contrast: Required minimum contrast ratio against every ref.
102    """
103    for step in range(_BACKGROUND_ADJUST_STEPS + 1):
104        factor = step / _BACKGROUND_ADJUST_STEPS
105        candidate = _mix(color, mix_toward, factor)
106        if all(_contrast_ratio(candidate, ref) >= min_contrast for ref in refs):
107            return candidate
108    return None
109
110
111def _adjust_with_fallback(color: _RGB, mix_toward: _RGB, refs: tuple[_RGB, ...]) -> _RGB | None:
112    """Try _PREFERRED_CONTRAST first, fall back to _MIN_CONTRAST if needed."""
113    return _adjust_until_contrast(
114        color, mix_toward, refs, _PREFERRED_CONTRAST
115    ) or _adjust_until_contrast(color, mix_toward, refs, _MIN_CONTRAST)
116
117
118def _extract_candidates(image_bytes: bytes) -> list[_RGB]:
119    """Extract a dominant-color palette via MMCQ (matches the colorthief JS lib)."""
120    palette = _mmcq_palette(
121        image_bytes, color_count=_PALETTE_QUANTIZE_COLORS, quality=_COLORTHIEF_QUALITY
122    )
123    return [(r, g, b) for r, g, b in palette]
124
125
126def _pick_on_color(
127    candidates: list[_RGB],
128    target: _RGB,
129    min_contrast: float = _MIN_CONTRAST,
130    max_contrast: float = float("inf"),
131) -> _RGB | None:
132    """Pick the candidate with highest contrast vs target in [min, max]."""
133    best: _RGB | None = None
134    best_ratio = 0.0
135    for rgb in candidates:
136        ratio = _contrast_ratio(rgb, target)
137        if min_contrast <= ratio <= max_contrast and ratio > best_ratio:
138            best = rgb
139            best_ratio = ratio
140    return best
141
142
143def _pick_on_color_with_fallback(
144    candidates: list[_RGB], target: _RGB, max_contrast: float = float("inf")
145) -> _RGB | None:
146    """Try _PREFERRED_CONTRAST first, fall back to _MIN_CONTRAST if needed."""
147    return _pick_on_color(candidates, target, _PREFERRED_CONTRAST, max_contrast) or _pick_on_color(
148        candidates, target, _MIN_CONTRAST, max_contrast
149    )
150
151
152def _pick_accent(primary: _RGB, candidates: list[_RGB]) -> _RGB | None:
153    """Return the first candidate that is hue-distant from primary, or None."""
154    for rgb in candidates:
155        if rgb == primary:
156            continue
157        if _color_distance_sq(rgb, primary) >= _SIMILARITY_THRESHOLD**2:
158            return rgb
159    return None
160
161
162def _derive_palette(candidates: list[_RGB]) -> MediaItemPalette:
163    if not candidates:
164        return MediaItemPalette()
165    primary = candidates[0]
166
167    # On-colors are picked from image candidates against pure black/white (the
168    # natural reference for a "light color" or "dark color"), this matches the previous algorithm from the frontend.
169    # on_light caps at _MAX_DARK_PICK_CONTRAST to avoid near-black picks (e.g. text outlines) winning over genuinely dark
170    # image colors.
171    on_dark = _pick_on_color_with_fallback(candidates, (0, 0, 0))
172    on_light = _pick_on_color_with_fallback(candidates, (255, 255, 255), _MAX_DARK_PICK_CONTRAST)
173
174    # Synthesize a fallback when no candidate cleared the contrast bar so the
175    # field is always emitted. Spec dual-use: on_dark must clear 4.5:1 vs black
176    # text (so it can also act as a light bg), on_light must clear vs white.
177    if on_dark is None:
178        on_dark = _adjust_until_contrast(primary, (255, 255, 255), ((0, 0, 0),))
179    if on_light is None:
180        on_light = _adjust_until_contrast(primary, (0, 0, 0), ((255, 255, 255),))
181
182    # Adjust backgrounds so they clear MIN_CONTRAST vs both the matching text
183    # color (white/black) and the chosen on-color, per spec.
184    bg_dark_refs: tuple[_RGB, ...] = ((255, 255, 255),)
185    if on_dark is not None:
186        bg_dark_refs = ((255, 255, 255), on_dark)
187    background_dark = _adjust_with_fallback(primary, (0, 0, 0), bg_dark_refs)
188
189    bg_light_refs: tuple[_RGB, ...] = ((0, 0, 0),)
190    if on_light is not None:
191        bg_light_refs = ((0, 0, 0), on_light)
192    background_light = _adjust_with_fallback(primary, (255, 255, 255), bg_light_refs)
193
194    # Accent: a hue-distant secondary from the image. This is not adjusted for contrast.
195    accent = _pick_accent(primary, candidates)
196
197    return MediaItemPalette(
198        background_dark=background_dark,
199        background_light=background_light,
200        primary=primary,
201        accent=accent,
202        on_dark=on_dark,
203        on_light=on_light,
204    )
205
206
207def extract_palette(image_bytes: bytes) -> MediaItemPalette:
208    """
209    Extract a MediaItemPalette from raw image bytes.
210
211    :param image_bytes: Raw image data (PNG, JPEG, etc.).
212    """
213    try:
214        candidates = _extract_candidates(image_bytes)
215    except ValueError, OSError:
216        return MediaItemPalette()
217    return _derive_palette(candidates)
218
219
220async def _extract_and_cache(
221    mass: MusicAssistant, path_or_url: str, provider: str, key: str
222) -> MediaItemPalette:
223    try:
224        img_data = await get_image_data(mass, path_or_url, provider)
225    except FileNotFoundError, MusicAssistantError:
226        # the image is unavailable (e.g. a stale artwork URL that 404s); the
227        # empty palette is not cached below, so extraction is retried once the
228        # image becomes available again
229        return MediaItemPalette()
230    palette = await asyncio.to_thread(extract_palette, img_data)
231    # Only persist a palette that actually yielded colors; an empty result is
232    # usually a transient decode/download failure that should be retried rather
233    # than cached for weeks.
234    if palette.primary is not None:
235        await mass.cache.set(
236            key,
237            palette.to_dict(),
238            provider=_CACHE_PROVIDER,
239            expiration=_CACHE_EXPIRATION,
240        )
241    return palette
242
243
244async def get_palette(
245    mass: MusicAssistant, path_or_url: str, provider: str
246) -> MediaItemPalette | None:
247    """
248    Get the color palette for an image, backed by the cache controller.
249
250    :param mass: The MusicAssistant instance.
251    :param path_or_url: Image path or URL (same format as get_image_data).
252    :param provider: Provider identifier for the image source.
253    """
254    if not path_or_url:
255        return None
256    key = create_thumb_hash(provider, path_or_url)
257
258    cached: MediaItemPalette | None = await mass.cache.get(
259        key, provider=_CACHE_PROVIDER, base_class=MediaItemPalette
260    )
261    if cached is not None:
262        return cached
263
264    # Dedupe concurrent extraction (e.g. now-playing + prefetch) for the same image.
265    task: asyncio.Task[MediaItemPalette] = mass.create_task(
266        _extract_and_cache,
267        mass,
268        path_or_url,
269        provider,
270        key,
271        task_id=f"palette.{key}",
272        abort_existing=False,
273    )
274    return await join_task(task)
275
276
277async def invalidate_cached_palette(mass: MusicAssistant, provider: str, path_or_url: str) -> None:
278    """
279    Remove the cached palette for an image so the next request re-extracts it.
280
281    :param mass: The MusicAssistant instance.
282    :param provider: Provider identifier for the image source.
283    :param path_or_url: Image path or URL (same format as get_palette).
284    """
285    await mass.cache.delete(key=create_thumb_hash(provider, path_or_url), provider=_CACHE_PROVIDER)
286
287
288async def get_palette_for_url(
289    mass: MusicAssistant, image_url: str | None
290) -> MediaItemPalette | None:
291    """Resolve an imageproxy URL to (path, provider) and return its palette."""
292    if not image_url:
293        return None
294    # /imageproxy/<id> form: async-resolve the id back to (provider, path).
295    if image_id := _extract_imageproxy_id(image_url):
296        resolved = await mass.metadata.resolve_image_id(image_id)
297        if resolved is None:
298            return None
299        provider, path = resolved
300    else:
301        path, provider = image_url, "builtin"
302    try:
303        return await get_palette(mass, path, provider)
304    except FileNotFoundError, OSError:
305        return None
306