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