music-assistant-server

1"""
2Tests for crossfade transition timing math.
3
4Covers the ``CrossfadeTimingInfo`` contract that drives lyrics-sync correctness:
5
6  pre_crossfade_duration  + crossfade_duration         = portion attributed to A
7  fadein_trimmed_duration + crossfade_duration         = where B's listener actually is
8                                                          when CF ends (the value the
9                                                          flow loop writes into
10                                                          streamdetails.seek_position)
11
12Pure math tests use a fake ``SmartFade`` subclass that just assigns the test-provided
13timing in its ``build``. Smart/standard end-to-end behavior is exercised through
14``SmartFadesMixer.build``.
15"""
16
17from __future__ import annotations
18
19import logging
20from collections.abc import AsyncGenerator
21from unittest.mock import AsyncMock, MagicMock
22
23import numpy as np
24import pytest
25from music_assistant_models.enums import ContentType, CrossfadeMode, MediaType, StreamType
26from music_assistant_models.media_items import AudioFormat
27from music_assistant_models.streamdetails import StreamDetails
28
29import music_assistant.controllers.streams.smart_fades.mixer as mixer_module
30from music_assistant.controllers.streams.smart_fades.fades import (
31    CrossfadeTimingInfo,
32    SmartCrossFade,
33    SmartFade,
34    SmartFadeNotApplicable,
35    StandardCrossFade,
36)
37from music_assistant.controllers.streams.smart_fades.filters import (
38    CrossfadeFilter,
39    FadeOutTrimFilter,
40    GradualTimeStretchFilter,
41)
42from music_assistant.controllers.streams.smart_fades.helpers import SMART_CROSSFADE_DURATION
43from music_assistant.controllers.streams.smart_fades.mixer import SmartFadesMixer
44from music_assistant.models.audio_analysis import AudioAnalysisData
45
46PCM = AudioFormat(
47    content_type=ContentType.PCM_S16LE,
48    sample_rate=44100,
49    bit_depth=16,
50    channels=2,
51)
52SAMPLE_SIZE = PCM.pcm_sample_size  # bytes per second
53
54
55def _seconds(seconds: float) -> int:
56    """Return the number of bytes that represent ``seconds`` of PCM audio."""
57    return int(seconds * SAMPLE_SIZE)
58
59
60def _streamdetails(item_id: str = "test", provider: str = "test") -> StreamDetails:
61    """Return a minimal StreamDetails object for mixer.build() input."""
62    return StreamDetails(
63        provider=provider,
64        item_id=item_id,
65        audio_format=PCM,
66        media_type=MediaType.TRACK,
67        stream_type=StreamType.HTTP,
68    )
69
70
71def _make_mixer(analysis_for: dict[str, AudioAnalysisData] | None = None) -> SmartFadesMixer:
72    """Build a SmartFadesMixer with a stubbed StreamsController."""
73    analysis_for = analysis_for or {}
74    streams = MagicMock()
75    streams.logger = logging.getLogger("test_smartfade_transition_timings")
76    streams.audio_analysis = MagicMock()
77
78    async def _get_analysis(item_id: str, _provider: str, **_kwargs: object) -> object:
79        return analysis_for.get(item_id)
80
81    streams.audio_analysis.get_audio_analysis = AsyncMock(side_effect=_get_analysis)
82    return SmartFadesMixer(streams)
83
84
85def _beats(start: float, count: int, interval: float) -> np.ndarray:
86    """Return ``count`` beat positions starting at ``start`` spaced by ``interval`` seconds."""
87    return np.arange(count, dtype=np.float32) * interval + start
88
89
90def _analysis(
91    bpm: float,
92    beats_start: float = 0.0,
93    beats_count: int = 200,
94    duration: float | None = None,
95    rms_energy: np.ndarray | None = None,
96) -> AudioAnalysisData:
97    """Synthetic analysis data with enough beats for SmartCrossFade.build() to succeed."""
98    interval = 60.0 / bpm  # seconds per beat
99    # When an explicit duration is given, generate beats spanning the full track so the
100    # buffer-local shift (duration - 45s) leaves real beats inside the 45s window.
101    if duration is not None:
102        count = max(beats_count, int(duration / interval) + 1)
103        beats = _beats(beats_start, count, interval)
104    else:
105        beats = _beats(beats_start, beats_count, interval)
106        duration = float(beats[-1] + interval)
107    downbeats = beats[::4]  # 4/4 time signature
108    return AudioAnalysisData(
109        duration=duration,
110        bpm=bpm,
111        beats=beats.tolist(),
112        downbeats=downbeats.tolist(),
113        rms_energy=rms_energy.tolist() if rms_energy is not None else None,
114    )
115
116
117def _with_vocal_activity(
118    analysis: AudioAnalysisData,
119    windows: list[tuple[float, float]],
120) -> AudioAnalysisData:
121    """
122    Add a valid 1800-bin vocal probability timeline to an analysis row.
123
124    :param analysis: Analysis row to update.
125    :param windows: Vocal windows in full-track media seconds.
126    """
127    assert analysis.duration is not None
128    frame_duration = analysis.duration / 1800
129    probabilities = [0.0] * 1800
130    for start, end in windows:
131        for index in range(int(start / frame_duration), int(end / frame_duration)):
132            probabilities[index] = 0.9
133    analysis.extra_data = {"vocal_activity": probabilities}
134    return analysis
135
136
137class _FixedTimingFade(SmartFade):
138    """Test-only SmartFade whose build just assigns a caller-provided timing."""
139
140    def __init__(self, timing: CrossfadeTimingInfo) -> None:
141        super().__init__(logging.getLogger("test_fixed_timing_fade"))
142        self._fixed_timing = timing
143
144    def build(
145        self,
146        fade_out_bytes_len: int,
147        fade_in_bytes_len: int,
148        pcm_format: AudioFormat,
149    ) -> None:
150        """Assign the timing supplied at construction time."""
151        self.filters = []  # non-empty would normally be required, but unused here
152        self.timing_info = self._fixed_timing
153
154
155# ---------------------------------------------------------------------------
156# CrossfadeTimingInfo dataclass
157# ---------------------------------------------------------------------------
158
159
160class TestCrossfadeTimingInfo:
161    """Cover the dataclass surface used by callers."""
162
163    def test_fields_are_set(self) -> None:
164        """Constructor stores every duration on the dataclass."""
165        timing = CrossfadeTimingInfo(
166            pre_crossfade_duration=1.0,
167            crossfade_duration=2.0,
168            fadein_trimmed_duration=3.0,
169            post_crossfade_duration=4.0,
170        )
171        assert timing.pre_crossfade_duration == 1.0
172        assert timing.crossfade_duration == 2.0
173        assert timing.fadein_trimmed_duration == 3.0
174        assert timing.post_crossfade_duration == 4.0
175
176    def test_default_values(self) -> None:
177        """All fields default to 0.0 so build can populate them incrementally."""
178        timing = CrossfadeTimingInfo()
179        assert timing.pre_crossfade_duration == 0.0
180        assert timing.crossfade_duration == 0.0
181        assert timing.fadein_trimmed_duration == 0.0
182        assert timing.post_crossfade_duration == 0.0
183
184
185# ---------------------------------------------------------------------------
186# StandardCrossFade.build — timing math via the real subclass
187# ---------------------------------------------------------------------------
188
189
190class TestStandardCrossFadeBuild:
191    """StandardCrossFade.build must produce the expected timing for given inputs."""
192
193    def _build(
194        self,
195        crossfade_duration: float,
196        fade_out_seconds: float,
197        fade_in_seconds: float,
198    ) -> CrossfadeTimingInfo:
199        fade = StandardCrossFade(logger=logging.getLogger(), crossfade_duration=crossfade_duration)
200        fade.build(_seconds(fade_out_seconds), _seconds(fade_in_seconds), PCM)
201        return fade.timing_info
202
203    def test_symmetric_buffers(self) -> None:
204        """Standard with full symmetric buffers — TRIM stays 0."""
205        timing = self._build(crossfade_duration=10.0, fade_out_seconds=30, fade_in_seconds=30)
206        assert timing.crossfade_duration == pytest.approx(10.0)
207        assert timing.fadein_trimmed_duration == 0.0
208        assert timing.pre_crossfade_duration == pytest.approx(20.0)
209        assert timing.post_crossfade_duration == pytest.approx(20.0)
210
211    def test_buffer_equals_overlap(self) -> None:
212        """Standard with X == CF leaves no PRE or POST — mix output is pure overlap."""
213        timing = self._build(crossfade_duration=10.0, fade_out_seconds=10, fade_in_seconds=10)
214        assert timing.pre_crossfade_duration == pytest.approx(0.0)
215        assert timing.crossfade_duration == pytest.approx(10.0)
216        assert timing.fadein_trimmed_duration == 0.0
217        assert timing.post_crossfade_duration == pytest.approx(0.0)
218
219    def test_short_fadein_clamps_overlap(self) -> None:
220        """Fade-in shorter than the configured CF clamps the effective CF down."""
221        timing = self._build(crossfade_duration=10.0, fade_out_seconds=20, fade_in_seconds=4)
222        assert timing.crossfade_duration == pytest.approx(4.0)
223        assert timing.pre_crossfade_duration == pytest.approx(16.0)
224        assert timing.post_crossfade_duration == pytest.approx(0.0)
225        assert timing.fadein_trimmed_duration == 0.0
226
227    def test_short_fadeout_clamps_overlap(self) -> None:
228        """Fade-out shorter than the configured CF clamps the effective CF down."""
229        timing = self._build(crossfade_duration=10.0, fade_out_seconds=3, fade_in_seconds=20)
230        assert timing.crossfade_duration == pytest.approx(3.0)
231        assert timing.pre_crossfade_duration == pytest.approx(0.0)
232        assert timing.post_crossfade_duration == pytest.approx(17.0)
233
234    def test_filter_duration_matches_clamped_timing(self) -> None:
235        """The acrossfade filter must use the clamped duration, not the configured one."""
236        fade = StandardCrossFade(logger=logging.getLogger(), crossfade_duration=10.0)
237        # only 6s of (stripped) fade-out audio available
238        fade.build(_seconds(6), _seconds(45), PCM)
239        assert fade.timing_info.crossfade_duration == pytest.approx(6.0)
240        crossfade_filter = fade.filters[0]
241        assert isinstance(crossfade_filter, CrossfadeFilter)
242        assert crossfade_filter.crossfade_samples == int(6.0 * PCM.sample_rate)
243
244    def test_fractional_overlap_keeps_filter_aligned_to_buffer(self) -> None:
245        """
246        A non-integer clamped overlap keeps acrossfade ``ns=`` aligned to the buffer.
247
248        Regression for the silent "FFmpeg produced no output" fallback: a fractional
249        effective crossfade made the byte slice a fraction of a sample shorter than the
250        ``d=`` the filter requested, so ffmpeg's acrossfade emitted nothing.
251        """
252        frame_size = (PCM.bit_depth // 8) * PCM.channels
253        # ~6.3333s of audible fade-out: a real PCM buffer is frame-aligned, yet still not a
254        # whole number of seconds, so the effective crossfade stays fractional
255        fade_out_len = _seconds(6.3333) // frame_size * frame_size
256        fade = StandardCrossFade(logger=logging.getLogger(), crossfade_duration=10.0)
257        fade.build(fade_out_len, _seconds(45), PCM)
258        crossfade_filter = fade.filters[0]
259        assert isinstance(crossfade_filter, CrossfadeFilter)
260        # the source-of-truth byte size is frame-aligned ...
261        assert fade.crossfade_size % frame_size == 0
262        # ... and the acrossfade sample count is exactly that buffer, in samples
263        assert crossfade_filter.crossfade_samples == fade.crossfade_size // frame_size
264        # the timing duration round-trips from the same integer, never the other way
265        assert fade.timing_info.crossfade_duration == pytest.approx(
266            fade.crossfade_size / PCM.pcm_sample_size
267        )
268
269
270# ---------------------------------------------------------------------------
271# StandardCrossFade.apply — byte slicing must follow the clamped timing
272# ---------------------------------------------------------------------------
273
274
275class TestStandardCrossFadeApplySlicing:
276    """apply() must slice the fade-out buffer by the clamped duration, not the configured one."""
277
278    @pytest.mark.asyncio
279    async def test_apply_slices_with_clamped_duration(
280        self, monkeypatch: pytest.MonkeyPatch
281    ) -> None:
282        """A 6s fade-out with a 10s configured CF must hand the base mixer all 6s, no more."""
283        captured: dict[str, bytes] = {}
284        crossfade_marker = b"crossfade-output"
285
286        async def fake_base_apply(
287            _self: SmartFade,
288            fade_out_part: bytes,
289            _fade_in_part: bytes | AsyncGenerator[bytes],
290            _pcm_format: AudioFormat,
291        ) -> AsyncGenerator[bytes]:
292            captured["fade_out"] = fade_out_part
293            yield crossfade_marker
294
295        monkeypatch.setattr(SmartFade, "apply", fake_base_apply)
296        fade = StandardCrossFade(logger=logging.getLogger(), crossfade_duration=10.0)
297        fade.build(_seconds(6), _seconds(45), PCM)
298        chunks = [
299            chunk async for chunk in fade.apply(b"\x00" * _seconds(6), b"\x00" * _seconds(45), PCM)
300        ]
301        assert len(captured["fade_out"]) == _seconds(6)
302        # nothing precedes the crossfade — the 6s buffer is consumed entirely by the overlap
303        assert chunks[0] == crossfade_marker
304
305    @pytest.mark.asyncio
306    async def test_apply_feeds_exactly_the_filter_sample_count(
307        self, monkeypatch: pytest.MonkeyPatch
308    ) -> None:
309        """
310        apply() must feed the base mixer exactly the acrossfade ``ns=`` sample count.
311
312        Otherwise ffmpeg's acrossfade receives fewer samples than requested and emits
313        nothing — the silent crossfade failure this regression guards against.
314        """
315        captured: dict[str, bytes] = {}
316
317        async def fake_base_apply(
318            _self: SmartFade,
319            fade_out_part: bytes,
320            fade_in_part: bytes | AsyncGenerator[bytes],
321            _pcm_format: AudioFormat,
322        ) -> AsyncGenerator[bytes]:
323            captured["fade_out"] = fade_out_part
324            assert isinstance(fade_in_part, bytes)
325            captured["fade_in"] = fade_in_part
326            yield b"crossfade-output"
327
328        monkeypatch.setattr(SmartFade, "apply", fake_base_apply)
329        frame_size = (PCM.bit_depth // 8) * PCM.channels
330        # frame-aligned like a real PCM buffer, but a fractional number of seconds
331        fade_out_len = _seconds(6.3333) // frame_size * frame_size
332        fade = StandardCrossFade(logger=logging.getLogger(), crossfade_duration=10.0)
333        fade.build(fade_out_len, _seconds(45), PCM)
334        crossfade_filter = fade.filters[0]
335        assert isinstance(crossfade_filter, CrossfadeFilter)
336        assert crossfade_filter.crossfade_samples is not None
337        async for _ in fade.apply(b"\x00" * fade_out_len, b"\x11" * _seconds(45), PCM):
338            pass
339        expected_bytes = crossfade_filter.crossfade_samples * frame_size
340        assert len(captured["fade_out"]) == expected_bytes
341        assert len(captured["fade_in"]) == expected_bytes
342
343    @pytest.mark.asyncio
344    async def test_apply_before_build_fails_fast(self) -> None:
345        """apply() without a prior build() must error, not silently hard-cut."""
346        fade = StandardCrossFade(logger=logging.getLogger(), crossfade_duration=10.0)
347        with pytest.raises(RuntimeError, match="not built"):
348            async for _ in fade.apply(b"\x00" * _seconds(5), b"\x11" * _seconds(5), PCM):
349                pass
350
351    @pytest.mark.asyncio
352    async def test_zero_crossfade_skips_ffmpeg(self, monkeypatch: pytest.MonkeyPatch) -> None:
353        """When crossfade_duration == 0, apply() must concatenate without calling ffmpeg."""
354        base_apply_invoked: list[bool] = []
355
356        async def _base_apply_sentinel(
357            _self: SmartFade,
358            _fade_out_part: bytes,
359            _fade_in_part: bytes | AsyncGenerator[bytes],
360            _pcm_format: AudioFormat,
361        ) -> AsyncGenerator[bytes]:
362            base_apply_invoked.append(True)
363            yield b""
364
365        monkeypatch.setattr(SmartFade, "apply", _base_apply_sentinel)
366
367        fade = StandardCrossFade(logger=logging.getLogger(), crossfade_duration=10.0)
368        # fade_out_bytes_len=0 → effective_cf = min(10, 0, 45) = 0 → crossfade_duration == 0
369        fade.build(0, _seconds(45), PCM)
370        assert fade.timing_info.crossfade_duration == 0.0
371
372        fade_out_data = b"\x00" * _seconds(5)
373        fade_in_data = b"\x11" * _seconds(5)
374        chunks = [chunk async for chunk in fade.apply(fade_out_data, fade_in_data, PCM)]
375        combined = b"".join(chunks)
376        assert len(combined) == _seconds(10), f"Expected {_seconds(10)} bytes, got {len(combined)}"
377        assert combined[0:1] == b"\x00", "fade_out bytes should come first"
378        assert combined[-1:] == b"\x11", "fade_in bytes should come last"
379        assert not base_apply_invoked, (
380            "SmartFade.apply (ffmpeg path) must not be called for zero-length crossfade"
381        )
382
383
384# ---------------------------------------------------------------------------
385# Pure math/invariant tests via _FixedTimingFade (no beat-alignment dependency)
386# ---------------------------------------------------------------------------
387
388
389class TestLyricsSyncInvariants:
390    """Invariants the flow loop and per-track loop rely on."""
391
392    def _continuation_offset(self, t: CrossfadeTimingInfo) -> float:
393        """Return the value the flow loop writes into streamdetails.seek_position."""
394        return t.fadein_trimmed_duration + t.crossfade_duration
395
396    def _fadeout_share(self, t: CrossfadeTimingInfo) -> float:
397        """Return the seconds of mix output attributed to the outgoing track."""
398        return t.pre_crossfade_duration + t.crossfade_duration
399
400    def test_continuation_offset_no_trim(self) -> None:
401        """Without trim the listener is exactly CF seconds into the incoming track."""
402        timing = CrossfadeTimingInfo(
403            pre_crossfade_duration=20.0,
404            crossfade_duration=10.0,
405            fadein_trimmed_duration=0.0,
406            post_crossfade_duration=20.0,
407        )
408        assert self._continuation_offset(timing) == pytest.approx(10.0)
409
410    def test_continuation_offset_with_trim(self) -> None:
411        """With trim the listener is TRIM + CF seconds into the incoming track."""
412        timing = CrossfadeTimingInfo(
413            pre_crossfade_duration=29.0,
414            crossfade_duration=16.0,
415            fadein_trimmed_duration=3.0,
416            post_crossfade_duration=26.0,
417        )
418        assert self._continuation_offset(timing) == pytest.approx(19.0)
419
420    def test_fadeout_share_accounts_for_full_outgoing_input(self) -> None:
421        """PRE + CF equals the full outgoing input — A is fully accounted for."""
422        timing = CrossfadeTimingInfo(
423            pre_crossfade_duration=29.0,
424            crossfade_duration=16.0,
425            fadein_trimmed_duration=3.0,
426            post_crossfade_duration=26.0,
427        )
428        # fade_out_seconds = PRE + CF = 45
429        assert self._fadeout_share(timing) == pytest.approx(45.0)
430
431    def test_fixed_timing_fade_round_trip(self) -> None:
432        """_FixedTimingFade.timing_info returns whatever was passed in."""
433        original = CrossfadeTimingInfo(
434            pre_crossfade_duration=1.0,
435            crossfade_duration=2.0,
436            fadein_trimmed_duration=3.0,
437            post_crossfade_duration=4.0,
438        )
439        fade = _FixedTimingFade(original)
440        fade.build(0, 0, PCM)
441        assert fade.timing_info == original
442
443
444# ---------------------------------------------------------------------------
445# SmartFadesMixer.build() — the entry point the flow loop calls
446# ---------------------------------------------------------------------------
447
448
449class TestMixerBuild:
450    """build() resolves smart vs standard, primes filters, and returns a SmartFade."""
451
452    @pytest.mark.asyncio
453    async def test_standard_mode_returns_standard_crossfade(
454        self, monkeypatch: pytest.MonkeyPatch
455    ) -> None:
456        """Standard mode always builds a StandardCrossFade with the configured duration."""
457
458        async def identity_strip(audio_data: bytes, **_kwargs: object) -> bytes:
459            return audio_data  # no stripping — test pure timing math
460
461        monkeypatch.setattr(mixer_module, "strip_silence", identity_strip)
462        mixer = _make_mixer()
463        fade = await mixer.build(
464            fade_in_streamdetails=_streamdetails("in"),
465            fade_out_streamdetails=_streamdetails("out"),
466            pcm_format=PCM,
467            standard_crossfade_duration=8,
468            mode=CrossfadeMode.STANDARD_CROSSFADE,
469            fade_out_data=b"\x00" * _seconds(20),
470            fade_in_bytes_len=_seconds(20),
471        )
472        assert isinstance(fade, StandardCrossFade)
473        timing = fade.timing_info
474        assert timing.crossfade_duration == pytest.approx(8.0)
475        assert timing.fadein_trimmed_duration == 0.0
476        assert timing.pre_crossfade_duration == pytest.approx(12.0)
477        assert timing.post_crossfade_duration == pytest.approx(12.0)
478        # continuation offset for standard fades is just CF
479        assert timing.fadein_trimmed_duration + timing.crossfade_duration == pytest.approx(8.0)
480
481    @pytest.mark.asyncio
482    async def test_smart_mode_returns_smart_crossfade_when_analysis_available(self) -> None:
483        """With audio analysis on both tracks, build() returns a SmartCrossFade."""
484        analysis_out = _analysis(120.0)
485        analysis_in = _analysis(124.0, beats_start=0.4)
486        mixer = _make_mixer({"out": analysis_out, "in": analysis_in})
487        fade_out_data = b"\x00" * _seconds(SMART_CROSSFADE_DURATION)
488        fade = await mixer.build(
489            fade_in_streamdetails=_streamdetails("in"),
490            fade_out_streamdetails=_streamdetails("out"),
491            pcm_format=PCM,
492            standard_crossfade_duration=10,
493            mode=CrossfadeMode.SMART_CROSSFADE,
494            fade_out_data=fade_out_data,
495            fade_in_bytes_len=_seconds(SMART_CROSSFADE_DURATION),
496        )
497        assert isinstance(fade, SmartCrossFade)
498        timing = fade.timing_info
499        # SmartCrossFade applies a beat-aligned trim, so TRIM is non-zero.
500        assert timing.fadein_trimmed_duration > 0
501        assert timing.crossfade_duration > 0
502        # Invariants the flow loop depends on:
503        #   PRE + CF == rendered_fade_out_seconds  (A's audio fully accounted for)
504        #   TRIM + CF + POST == fade_in_seconds    (B's audio fully accounted for)
505        # When time-stretch is active, rendered_fade_out_seconds < buffer_duration because
506        # rubberband compresses the tail; savings come from the plan's TempoPlan.
507        rendered_fade_out_seconds = fade.effective_end - _savings_until(fade, fade.effective_end)
508        fade_in_seconds = float(SMART_CROSSFADE_DURATION)
509        assert timing.pre_crossfade_duration + timing.crossfade_duration == pytest.approx(
510            rendered_fade_out_seconds, abs=0.05
511        )
512        assert (
513            timing.fadein_trimmed_duration
514            + timing.crossfade_duration
515            + timing.post_crossfade_duration
516            == pytest.approx(fade_in_seconds, abs=0.01)
517        )
518
519    @pytest.mark.asyncio
520    async def test_smart_mode_falls_back_when_no_analysis(
521        self, monkeypatch: pytest.MonkeyPatch
522    ) -> None:
523        """No audio analysis -> build() falls back to StandardCrossFade."""
524
525        async def identity_strip(audio_data: bytes, **_kwargs: object) -> bytes:
526            return audio_data  # no stripping — test pure timing math
527
528        monkeypatch.setattr(mixer_module, "strip_silence", identity_strip)
529        mixer = _make_mixer()
530        fade = await mixer.build(
531            fade_in_streamdetails=_streamdetails("in"),
532            fade_out_streamdetails=_streamdetails("out"),
533            pcm_format=PCM,
534            standard_crossfade_duration=10,
535            mode=CrossfadeMode.SMART_CROSSFADE,
536            fade_out_data=b"\x00" * _seconds(45),
537            fade_in_bytes_len=_seconds(45),
538        )
539        assert isinstance(fade, StandardCrossFade)
540        timing = fade.timing_info
541        assert timing.crossfade_duration == pytest.approx(10.0)
542        assert timing.fadein_trimmed_duration == 0.0
543        assert timing.pre_crossfade_duration == pytest.approx(35.0)
544        assert timing.post_crossfade_duration == pytest.approx(35.0)
545
546    @pytest.mark.asyncio
547    async def test_smart_mode_falls_back_when_analysis_missing_bpm(
548        self, monkeypatch: pytest.MonkeyPatch
549    ) -> None:
550        """Analysis present but missing bpm/beats -> falls back to standard."""
551
552        async def identity_strip(audio_data: bytes, **_kwargs: object) -> bytes:
553            return audio_data
554
555        monkeypatch.setattr(mixer_module, "strip_silence", identity_strip)
556        incomplete = AudioAnalysisData(duration=180.0, bpm=None, beats=None)
557        mixer = _make_mixer({"out": incomplete, "in": incomplete})
558        fade = await mixer.build(
559            fade_in_streamdetails=_streamdetails("in"),
560            fade_out_streamdetails=_streamdetails("out"),
561            pcm_format=PCM,
562            standard_crossfade_duration=10,
563            mode=CrossfadeMode.SMART_CROSSFADE,
564            fade_out_data=b"\x00" * _seconds(30),
565            fade_in_bytes_len=_seconds(30),
566        )
567        assert isinstance(fade, StandardCrossFade)
568        assert fade.timing_info.fadein_trimmed_duration == 0.0
569
570    @pytest.mark.asyncio
571    async def test_build_returns_fade_with_timing_info_readable(
572        self, monkeypatch: pytest.MonkeyPatch
573    ) -> None:
574        """timing_info is queryable immediately after build() — no apply() needed."""
575
576        async def identity_strip(audio_data: bytes, **_kwargs: object) -> bytes:
577            return audio_data
578
579        monkeypatch.setattr(mixer_module, "strip_silence", identity_strip)
580        mixer = _make_mixer()
581        fade = await mixer.build(
582            fade_in_streamdetails=_streamdetails("in"),
583            fade_out_streamdetails=_streamdetails("out"),
584            pcm_format=PCM,
585            standard_crossfade_duration=10,
586            mode=CrossfadeMode.STANDARD_CROSSFADE,
587            fade_out_data=b"\x00" * _seconds(15),
588            fade_in_bytes_len=_seconds(15),
589        )
590        assert isinstance(fade.timing_info, CrossfadeTimingInfo)
591
592    @pytest.mark.asyncio
593    async def test_standard_mode_strips_trailing_silence_before_timing(
594        self, monkeypatch: pytest.MonkeyPatch
595    ) -> None:
596        """Timing must be computed from the stripped length; the plan is stored on the fade."""
597
598        async def fake_strip(
599            audio_data: bytes, *, reverse: bool = False, **_kwargs: object
600        ) -> bytes:
601            assert reverse is True
602            return audio_data[: -_seconds(3)]  # pretend 3s of trailing silence
603
604        monkeypatch.setattr(mixer_module, "strip_silence", fake_strip)
605        mixer = _make_mixer()
606        fade_out_data = b"\x00" * _seconds(45)
607        smart_fade = await mixer.build(
608            fade_in_streamdetails=_streamdetails("b"),
609            fade_out_streamdetails=_streamdetails("a"),
610            pcm_format=PCM,
611            standard_crossfade_duration=10,
612            mode=CrossfadeMode.STANDARD_CROSSFADE,
613            fade_out_data=fade_out_data,
614            fade_in_bytes_len=_seconds(45),
615        )
616        assert isinstance(smart_fade, StandardCrossFade)
617        assert smart_fade.trailing_silence_bytes == _seconds(3)
618        timing = smart_fade.timing_info
619        assert timing.pre_crossfade_duration + timing.crossfade_duration == pytest.approx(42.0)
620
621    @pytest.mark.asyncio
622    async def test_smart_mode_never_measures_silence(self, monkeypatch: pytest.MonkeyPatch) -> None:
623        """The smart path must never call strip_silence — beat coordinates map onto the full buffer."""
624
625        async def fail_strip(*_args: object, **_kwargs: object) -> bytes:
626            raise AssertionError("strip_silence must not be called on the smart path")
627
628        monkeypatch.setattr(mixer_module, "strip_silence", fail_strip)
629        mixer = _make_mixer(analysis_for={"a": _analysis(bpm=120.0), "b": _analysis(bpm=120.0)})
630        fade_out_data = b"\x00" * _seconds(45)
631        smart_fade = await mixer.build(
632            fade_in_streamdetails=_streamdetails("b"),
633            fade_out_streamdetails=_streamdetails("a"),
634            pcm_format=PCM,
635            standard_crossfade_duration=10,
636            mode=CrossfadeMode.SMART_CROSSFADE,
637            fade_out_data=fade_out_data,
638            fade_in_bytes_len=_seconds(45),
639        )
640        assert isinstance(smart_fade, SmartCrossFade)
641
642    @pytest.mark.asyncio
643    async def test_smart_fallback_to_standard_strips(self, monkeypatch: pytest.MonkeyPatch) -> None:
644        """Smart mode without analysis falls back to standard — which must measure silence."""
645
646        async def fake_strip(audio_data: bytes, **_kwargs: object) -> bytes:
647            return audio_data[: -_seconds(5)]
648
649        monkeypatch.setattr(mixer_module, "strip_silence", fake_strip)
650        mixer = _make_mixer(analysis_for={})  # no analysis available
651        smart_fade = await mixer.build(
652            fade_in_streamdetails=_streamdetails("b"),
653            fade_out_streamdetails=_streamdetails("a"),
654            pcm_format=PCM,
655            standard_crossfade_duration=10,
656            mode=CrossfadeMode.SMART_CROSSFADE,
657            fade_out_data=b"\x00" * _seconds(45),
658            fade_in_bytes_len=_seconds(45),
659        )
660        assert isinstance(smart_fade, StandardCrossFade)
661        assert smart_fade.trailing_silence_bytes == _seconds(5)
662
663    @pytest.mark.asyncio
664    async def test_smart_fallback_retains_an_audible_outgoing_vocal(
665        self, monkeypatch: pytest.MonkeyPatch
666    ) -> None:
667        """Validated FireRed activity extends a fallback trim only within audible RMS energy."""
668
669        async def fake_strip(audio_data: bytes, **_kwargs: object) -> bytes:
670            return audio_data[: _seconds(30)]
671
672        def fail_smart_build(*_args: object, **_kwargs: object) -> None:
673            raise SmartFadeNotApplicable("forced fallback")
674
675        monkeypatch.setattr(mixer_module, "strip_silence", fake_strip)
676        monkeypatch.setattr(SmartCrossFade, "build", fail_smart_build)
677        outgoing = _with_vocal_activity(
678            _analysis(
679                120.0,
680                duration=240.0,
681                rms_energy=_rms_with_silent_tail(240.0, 5.0),
682            ),
683            [(228.0, 232.0)],
684        )
685        mixer = _make_mixer({"a": outgoing, "b": _analysis(120.0, duration=240.0)})
686
687        fade = await mixer.build(
688            fade_in_streamdetails=_streamdetails("b"),
689            fade_out_streamdetails=_streamdetails("a"),
690            pcm_format=PCM,
691            standard_crossfade_duration=10,
692            mode=CrossfadeMode.SMART_CROSSFADE,
693            fade_out_data=b"\x00" * _seconds(45),
694            fade_in_bytes_len=_seconds(45),
695        )
696
697        assert isinstance(fade, StandardCrossFade)
698        retained_seconds = (_seconds(45) - fade.trailing_silence_bytes) / SAMPLE_SIZE
699        assert retained_seconds == pytest.approx(37.75, abs=1 / PCM.sample_rate)
700
701    @pytest.mark.asyncio
702    async def test_smart_fallback_invalid_vocal_data_matches_missing_data(
703        self, monkeypatch: pytest.MonkeyPatch
704    ) -> None:
705        """Missing and stale vocal metadata keep the exact standard silence trim."""
706
707        async def fake_strip(audio_data: bytes, **_kwargs: object) -> bytes:
708            return audio_data[: _seconds(30)]
709
710        def fail_smart_build(*_args: object, **_kwargs: object) -> None:
711            raise SmartFadeNotApplicable("forced fallback")
712
713        monkeypatch.setattr(mixer_module, "strip_silence", fake_strip)
714        monkeypatch.setattr(SmartCrossFade, "build", fail_smart_build)
715        missing = _analysis(120.0, duration=240.0, rms_energy=_rms_with_silent_tail(240.0, 5.0))
716        stale = _analysis(120.0, duration=240.0, rms_energy=_rms_with_silent_tail(240.0, 5.0))
717        stale.extra_data = {
718            "vocal_activity": {
719                "model": "firered_aed",
720                "frame_duration": 0.1,
721                "probabilities": [0.9] * 2400,
722            }
723        }
724
725        trims: list[int] = []
726        for outgoing in (missing, stale):
727            mixer = _make_mixer({"a": outgoing, "b": _analysis(120.0, duration=240.0)})
728            fade = await mixer.build(
729                fade_in_streamdetails=_streamdetails("b"),
730                fade_out_streamdetails=_streamdetails("a"),
731                pcm_format=PCM,
732                standard_crossfade_duration=10,
733                mode=CrossfadeMode.SMART_CROSSFADE,
734                fade_out_data=b"\x00" * _seconds(45),
735                fade_in_bytes_len=_seconds(45),
736            )
737            assert isinstance(fade, StandardCrossFade)
738            trims.append(fade.trailing_silence_bytes)
739
740        assert trims == [_seconds(15), _seconds(15)]
741
742    @pytest.mark.asyncio
743    async def test_smart_fallback_caps_vocal_retention_at_the_rms_boundary(
744        self, monkeypatch: pytest.MonkeyPatch
745    ) -> None:
746        """FireRed cannot restore a long low-energy tail beyond the audible RMS boundary."""
747
748        async def fake_strip(audio_data: bytes, **_kwargs: object) -> bytes:
749            return audio_data[: _seconds(20)]
750
751        def fail_smart_build(*_args: object, **_kwargs: object) -> None:
752            raise SmartFadeNotApplicable("forced fallback")
753
754        monkeypatch.setattr(mixer_module, "strip_silence", fake_strip)
755        monkeypatch.setattr(SmartCrossFade, "build", fail_smart_build)
756        outgoing = _with_vocal_activity(
757            _analysis(
758                120.0,
759                duration=240.0,
760                rms_energy=_rms_with_silent_tail(240.0, 20.0),
761            ),
762            [(225.0, 235.0)],
763        )
764        mixer = _make_mixer({"a": outgoing, "b": _analysis(120.0, duration=240.0)})
765
766        fade = await mixer.build(
767            fade_in_streamdetails=_streamdetails("b"),
768            fade_out_streamdetails=_streamdetails("a"),
769            pcm_format=PCM,
770            standard_crossfade_duration=10,
771            mode=CrossfadeMode.SMART_CROSSFADE,
772            fade_out_data=b"\x00" * _seconds(45),
773            fade_in_bytes_len=_seconds(45),
774        )
775
776        assert isinstance(fade, StandardCrossFade)
777        retained_seconds = (_seconds(45) - fade.trailing_silence_bytes) / SAMPLE_SIZE
778        assert retained_seconds == pytest.approx(20.0)
779
780    @pytest.mark.asyncio
781    async def test_standard_mode_fully_silent_tail(self, monkeypatch: pytest.MonkeyPatch) -> None:
782        """A fully silent tail — timing collapses to zero; trailing_silence_bytes is the full input."""
783
784        async def fake_strip(_audio_data: bytes, **_kwargs: object) -> bytes:
785            return b""
786
787        monkeypatch.setattr(mixer_module, "strip_silence", fake_strip)
788        mixer = _make_mixer()
789        smart_fade = await mixer.build(
790            fade_in_streamdetails=_streamdetails("b"),
791            fade_out_streamdetails=_streamdetails("a"),
792            pcm_format=PCM,
793            standard_crossfade_duration=10,
794            mode=CrossfadeMode.STANDARD_CROSSFADE,
795            fade_out_data=b"\x00" * _seconds(45),
796            fade_in_bytes_len=_seconds(45),
797        )
798        assert isinstance(smart_fade, StandardCrossFade)
799        assert smart_fade.trailing_silence_bytes == _seconds(45)
800        timing = smart_fade.timing_info
801        assert timing.pre_crossfade_duration == 0.0
802        assert timing.crossfade_duration == 0.0
803
804    @pytest.mark.asyncio
805    async def test_standard_mode_measurement_failure_degrades_gracefully(
806        self, monkeypatch: pytest.MonkeyPatch
807    ) -> None:
808        """A strip_silence failure must not propagate — build degrades with trailing_silence_bytes=0."""
809
810        async def broken_strip(*_args: object, **_kwargs: object) -> bytes:
811            raise OSError("ffmpeg spawn failed")
812
813        monkeypatch.setattr(mixer_module, "strip_silence", broken_strip)
814        mixer = _make_mixer()
815        fade_out_data = b"\x00" * _seconds(45)
816        smart_fade = await mixer.build(
817            fade_in_streamdetails=_streamdetails("b"),
818            fade_out_streamdetails=_streamdetails("a"),
819            pcm_format=PCM,
820            standard_crossfade_duration=10,
821            mode=CrossfadeMode.STANDARD_CROSSFADE,
822            fade_out_data=fade_out_data,
823            fade_in_bytes_len=_seconds(45),
824        )
825        assert isinstance(smart_fade, StandardCrossFade)
826        assert smart_fade.trailing_silence_bytes == 0
827        timing = smart_fade.timing_info
828        assert timing.pre_crossfade_duration + timing.crossfade_duration == pytest.approx(45.0)
829
830    @pytest.mark.asyncio
831    async def test_apply_executes_silence_trim_plan(self, monkeypatch: pytest.MonkeyPatch) -> None:
832        """apply() must slice out trailing_silence_bytes before crossfading."""
833
834        async def fake_strip(audio_data: bytes, **_kwargs: object) -> bytes:
835            return audio_data[: -_seconds(3)]  # pretend 3s of trailing silence
836
837        monkeypatch.setattr(mixer_module, "strip_silence", fake_strip)
838
839        captured: dict[str, bytes] = {}
840        crossfade_marker = b"crossfade-output"
841
842        async def fake_base_apply(
843            _self: SmartFade,
844            fade_out_part: bytes,
845            _fade_in_part: bytes | AsyncGenerator[bytes],
846            _pcm_format: AudioFormat,
847        ) -> AsyncGenerator[bytes]:
848            captured["fade_out"] = fade_out_part
849            yield crossfade_marker
850
851        monkeypatch.setattr(SmartFade, "apply", fake_base_apply)
852
853        mixer = _make_mixer()
854        fade_out_data = b"\x00" * _seconds(45)
855        smart_fade = await mixer.build(
856            fade_in_streamdetails=_streamdetails("b"),
857            fade_out_streamdetails=_streamdetails("a"),
858            pcm_format=PCM,
859            standard_crossfade_duration=10,
860            mode=CrossfadeMode.STANDARD_CROSSFADE,
861            fade_out_data=fade_out_data,
862            fade_in_bytes_len=_seconds(45),
863        )
864        assert isinstance(smart_fade, StandardCrossFade)
865
866        chunks = [
867            chunk
868            async for chunk in smart_fade.apply(
869                fade_out_part=fade_out_data,
870                fade_in_part=b"\x00" * _seconds(45),
871                pcm_format=PCM,
872            )
873        ]
874
875        # The yielded pre-crossfade bytes are where the trim actually lands:
876        # 45s input - 3s measured silence - 10s clamped CF = 32s. Without the
877        # trim apply() would yield 35s here.
878        marker_idx = chunks.index(crossfade_marker)
879        pre_crossfade_bytes = sum(len(chunk) for chunk in chunks[:marker_idx])
880        assert pre_crossfade_bytes == _seconds(32)
881        assert len(captured["fade_out"]) == _seconds(10)
882
883
884# ---------------------------------------------------------------------------
885# SmartCrossFade — silence-aware fade-out anchoring
886# ---------------------------------------------------------------------------
887
888LOGGER = logging.getLogger(__name__)
889
890
891def _rms_with_silent_tail(track_duration: float, silent_tail: float) -> np.ndarray:
892    bins = np.full(1800, 0.5, dtype=np.float32)
893    bins[0] = 1.0
894    if silent_tail > 0:
895        bins[-int(silent_tail / track_duration * 1800) :] = 0.001
896    return bins
897
898
899class TestSilenceAwareAnchoring:
900    """SmartCrossFade must anchor the fade where audible content ends."""
901
902    def _build_fade(self, silent_tail: float) -> SmartCrossFade:
903        duration = 240.0
904        fade = SmartCrossFade(
905            logger=LOGGER,
906            fade_out_analysis=_analysis(
907                bpm=120.0,
908                duration=duration,
909                rms_energy=_rms_with_silent_tail(duration, silent_tail),
910            ),
911            fade_in_analysis=_analysis(bpm=120.0, duration=duration),
912        )
913        fade.build(_seconds(45), _seconds(45), PCM)
914        return fade
915
916    def test_silent_tail_moves_the_anchor(self) -> None:
917        """A 10s silent tail shortens the audible anchor to ~35s and inserts FadeOutTrimFilter."""
918        fade = self._build_fade(silent_tail=10.0)
919        assert fade.effective_end == pytest.approx(35.0, abs=0.3)
920        # the rendered fade-out covers only the audible region
921        timing = fade.timing_info
922        assert timing.pre_crossfade_duration + timing.crossfade_duration == pytest.approx(
923            fade.effective_end, abs=0.05
924        )
925        # tail trim is the FIRST filter so later schedules see the trimmed stream
926        assert isinstance(fade.filters[0], FadeOutTrimFilter)
927        assert fade.filters[0].fadeout_end_pos == pytest.approx(fade.effective_end)
928
929    def test_no_silence_keeps_buffer_end_anchor(self) -> None:
930        """Without silence, effective_end equals the full buffer and no trim filter is added."""
931        fade = self._build_fade(silent_tail=0.0)
932        assert fade.effective_end == pytest.approx(45.0, abs=0.3)
933        assert not any(isinstance(f, FadeOutTrimFilter) for f in fade.filters)
934
935    def test_sub_tolerance_silent_tail_snaps_anchor_to_buffer_end(self) -> None:
936        """A silent tail below the trim tolerance keeps the anchor at the rendered buffer end."""
937        fade = self._build_fade(silent_tail=0.4)
938        assert fade.effective_end == pytest.approx(45.0)
939        assert not any(isinstance(f, FadeOutTrimFilter) for f in fade.filters)
940
941    def test_mostly_silent_tail_raises_for_fallback(self) -> None:
942        """A tail with only ~5s of audible content raises SmartFadeNotApplicable so the caller falls back."""
943        with pytest.raises(SmartFadeNotApplicable, match="silent"):
944            self._build_fade(silent_tail=40.0)
945
946    def test_partial_buffer_keeps_beats_aligned(self) -> None:
947        """
948        The live holdback buffer is rarely exactly 45s.
949
950        Beat coordinates must use the actual buffer length or every downbeat snap
951        is off by the difference.
952        """
953        duration = 240.0
954        fade = SmartCrossFade(
955            logger=LOGGER,
956            fade_out_analysis=_analysis(bpm=120.0, duration=duration),
957            fade_in_analysis=_analysis(bpm=120.0, duration=duration),
958        )
959        # 44.3s buffer: 0.7s short of the constant, like a real partial final chunk
960        partial_bytes = int(PCM.pcm_sample_size * 44.3)
961        frame_size = (PCM.bit_depth // 8) * PCM.channels
962        partial_bytes = (partial_bytes // frame_size) * frame_size
963        fade.build(partial_bytes, _seconds(45), PCM)
964        buffer_duration = partial_bytes / PCM.pcm_sample_size
965        # beats are on a strict 0.5s grid from t=0 in the analysis fixture;
966        # in real buffer coordinates each beat must satisfy
967        # (beat + duration - buffer_duration) % 0.5 == 0
968        offset = duration - buffer_duration
969        for beat in fade.fade_out_beats[:8]:
970            track_pos = beat + offset
971            assert abs(track_pos % 0.5) < 1e-3 or abs(track_pos % 0.5 - 0.5) < 1e-3, (
972                f"beat {beat:.4f} maps to track_pos {track_pos:.4f}, "
973                f"not on 0.5s grid (offset={offset:.4f})"
974            )
975        # the snapped crossfade start must land on a real downbeat (2s grid), not 0.7s off
976        crossfade_start = fade.effective_end - fade.timing_info.crossfade_duration
977        start_track_pos = crossfade_start + offset
978        assert abs(start_track_pos % 2.0) < 0.02 or abs(start_track_pos % 2.0 - 2.0) < 0.02, (
979            f"crossfade start {crossfade_start:.4f} maps to track_pos {start_track_pos:.4f}, "
980            f"not on 2s downbeat grid (offset={offset:.4f})"
981        )
982
983    def test_fadeout_beats_are_masked_to_effective_end(self) -> None:
984        """Beats in the silent tail are dropped so no downbeat sits beyond effective_end."""
985        fade = self._build_fade(silent_tail=10.0)
986        assert fade.fade_out_beats.min() >= 0.0
987        assert fade.fade_out_beats.max() <= fade.effective_end + 0.01
988
989    def test_short_audible_tail_keeps_crossfade_inside_it(self) -> None:
990        """A crossfade longer than the audible tail is capped so no schedule goes negative."""
991        duration = 240.0
992        fade = SmartCrossFade(
993            logger=LOGGER,
994            fade_out_analysis=_analysis(
995                bpm=120.0,
996                duration=duration,
997                rms_energy=_rms_with_silent_tail(duration, 33.0),
998            ),
999            # 115 vs 120 BPM is within the stretch threshold, so the tempo ramp is active
1000            fade_in_analysis=_analysis(bpm=115.0, duration=duration),
1001        )
1002        fade.build(_seconds(45), _seconds(45), PCM)
1003        assert fade.timing_info.crossfade_duration <= fade.effective_end + 1e-6
1004        # the capped crossfade consumes the whole audible tail, so the stretch is skipped
1005        assert not any(isinstance(f, GradualTimeStretchFilter) for f in fade.filters)
1006
1007
1008# ---------------------------------------------------------------------------
1009# SmartCrossFade — rubberband stretch savings compensation
1010# ---------------------------------------------------------------------------
1011
1012
1013def _savings_until(fade: SmartCrossFade, t: float) -> float:
1014    """Rendered-time savings of the built fade's tempo plan up to input time t."""
1015    assert fade.plan is not None
1016    return fade.plan.tempo_plan.savings_until(t)
1017
1018
1019class TestStretchSavings:
1020    """
1021    Rendered-time savings from the stretch must reach the timing bookkeeping.
1022
1023    The savings integration math itself is unit-tested on TempoPlan in
1024    tests/controllers/streams/smart_fades/test_models.py.
1025    """
1026
1027    def _stretched_fade(self) -> SmartCrossFade:
1028        duration = 240.0
1029        # 4% BPM difference with >4 bars available -> stretch is applied
1030        fade = SmartCrossFade(
1031            logger=LOGGER,
1032            fade_out_analysis=_analysis(bpm=120.0, duration=duration),
1033            fade_in_analysis=_analysis(bpm=124.8, duration=duration),
1034        )
1035        fade.build(_seconds(45), _seconds(45), PCM)
1036        return fade
1037
1038    def test_pre_plus_cf_equals_rendered_tail(self) -> None:
1039        """PRE + CF equals the rendered tail duration (buffer minus stretch savings)."""
1040        fade = self._stretched_fade()
1041        assert fade.tempo_steps, "test requires the stretch to be active"
1042        total_savings = _savings_until(fade, fade.effective_end)
1043        assert total_savings > 0.0
1044        timing = fade.timing_info
1045        assert timing.pre_crossfade_duration + timing.crossfade_duration == pytest.approx(
1046            fade.effective_end - total_savings, abs=0.05
1047        )
1048
1049    def test_pre_plus_cf_equals_rendered_tail_when_slowing_down(self) -> None:
1050        """A slower incoming track lengthens the rendered tail — PRE + CF exceeds effective_end."""
1051        duration = 240.0
1052        # ~3.8% BPM difference downwards -> stretch slows the outgoing track
1053        fade = SmartCrossFade(
1054            logger=LOGGER,
1055            fade_out_analysis=_analysis(bpm=120.0, duration=duration),
1056            fade_in_analysis=_analysis(bpm=115.4, duration=duration),
1057        )
1058        fade.build(_seconds(45), _seconds(45), PCM)
1059        assert fade.tempo_steps, "test requires the stretch to be active"
1060        total_savings = _savings_until(fade, fade.effective_end)
1061        assert total_savings < 0.0
1062        timing = fade.timing_info
1063        assert timing.pre_crossfade_duration + timing.crossfade_duration == pytest.approx(
1064            fade.effective_end - total_savings, abs=0.05
1065        )
1066
1067    def test_bass_kill_completes_at_the_anchor(self) -> None:
1068        """
1069        The outgoing low-shelf kill reaches full depth at or before the audible end.
1070
1071        A-side shelves render BEFORE the rubberband stretch, so their schedules
1072        live in musical input time — no rendered-time remap needed (unlike the
1073        old post-stretch frequency sweeps).
1074        """
1075        fade = self._stretched_fade()
1076        assert fade.plan is not None
1077        low_out = fade.plan.eq_plan.low_out
1078        assert low_out is not None
1079        assert low_out.steps[-1][1] == pytest.approx(-26.0)
1080        assert low_out.steps[-1][0] <= fade.effective_end + 0.05
1081
1082    def test_trim_and_stretch_combined(self) -> None:
1083        """A trimmed silent tail and an active stretch compose: both anchor on the rendered end."""
1084        duration = 240.0
1085        fade = SmartCrossFade(
1086            logger=LOGGER,
1087            fade_out_analysis=_analysis(
1088                bpm=120.0,
1089                duration=duration,
1090                rms_energy=_rms_with_silent_tail(duration, 10.0),
1091            ),
1092            fade_in_analysis=_analysis(bpm=124.8, duration=duration),
1093        )
1094        fade.build(_seconds(45), _seconds(45), PCM)
1095        # tail trim must come first so every later schedule sees the trimmed stream
1096        assert isinstance(fade.filters[0], FadeOutTrimFilter)
1097        assert fade.tempo_steps, "test requires the stretch to be active"
1098        rendered_end = fade.effective_end - _savings_until(fade, fade.effective_end)
1099        timing = fade.timing_info
1100        assert timing.pre_crossfade_duration + timing.crossfade_duration == pytest.approx(
1101            rendered_end, abs=0.05
1102        )
1103
1104    def test_unstretched_fade_has_zero_savings(self) -> None:
1105        """Without a stretch, savings are zero and PRE + CF equals effective_end exactly."""
1106        fade = SmartCrossFade(
1107            logger=LOGGER,
1108            fade_out_analysis=_analysis(bpm=120.0, duration=240.0),
1109            fade_in_analysis=_analysis(bpm=120.0, duration=240.0),
1110        )
1111        fade.build(_seconds(45), _seconds(45), PCM)
1112        assert fade.tempo_steps == []
1113        assert _savings_until(fade, fade.effective_end) == 0.0
1114        timing = fade.timing_info
1115        assert timing.pre_crossfade_duration + timing.crossfade_duration == pytest.approx(
1116            fade.effective_end, abs=0.05
1117        )
1118