music-assistant-server

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