music-assistant-server

1"""Controller for distributing audio analysis to providers."""
2
3from __future__ import annotations
4
5import asyncio
6import contextlib
7import dataclasses
8import logging
9import os
10import sys
11import time
12from collections.abc import AsyncGenerator, Iterable, Mapping
13from concurrent.futures import ThreadPoolExecutor
14from math import isfinite
15from typing import TYPE_CHECKING, Any
16
17from music_assistant_models.audio_analysis import AudioAnalysisCoverage
18from music_assistant_models.auth import Scope
19from music_assistant_models.background_task import TaskSchedule
20from music_assistant_models.enums import ContentType, MediaType, ProviderType, StreamType
21from music_assistant_models.errors import ProviderUnavailableError
22from music_assistant_models.media_items import AudioMetadata
23
24from music_assistant.constants import (
25    CONF_BACKGROUND_SCAN_CONCURRENCY,
26    DB_TABLE_AUDIO_ANALYSIS,
27    DB_TABLE_AUDIO_ANALYSIS_FAILURES,
28    DB_TABLE_PROVIDER_MAPPINGS,
29    DEFAULT_BACKGROUND_SCAN_CONCURRENCY,
30    LOUDNESS_MEASUREMENT_MIN_LUFS,
31    MASS_LOGGER_NAME,
32)
33from music_assistant.controllers.streams.audio_buffer import AudioBufferDiscarded, AudioBufferEOF
34from music_assistant.helpers.api import api_command
35from music_assistant.helpers.datetime import local_clock_time_to_utc, utc_timestamp
36from music_assistant.helpers.json import json_dumps, json_loads
37from music_assistant.helpers.util import inference_thread_budget, is_arm
38from music_assistant.models.audio_analysis import AudioAnalysisData
39from music_assistant.models.audio_analysis_provider import (
40    AudioAnalysisProvider,
41    InstrumentedSemaphore,
42)
43from music_assistant.models.music_provider import MusicProvider
44
45LOUDNESS_ANALYSIS_DOMAIN = "loudness_analysis"
46SMART_FADES_ANALYSIS_DOMAIN = "smart_fades"
47SONIC_ANALYSIS_DOMAIN = "sonic_analysis"
48# AA domains trusted for frontend-facing track data (bpm/key/waveform), authoritative first.
49TRACK_EXPORT_AA_PRIORITY = (SMART_FADES_ANALYSIS_DOMAIN, SONIC_ANALYSIS_DOMAIN)
50BACKGROUND_SCAN_TASK_ID = "audio_analysis_background_scan"
51BACKGROUND_PER_TRACK_TIMEOUT_SECONDS = 300
52BACKGROUND_PER_TRACK_TIMEOUT_DURATION_MULTIPLIER = 1.5
53# Per-run wall-clock cap; in-flight tracks finish, new ones defer to the next run.
54BACKGROUND_SCAN_RUN_BUDGET_SECONDS = 4 * 3600
55# Per-chunk processing ceiling for live and background analysis; a provider that exceeds it is
56# treated as stuck and evicted. Generous because analysis runs one offload at a time while a
57# player streams, so a chunk may wait behind other work before it computes.
58CHUNK_HANG_GUARD_SECONDS = 120.0
59# Floor on wall-seconds between consecutive background chunk dispatches (one chunk = one
60# audio-second), capping each scanned track at ~4x realtime so a background analyse doesn't
61# consume all resources. Nice and slow is preferred for nightly background scans.
62BACKGROUND_PACE_INTERVAL_SECONDS_FLOOR = 0.250
63# OS nice value for analysis worker threads (Linux): keeps analysis below playback so the
64# scheduler favors the event loop and ffmpeg under contention.
65ANALYSIS_THREAD_NICE = 10
66# Cap on concurrent realtime analysis sessions (the playing track plus the preloaded next).
67# Rapid track skipping would otherwise spawn an analysis per abandoned track; the oldest is
68# evicted to keep the count bounded.
69REALTIME_ANALYSIS_MAX_SESSIONS = 2
70# Minimum fraction of the expected track duration that must have been received before an
71# ended stream is finalized. A source that ends far short of it (e.g. a stream that died
72# without raising an error) is discarded instead, so no truncated analysis is persisted.
73ANALYSIS_MIN_COMPLETENESS_RATIO = 0.9
74# Free the heavy analysis models after this long with no analysis activity; they are reloaded
75# on the next track. Long enough that gaps between tracks/sessions don't thrash the reload.
76MODEL_IDLE_UNLOAD_SECONDS = 300
77MODEL_IDLE_CHECK_INTERVAL_SECONDS = 60
78FILESYSTEM_PROVIDER_DOMAINS: tuple[str, ...] = (
79    "filesystem_local",
80    "filesystem_smb",
81    "filesystem_nfs",
82)
83
84LOGGER = logging.getLogger(f"{MASS_LOGGER_NAME}.audio_analysis")
85
86if TYPE_CHECKING:
87    from datetime import datetime
88
89    from music_assistant_models.media_items import AudioFormat, Track
90    from music_assistant_models.streamdetails import StreamDetails
91
92    from music_assistant.controllers.streams.audio_buffer import AudioBuffer
93    from music_assistant.controllers.streams.controller import StreamsController
94
95
96def _get_row_value(row: Mapping[str, Any], key: str) -> Any:
97    """Return a database row value without assuming dict-only helpers."""
98    try:
99        return row[key]
100    except IndexError, KeyError, TypeError:
101        return None
102
103
104def _parse_row(
105    row: Mapping[str, Any],
106    unparsable_ids: list[Any] | None = None,
107) -> AudioAnalysisData | None:
108    """
109    Parse a single audio_analysis row's analysis_data, logging and skipping on error.
110
111    :param row: The audio_analysis row to parse.
112    :param unparsable_ids: When given, the id of a row that fails to parse is appended.
113    """
114    try:
115        return AudioAnalysisData.from_dict(json_loads(row["analysis_data"]))
116    except (IndexError, KeyError, TypeError, ValueError) as err:
117        row_id = _get_row_value(row, "id")
118        # the error itself may embed the full (huge) field value, so log only
119        # the error type plus the offending field name when available
120        error_detail = type(err).__name__
121        if field_name := getattr(err, "field_name", None):
122            error_detail = f"{error_detail} in field {field_name}"
123        LOGGER.warning(
124            "Skipping unparsable audio_analysis row (id=%s, domain=%s, error=%s)",
125            row_id,
126            _get_row_value(row, "aa_provider_domain"),
127            error_detail,
128        )
129        if unparsable_ids is not None and row_id is not None:
130            unparsable_ids.append(row_id)
131        return None
132
133
134def _merged_from_rows(
135    rows: Iterable[Mapping[str, Any]],
136    available_aa_domains: set[str],
137    priority: tuple[str, ...] | None = None,
138    unparsable_ids: list[Any] | None = None,
139) -> AudioAnalysisData | None:
140    """
141    Fold audio_analysis rows into one merged result.
142
143    Rows from AA providers not in available_aa_domains, and rows whose analysis_data
144    is unparsable, are always skipped. Returns None when no usable row remains.
145
146    :param rows: audio_analysis rows ordered oldest-first; each must carry
147        aa_provider_domain and analysis_data.
148    :param available_aa_domains: AA provider domains currently available.
149    :param priority: When None, merge all available providers' rows with latest-write-wins
150        (non-None fields). When a tuple of AA provider domains is given, only those domains
151        are considered and the first-listed domain wins each per-field conflict.
152    :param unparsable_ids: When given, ids of rows whose analysis_data fails to parse
153        are appended.
154    """
155    merged = AudioAnalysisData()
156    found = False
157    if priority is None:
158        for row in rows:
159            if row["aa_provider_domain"] not in available_aa_domains:
160                continue
161            if (row_data := _parse_row(row, unparsable_ids)) is None:
162                continue
163            merged.update(row_data)
164            found = True
165        return merged if found else None
166
167    # priority given: merge only these domains, first-listed wins each field.
168    wanted = tuple(d for d in priority if d in available_aa_domains)
169    wanted_set = set(wanted)
170    by_domain: dict[str, AudioAnalysisData] = {}
171    for row in rows:
172        domain = row["aa_provider_domain"]
173        if domain not in wanted_set or domain in by_domain:
174            continue
175        if (row_data := _parse_row(row, unparsable_ids)) is None:
176            continue
177        by_domain[domain] = row_data
178    for domain in reversed(wanted):
179        if (row_data := by_domain.get(domain)) is not None:
180            merged.update(row_data)
181            found = True
182    return merged if found else None
183
184
185def _first_non_finite_field(analysis: AudioAnalysisData) -> str | None:
186    """Return the name of the first float field holding a non-finite value, if any."""
187    for fld in dataclasses.fields(analysis):
188        value = getattr(analysis, fld.name)
189        if isinstance(value, float):
190            if not isfinite(value):
191                return fld.name
192        elif isinstance(value, list) and any(
193            isinstance(item, float) and not isfinite(item) for item in value
194        ):
195            return fld.name
196    return None
197
198
199def _nice_analysis_worker() -> None:
200    """
201    Lower the OS scheduling priority of the calling analysis worker thread.
202
203    Runs once per worker thread (ThreadPoolExecutor initializer). Linux-only, where the nice
204    value is per-thread and so affects just this pool; a no-op on other platforms.
205    """
206    if sys.platform != "linux" or not hasattr(os, "setpriority"):
207        return
208    with contextlib.suppress(OSError):
209        os.setpriority(os.PRIO_PROCESS, 0, ANALYSIS_THREAD_NICE)
210
211
212class AudioAnalysisController:
213    """Controller that distributes PCM chunks to all registered AudioAnalysisProviders."""
214
215    def __init__(self, streams: StreamsController) -> None:
216        """Initialize the AudioAnalysisController."""
217        self.streams = streams
218        self.mass = streams.mass
219        self.logger = self.mass.logger.getChild("audio_analysis")
220        self._active_sessions: dict[str, set[str]] = {}
221        self._workers: dict[str, asyncio.Task[None]] = {}
222        # Realtime session key -> queue id, insertion-ordered, so the session cap is applied
223        # per queue (concurrent queues don't evict each other's still-playing analysis).
224        self._session_queues: dict[str, str] = {}
225        self._inference_runtime_configured = False
226        # Bounds how many analysis offloads run concurrently to half the cores; created in
227        # ensure_inference_runtime_configured once the core count is known (None until then),
228        # and honored by AudioAnalysisProvider._run_offloaded.
229        self.analysis_semaphore: InstrumentedSemaphore | None = None
230        # Held by an analysis offload while any player streams, capping analysis to one offload
231        # at a time; honored by AudioAnalysisProvider._run_offloaded.
232        self.analysis_solo_lock: asyncio.Lock | None = None
233        # Niced worker pool that runs analysis offloads, so the lower priority applies to
234        # analysis threads only; created in ensure_inference_runtime_configured.
235        self.analysis_executor: ThreadPoolExecutor | None = None
236        # Monotonic time of the last analysis start, and the monitor that unloads idle models.
237        self._last_analysis_activity: float = 0.0
238        self._idle_unload_task: asyncio.Task[None] | None = None
239
240    def setup(self) -> None:
241        """Register the nightly background scan task."""
242        utc_hour, utc_minute = local_clock_time_to_utc(0, 0)
243        self.mass.tasks.register_scheduled_task(
244            task_id=BACKGROUND_SCAN_TASK_ID,
245            name="Audio analysis — background scan of local files",
246            handler=self._run_background_scan,
247            schedule=TaskSchedule.daily(hour=utc_hour, minute=utc_minute),
248            metadata={"task_domain": "audio_analysis"},
249            allow_retry=True,
250        )
251
252    async def close(self) -> None:
253        """Drain in-flight sessions and chunk workers on shutdown."""
254        tasks = list(self._workers.values())
255        self._workers.clear()
256        if self._idle_unload_task is not None:
257            tasks.append(self._idle_unload_task)
258            self._idle_unload_task = None
259        for task in tasks:
260            if not task.done():
261                task.cancel()
262        for session_key in list(self._active_sessions):
263            self._cancel_providers(session_key)
264        if tasks:
265            await asyncio.gather(*tasks, return_exceptions=True)
266        if self.analysis_executor is not None:
267            # A running CPU-bound thread can't be cancelled, so shut down without waiting on it.
268            self.analysis_executor.shutdown(wait=False, cancel_futures=True)
269            self.analysis_executor = None
270
271    def ensure_inference_runtime_configured(self) -> None:
272        """
273        Configure the on-device inference runtime for analysis (process-wide, applied once).
274
275        Torch-backed analysis providers call this at the start of their handle_async_init,
276        before loading their models.
277        """
278        if self._inference_runtime_configured:
279            return
280        # Lazy import: only torch-backed providers call this, so a host running no such
281        # provider never imports torch. Running before the first model load also lets
282        # set_num_interop_threads take effect (only settable before the first op).
283        import torch  # noqa: PLC0415
284
285        budget = self._aa_thread_budget()
286        torch.set_num_threads(budget)
287        with contextlib.suppress(RuntimeError):
288            # set_num_interop_threads can only be called before the first torch op
289            torch.set_num_interop_threads(1)
290        # torch.set_num_threads only governs torch's own ops. The per-block librosa/numpy
291        # feature extraction runs through the native BLAS pool (OpenBLAS), which is capped to
292        # the same budget from the environment at process start (cap_native_thread_pools);
293        # it cannot be capped from here without deadlocking against a concurrent import.
294        arm = is_arm()
295        if arm:
296            # NNPACK frequently fails to initialize on ARM SBCs (e.g. Raspberry Pi); torch
297            # then re-logs "Could not initialize NNPACK" to stderr on every conv op. The fp32
298            # conv fallback is used on those hosts regardless, so disabling it only removes
299            # the log spam.
300            with contextlib.suppress(RuntimeError):
301                torch.backends.nnpack.set_flags(False)  # type: ignore[no-untyped-call]
302        # Cap concurrent analysis offloads to half the cores so analysis (live or background)
303        # never occupies the whole box and starves playback/the host — slow and steady on any
304        # machine. Applies to every host; honored by AudioAnalysisProvider._run_offloaded.
305        concurrency_cap = max(1, self._cpu_count() // 2)
306        self.analysis_semaphore = InstrumentedSemaphore(concurrency_cap)
307        self.analysis_solo_lock = asyncio.Lock()
308        # Niced pool sized to the idle cap plus headroom; the semaphore and solo lock bound
309        # how many of its threads run at once.
310        self.analysis_executor = ThreadPoolExecutor(
311            max_workers=max(2, self._cpu_count()),
312            thread_name_prefix="analysis",
313            initializer=_nice_analysis_worker,
314        )
315        self.logger.info(
316            "AudioAnalysis runtime: torch intra=%d interop=%d, blas<=%s, "
317            "analysis concurrency<=%d (1 while a player streams), nnpack=%s",
318            torch.get_num_threads(),
319            torch.get_num_interop_threads(),
320            os.environ.get("OPENBLAS_NUM_THREADS", "uncapped"),
321            concurrency_cap,
322            "off" if arm else "on",
323        )
324        # Only mark done once configuration actually succeeded, so a failure retries.
325        self._inference_runtime_configured = True
326
327    @property
328    def providers(self) -> list[AudioAnalysisProvider]:
329        """Return all available audio analysis providers."""
330        return [
331            prov
332            for prov in self.mass.get_providers(ProviderType.AUDIO_ANALYSIS)
333            if isinstance(prov, AudioAnalysisProvider) and prov.available
334        ]
335
336    @property
337    def smart_fades_provider_available(self) -> bool:
338        """Return whether the smart fades audio analysis provider is loaded and available."""
339        return any(prov.domain == SMART_FADES_ANALYSIS_DOMAIN for prov in self.providers)
340
341    def playback_active(self) -> bool:
342        """Return whether a queue stream is actively serving a player right now."""
343        return self.streams.output_stream_active()
344
345    async def start_analysis(
346        self,
347        audio_buffer: AudioBuffer,
348        streamdetails: StreamDetails,
349    ) -> None:
350        """
351        Start analysis session for a track across all providers.
352
353        :param audio_buffer: The shared playback AudioBuffer the analysis reads PCM from.
354        :param streamdetails: The stream details for the item being analyzed.
355        """
356        providers = self.providers
357        if not providers:
358            self.logger.debug("No audio analysis providers available")
359            return
360
361        session_key = streamdetails.uri
362
363        # Skip if another queue already has an analysis running for the same item
364        if session_key in self._active_sessions:
365            self.logger.debug(
366                "Analysis session already active for %s, ignoring start request",
367                session_key,
368            )
369            return
370
371        provider_ids = await self._start_analysis_on_providers(
372            session_key, streamdetails, audio_buffer.pcm_format, providers
373        )
374        if not provider_ids:
375            self.logger.debug("No providers accepted analysis for %s", session_key)
376            return
377
378        # Bound concurrent realtime sessions per queue, evicting the oldest in this queue (the
379        # current track and its preloaded next are the youngest, so they survive a burst of
380        # skips). Scoping per queue keeps simultaneous queues from evicting each other.
381        queue_id = streamdetails.queue_id or session_key
382        in_queue = [key for key, qid in self._session_queues.items() if qid == queue_id]
383        for stale_key in in_queue[: max(0, len(in_queue) - REALTIME_ANALYSIS_MAX_SESSIONS + 1)]:
384            self._evict_realtime_session(stale_key)
385
386        self._active_sessions[session_key] = provider_ids
387        self._session_queues[session_key] = queue_id
388        worker = self.mass.create_task(
389            self._buffer_reader_worker(session_key, audio_buffer, streamdetails.duration)
390        )
391        self._workers[session_key] = worker
392
393        def _on_cancel() -> None:
394            # Buffer torn down (track skipped / inactivity) — free the session.
395            self._evict_realtime_session(session_key)
396
397        audio_buffer.register_cancel_callback(_on_cancel)
398
399    async def set_audio_analysis(
400        self,
401        item_id: str,
402        provider_instance_id_or_domain: str,
403        aa_provider_domain: str,
404        analysis: AudioAnalysisData,
405        analysis_version: int = 1,
406        media_type: MediaType = MediaType.TRACK,
407    ) -> None:
408        """
409        Store audio analysis results from an Audio Analysis provider.
410
411        :param item_id: Provider-native item ID from streamdetails.item_id.
412        :param provider_instance_id_or_domain: Music provider instance ID or domain.
413        :param aa_provider_domain: Domain of the AA provider that produced the data.
414        :param analysis: The analysis data to store.
415        :param analysis_version: Version of the AA provider's algorithm.
416        :param media_type: The media type of the item being analyzed.
417        :raises ValueError: When a float field of the analysis holds a non-finite value.
418        """
419        # non-finite floats serialize to JSON null, which corrupts the stored row;
420        # refuse them here so a bad payload can never poison the database
421        if (field_name := _first_non_finite_field(analysis)) is not None:
422            raise ValueError(
423                f"audio analysis for {item_id} contains a non-finite value in {field_name}"
424            )
425        provider = self.mass.get_provider(provider_instance_id_or_domain)
426        if not isinstance(provider, MusicProvider):
427            return
428        prov_key = provider.domain if provider.is_streaming_provider else provider.instance_id
429        data_json = json_dumps(analysis.to_dict())
430        await self.mass.music.database.insert_or_replace(
431            DB_TABLE_AUDIO_ANALYSIS,
432            {
433                "media_type": media_type.value,
434                "item_id": item_id,
435                "provider": prov_key,
436                "aa_provider_domain": aa_provider_domain,
437                "analysis_data": data_json,
438                "analysis_version": analysis_version,
439            },
440        )
441        await self.clear_analysis_failure(
442            item_id=item_id,
443            provider_instance_id_or_domain=provider_instance_id_or_domain,
444            aa_provider_domain=aa_provider_domain,
445            media_type=media_type,
446        )
447
448    async def record_analysis_failure(
449        self,
450        item_id: str,
451        provider_instance_id_or_domain: str,
452        aa_provider_domain: str,
453        reason: str,
454        retry_at: datetime | None = None,
455        analysis_version: int = 1,
456        media_type: MediaType = MediaType.TRACK,
457    ) -> None:
458        """
459        Record an analysis failure for a track.
460
461        No-op when the provider does not resolve to a loaded music provider.
462
463        :param item_id: Provider-native item ID from streamdetails.item_id.
464        :param provider_instance_id_or_domain: Music provider instance ID or domain.
465        :param aa_provider_domain: Domain of the AA provider that failed.
466        :param reason: Human-readable failure reason.
467        :param retry_at: Timezone-aware datetime when to allow a retry; None (default)
468            means never auto-retry.
469        :param analysis_version: The AA provider's algorithm version at failure time.
470        :param media_type: The media type of the item.
471        """
472        provider = self.mass.get_provider(provider_instance_id_or_domain)
473        if not isinstance(provider, MusicProvider):
474            self.logger.debug(
475                "Skipping failure record for %s: not a loaded music provider",
476                provider_instance_id_or_domain,
477            )
478            return
479        prov_key = provider.domain if provider.is_streaming_provider else provider.instance_id
480        await self.mass.music.database.insert_or_replace(
481            DB_TABLE_AUDIO_ANALYSIS_FAILURES,
482            {
483                "media_type": media_type.value,
484                "item_id": item_id,
485                "provider": prov_key,
486                "aa_provider_domain": aa_provider_domain,
487                "reason": reason,
488                "analysis_version": analysis_version,
489                "next_retry": int(retry_at.timestamp()) if retry_at is not None else None,
490            },
491        )
492
493    async def clear_analysis_failure(
494        self,
495        item_id: str,
496        provider_instance_id_or_domain: str,
497        aa_provider_domain: str,
498        media_type: MediaType = MediaType.TRACK,
499    ) -> None:
500        """
501        Delete a recorded analysis failure (e.g. after a later success).
502
503        No-op when the provider does not resolve to a loaded music provider.
504
505        :param item_id: Provider-native item ID from streamdetails.item_id.
506        :param provider_instance_id_or_domain: Music provider instance ID or domain.
507        :param aa_provider_domain: Domain of the AA provider whose failure to clear.
508        :param media_type: The media type of the item.
509        """
510        provider = self.mass.get_provider(provider_instance_id_or_domain)
511        if not isinstance(provider, MusicProvider):
512            self.logger.debug(
513                "Skipping failure clear for %s: not a loaded music provider",
514                provider_instance_id_or_domain,
515            )
516            return
517        prov_key = provider.domain if provider.is_streaming_provider else provider.instance_id
518        await self.mass.music.database.delete(
519            DB_TABLE_AUDIO_ANALYSIS_FAILURES,
520            {
521                "item_id": item_id,
522                "provider": prov_key,
523                "aa_provider_domain": aa_provider_domain,
524                "media_type": media_type.value,
525            },
526        )
527
528    async def get_audio_analysis(
529        self,
530        item_id: str,
531        provider_instance_id_or_domain: str,
532        media_type: MediaType = MediaType.TRACK,
533        priority: tuple[str, ...] | None = None,
534    ) -> AudioAnalysisData | None:
535        """
536        Get merged audio analysis data for a track.
537
538        Only rows from currently available AA providers are included. Rows that fail
539        to parse are deleted, so the track can be re-analyzed.
540
541        :param item_id: Provider-native item ID from streamdetails.item_id.
542        :param provider_instance_id_or_domain: Music provider instance ID or domain.
543        :param media_type: The media type of the item.
544        :param priority: AA provider domains the values must come from. When None, all
545            available providers are merged latest-write-wins. With a single domain, only
546            that provider's values are used. With multiple domains, only those are merged
547            and the first-listed domain wins each per-field conflict. Use this when a field
548            (e.g. loudness_integrated) is written by several providers with different
549            semantics, so the authoritative source is selected.
550        """
551        provider = self.mass.get_provider(provider_instance_id_or_domain)
552        if not isinstance(provider, MusicProvider):
553            return None
554        prov_key = provider.domain if provider.is_streaming_provider else provider.instance_id
555        rows = await self.mass.music.database.get_rows(
556            DB_TABLE_AUDIO_ANALYSIS,
557            {
558                "item_id": item_id,
559                "provider": prov_key,
560                "media_type": media_type.value,
561            },
562            order_by="timestamp_created ASC",
563        )
564        if not rows:
565            return None
566
567        available_aa_domains = {
568            p.domain for p in self.mass.get_providers(ProviderType.AUDIO_ANALYSIS) if p.available
569        }
570        unparsable_ids: list[Any] = []
571        merged = _merged_from_rows(rows, available_aa_domains, priority, unparsable_ids)
572        # corrupt rows would otherwise block re-analysis forever (their stored
573        # analysis_version still gates new sessions), so drop them right away
574        for row_id in unparsable_ids:
575            await self.mass.music.database.delete(DB_TABLE_AUDIO_ANALYSIS, {"id": row_id})
576        if unparsable_ids:
577            self.logger.info(
578                "Deleted %d corrupt audio_analysis row(s) for %s/%s; "
579                "the item is eligible for re-analysis",
580                len(unparsable_ids),
581                prov_key,
582                item_id,
583            )
584        return merged
585
586    async def get_track_audio_metadata(self, track: Track) -> AudioMetadata | None:
587        """
588        Return AudioMetadata (bpm, musical key) for a track, or None when no analysis exists.
589
590        Provider mappings are tried best-quality first; per field the Smart Fades AA
591        provider is preferred over other AA providers.
592
593        :param track: The track to look up stored analysis data for.
594        """
595        priority = TRACK_EXPORT_AA_PRIORITY
596        for mapping in sorted(track.provider_mappings, key=lambda m: m.quality, reverse=True):
597            analysis = await self.get_audio_analysis(
598                mapping.item_id, mapping.provider_instance, priority=priority
599            )
600            if analysis is None or (analysis.bpm is None and analysis.key is None):
601                continue
602            musical_key: str | None = None
603            if analysis.key is not None:
604                musical_key = f"{analysis.key} {analysis.mode}" if analysis.mode else analysis.key
605            return AudioMetadata(bpm=analysis.bpm, musical_key=musical_key)
606        return None
607
608    @api_command("audio_analysis/wave_form")
609    async def get_wave_form(
610        self,
611        item_id: str,
612        provider_instance_id_or_domain: str,
613    ) -> list[float] | None:
614        """
615        Return the RMS energy waveform for a track, or None when no analysis exists.
616
617        The waveform is a fixed array of 1800 bins (normalized 0.0-1.0) evenly covering
618        the track duration. Values come from the Smart Fades AA provider when available,
619        falling back to any other AA provider that stored RMS energy.
620
621        :param item_id: Provider-native item ID.
622        :param provider_instance_id_or_domain: Music provider instance ID or domain.
623        """
624        analysis = await self.get_audio_analysis(
625            item_id,
626            provider_instance_id_or_domain,
627            priority=TRACK_EXPORT_AA_PRIORITY,
628        )
629        if analysis is None or analysis.rms_energy is None:
630            return None
631        return [float(value) for value in analysis.rms_energy]
632
633    async def set_track_loudness(
634        self,
635        item_id: str,
636        provider_instance_id_or_domain: str,
637        loudness: float,
638        loudness_album: float | None = None,
639        media_type: MediaType = MediaType.TRACK,
640    ) -> None:
641        """
642        Store track loudness measurement from an external source (tags, ReplayGain, etc).
643
644        Persists the loudness values under the builtin loudness_analysis provider so
645        the runtime ebur128 analysis will not re-analyze the track on playback.
646
647        :param item_id: Provider-native item ID.
648        :param provider_instance_id_or_domain: Music provider instance ID or domain.
649        :param loudness: Integrated track loudness in LUFS.
650        :param loudness_album: Optional album-level integrated loudness in LUFS.
651        :param media_type: The media type of the item.
652        """
653        if loudness is None or not isfinite(loudness) or loudness <= LOUDNESS_MEASUREMENT_MIN_LUFS:
654            return
655        if (
656            loudness_album is None
657            or not isfinite(loudness_album)
658            or loudness_album <= LOUDNESS_MEASUREMENT_MIN_LUFS
659        ):
660            loudness_album = None
661        analysis = AudioAnalysisData(
662            loudness_integrated=loudness,
663            loudness_album=loudness_album,
664        )
665        await self.set_audio_analysis(
666            item_id=item_id,
667            provider_instance_id_or_domain=provider_instance_id_or_domain,
668            aa_provider_domain=LOUDNESS_ANALYSIS_DOMAIN,
669            analysis=analysis,
670            media_type=media_type,
671        )
672
673    async def get_extra_data_for_album_tracks(
674        self,
675        track_item_ids: list[str],
676        provider_instance_id_or_domain: str,
677        aa_provider_domain: str,
678    ) -> list[dict[str, Any]]:
679        """
680        Return one AA provider's ``extra_data`` for each given track that has one.
681
682        :param track_item_ids: Provider-native track IDs to look up.
683        :param provider_instance_id_or_domain: Music provider instance ID or domain.
684        :param aa_provider_domain: Domain of the AA provider whose rows to fetch.
685        """
686        if not track_item_ids:
687            return []
688        provider = self.mass.get_provider(
689            provider_instance_id_or_domain, provider_type=MusicProvider
690        )
691        if provider is None:
692            return []
693        prov_key = provider.domain if provider.is_streaming_provider else provider.instance_id
694
695        placeholders = ",".join(f":id{i}" for i in range(len(track_item_ids)))
696        params: dict[str, Any] = {f"id{i}": tid for i, tid in enumerate(track_item_ids)}
697        params["provider"] = prov_key
698        params["domain"] = aa_provider_domain
699        params["media_type"] = MediaType.TRACK.value
700
701        query = (
702            f"SELECT analysis_data FROM {DB_TABLE_AUDIO_ANALYSIS} "
703            f"WHERE aa_provider_domain = :domain "
704            f"AND media_type = :media_type "
705            f"AND provider = :provider "
706            f"AND item_id IN ({placeholders})"
707        )
708        rows = await self.mass.music.database.get_rows_from_query(
709            query, params, limit=len(track_item_ids)
710        )
711
712        results: list[dict[str, Any]] = []
713        for row in rows:
714            try:
715                data = json_loads(row["analysis_data"])
716            except ValueError, TypeError:
717                continue
718            if not isinstance(data, dict):
719                continue
720            extra = data.get("extra_data")
721            if isinstance(extra, dict):
722                results.append(extra)
723        return results
724
725    async def get_audio_analysis_version(
726        self,
727        item_id: str,
728        provider_instance_id_or_domain: str,
729        aa_provider_domain: str,
730        media_type: MediaType = MediaType.TRACK,
731    ) -> int | None:
732        """
733        Get the stored analysis version for a specific AA provider and track.
734
735        :param item_id: Provider-native item ID from streamdetails.item_id.
736        :param provider_instance_id_or_domain: Music provider instance ID or domain.
737        :param aa_provider_domain: Domain of the AA provider.
738        :param media_type: The media type of the item.
739        """
740        provider = self.mass.get_provider(provider_instance_id_or_domain)
741        if not isinstance(provider, MusicProvider):
742            return None
743        prov_key = provider.domain if provider.is_streaming_provider else provider.instance_id
744        row = await self.mass.music.database.get_row(
745            DB_TABLE_AUDIO_ANALYSIS,
746            {
747                "item_id": item_id,
748                "provider": prov_key,
749                "aa_provider_domain": aa_provider_domain,
750                "media_type": media_type.value,
751            },
752        )
753        if not row:
754            return None
755        return int(row["analysis_version"])
756
757    async def get_audio_analysis_count(
758        self,
759        aa_provider_domain: str,
760        media_type: MediaType = MediaType.TRACK,
761    ) -> int:
762        """
763        Count audio_analysis rows for a given aa_provider_domain.
764
765        :param aa_provider_domain: Domain of the AA provider whose rows to count.
766        :param media_type: The media type to count rows for.
767        """
768        return await self.mass.music.database.get_count_from_query(
769            f"SELECT id FROM {DB_TABLE_AUDIO_ANALYSIS} "
770            f"WHERE aa_provider_domain = :aa_provider_domain AND media_type = :media_type",
771            {"aa_provider_domain": aa_provider_domain, "media_type": media_type.value},
772        )
773
774    async def iter_audio_analysis_rows(
775        self,
776        aa_provider_domain: str,
777        media_type: MediaType = MediaType.TRACK,
778    ) -> AsyncGenerator[Mapping[str, Any]]:
779        """
780        Stream audio_analysis rows for a given aa_provider_domain.
781
782        :param aa_provider_domain: Domain of the AA provider whose rows to yield.
783        :param media_type: The media type to filter rows by.
784        """
785        query = (
786            f"SELECT * FROM {DB_TABLE_AUDIO_ANALYSIS} "
787            f"WHERE aa_provider_domain = :aa_provider_domain AND media_type = :media_type"
788        )
789        async for row in self.mass.music.database.iter_rows_from_query(
790            query,
791            {"aa_provider_domain": aa_provider_domain, "media_type": media_type.value},
792        ):
793            yield row
794
795    async def iter_merged_audio_analysis_rows(
796        self,
797        primary_aa_domain: str,
798        media_type: MediaType = MediaType.TRACK,
799        priority: tuple[str, ...] | None = None,
800    ) -> AsyncGenerator[tuple[str, str, AudioAnalysisData]]:
801        """
802        Yield one merged AudioAnalysisData per track present in primary_aa_domain.
803
804        Unlike get_audio_analysis, the music provider need not be loaded — rows
805        are merged purely from the database, gated only on AA-provider
806        availability. Used by bulk consumers (e.g. similarity index rebuild).
807
808        Rows are streamed and grouped on the fly: only the rows for the
809        currently-folding (item_id, provider) pair are held in memory at once,
810        so peak memory is proportional to one track, not the whole library.
811
812        If primary_aa_domain is not currently available, no rows can satisfy
813        the availability gate and the generator yields nothing (a WARNING is
814        logged so callers can distinguish "offline" from "empty").
815
816        :param primary_aa_domain: AA provider domain that defines the universe of
817            tracks to yield. Only (item_id, provider) pairs with at least one
818            row in this domain are emitted.
819        :param media_type: The media type to filter on.
820        :param priority: AA provider domains the merged values must come from, first-listed
821            wins per-field conflicts (see get_audio_analysis). When None, all available
822            providers are merged latest-write-wins.
823        """
824        available_aa_domains = {
825            p.domain for p in self.mass.get_providers(ProviderType.AUDIO_ANALYSIS) if p.available
826        }
827        if primary_aa_domain not in available_aa_domains:
828            LOGGER.warning(
829                "iter_merged_audio_analysis_rows called with offline primary AA domain "
830                "%r; yielding no rows. Available domains: %s",
831                primary_aa_domain,
832                sorted(available_aa_domains),
833            )
834            return
835        # EXISTS subquery scopes to the primary domain's universe at the DB level;
836        # ORDER BY (item_id, provider, ts) lets us fold each track in one streaming pass.
837        query = (
838            f"SELECT item_id, provider, aa_provider_domain, analysis_data, id "
839            f"FROM {DB_TABLE_AUDIO_ANALYSIS} aa1 "
840            f"WHERE aa1.media_type = :media_type "
841            f"AND EXISTS ("
842            f"    SELECT 1 FROM {DB_TABLE_AUDIO_ANALYSIS} aa2 "
843            f"    WHERE aa2.item_id = aa1.item_id "
844            f"    AND aa2.provider = aa1.provider "
845            f"    AND aa2.aa_provider_domain = :primary_aa_domain "
846            f"    AND aa2.media_type = :media_type"
847            f") "
848            f"ORDER BY aa1.item_id, aa1.provider, aa1.timestamp_created ASC"
849        )
850        current_key: tuple[str, str] | None = None
851        current_group: list[Mapping[str, Any]] = []
852        async for row in self.mass.music.database.iter_rows_from_query(
853            query,
854            {"media_type": media_type.value, "primary_aa_domain": primary_aa_domain},
855        ):
856            key = (row["item_id"], row["provider"])
857            if current_key is not None and key != current_key:
858                merged = _merged_from_rows(current_group, available_aa_domains, priority)
859                if merged is not None:
860                    yield (*current_key, merged)
861                current_group = []
862            current_key = key
863            current_group.append(row)
864        if current_key is not None:
865            merged = _merged_from_rows(current_group, available_aa_domains, priority)
866            if merged is not None:
867                yield (*current_key, merged)
868
869    @api_command("audio_analysis/coverage", required_scope=Scope.SYSTEM_MANAGE)
870    async def get_coverage(self, aa_domain: str) -> AudioAnalysisCoverage:
871        """
872        Return analysis-coverage health counts for an AA provider.
873
874        :param aa_domain: AA provider domain to query.
875        :returns: Counts where ``pending`` reflects filesystem-source tracks only;
876            streaming-provider tracks are never considered for background analysis
877            and are excluded.
878        """
879        provider = self.mass.get_provider(
880            aa_domain,
881            provider_type=AudioAnalysisProvider,  # type: ignore[type-abstract]
882        )
883        if provider is None:
884            raise ProviderUnavailableError(f"{aa_domain} is not available")
885
886        analyzed = await self.get_audio_analysis_count(aa_domain)
887        pending = await self._count_candidates_missing_analysis(
888            aa_domain, provider.analysis_version
889        )
890        # NULL analysis_version (pre-versioning rows) is treated as stale: SQLite
891        # evaluates `NULL < N` as NULL (falsy), so it must be matched explicitly.
892        stale_query = (
893            f"SELECT id FROM {DB_TABLE_AUDIO_ANALYSIS} "
894            f"WHERE aa_provider_domain = :aa_domain "
895            f"  AND media_type = :media_type "
896            f"  AND (analysis_version IS NULL OR analysis_version < :current_version)"
897        )
898        stale_version = await self.mass.music.database.get_count_from_query(
899            stale_query,
900            {
901                "aa_domain": aa_domain,
902                "media_type": MediaType.TRACK.value,
903                "current_version": provider.analysis_version,
904            },
905        )
906        return AudioAnalysisCoverage(
907            analyzed=analyzed,
908            pending=pending,
909            stale_version=stale_version,
910            analysis_version=provider.analysis_version,
911        )
912
913    @api_command("audio_analysis/failures", required_scope=Scope.SYSTEM_MANAGE)
914    async def get_failures(self, aa_domain: str | None = None) -> list[dict[str, Any]]:
915        """
916        Return recorded analysis failures, optionally filtered by AA provider domain.
917
918        :param aa_domain: When given, only failures for this AA provider domain are returned.
919        """
920        match = {"aa_provider_domain": aa_domain} if aa_domain is not None else None
921        rows = await self.mass.music.database.get_rows(
922            DB_TABLE_AUDIO_ANALYSIS_FAILURES, match, limit=0
923        )
924        return [
925            {
926                "item_id": r["item_id"],
927                "provider": r["provider"],
928                "aa_provider_domain": r["aa_provider_domain"],
929                "reason": r["reason"],
930                "next_retry": r["next_retry"],
931                "timestamp_created": r["timestamp_created"],
932            }
933            for r in rows
934        ]
935
936    @api_command("audio_analysis/failures/clear", required_scope=Scope.SYSTEM_MANAGE)
937    async def clear_failures(
938        self,
939        item_id: str | None = None,
940        provider: str | None = None,
941        aa_domain: str | None = None,
942    ) -> int:
943        """
944        Delete recorded failures matching the given filters; returns the number deleted.
945
946        At least one filter is required; a call with all filters None deletes nothing.
947
948        :param item_id: Provider-native item ID to clear.
949        :param provider: Stored music-provider key (domain or instance_id) to clear.
950        :param aa_domain: AA provider domain to clear.
951        """
952        match: dict[str, Any] = {}
953        if item_id is not None:
954            match["item_id"] = item_id
955        if provider is not None:
956            match["provider"] = provider
957        if aa_domain is not None:
958            match["aa_provider_domain"] = aa_domain
959        if not match:
960            return 0
961        rows = await self.mass.music.database.get_rows(
962            DB_TABLE_AUDIO_ANALYSIS_FAILURES, match, limit=0
963        )
964        count = len(rows)
965        if count:
966            await self.mass.music.database.delete(DB_TABLE_AUDIO_ANALYSIS_FAILURES, match)
967        return count
968
969    async def _run_background_scan(self) -> None:
970        """Run the scan as decode-once-fan-out streaming over candidate tracks."""
971        providers = self.providers
972        if not providers:
973            return
974
975        provider_versions = {p.domain: p.analysis_version for p in providers}
976        candidates = await self._find_candidates_missing_analysis(provider_versions, limit=0)
977        if not candidates:
978            return
979
980        scan_started = time.monotonic()
981        run_deadline = scan_started + BACKGROUND_SCAN_RUN_BUDGET_SECONDS
982        self.logger.info(
983            "Background analysis (streaming): %d track(s) pending across %d provider(s); "
984            "run budget %.1fh",
985            len(candidates),
986            len(providers),
987            BACKGROUND_SCAN_RUN_BUDGET_SECONDS / 3600,
988        )
989
990        concurrency = self._get_scan_concurrency()
991        semaphore = asyncio.Semaphore(concurrency)
992        provider_by_domain = {p.domain: p for p in providers}
993
994        processed = 0
995        deferred = 0
996
997        async def _run_one(candidate: dict[str, Any]) -> None:
998            nonlocal processed, deferred
999            async with semaphore:
1000                if time.monotonic() >= run_deadline:
1001                    deferred += 1
1002                    return
1003
1004                item_id = candidate["item_id"]
1005                provider_instance = candidate["provider_instance"]
1006                missing = candidate["missing_domains"]
1007
1008                music_prov = self.mass.get_provider(provider_instance, provider_type=MusicProvider)
1009                if music_prov is None or not music_prov.available:
1010                    self.logger.debug(
1011                        "Skipping %s: music provider %s unavailable", item_id, provider_instance
1012                    )
1013                    return
1014
1015                try:
1016                    streamdetails = await music_prov.get_stream_details(item_id, MediaType.TRACK)
1017                except Exception as err:
1018                    # Provider method with an open-ended failure surface; any failure
1019                    # just skips this scan candidate.
1020                    self.logger.debug("Skipping %s: stream details failed: %s", item_id, err)
1021                    return
1022
1023                if streamdetails.stream_type != StreamType.LOCAL_FILE:
1024                    return
1025                if not isinstance(streamdetails.path, str) or not streamdetails.path:
1026                    return
1027
1028                providers_for_track = [
1029                    p
1030                    for p in (provider_by_domain.get(d) for d in missing)
1031                    if p is not None and p.available
1032                ]
1033                if not providers_for_track:
1034                    return
1035
1036                await self._run_background_streaming_for_track(
1037                    streamdetails,
1038                    providers_for_track,
1039                )
1040                processed += 1
1041
1042        await asyncio.gather(*(_run_one(c) for c in candidates))
1043
1044        elapsed = time.monotonic() - scan_started
1045        if deferred:
1046            self.logger.info(
1047                "Background analysis: run-budget reached "
1048                "(%d processed, %d deferred to next run, %.1fs elapsed)",
1049                processed,
1050                deferred,
1051                elapsed,
1052            )
1053        else:
1054            self.logger.info(
1055                "Background analysis: complete (%d candidates processed in %.1fs)",
1056                processed,
1057                elapsed,
1058            )
1059
1060    async def _run_background_streaming_for_track(
1061        self,
1062        streamdetails: StreamDetails,
1063        providers: list[AudioAnalysisProvider],
1064    ) -> None:
1065        """
1066        Run a single track through the streaming pipeline using ffmpeg as the source.
1067
1068        :param streamdetails: Stream details for the track being analyzed.
1069        :param providers: Audio analysis providers to dispatch chunks to.
1070        """
1071        session_key = streamdetails.uri
1072        if session_key in self._active_sessions:
1073            self.logger.debug(
1074                "Background streaming: session already active for %s, skipping", session_key
1075            )
1076            return
1077
1078        # Floor at the fixed budget so short tracks keep ffmpeg-startup headroom.
1079        timeout_seconds = max(
1080            BACKGROUND_PER_TRACK_TIMEOUT_SECONDS,
1081            int((streamdetails.duration or 0) * BACKGROUND_PER_TRACK_TIMEOUT_DURATION_MULTIPLIER),
1082        )
1083
1084        try:
1085            await asyncio.wait_for(
1086                self._run_background_streaming_inner(session_key, streamdetails, providers),
1087                timeout=timeout_seconds,
1088            )
1089        except asyncio.CancelledError:
1090            # CancelledError inherits from BaseException — the broad except below
1091            # does not catch it. Clean up the session, then re-raise.
1092            self.logger.debug("Background analysis cancelled for %s", session_key)
1093            self._cancel_providers(session_key)
1094            raise
1095        except TimeoutError:
1096            self.logger.warning(
1097                "Background analysis exceeded %ds budget for %s, skipping",
1098                timeout_seconds,
1099                session_key,
1100            )
1101            self._cancel_providers(session_key)
1102            self.mass.tasks.add_task_failure(
1103                BACKGROUND_SCAN_TASK_ID,
1104                f"Timed out after {timeout_seconds}s: {session_key}",
1105            )
1106        except Exception as err:
1107            self.logger.warning("Background analysis failed for %s: %s", session_key, err)
1108            self._cancel_providers(session_key)
1109            self.mass.tasks.add_task_failure(
1110                BACKGROUND_SCAN_TASK_ID,
1111                f"Failed: {session_key}: {err}",
1112            )
1113
1114    async def _run_background_streaming_inner(
1115        self,
1116        session_key: str,
1117        streamdetails: StreamDetails,
1118        providers: list[AudioAnalysisProvider],
1119    ) -> None:
1120        """
1121        Inner body of _run_background_streaming_for_track, wrapped by wait_for.
1122
1123        :param session_key: Active-session key for this track.
1124        :param streamdetails: Stream details for the track being analyzed.
1125        :param providers: Audio analysis providers to dispatch chunks to.
1126        """
1127        if not isinstance(streamdetails.path, str) or not streamdetails.path:
1128            return
1129
1130        # Override content_type so ffmpeg decodes rather than re-muxing the source codec.
1131        pcm_format = dataclasses.replace(
1132            streamdetails.audio_format,
1133            content_type=ContentType.from_bit_depth(streamdetails.audio_format.bit_depth),
1134        )
1135
1136        accepted = await self._start_analysis_on_providers(
1137            session_key, streamdetails, pcm_format, providers
1138        )
1139        if not accepted:
1140            self.logger.debug("No providers accepted background analysis for %s", session_key)
1141            return
1142        self._active_sessions[session_key] = accepted
1143
1144        audio_source = self.mass.streams.audio.get_media_stream(streamdetails, pcm_format)
1145        next_allowed = time.monotonic()
1146        # aclosing guarantees the source (and any provider stream slot it holds)
1147        # is released promptly when the loop breaks out early
1148        async with contextlib.aclosing(audio_source):
1149            async for chunk in audio_source:
1150                if session_key not in self._active_sessions:
1151                    # all providers evicted — bail early
1152                    break
1153                now = time.monotonic()
1154                if now < next_allowed:
1155                    await asyncio.sleep(next_allowed - now)
1156                await self._distribute_chunk(
1157                    session_key, chunk, max_interval=CHUNK_HANG_GUARD_SECONDS
1158                )
1159                next_allowed = time.monotonic() + BACKGROUND_PACE_INTERVAL_SECONDS_FLOOR
1160        if session_key in self._active_sessions:
1161            self._finalize_providers(session_key)
1162
1163    def _available_filesystem_domains(self) -> tuple[str, ...]:
1164        """Return configured filesystem provider domains that are currently available."""
1165        return tuple(
1166            domain
1167            for domain in FILESYSTEM_PROVIDER_DOMAINS
1168            if any(
1169                p.domain == domain and p.available
1170                for p in self.mass.get_providers(ProviderType.MUSIC)
1171            )
1172        )
1173
1174    async def _find_candidates_missing_analysis(
1175        self,
1176        aa_provider_versions: Mapping[str, int],
1177        limit: int,
1178    ) -> list[dict[str, Any]]:
1179        """
1180        Return tracks that need (re)analysis for one or more AA providers.
1181
1182        A track is a candidate for a given AA provider domain when it has no analysis row for
1183        that domain, when its stored row predates the provider's current analysis_version (a
1184        NULL stored version, from pre-versioning rows, is also treated as stale), and when no
1185        blocking failure row exists (a failure at the current-or-newer analysis_version whose
1186        retry is NULL or still in the future). The version check mirrors the per-track gate in
1187        AudioAnalysisProvider.start_analysis so a provider bumping its analysis_version triggers
1188        a background re-scan.
1189
1190        :param aa_provider_versions: Mapping of AA provider domain to the provider's current
1191            analysis_version.
1192        :param limit: Maximum number of candidate rows to return (0 for no limit).
1193        :returns: Rows {item_id, provider_instance, missing_domains} where missing_domains
1194            lists the AA provider domains needing analysis.
1195        """
1196        if not aa_provider_versions:
1197            return []
1198
1199        filesystem_domains = self._available_filesystem_domains()
1200        if not filesystem_domains:
1201            return []
1202
1203        # CROSS JOIN (track x possible domain), keep pairs with no up-to-date analysis row and
1204        # no blocking failure row, then GROUP_CONCAT the missing domains per track. An analysis
1205        # row counts as up-to-date only when its analysis_version is non-NULL and >= the
1206        # provider's current version, so missing and stale-version rows both surface.
1207        aa_domains = list(aa_provider_versions)
1208        fs_inline = ", ".join(f"'{d}'" for d in filesystem_domains)
1209        aa_select_terms = " UNION ALL ".join(
1210            f"SELECT :aa_{i} AS aa_provider_domain, :ver_{i} AS current_version"
1211            for i in range(len(aa_domains))
1212        )
1213        params: dict[str, Any] = {
1214            "media_type": MediaType.TRACK.value,
1215            "now": int(utc_timestamp()),
1216            **{f"aa_{i}": d for i, d in enumerate(aa_domains)},
1217            **{f"ver_{i}": aa_provider_versions[d] for i, d in enumerate(aa_domains)},
1218        }
1219        # The NOT EXISTS gate only counts an analysis row as up-to-date when its
1220        # analysis_version is non-NULL and >= the provider's current version, so
1221        # missing rows and stale-version rows both surface as candidates.
1222        query = (
1223            f"SELECT pm.provider_item_id AS item_id, "
1224            f"       pm.provider_instance AS provider_instance, "
1225            f"       GROUP_CONCAT(possible.aa_provider_domain) AS missing_domains "
1226            f"FROM {DB_TABLE_PROVIDER_MAPPINGS} pm "
1227            f"CROSS JOIN ({aa_select_terms}) possible "
1228            f"WHERE pm.media_type = :media_type "
1229            f"  AND pm.provider_domain IN ({fs_inline}) "
1230            f"  AND NOT EXISTS ("
1231            f"    SELECT 1 FROM {DB_TABLE_AUDIO_ANALYSIS} aa "
1232            f"    WHERE aa.item_id = pm.provider_item_id "
1233            f"      AND aa.provider = pm.provider_instance "
1234            f"      AND aa.aa_provider_domain = possible.aa_provider_domain "
1235            f"      AND aa.media_type = :media_type "
1236            f"      AND aa.analysis_version IS NOT NULL "
1237            f"      AND aa.analysis_version >= possible.current_version"
1238            f"  ) "
1239            f"  AND NOT EXISTS ("
1240            f"    SELECT 1 FROM {DB_TABLE_AUDIO_ANALYSIS_FAILURES} f "
1241            f"    WHERE f.item_id = pm.provider_item_id "
1242            f"      AND f.provider = pm.provider_instance "
1243            f"      AND f.aa_provider_domain = possible.aa_provider_domain "
1244            f"      AND f.media_type = :media_type "
1245            f"      AND f.analysis_version >= possible.current_version "
1246            f"      AND (f.next_retry IS NULL OR f.next_retry > :now)"
1247            f"  ) "
1248            f"GROUP BY pm.provider_item_id, pm.provider_instance"
1249        )
1250        rows = await self.mass.music.database.get_rows_from_query(query, params, limit=limit)
1251        results: list[dict[str, Any]] = []
1252        for r in rows:
1253            missing_raw = r["missing_domains"]
1254            if not missing_raw:
1255                continue
1256            results.append(
1257                {
1258                    "item_id": str(r["item_id"]),
1259                    "provider_instance": str(r["provider_instance"]),
1260                    "missing_domains": sorted(set(missing_raw.split(","))),
1261                }
1262            )
1263        return results
1264
1265    async def _count_candidates_missing_analysis(self, aa_domain: str, current_version: int) -> int:
1266        """Count filesystem candidate tracks lacking a current analysis row or blocking failure."""
1267        filesystem_domains = self._available_filesystem_domains()
1268        if not filesystem_domains:
1269            return 0
1270        fs_inline = ", ".join(f"'{d}'" for d in filesystem_domains)
1271        query = (
1272            f"SELECT pm.provider_item_id FROM {DB_TABLE_PROVIDER_MAPPINGS} pm "
1273            f"WHERE pm.media_type = :media_type "
1274            f"  AND pm.provider_domain IN ({fs_inline}) "
1275            f"  AND NOT EXISTS ("
1276            f"    SELECT 1 FROM {DB_TABLE_AUDIO_ANALYSIS} aa "
1277            f"    WHERE aa.item_id = pm.provider_item_id "
1278            f"      AND aa.provider = pm.provider_instance "
1279            f"      AND aa.aa_provider_domain = :aa_domain "
1280            f"      AND aa.media_type = :media_type "
1281            f"      AND aa.analysis_version IS NOT NULL "
1282            f"      AND aa.analysis_version >= :current_version"
1283            f"  ) "
1284            f"  AND NOT EXISTS ("
1285            f"    SELECT 1 FROM {DB_TABLE_AUDIO_ANALYSIS_FAILURES} f "
1286            f"    WHERE f.item_id = pm.provider_item_id "
1287            f"      AND f.provider = pm.provider_instance "
1288            f"      AND f.aa_provider_domain = :aa_domain "
1289            f"      AND f.media_type = :media_type "
1290            f"      AND f.analysis_version >= :current_version "
1291            f"      AND (f.next_retry IS NULL OR f.next_retry > :now)"
1292            f"  )"
1293        )
1294        return await self.mass.music.database.get_count_from_query(
1295            query,
1296            {
1297                "media_type": MediaType.TRACK.value,
1298                "aa_domain": aa_domain,
1299                "current_version": current_version,
1300                "now": int(utc_timestamp()),
1301            },
1302        )
1303
1304    async def _start_analysis_on_providers(
1305        self,
1306        session_key: str,
1307        streamdetails: StreamDetails,
1308        audio_format: AudioFormat,
1309        providers: list[AudioAnalysisProvider],
1310    ) -> set[str]:
1311        """Call start_analysis on each provider, returning IDs of those that accepted."""
1312        self._mark_analysis_activity()
1313        provider_ids: set[str] = set()
1314        for provider in providers:
1315            try:
1316                if await provider.start_analysis(
1317                    session_id=session_key,
1318                    streamdetails=streamdetails,
1319                    audio_format=audio_format,
1320                ):
1321                    provider_ids.add(provider.instance_id)
1322            except Exception as err:
1323                # provider.start_analysis is provider-implemented; skip the one that
1324                # fails to start and keep the rest of the session going.
1325                self.logger.warning(
1326                    "Failed to start analysis on provider %s: %s", provider.name, err
1327                )
1328        return provider_ids
1329
1330    def _finalize_providers(self, session_key: str) -> None:
1331        """Finalize each provider in the session."""
1332        provider_ids = self._active_sessions.pop(session_key, None)
1333        if not provider_ids:
1334            return
1335        for provider_id in provider_ids:
1336            provider = self.mass.get_provider(provider_id)
1337            if provider and isinstance(provider, AudioAnalysisProvider) and provider.available:
1338                self.mass.create_task(provider.finalize(session_key))
1339
1340    def _cancel_providers(self, session_key: str) -> None:
1341        """Cancel each provider in the session."""
1342        provider_ids = self._active_sessions.pop(session_key, None)
1343        if not provider_ids:
1344            return
1345        for provider_id in provider_ids:
1346            provider = self.mass.get_provider(provider_id)
1347            if provider and isinstance(provider, AudioAnalysisProvider) and provider.available:
1348                self.mass.create_task(provider.cancel(session_key))
1349
1350    def _evict_realtime_session(self, session_key: str) -> None:
1351        """Stop a realtime analysis worker and cancel its providers, freeing the session slot."""
1352        self._session_queues.pop(session_key, None)
1353        worker = self._workers.pop(session_key, None)
1354        if worker is not None and not worker.done():
1355            worker.cancel()
1356        # Cancel providers directly: a task cancelled before it first runs has no finally to run.
1357        self._cancel_providers(session_key)
1358        self.logger.debug("Stopped realtime analysis session %s", session_key)
1359
1360    def _mark_analysis_activity(self) -> None:
1361        """Record analysis activity and ensure the idle-model monitor is running."""
1362        self._last_analysis_activity = time.monotonic()
1363        if self._idle_unload_task is None or self._idle_unload_task.done():
1364            self._idle_unload_task = self.mass.create_task(self._monitor_idle_models())
1365
1366    async def _monitor_idle_models(self) -> None:
1367        """Unload heavy models once no analysis has run for MODEL_IDLE_UNLOAD_SECONDS."""
1368        while True:
1369            await asyncio.sleep(MODEL_IDLE_CHECK_INTERVAL_SECONDS)
1370            if self._active_sessions:
1371                # Keep the timer fresh while analysis is running.
1372                self._last_analysis_activity = time.monotonic()
1373                continue
1374            if time.monotonic() - self._last_analysis_activity < MODEL_IDLE_UNLOAD_SECONDS:
1375                continue
1376            await self._unload_idle_models()
1377            return  # stop until the next analysis restarts the monitor
1378
1379    async def _unload_idle_models(self) -> None:
1380        """Free heavy models on every provider that supports unloading them."""
1381        for provider in self.providers:
1382            if not provider.has_unloadable_models:
1383                continue
1384            try:
1385                await provider.unload_idle_models()
1386            except Exception as err:
1387                self.logger.warning("Failed to unload models for %s: %s", provider.name, err)
1388
1389    async def _distribute_chunk(
1390        self,
1391        session_key: str,
1392        pcm_data: bytes,
1393        max_interval: float = CHUNK_HANG_GUARD_SECONDS,
1394    ) -> None:
1395        """
1396        Fan a single PCM chunk to every provider in the session.
1397
1398        :param session_key: Active-session key for the dispatch.
1399        :param pcm_data: The 1-second PCM chunk to hand to each provider.
1400        :param max_interval: Per-provider processing timeout; providers exceeding this are evicted.
1401        """
1402        provider_ids = self._active_sessions.get(session_key)
1403        if not provider_ids:
1404            return
1405
1406        async def _process(prov_id: str) -> str | None:
1407            try:
1408                provider = self.mass.get_provider(prov_id)
1409                if not (
1410                    provider and isinstance(provider, AudioAnalysisProvider) and provider.available
1411                ):
1412                    return None
1413                await asyncio.wait_for(
1414                    provider.process_pcm_chunk(session_key, pcm_data),
1415                    timeout=max_interval,
1416                )
1417            except TimeoutError:
1418                sem = self.analysis_semaphore
1419                contention = (
1420                    f"{sem.in_flight}/{sem.capacity} permits in use, {sem.waiters} queued"
1421                    if isinstance(sem, InstrumentedSemaphore)
1422                    else "concurrency gauge unavailable"
1423                )
1424                self.logger.warning(
1425                    "Provider %s timed out after %.1fs processing chunk for %s "
1426                    "(%s, %d active sessions), removing from session",
1427                    prov_id,
1428                    max_interval,
1429                    session_key,
1430                    contention,
1431                    len(self._active_sessions),
1432                )
1433                return prov_id
1434            except Exception as err:
1435                # process_pcm_chunk is provider-implemented (torch/numpy/ffmpeg); evict
1436                # the provider that fails on a chunk rather than crashing the session.
1437                self.logger.warning("Error processing PCM chunk on provider %s: %s", prov_id, err)
1438                return prov_id
1439            return None
1440
1441        results = await asyncio.gather(*[_process(prov_id) for prov_id in provider_ids])
1442        evicted = {prov_id for prov_id in results if prov_id is not None}
1443        if evicted:
1444            for prov_id in evicted:
1445                provider = self.mass.get_provider(prov_id)
1446                if provider and isinstance(provider, AudioAnalysisProvider) and provider.available:
1447                    self.mass.create_task(provider.cancel(session_key))
1448            provider_ids -= evicted
1449            if not provider_ids:
1450                self._active_sessions.pop(session_key, None)
1451
1452    async def _buffer_reader_worker(
1453        self,
1454        session_key: str,
1455        audio_buffer: AudioBuffer,
1456        expected_duration: float | None,
1457    ) -> None:
1458        """
1459        Read PCM straight from the shared playback buffer and distribute it to providers.
1460
1461        Reads at its own pace from the buffer's retained window. On clean end-of-stream the
1462        providers are finalized, unless the source ended far short of the expected duration.
1463        If the reader falls a full window behind playback (the chunk it needs has been
1464        evicted) or the buffer is torn down first, the session is dropped.
1465
1466        :param session_key: Active-session key for this worker.
1467        :param audio_buffer: The shared playback buffer to read PCM from.
1468        :param expected_duration: Expected track duration in seconds (None when unknown,
1469            e.g. radio), used to discard sessions of streams that ended prematurely.
1470        """
1471        start_chunk = audio_buffer.first_buffered_chunk
1472        cursor = start_chunk
1473        completed = False
1474        try:
1475            while session_key in self._active_sessions:
1476                try:
1477                    chunk = await audio_buffer.read_chunk_for_analysis(cursor)
1478                except AudioBufferEOF:
1479                    completed = True
1480                    break
1481                except AudioBufferDiscarded:
1482                    self.logger.debug(
1483                        "Analysis fell behind the playback buffer for %s (chunk %d evicted); "
1484                        "dropping session",
1485                        session_key,
1486                        cursor,
1487                    )
1488                    break
1489                except Exception as err:
1490                    self.logger.debug("Analysis read failed for %s: %s", session_key, err)
1491                    break
1492                await self._distribute_chunk(
1493                    session_key, chunk, max_interval=CHUNK_HANG_GUARD_SECONDS
1494                )
1495                cursor += 1
1496        finally:
1497            self._workers.pop(session_key, None)
1498            self._session_queues.pop(session_key, None)
1499            # one chunk equals one second of audio
1500            received_seconds = cursor - start_chunk
1501            if (
1502                completed
1503                and expected_duration
1504                and received_seconds < expected_duration * ANALYSIS_MIN_COMPLETENESS_RATIO
1505            ):
1506                self.logger.debug(
1507                    "Analysis received only %ds of the expected %ds for %s; "
1508                    "discarding incomplete session",
1509                    received_seconds,
1510                    expected_duration,
1511                    session_key,
1512                )
1513                completed = False
1514            if completed:
1515                self._finalize_providers(session_key)
1516            else:
1517                self._cancel_providers(session_key)
1518
1519    def _cpu_count(self) -> int:
1520        """Return the CPU core count available to this process (fallback 4 when unknown)."""
1521        return os.process_cpu_count() or os.cpu_count() or 4
1522
1523    def _aa_thread_budget(self) -> int:
1524        """Return the per-op PyTorch intra-op thread budget for inference (~25% of cpu_count)."""
1525        # Shared with the native BLAS cap applied at process start, so torch and BLAS agree.
1526        return inference_thread_budget()
1527
1528    def _get_scan_concurrency(self) -> int:
1529        """Read background scan concurrency from config, clamped to [1, 16]."""
1530        try:
1531            value = int(
1532                self.mass.config.get_raw_core_config_value(
1533                    "streams",
1534                    CONF_BACKGROUND_SCAN_CONCURRENCY,
1535                    DEFAULT_BACKGROUND_SCAN_CONCURRENCY,
1536                )
1537                or DEFAULT_BACKGROUND_SCAN_CONCURRENCY
1538            )
1539        except ValueError, TypeError:
1540            value = DEFAULT_BACKGROUND_SCAN_CONCURRENCY
1541        return max(1, min(value, 16))
1542