Fix standard crossfade falling back to a hard cut on some tracks (#4253)

Author: marcelveldt

music_assistant/controllers/streams/smart_fades/fades.py

@@ -210,11 +210,9 @@ async def _drain_stderr() -> None:
 
             if proc.returncode != 0:
                 stderr_msg = "; ".join(stderr_lines) if stderr_lines else "(no stderr)"
-                raise RuntimeError(
-                    f"Smart crossfade FFmpeg failed (rc={proc.returncode}): {stderr_msg}"
-                )
+                raise RuntimeError(f"Crossfade FFmpeg failed (rc={proc.returncode}): {stderr_msg}")
             if not got_output:
-                msg = "Smart crossfade FFmpeg produced no output"
+                msg = "Crossfade FFmpeg produced no output"
                 if stderr_lines:
                     msg += f": {'; '.join(stderr_lines)}"
                 raise RuntimeError(msg)
@@ -768,6 +766,7 @@ def __init__(self, logger: logging.Logger, crossfade_duration: float = 10.0) ->
         super().__init__(logger)
         self.crossfade_duration = crossfade_duration
         self.trailing_silence_bytes: int = 0
+        self.crossfade_size: int = 0
 
     def _build(
         self,
@@ -780,14 +779,24 @@ def _build(
         fade_in_seconds = fade_in_bytes_len / pcm_format.pcm_sample_size
         # clamp CF to fit shorter inputs (defensive — normally full buffers)
         effective_cf = min(self.crossfade_duration, fade_out_seconds, fade_in_seconds)
+        # Quantize the overlap to a whole number of PCM frames and drive both the
+        # byte slice (in apply) and the acrossfade length from this one integer.
+        # apply slices the buffers on frame boundaries, so a fractional effective_cf
+        # leaves the rendered buffer a fraction of a sample short of the acrossfade
+        # duration — and acrossfade then silently produces no output at all.
+        frame_size = (pcm_format.bit_depth // 8) * pcm_format.channels
+        crossfade_bytes = int(pcm_format.pcm_sample_size * effective_cf)
+        self.crossfade_size = crossfade_bytes // frame_size * frame_size
+        crossfade_samples = self.crossfade_size // frame_size
+        effective_cf = self.crossfade_size / pcm_format.pcm_sample_size
         self.timing_info = CrossfadeTimingInfo(
             pre_crossfade_duration=max(0.0, fade_out_seconds - effective_cf),
             crossfade_duration=effective_cf,
             fadein_trimmed_duration=0.0,
             post_crossfade_duration=max(0.0, fade_in_seconds - effective_cf),
         )
         self.filters = [
-            CrossfadeFilter(logger=self.logger, crossfade_duration=effective_cf),
+            CrossfadeFilter(logger=self.logger, crossfade_samples=crossfade_samples),
         ]
 
     async def apply(
@@ -801,11 +810,16 @@ async def apply(
 
         Only the overlapping portions are crossfaded, not the full buffers.
         """
+        # crossfade_size legitimately ends up 0 for a silent/tiny buffer, so guard on
+        # the filter chain (set in _build) to still fail fast on apply-before-build,
+        # consistent with SmartFade._get_ffmpeg_filters()
+        if not self.filters:
+            raise RuntimeError("SmartFade not built — call Mixer.build() first")
         if self.trailing_silence_bytes:
             fade_out_part = fade_out_part[: len(fade_out_part) - self.trailing_silence_bytes]
-        frame_size = (pcm_format.bit_depth // 8) * pcm_format.channels
-        crossfade_size = int(pcm_format.pcm_sample_size * self.timing_info.crossfade_duration)
-        crossfade_size = (crossfade_size // frame_size) * frame_size
+        # frame-aligned overlap computed once in _build, so it exactly matches the
+        # acrossfade `ns=` length the filter was built with
+        crossfade_size = self.crossfade_size
         if crossfade_size == 0:
             # nothing to blend — concatenate without spawning ffmpeg
             for pcm_slice in iter_pcm_slices(fade_out_part, pcm_format, 1000):

music_assistant/controllers/streams/smart_fades/filters.py

@@ -239,19 +239,40 @@ class CrossfadeFilter(Filter):
     output_fadeout_label: str = "crossfade"
     output_fadein_label: str = "crossfade"
 
-    def __init__(self, logger: logging.Logger, crossfade_duration: float):
-        """Initialize crossfade filter."""
+    def __init__(
+        self,
+        logger: logging.Logger,
+        crossfade_duration: float | None = None,
+        crossfade_samples: int | None = None,
+    ):
+        """
+        Initialize crossfade filter.
+
+        :param crossfade_duration: Overlap length in seconds (emits acrossfade ``d=``).
+        :param crossfade_samples: Overlap length in PCM samples (emits acrossfade ``ns=``).
+            Prefer this when the inputs are pre-trimmed to exactly the overlap region:
+            acrossfade silently emits nothing when its requested length exceeds the
+            buffer it is fed, and a fractional ``d`` can round just past a
+            frame-aligned buffer. A sample count cannot.
+        """
+        if (crossfade_duration is None) == (crossfade_samples is None):
+            raise ValueError("Provide exactly one of crossfade_duration or crossfade_samples")
         self.crossfade_duration = crossfade_duration
+        self.crossfade_samples = crossfade_samples
         super().__init__(logger)
 
     def apply(self, input_fadein_label: str, input_fadeout_label: str) -> list[str]:
         """Apply the acrossfade filter."""
+        overlap = (
+            f"ns={self.crossfade_samples}"
+            if self.crossfade_samples is not None
+            else f"d={self.crossfade_duration}"
+        )
         # equal-power qsin curves; the default tri/tri dips ~3dB mid-fade on uncorrelated material
-        return [
-            f"{input_fadeout_label}{input_fadein_label}"
-            f"acrossfade=d={self.crossfade_duration}:c1=qsin:c2=qsin"
-        ]
+        return [f"{input_fadeout_label}{input_fadein_label}acrossfade={overlap}:c1=qsin:c2=qsin"]
 
     def __repr__(self) -> str:
         """Return string representation of CrossfadeFilter."""
+        if self.crossfade_samples is not None:
+            return f"Crossfade(ns={self.crossfade_samples})"
         return f"Crossfade(d={self.crossfade_duration:.1f}s)"

tests/controllers/streams/smart_fades/test_filters.py

@@ -201,3 +201,24 @@ def test_crossfade_uses_equal_power_curves() -> None:
     crossfade = CrossfadeFilter(logger=LOGGER, crossfade_duration=12.5)
     filter_strings = crossfade.apply("[fadein]", "[fadeout]")
     assert filter_strings == ["[fadeout][fadein]acrossfade=d=12.5:c1=qsin:c2=qsin"]
+
+
+def test_crossfade_sample_count_uses_ns() -> None:
+    """
+    A sample-count crossfade must emit ``acrossfade=ns=`` rather than ``d=``.
+
+    ffmpeg's ``acrossfade`` silently produces no output when its requested length
+    overruns the buffer it is fed; an integer sample count matches a frame-aligned
+    buffer exactly, where a fractional ``d`` can round just past it.
+    """
+    crossfade = CrossfadeFilter(logger=LOGGER, crossfade_samples=441000)
+    filter_strings = crossfade.apply("[fadein]", "[fadeout]")
+    assert filter_strings == ["[fadeout][fadein]acrossfade=ns=441000:c1=qsin:c2=qsin"]
+
+
+def test_crossfade_requires_exactly_one_length() -> None:
+    """CrossfadeFilter must reject ambiguous or missing length specifications."""
+    with pytest.raises(ValueError, match="exactly one"):
+        CrossfadeFilter(logger=LOGGER)
+    with pytest.raises(ValueError, match="exactly one"):
+        CrossfadeFilter(logger=LOGGER, crossfade_duration=5.0, crossfade_samples=220500)

tests/core/test_smartfade_transition_timings.py

@@ -220,7 +220,32 @@ def test_filter_duration_matches_clamped_timing(self) -> None:
         assert fade.timing_info.crossfade_duration == pytest.approx(6.0)
         crossfade_filter = fade.filters[0]
         assert isinstance(crossfade_filter, CrossfadeFilter)
-        assert crossfade_filter.crossfade_duration == pytest.approx(6.0)
+        assert crossfade_filter.crossfade_samples == int(6.0 * PCM.sample_rate)
+
+    def test_fractional_overlap_keeps_filter_aligned_to_buffer(self) -> None:
+        """
+        A non-integer clamped overlap keeps acrossfade ``ns=`` aligned to the buffer.
+
+        Regression for the silent "FFmpeg produced no output" fallback: a fractional
+        effective crossfade made the byte slice a fraction of a sample shorter than the
+        ``d=`` the filter requested, so ffmpeg's acrossfade emitted nothing.
+        """
+        frame_size = (PCM.bit_depth // 8) * PCM.channels
+        # ~6.3333s of audible fade-out: a real PCM buffer is frame-aligned, yet still not a
+        # whole number of seconds, so the effective crossfade stays fractional
+        fade_out_len = _seconds(6.3333) // frame_size * frame_size
+        fade = StandardCrossFade(logger=logging.getLogger(), crossfade_duration=10.0)
+        fade._build(fade_out_len, _seconds(45), PCM)
+        crossfade_filter = fade.filters[0]
+        assert isinstance(crossfade_filter, CrossfadeFilter)
+        # the source-of-truth byte size is frame-aligned ...
+        assert fade.crossfade_size % frame_size == 0
+        # ... and the acrossfade sample count is exactly that buffer, in samples
+        assert crossfade_filter.crossfade_samples == fade.crossfade_size // frame_size
+        # the timing duration round-trips from the same integer, never the other way
+        assert fade.timing_info.crossfade_duration == pytest.approx(
+            fade.crossfade_size / PCM.pcm_sample_size
+        )
 
 
 # ---------------------------------------------------------------------------
@@ -258,6 +283,52 @@ async def fake_base_apply(
         # nothing precedes the crossfade — the 6s buffer is consumed entirely by the overlap
         assert chunks[0] == crossfade_marker
 
+    @pytest.mark.asyncio
+    async def test_apply_feeds_exactly_the_filter_sample_count(
+        self, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        """
+        apply() must feed the base mixer exactly the acrossfade ``ns=`` sample count.
+
+        Otherwise ffmpeg's acrossfade receives fewer samples than requested and emits
+        nothing — the silent crossfade failure this regression guards against.
+        """
+        captured: dict[str, bytes] = {}
+
+        async def fake_base_apply(
+            _self: SmartFade,
+            fade_out_part: bytes,
+            fade_in_part: bytes | AsyncGenerator[bytes],
+            _pcm_format: AudioFormat,
+        ) -> AsyncGenerator[bytes]:
+            captured["fade_out"] = fade_out_part
+            assert isinstance(fade_in_part, bytes)
+            captured["fade_in"] = fade_in_part
+            yield b"crossfade-output"
+
+        monkeypatch.setattr(SmartFade, "apply", fake_base_apply)
+        frame_size = (PCM.bit_depth // 8) * PCM.channels
+        # frame-aligned like a real PCM buffer, but a fractional number of seconds
+        fade_out_len = _seconds(6.3333) // frame_size * frame_size
+        fade = StandardCrossFade(logger=logging.getLogger(), crossfade_duration=10.0)
+        fade._build(fade_out_len, _seconds(45), PCM)
+        crossfade_filter = fade.filters[0]
+        assert isinstance(crossfade_filter, CrossfadeFilter)
+        assert crossfade_filter.crossfade_samples is not None
+        async for _ in fade.apply(b"\x00" * fade_out_len, b"\x11" * _seconds(45), PCM):
+            pass
+        expected_bytes = crossfade_filter.crossfade_samples * frame_size
+        assert len(captured["fade_out"]) == expected_bytes
+        assert len(captured["fade_in"]) == expected_bytes
+
+    @pytest.mark.asyncio
+    async def test_apply_before_build_fails_fast(self) -> None:
+        """apply() without a prior _build() must error, not silently hard-cut."""
+        fade = StandardCrossFade(logger=logging.getLogger(), crossfade_duration=10.0)
+        with pytest.raises(RuntimeError, match="not built"):
+            async for _ in fade.apply(b"\x00" * _seconds(5), b"\x11" * _seconds(5), PCM):
+                pass
+
     @pytest.mark.asyncio
     async def test_zero_crossfade_skips_ffmpeg(self, monkeypatch: pytest.MonkeyPatch) -> None:
         """When crossfade_duration == 0, apply() must concatenate without calling ffmpeg."""