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