Stream PCM to audio analysis providers during background scan

music_assistant/constants.py

@@ -140,11 +140,13 @@
 CONF_PROTOCOL_KEY_SPLITTER: Final[str] = "||protocol||"
 CONF_PROTOCOL_CATEGORY_PREFIX: Final[str] = "protocol"
 CONF_DEFAULT_PROVIDERS_SETUP: Final[str] = "default_providers_setup"
+CONF_BACKGROUND_SCAN_CONCURRENCY: Final[str] = "background_scan_concurrency"
 
 
 # config default values
 DEFAULT_HOST: Final[str] = "0.0.0.0"
 DEFAULT_PORT: Final[int] = 8095
+DEFAULT_BACKGROUND_SCAN_CONCURRENCY: Final[int] = 1
 
 
 # common db tables

music_assistant/controllers/streams/controller.py

@@ -30,6 +30,7 @@
 
 from music_assistant.constants import (
     ANNOUNCE_ALERT_FILE,
+    CONF_BACKGROUND_SCAN_CONCURRENCY,
     CONF_BIND_IP,
     CONF_BIND_PORT,
     CONF_CROSSFADE_DURATION,
@@ -43,6 +44,7 @@
     CONF_VOLUME_NORMALIZATION_FIXED_GAIN_TRACKS,
     CONF_VOLUME_NORMALIZATION_RADIO,
     CONF_VOLUME_NORMALIZATION_TRACKS,
+    DEFAULT_BACKGROUND_SCAN_CONCURRENCY,
     DEFAULT_STREAM_HEADERS,
     DLNA_CONTENT_FEATURES,
     DLNA_CONTENT_FEATURES_REALTIME,
@@ -252,6 +254,17 @@ async def get_config_entries(
                 category="generic",
                 advanced=True,
             ),
+            ConfigEntry(
+                key=CONF_BACKGROUND_SCAN_CONCURRENCY,
+                type=ConfigEntryType.INTEGER,
+                range=(1, 8),
+                default_value=DEFAULT_BACKGROUND_SCAN_CONCURRENCY,
+                label="Background analysis concurrency",
+                description="Maximum number of tracks analyzed concurrently during the nightly "
+                "background scan. Default 1 (serial). Increase only if your hardware can handle "
+                "concurrent torch/ffmpeg work.",
+                category="audio_analysis",
+            ),
         )
 
     async def setup(self, config: CoreConfig) -> None:
@@ -313,6 +326,7 @@ async def setup(self, config: CoreConfig) -> None:
 
     async def close(self) -> None:
         """Cleanup on exit."""
+        await self._audio_analysis.close()
         await self._server.close()
 
     async def resolve_stream_url(self, player_id: str, media: PlayerMedia) -> str:

music_assistant/models/audio_analysis_provider.py

@@ -28,15 +28,13 @@ class AnalysisSessionData:
 
 
 class AudioAnalysisProvider(Provider):
-    """Base representation of an Audio Analysis Provider.
-
-    Audio Analysis Provider implementations should inherit from this base model.
-    These providers receive PCM audio chunks during streaming and produce analysis
-    results such as beat tracking, key detection, phrase boundaries, etc.
+    """
+    Base representation of an Audio Analysis Provider.
 
-    The AudioAnalysisController creates session IDs and passes them to all methods.
-    Providers implement _start_analysis and _finalize as hooks — the base class
-    manages session lifecycle, version gating, and cleanup.
+    Receives PCM audio chunks during streaming and produces analysis results
+    such as beat tracking, key detection, or loudness. The same hooks drive
+    both live playback and background scans; providers do not need to know
+    which context they are running in.
     """
 
     # Version of the analysis algorithm. Providers should increment this when
@@ -61,10 +59,9 @@ async def start_analysis(
         streamdetails: StreamDetails,
         audio_format: AudioFormat,
     ) -> bool:
-        """Start analysis for a new session.
+        """
+        Start analysis for a new session.
 
-        Checks whether analysis is needed (version gating), stores session data,
-        and calls _start_analysis for provider-specific initialization.
         Returns True if the provider accepted the session.
 
         :param session_id: Session ID created by the AudioAnalysisController.
@@ -95,11 +92,10 @@ async def _start_analysis(
         streamdetails: StreamDetails,
         audio_format: AudioFormat,
     ) -> bool:
-        """Provider-specific initialization for a new analysis session.
+        """
+        Provider-specific initialization for a new analysis session.
 
-        Called by start_analysis after version gating and session storage.
         Return False to reject the session (e.g. unsupported format).
-        Session data is available in self._sessions[session_id].
 
         :param session_id: The analysis session ID.
         :param streamdetails: The stream details for the item being analyzed.
@@ -112,53 +108,60 @@ async def process_pcm_chunk(
         session_id: str,
         pcm_chunk: bytes,
     ) -> None:
-        """Process a PCM audio chunk.
+        """
+        Process a PCM audio chunk.
 
-        Called for each chunk of audio data during streaming.
+        Implementations MUST `await` all chunk-processing work; the controller
+        relies on this to backpressure the audio source.
 
         :param session_id: The analysis session ID.
         :param pcm_chunk: Raw PCM audio data.
         """
 
     @abstractmethod
-    async def _finalize(self, session_id: str) -> None:
-        """Finalize analysis and store results.
+    async def _finalize(self, session_id: str) -> AudioAnalysisData | None:
+        """
+        Finalize analysis and return the persisted analysis (or None to skip).
 
-        Called when the track has finished streaming. Providers are responsible
-        for storing their results via mass.streams.audio_analysis.set_audio_analysis().
+        Providers store results via mass.streams.audio_analysis.set_audio_analysis().
+        The returned value drives the post_analysis hook; return None to skip it.
 
         :param session_id: The analysis session ID.
         """
 
     async def finalize(self, session_id: str) -> None:
-        """Finalize analysis and clean up session state.
-
-        Calls _finalize, then removes the session from _sessions.
-        The controller calls this method — providers override _finalize.
-
-        :param session_id: The analysis session ID.
-        """
+        """Finalize analysis, optionally fire post_analysis, and clean up state."""
+        analysis: AudioAnalysisData | None = None
         try:
-            await self._finalize(session_id)
-        finally:
-            self._sessions.pop(session_id, None)
+            analysis = await self._finalize(session_id)
+        except Exception as err:
+            self.logger.error("_finalize raised for session %s: %s", session_id, err, exc_info=err)
+        session = self._sessions.get(session_id)
+        if analysis is not None and session is not None:
+            try:
+                await self.post_analysis(session.streamdetails, analysis)
+            except Exception as err:
+                self.logger.warning(
+                    "post_analysis raised for %s: %s", self.domain, err, exc_info=err
+                )
+        self._sessions.pop(session_id, None)
 
-    async def analyze_file(
+    async def post_analysis(
         self,
         streamdetails: StreamDetails,
-    ) -> AudioAnalysisData | None:
+        analysis: AudioAnalysisData,
+    ) -> None:
         """
-        Run analysis directly on a local audio file.
+        Run side effects after analysis is finalized and persisted.
 
-        Used by the AudioAnalysisController's background scan. Providers that can
-        analyze a file without going through live PCM streaming (e.g. by handing
-        the path to FFmpeg/librosa/etc.) should override this. Default returns
-        None, meaning the provider does not support file-based analysis.
+        Default is a no-op. Implementations MUST self-gate on whether
+        `streamdetails.path` is a writable filesystem path, since this hook
+        fires for both live and background-scan analyses.
 
-        :param streamdetails: StreamDetails for the item being analyzed.
-            Contains the local file path and audio format.
+        :param streamdetails: The stream details for the analyzed item.
+        :param analysis: The analysis data that was persisted by `_finalize`.
         """
-        return None
+        return
 
     async def cancel(self, session_id: str) -> None:
         """Cancel an in-progress analysis session."""

music_assistant/providers/loudness_analysis/provider.py

@@ -83,30 +83,6 @@ async def cancel(self, session_id: str) -> None:
                 await data.ffmpeg.close()
         await super().cancel(session_id)
 
-    async def analyze_file(self, streamdetails: StreamDetails) -> AudioAnalysisData | None:
-        """Run ebur128 directly on a local audio file and return the measurement."""
-        if not isinstance(streamdetails.path, str) or not streamdetails.path:
-            return None
-        metrics = await _run_ebur128_on_file(streamdetails.path, streamdetails.audio_format)
-        if metrics is None:
-            return None
-        loudness, loudness_range, true_peak = metrics
-        if loudness is None or loudness <= LOUDNESS_MEASUREMENT_MIN_LUFS:
-            return None
-        if self.config.get_value(CONF_WRITE_REPLAYGAIN_TAGS):
-            # ReplayGain 2.0: track_gain_db = -18 - loudness_lufs
-            track_gain_db = -18.0 - loudness
-            ok = await write_replaygain_track_gain(streamdetails.path, track_gain_db)
-            if ok:
-                self.logger.debug(
-                    "Background loudness: wrote ReplayGain tag to %s", streamdetails.path
-                )
-        return AudioAnalysisData(
-            loudness_integrated=round(loudness, 2),
-            loudness_range=round(loudness_range, 2) if loudness_range is not None else None,
-            true_peak=round(true_peak, 2) if true_peak is not None else None,
-        )
-
     async def _start_analysis(
         self,
         session_id: str,
@@ -131,26 +107,26 @@ async def _start_analysis(
         self._data[session_id] = LoudnessSessionData(ffmpeg=ffmpeg)
         return True
 
-    async def _finalize(self, session_id: str) -> None:
+    async def _finalize(self, session_id: str) -> AudioAnalysisData | None:
         """Persist the final loudness measurement for the session."""
         data = self._data.pop(session_id, None)
         if not data:
-            return
+            return None
 
         await self._send_eof(data)
         try:
             await data.ffmpeg.wait()
         except Exception as err:
             self.logger.debug("Loudness analysis ffmpeg failed: %s", err)
             await data.ffmpeg.close()
-            return
+            return None
 
         metrics = _parse_ebur128_metrics(data.ffmpeg.log_history)
         await data.ffmpeg.close()
 
         session = self._sessions.get(session_id)
         if session is None:
-            return
+            return None
 
         if data.chunks_received < MIN_DURATION_SECONDS:
             self.logger.debug(
@@ -160,15 +136,15 @@ async def _finalize(self, session_id: str) -> None:
                 data.chunks_received,
                 MIN_DURATION_SECONDS,
             )
-            return
+            return None
 
         loudness, loudness_range, true_peak = metrics
         if loudness is None:
             self.logger.debug(
                 "Could not determine loudness of %s from buffer analysis",
                 session.streamdetails.uri,
             )
-            return
+            return None
 
         if loudness <= LOUDNESS_MEASUREMENT_MIN_LUFS:
             # ebur128 reports ~-70 LUFS on near-silence / cancelled streams,
@@ -180,7 +156,7 @@ async def _finalize(self, session_id: str) -> None:
                 loudness,
                 LOUDNESS_MEASUREMENT_MIN_LUFS,
             )
-            return
+            return None
 
         analysis = AudioAnalysisData(
             loudness_integrated=round(loudness, 2),
@@ -205,6 +181,29 @@ async def _finalize(self, session_id: str) -> None:
             loudness_range,
             true_peak,
         )
+        return analysis
+
+    async def post_analysis(
+        self,
+        streamdetails: StreamDetails,
+        analysis: AudioAnalysisData,
+    ) -> None:
+        """Write the ReplayGain track-gain tag back to the source file when configured."""
+        if not isinstance(streamdetails.path, str) or not streamdetails.path:
+            return
+        if not self.config.get_value(CONF_WRITE_REPLAYGAIN_TAGS):
+            return
+        if analysis.loudness_integrated is None:
+            return
+        # ReplayGain 2.0: track_gain_db = -18 - loudness_lufs
+        track_gain_db = -18.0 - analysis.loudness_integrated
+        ok = await write_replaygain_track_gain(streamdetails.path, track_gain_db)
+        if ok:
+            self.logger.debug(
+                "Wrote ReplayGain tag to %s (gain=%.2f dB)",
+                streamdetails.path,
+                track_gain_db,
+            )
 
     async def _send_eof(self, data: LoudnessSessionData) -> None:
         """Signal end-of-input to the session's ffmpeg process (idempotent)."""
@@ -234,23 +233,3 @@ def _match_float(pattern: re.Pattern[str], text: str) -> float | None:
         return float(match.group(1))
     except ValueError:
         return None
-
-
-async def _run_ebur128_on_file(
-    file_path: str, audio_format: AudioFormat
-) -> tuple[float | None, float | None, float | None] | None:
-    """Run ebur128 on a local audio file and return the (I, LRA, TP) tuple."""
-    try:
-        async with FFMpeg(
-            audio_input=file_path,
-            input_format=audio_format,
-            output_format=audio_format,
-            audio_output="NULL",
-            filter_params=["ebur128=framelog=verbose"],
-            collect_log_history=True,
-            loglevel="info",
-        ) as ffmpeg:
-            await ffmpeg.wait()
-            return _parse_ebur128_metrics(ffmpeg.log_history)
-    except Exception:
-        return None