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