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