/
/
/
1"""On-device audio analysis: librosa scalars + Microsoft CLAP zero-shot, one audio load per track."""
2
3from __future__ import annotations
4
5import asyncio
6import time
7from dataclasses import dataclass, field
8from datetime import timedelta
9from typing import TYPE_CHECKING, Any
10
11import numpy as np
12import soxr
13from music_assistant_models.config_entries import ConfigEntry, ConfigValueOption
14from music_assistant_models.enums import ConfigEntryType, ContentType
15from music_assistant_models.errors import SetupFailedError, UnsupportedSystemError
16
17from music_assistant.helpers.datetime import utc
18from music_assistant.helpers.util import (
19 join_task,
20 system_meets_requirements,
21 verify_system_meets_requirements,
22)
23from music_assistant.models.audio_analysis import AudioAnalysisData, AudioAnalysisError
24from music_assistant.models.audio_analysis_provider import (
25 ACCUMULATING_ANALYSIS_MAX_DURATION_SECONDS,
26 AnalysisSessionData,
27 AudioAnalysisProvider,
28)
29
30from .clap_prompts import (
31 PRECOMPUTED_EMBEDDINGS_PATH,
32 SCALAR_PROMPT_PAIRS,
33 hash_scalar_prompt_pairs,
34 load_precomputed_prompt_embeddings,
35 score_scalars,
36)
37from .helpers import (
38 BlockFeatures,
39 collapse_to_analysis,
40 extract_block_features,
41 merge_block_features,
42)
43
44if TYPE_CHECKING:
45 from music_assistant_models.config_entries import ProviderConfig
46 from music_assistant_models.enums import ProviderFeature
47 from music_assistant_models.media_items import AudioFormat
48 from music_assistant_models.provider import ProviderManifest
49 from music_assistant_models.streamdetails import StreamDetails
50
51 from music_assistant.mass import MusicAssistant
52 from music_assistant.models import ProviderInstanceType
53
54BLOCK_SECONDS: int = 10
55# Must equal helpers._CHROMA_SR â filterbanks in helpers.py are baked at this rate.
56ANALYSIS_SAMPLE_RATE: int = 22050
57OVERLAP_SAMPLES: int = 2048
58
59EXTRA_DATA_CLAP_EMBEDDING: str = "clap_embedding"
60
61# CLAP's HTSAT audio encoder takes a fixed 7-second input at 44.1 kHz.
62CLAP_WINDOW_SECONDS: int = 7
63CLAP_SKIP_SECONDS: int = 45
64CLAP_MIN_WINDOW_SECONDS: float = 1.0
65
66CLAP_SAMPLING_FAST: str = "fast"
67CLAP_SAMPLING_BALANCED: str = "balanced"
68CLAP_SAMPLING_THOROUGH: str = "thorough"
69CLAP_WINDOW_COUNTS: dict[str, int] = {
70 CLAP_SAMPLING_FAST: 1,
71 CLAP_SAMPLING_BALANCED: 3,
72 CLAP_SAMPLING_THOROUGH: 8,
73}
74
75CONF_CLAP_SAMPLING: str = "clap_sampling"
76
77# Sonic Analysis runs on-device CLAP inference; gate it to capable hardware.
78# 4GB nominal; the gate's tolerance (meets_memory_target) admits genuine 4GB hosts,
79# which report ~3.8GB after the kernel/firmware reservation.
80MIN_RAM_GB: float = 4.0
81MIN_CPU_CORES: int = 2
82# Below the recommended thresholds the provider still runs, but we surface an
83# informational notice (see get_config_entries) as it may be tight under load.
84RECOMMENDED_RAM_GB: float = 6.0
85RECOMMENDED_CPU_CORES: int = 4
86
87MODEL_FAILURE_RETRY_DELAY: timedelta = timedelta(hours=24)
88
89# Bounds the wait, not the load: a slower first download keeps running and the
90# next setup attempt joins it rather than starting over.
91MODEL_SETUP_GRACE_SECONDS: int = 200
92
93
94@dataclass
95class SonicSessionData(AnalysisSessionData):
96 """Per-session state: PCM block buffer and accumulated per-block features."""
97
98 pcm_buffer: bytearray = field(default_factory=bytearray)
99 block_bytes: int = 0
100 resampler: soxr.ResampleStream | None = None
101 accumulated: BlockFeatures = field(default_factory=BlockFeatures)
102 total_samples: int = 0
103 overlap: np.ndarray | None = None
104 start_time: float = 0.0
105 peak_absolute: float = 0.0
106 # Per-window selective buffer for live CLAP. clap_target_starts is
107 # planned at session start from streamdetails.duration + preset; the
108 # buffers fill via _dispatch_clap_chunk and free on completion.
109 clap_target_starts: list[int] = field(default_factory=list)
110 clap_target_buffers: list[list[np.ndarray]] = field(default_factory=list)
111 clap_target_complete: list[bool] = field(default_factory=list)
112 clap_position_samples: int = 0
113 # Inference task handles + running sums for mean-pooling at finalize.
114 clap_inference_tasks: list[asyncio.Task[None]] = field(default_factory=list)
115 clap_sum_embedding: np.ndarray | None = None
116 clap_sum_similarities: np.ndarray | None = None
117 clap_completed_count: int = 0
118 # Timing accumulators for the finalize diagnostic breakdown. feature_seconds
119 # sums the inline librosa decode/extract/collapse work; clap_seconds sums each
120 # per-window CLAP await-to-completion span (timer starts before the offload
121 # hop, so it folds in scheduling/queue wait â and since windows run concurrently
122 # it is cumulative and can exceed wall-clock). clap_preset records the configured
123 # sampling preset for the same line.
124 feature_seconds: float = 0.0
125 clap_seconds: float = 0.0
126 clap_preset: str = ""
127
128
129async def setup(
130 mass: MusicAssistant, manifest: ProviderManifest, config: ProviderConfig
131) -> ProviderInstanceType:
132 """Initialize provider instance with given configuration."""
133 return SonicAnalysisProvider(mass, manifest, config)
134
135
136def compute_clap_target_starts(
137 track_duration_s: float,
138 preset_n: int,
139 source_sr: int,
140) -> list[int]:
141 """
142 Plan deterministic 7s window start offsets for the live CLAP path.
143
144 :param track_duration_s: Total track duration in seconds.
145 :param preset_n: Configured window count (Fast/Balanced/Thorough â 1/3/8).
146 :param source_sr: Sample rate the live PCM stream is delivered at.
147 :returns: Sample-position offsets at source_sr; length is the effective N
148 (capped at what the track length supports without near-duplicates).
149 """
150 if track_duration_s < CLAP_MIN_WINDOW_SECONDS:
151 return []
152 if track_duration_s < CLAP_WINDOW_SECONDS:
153 return [0]
154 if track_duration_s < CLAP_SKIP_SECONDS + CLAP_WINDOW_SECONDS:
155 start_seconds = (track_duration_s - CLAP_WINDOW_SECONDS) / 2.0
156 return [int(start_seconds * source_sr)]
157
158 usable_start = float(CLAP_SKIP_SECONDS)
159 usable_end = track_duration_s - CLAP_WINDOW_SECONDS
160 if preset_n <= 1:
161 return [int(usable_start * source_sr)]
162
163 usable_seconds = usable_end - usable_start
164 max_non_overlap = int(usable_seconds // CLAP_WINDOW_SECONDS) + 1
165 effective_n = max(1, min(preset_n, max_non_overlap))
166 if effective_n == 1:
167 return [int(usable_start * source_sr)]
168
169 positions = np.linspace(usable_start, usable_end, effective_n)
170 return [int(p * source_sr) for p in positions]
171
172
173def _store_clap_embedding(analysis: AudioAnalysisData, embedding: np.ndarray) -> None:
174 """Store the CLAP audio embedding on the analysis object for downstream consumers."""
175 if analysis.extra_data is None:
176 analysis.extra_data = {}
177 analysis.extra_data[EXTRA_DATA_CLAP_EMBEDDING] = embedding.tolist()
178
179
180def _dispatch_clap_chunk(
181 session: SonicSessionData,
182 decoded_audio: np.ndarray,
183 source_sr: int,
184) -> list[np.ndarray]:
185 """
186 Route a PCM chunk to active CLAP target windows; return any windows completed.
187
188 :param session: Active session; target buffers are mutated in place.
189 :param decoded_audio: Mono float32 PCM chunk at source_sr.
190 :param source_sr: Sample rate of decoded_audio.
191 """
192 if not session.clap_target_starts:
193 return []
194
195 chunk_start = session.clap_position_samples
196 chunk_end = chunk_start + len(decoded_audio)
197 session.clap_position_samples = chunk_end
198
199 window_samples = CLAP_WINDOW_SECONDS * source_sr
200 completed: list[np.ndarray] = []
201
202 for i, target_start in enumerate(session.clap_target_starts):
203 if session.clap_target_complete[i]:
204 continue
205 target_end = target_start + window_samples
206 if chunk_end <= target_start or chunk_start >= target_end:
207 continue
208 slice_start = max(0, target_start - chunk_start)
209 slice_end = min(len(decoded_audio), target_end - chunk_start)
210 session.clap_target_buffers[i].append(decoded_audio[slice_start:slice_end])
211
212 accumulated = sum(len(arr) for arr in session.clap_target_buffers[i])
213 if accumulated >= window_samples:
214 window_audio = np.concatenate(session.clap_target_buffers[i])[:window_samples]
215 session.clap_target_buffers[i] = []
216 session.clap_target_complete[i] = True
217 completed.append(window_audio)
218
219 return completed
220
221
222def _pcm_bytes_to_audio(audio_format: AudioFormat, pcm_chunk: bytes) -> np.ndarray:
223 """
224 Decode a raw PCM chunk to a mono float32 numpy array.
225
226 :param audio_format: The audio format describing the PCM data.
227 :param pcm_chunk: Raw PCM audio data.
228 """
229 import torch # noqa: PLC0415
230
231 content_type = audio_format.content_type
232 writable = bytearray(pcm_chunk)
233
234 if content_type == ContentType.PCM_F32LE:
235 audio = torch.frombuffer(writable, dtype=torch.float32).clone()
236 elif content_type == ContentType.PCM_F64LE:
237 audio = torch.frombuffer(writable, dtype=torch.float64).clone().to(torch.float32)
238 elif content_type == ContentType.PCM_S32LE:
239 audio = (
240 torch.frombuffer(writable, dtype=torch.int32).clone().to(torch.float32) / 2147483648.0
241 )
242 elif content_type == ContentType.PCM_S24LE:
243 raw = torch.frombuffer(writable, dtype=torch.uint8).clone()
244 raw = raw[: (raw.numel() // 3) * 3].reshape(-1, 3).to(torch.int32)
245 audio = raw[:, 0] | (raw[:, 1] << 8) | (raw[:, 2] << 16)
246 audio = torch.where(audio & 0x800000 != 0, audio - 0x1000000, audio)
247 audio = audio.to(torch.float32) / 8388608.0
248 else:
249 audio = torch.frombuffer(writable, dtype=torch.int16).clone().to(torch.float32) / 32768.0
250
251 channels = audio_format.channels
252 if channels > 1:
253 frame_samples = (audio.numel() // channels) * channels
254 audio = audio[:frame_samples].reshape(-1, channels).mean(dim=1)
255
256 return np.asarray(audio.numpy(), dtype=np.float32)
257
258
259def _decode_resample_extract(
260 audio_format: AudioFormat,
261 block_bytes: bytes,
262 overlap: np.ndarray | None,
263 sample_rate: int,
264 resampler: soxr.ResampleStream | None,
265 *,
266 is_last: bool = False,
267) -> tuple[np.ndarray, np.ndarray, BlockFeatures | None]:
268 """
269 Decode PCM bytes, optionally resample, and extract block features in one offloaded call.
270
271 :param audio_format: AudioFormat describing the PCM encoding of block_bytes.
272 :param block_bytes: Raw PCM bytes for one analysis block (or the final tail).
273 :param overlap: Post-resample samples from the previous block to prepend before
274 feature extraction, or None for the first block.
275 :param sample_rate: Sample rate expected by extract_block_features
276 (must equal ANALYSIS_SAMPLE_RATE after resampling).
277 :param resampler: Active ResampleStream, or None when source SR already matches
278 ANALYSIS_SAMPLE_RATE.
279 :param is_last: Pass True when processing the tail PCM in _finalize so the resampler
280 flushes its internal delay buffer.
281 :returns: A 3-tuple of (pre_resample_audio, post_resample_audio, block_features).
282 pre_resample_audio is at the source sample rate (for CLAP dispatch and duration/peak
283 accounting). post_resample_audio is at ANALYSIS_SAMPLE_RATE (for overlap bookkeeping).
284 block_features is None when the audio is too short for meaningful extraction.
285 """
286 pre_resample = _pcm_bytes_to_audio(audio_format, block_bytes)
287 if resampler is not None:
288 post_resample = resampler.resample_chunk(pre_resample, last=is_last)
289 else:
290 post_resample = pre_resample
291 audio_for_extract = (
292 np.concatenate([overlap, post_resample]) if overlap is not None else post_resample
293 )
294 bf = extract_block_features(audio_for_extract, sample_rate)
295 return pre_resample, post_resample, bf
296
297
298class SonicAnalysisProvider(AudioAnalysisProvider):
299 """Audio analysis provider running librosa scalars + CLAP zero-shot per track."""
300
301 analysis_version: int = 1
302 max_analysis_duration = ACCUMULATING_ANALYSIS_MAX_DURATION_SECONDS
303 has_unloadable_models = True
304
305 def __init__(
306 self,
307 mass: MusicAssistant,
308 manifest: ProviderManifest,
309 config: ProviderConfig,
310 supported_features: set[ProviderFeature] | None = None,
311 ) -> None:
312 """Initialize the provider."""
313 super().__init__(mass, manifest, config, supported_features)
314 self._clap_model: Any = None
315 self._clap_text_embeddings: Any = None
316 self._clap_prompt_order: list[tuple[str, tuple[str, str]]] = []
317
318 async def get_config_entries(self) -> tuple[ConfigEntry, ...]:
319 """Return Config entries to configure this provider."""
320 return (
321 ConfigEntry(
322 key="resource_warning",
323 type=ConfigEntryType.ALERT,
324 required=False,
325 hidden=system_meets_requirements(
326 min_memory_gb=RECOMMENDED_RAM_GB,
327 min_cpu_cores=RECOMMENDED_CPU_CORES,
328 ),
329 ),
330 ConfigEntry(
331 key=CONF_CLAP_SAMPLING,
332 type=ConfigEntryType.STRING,
333 default_value=CLAP_SAMPLING_FAST,
334 options=[
335 ConfigValueOption(CLAP_SAMPLING_FAST),
336 ConfigValueOption(CLAP_SAMPLING_BALANCED),
337 ConfigValueOption(CLAP_SAMPLING_THOROUGH),
338 ],
339 required=False,
340 ),
341 )
342
343 async def handle_async_init(self) -> None:
344 """
345 Async initialization of the provider.
346
347 Loads the CLAP model so provider.available only goes True once analysis can run.
348
349 :raises SetupFailedError: When the model is not ready within the grace period,
350 or loading it failed.
351 """
352 await verify_system_meets_requirements(
353 feature_name="Sonic Analysis",
354 min_memory_gb=MIN_RAM_GB,
355 min_cpu_cores=MIN_CPU_CORES,
356 require_ml_inference=True,
357 )
358 # Configure the inference runtime before loading the model (see the controller method).
359 self.mass.streams.audio_analysis.ensure_inference_runtime_configured()
360 # The ~690MB first load outlives an attempt that gives up: cancelling would not stop
361 # the worker thread, and the task_id makes the next attempt join this same load.
362 task = self.mass.create_task(
363 asyncio.to_thread(self._load_clap),
364 task_id=f"sonic_analysis.model_setup.{self.instance_id}",
365 )
366 try:
367 # The result comes back through the task: a retry is a different instance.
368 models = await join_task(task, timeout=MODEL_SETUP_GRACE_SECONDS)
369 except TimeoutError:
370 raise SetupFailedError(
371 "The Sonic Analysis model is still being prepared. "
372 "It keeps running in the background; reload the provider once it has "
373 "had time to finish.",
374 translation_key="model_setup_pending",
375 translation_owner=self.translation_owner,
376 ) from None
377 self._clap_model, self._clap_text_embeddings, self._clap_prompt_order = models
378 self._models_loaded = True
379
380 async def cancel(self, session_id: str) -> None:
381 """Cancel pending CLAP inferences and free per-window buffers."""
382 session = self._sessions.get(session_id)
383 if isinstance(session, SonicSessionData):
384 for task in session.clap_inference_tasks:
385 if not task.done():
386 task.cancel()
387 session.clap_target_buffers.clear()
388 await super().cancel(session_id)
389
390 async def process_pcm_chunk(
391 self,
392 session_id: str,
393 pcm_chunk: bytes,
394 ) -> None:
395 """
396 Accumulate PCM and run feature extraction once a 10-second block is full.
397
398 :param session_id: The analysis session ID.
399 :param pcm_chunk: Raw PCM audio data.
400 """
401 if session_id not in self._sessions:
402 return
403 session = self._sessions[session_id]
404 assert isinstance(session, SonicSessionData)
405 session.pcm_buffer.extend(pcm_chunk)
406 af = session.audio_format
407 if len(session.pcm_buffer) >= session.block_bytes:
408 block_bytes = bytes(session.pcm_buffer[: session.block_bytes])
409 del session.pcm_buffer[: session.block_bytes]
410 t0 = time.monotonic()
411 pre_audio, post_audio, bf = await self._run_offloaded(
412 _decode_resample_extract,
413 af,
414 block_bytes,
415 session.overlap,
416 ANALYSIS_SAMPLE_RATE,
417 session.resampler,
418 )
419 session.feature_seconds += time.monotonic() - t0
420 session.total_samples += len(pre_audio)
421 session.peak_absolute = max(session.peak_absolute, float(np.max(np.abs(pre_audio))))
422 self._dispatch_clap_to_targets(session, pre_audio, af.sample_rate)
423 session.overlap = post_audio[-OVERLAP_SAMPLES:].copy()
424 if bf is not None:
425 merge_block_features(session.accumulated, bf)
426
427 async def _load_models(self) -> None:
428 """Load the CLAP model and prompt embeddings into memory."""
429 (
430 self._clap_model,
431 self._clap_text_embeddings,
432 self._clap_prompt_order,
433 ) = await asyncio.to_thread(self._load_clap)
434
435 def _free_models(self) -> None:
436 """Release the CLAP model and prompt embeddings."""
437 self._clap_model = None
438 self._clap_text_embeddings = None
439
440 def _load_clap(
441 self,
442 ) -> tuple[Any, Any, list[tuple[str, tuple[str, str]]]]:
443 """
444 Load and return the CLAP model, text embeddings, and prompt ordering.
445
446 Downloads the checkpoint on first use, so only call this off the event loop.
447
448 :raises SetupFailedError: When the checkpoint could not be downloaded.
449 :raises UnsupportedSystemError: When the shipped prompt embeddings are unusable.
450 """
451 # httpx arrives with huggingface-hub; named only for the errors it re-raises below.
452 import httpx # noqa: PLC0415
453 import torch # noqa: PLC0415
454
455 from .vendored_clap import CLAP # noqa: PLC0415
456
457 prompt_order: list[tuple[str, tuple[str, str]]] = list(SCALAR_PROMPT_PAIRS.items())
458
459 cached = self._try_load_cached_prompt_embeddings()
460 if cached is None:
461 # Falling back to a text-enabled model here would download the GPT2 encoder.
462 raise UnsupportedSystemError(
463 "The precomputed CLAP prompt embeddings are missing or out of date. "
464 "Re-run scripts/precompute_clap_prompt_embeddings.py to regenerate them.",
465 translation_key="prompt_embeddings_unavailable",
466 translation_owner=self.translation_owner,
467 )
468 try:
469 model = CLAP(version="2023", use_cuda=False, text_enabled=False)
470 except (OSError, httpx.HTTPError) as err:
471 # The hub reports most failures as OSError, but re-raises httpx transport errors
472 # verbatim once its resume attempts are spent; only typed errors get retried.
473 raise SetupFailedError(
474 f"Failed to download the Sonic Analysis model: {err}",
475 translation_key="model_assets_download_failed",
476 translation_owner=self.translation_owner,
477 ) from err
478 self.logger.info("CLAP model loaded; %d prompt pairs ready", len(prompt_order))
479 return model, torch.from_numpy(cached), prompt_order
480
481 def _try_load_cached_prompt_embeddings(self) -> np.ndarray | None:
482 """Return shipped prompt embeddings if present and hash-current, else None."""
483 try:
484 cached_embeddings, cached_hash = load_precomputed_prompt_embeddings(
485 PRECOMPUTED_EMBEDDINGS_PATH
486 )
487 except FileNotFoundError:
488 self.logger.warning(
489 "Precomputed CLAP prompt embeddings missing at %s; "
490 "re-run scripts/precompute_clap_prompt_embeddings.py",
491 PRECOMPUTED_EMBEDDINGS_PATH,
492 )
493 return None
494 expected_hash = hash_scalar_prompt_pairs(SCALAR_PROMPT_PAIRS)
495 if cached_hash != expected_hash:
496 self.logger.warning(
497 "Precomputed CLAP prompt embeddings hash mismatch (%s != %s); "
498 "re-run scripts/precompute_clap_prompt_embeddings.py",
499 cached_hash[:12],
500 expected_hash[:12],
501 )
502 return None
503 return cached_embeddings
504
505 async def _start_analysis(
506 self,
507 session_id: str,
508 streamdetails: StreamDetails,
509 audio_format: AudioFormat,
510 ) -> bool:
511 """
512 Initialize a new analysis session.
513
514 :param session_id: Unique session ID from the controller.
515 :param streamdetails: Stream details for the item being analyzed.
516 :param audio_format: PCM format of the audio stream.
517 """
518 if self._clap_model is None:
519 self.logger.debug(
520 "Skipping analysis for %s: CLAP model not yet available",
521 streamdetails.item_id,
522 )
523 return False
524 if not streamdetails.duration:
525 # Without a known duration we can't plan CLAP windows, and the result
526 # would be librosa-only â unusable for similarity. Reject so the next
527 # analysis attempt (with duration filled in) can succeed instead of
528 # caching an incomplete record.
529 self.logger.debug(
530 "Skipping analysis for %s: streamdetails.duration missing or zero",
531 streamdetails.item_id,
532 )
533 return False
534 bytes_per_sample = audio_format.bit_depth // 8
535 block_bytes = (
536 audio_format.sample_rate * bytes_per_sample * audio_format.channels * BLOCK_SECONDS
537 )
538 if block_bytes <= 0:
539 self.logger.warning(
540 "Invalid audio format for session %s (sample_rate=%d, bit_depth=%d, channels=%d)"
541 " â skipping analysis",
542 session_id,
543 audio_format.sample_rate,
544 audio_format.bit_depth,
545 audio_format.channels,
546 )
547 return False
548 preset = str(self.config.get_value(CONF_CLAP_SAMPLING, CLAP_SAMPLING_FAST))
549 preset_n = CLAP_WINDOW_COUNTS.get(preset, 1)
550 target_starts = compute_clap_target_starts(
551 streamdetails.duration, preset_n, audio_format.sample_rate
552 )
553
554 resampler: soxr.ResampleStream | None = None
555 if audio_format.sample_rate != ANALYSIS_SAMPLE_RATE:
556 resampler = soxr.ResampleStream(
557 in_rate=audio_format.sample_rate,
558 out_rate=ANALYSIS_SAMPLE_RATE,
559 num_channels=1,
560 dtype="float32",
561 )
562 self._sessions[session_id] = SonicSessionData(
563 streamdetails=streamdetails,
564 audio_format=audio_format,
565 block_bytes=block_bytes,
566 resampler=resampler,
567 start_time=time.monotonic(),
568 clap_target_starts=target_starts,
569 clap_target_buffers=[[] for _ in target_starts],
570 clap_target_complete=[False] * len(target_starts),
571 clap_preset=preset,
572 )
573 self.logger.debug(
574 "Started sonic analysis for %s/%s (preset=%s, %d CLAP target windows)",
575 streamdetails.provider,
576 streamdetails.item_id,
577 preset,
578 len(target_starts),
579 )
580 return True
581
582 def _dispatch_clap_to_targets(
583 self, session: SonicSessionData, audio: np.ndarray, source_sr: int
584 ) -> None:
585 """Route a decoded block to active CLAP target windows; spawn inference per completion."""
586 if not session.clap_target_starts:
587 return
588 completed = _dispatch_clap_chunk(session, audio, source_sr)
589 for window_audio in completed:
590 task = self.mass.create_task(
591 self._run_single_clap_window(session, window_audio, source_sr)
592 )
593 session.clap_inference_tasks.append(task)
594
595 async def _run_live_clap_if_eligible(
596 self, session: SonicSessionData, analysis: AudioAnalysisData
597 ) -> None:
598 """
599 Finalize CLAP analysis, writing the scalar attributes and embedding onto the analysis.
600
601 :param session: The analysis session.
602 :param analysis: Analysis object the CLAP scalars and embedding are written onto.
603 :raises AudioAnalysisError: Retryable, when fewer windows completed than were planned.
604 """
605 if not session.clap_target_starts:
606 return
607 if session.clap_inference_tasks:
608 await asyncio.gather(*session.clap_inference_tasks, return_exceptions=True)
609 n = session.clap_completed_count
610 planned = len(session.clap_target_starts)
611 sd = session.streamdetails
612 if (
613 n < planned
614 or session.clap_sum_embedding is None
615 or session.clap_sum_similarities is None
616 ):
617 self.logger.warning(
618 "Live CLAP for %s/%s: only %d of %d planned windows completed â "
619 "failing the analysis so it is retried",
620 sd.provider,
621 sd.item_id,
622 n,
623 planned,
624 )
625 raise AudioAnalysisError(
626 f"live CLAP completed {n} of {planned} planned windows",
627 retry_at=utc() + MODEL_FAILURE_RETRY_DELAY,
628 )
629
630 mean_emb = session.clap_sum_embedding / n
631 norm = float(np.linalg.norm(mean_emb))
632 if norm > 0:
633 mean_emb = mean_emb / norm
634 mean_sim = session.clap_sum_similarities / n
635
636 for scalar_name, value in score_scalars(mean_sim).items():
637 setattr(analysis, scalar_name, value)
638
639 _store_clap_embedding(analysis, mean_emb)
640 self.logger.debug(
641 "Live CLAP for %s/%s: %d/%d windows completed",
642 sd.provider,
643 sd.item_id,
644 n,
645 len(session.clap_target_starts),
646 )
647
648 async def _finalize(self, session_id: str) -> AudioAnalysisData | None:
649 """
650 Flush remaining PCM, collapse features, and return the analysis result.
651
652 Returns the analysis for the base class to persist, or None to skip persistence.
653
654 :param session_id: The analysis session ID.
655 """
656 if session_id not in self._sessions:
657 self.logger.debug("Finalize called for unknown session %s", session_id)
658 return None
659 session = self._sessions[session_id]
660 assert isinstance(session, SonicSessionData)
661 sd = session.streamdetails
662 af = session.audio_format
663
664 if session.pcm_buffer:
665 t0 = time.monotonic()
666 pre_audio, _post_audio, bf = await self._run_offloaded(
667 _decode_resample_extract,
668 af,
669 bytes(session.pcm_buffer),
670 session.overlap,
671 ANALYSIS_SAMPLE_RATE,
672 session.resampler,
673 is_last=True,
674 )
675 session.feature_seconds += time.monotonic() - t0
676 session.total_samples += len(pre_audio)
677 session.peak_absolute = max(session.peak_absolute, float(np.max(np.abs(pre_audio))))
678 self._dispatch_clap_to_targets(session, pre_audio, af.sample_rate)
679 if bf is not None:
680 merge_block_features(session.accumulated, bf)
681 session.pcm_buffer.clear()
682
683 if not session.accumulated.rms_frames:
684 raise AudioAnalysisError("no usable audio frames extracted")
685
686 self._flush_incomplete_clap_windows(session, af.sample_rate)
687
688 t0 = time.monotonic()
689 analysis = await self._run_offloaded(
690 collapse_to_analysis, session.accumulated, ANALYSIS_SAMPLE_RATE
691 )
692 session.feature_seconds += time.monotonic() - t0
693
694 analysis.duration = session.total_samples / af.sample_rate
695 if session.peak_absolute > 0:
696 analysis.true_peak = float(20.0 * np.log10(session.peak_absolute))
697 else:
698 analysis.true_peak = -96.0
699
700 await self._run_live_clap_if_eligible(session, analysis)
701
702 elapsed = time.monotonic() - session.start_time
703 self.logger.debug(
704 "Stored analysis for %s/%s (%.1fs elapsed: feature=%.1fs, "
705 "clap=%.1fs cumulative over %d/%d windows, preset=%s)",
706 sd.provider,
707 sd.item_id,
708 elapsed,
709 session.feature_seconds,
710 session.clap_seconds,
711 session.clap_completed_count,
712 len(session.clap_target_starts),
713 session.clap_preset,
714 )
715 return analysis
716
717 def _flush_incomplete_clap_windows(self, session: SonicSessionData, source_sr: int) -> None:
718 """
719 Dispatch every planned CLAP window that buffered audio but never filled to 7 seconds.
720
721 :param session: The analysis session; the buffers of flushed windows are consumed.
722 :param source_sr: Sample rate the buffered PCM was captured at.
723 """
724 # The vendored wrapper repeat-pads short input, so anything below the floor would
725 # blow a sliver of audio up into a full window and embed noise.
726 min_samples = int(CLAP_MIN_WINDOW_SECONDS * source_sr)
727 for i, buffered in enumerate(session.clap_target_buffers):
728 if session.clap_target_complete[i]:
729 continue
730 if sum(len(arr) for arr in buffered) < min_samples:
731 continue
732 window_audio = np.concatenate(buffered)
733 session.clap_target_buffers[i] = []
734 session.clap_target_complete[i] = True
735 task = self.mass.create_task(
736 self._run_single_clap_window(session, window_audio, source_sr)
737 )
738 session.clap_inference_tasks.append(task)
739
740 def _single_window_inference_sync(
741 self,
742 window_audio: np.ndarray,
743 source_sr: int,
744 ) -> tuple[np.ndarray, np.ndarray] | None:
745 """
746 Run CLAP inference on a single 7-second window.
747
748 :param window_audio: Mono float32 audio at source_sr.
749 :param source_sr: Sample rate of window_audio.
750 :returns: (1024-dim embedding, similarity logit row), or None if the model is unloaded.
751 """
752 import torch # noqa: PLC0415
753
754 model = self._clap_model
755 text_embeddings = self._clap_text_embeddings
756 if model is None or text_embeddings is None:
757 return None
758 window_tensor = torch.from_numpy(window_audio)
759 audio_embs = model.get_audio_embeddings_from_tensor([window_tensor], source_sr)
760 similarities = model.compute_similarity(audio_embs, text_embeddings)
761 embedding = audio_embs[0].detach().cpu().numpy().astype(np.float32).reshape(-1)
762 similarity_row = similarities[0].detach().cpu().numpy().astype(np.float32).reshape(-1)
763 return embedding, similarity_row
764
765 async def _run_single_clap_window(
766 self,
767 session: SonicSessionData,
768 window_audio: np.ndarray,
769 source_sr: int,
770 ) -> None:
771 """Run CLAP on a single window off-thread and accumulate running sums."""
772 if self._clap_model is None:
773 return
774 t0 = time.monotonic()
775 try:
776 result = await self._run_offloaded(
777 self._single_window_inference_sync, window_audio, source_sr
778 )
779 except asyncio.CancelledError:
780 raise
781 except Exception as err:
782 # CLAP inference runs torch ops off-thread; the failure surface is broad and
783 # version-dependent, so any failure just drops this window's contribution.
784 self.logger.debug("CLAP single-window inference failed: %s", err)
785 return
786 finally:
787 session.clap_seconds += time.monotonic() - t0
788 if result is None:
789 self.logger.debug("CLAP inference skipped â model unloaded mid-flight")
790 return
791 embedding, similarity_row = result
792 if session.clap_sum_embedding is None:
793 session.clap_sum_embedding = np.zeros_like(embedding)
794 session.clap_sum_similarities = np.zeros_like(similarity_row)
795 assert session.clap_sum_similarities is not None # narrowed by line above
796 session.clap_sum_embedding += embedding
797 session.clap_sum_similarities += similarity_row
798 session.clap_completed_count += 1
799