music-assistant-server

1"""Tests for crossfade degradation when incoming source capacity is unavailable."""
2
3from __future__ import annotations
4
5from collections.abc import AsyncGenerator
6from types import SimpleNamespace
7from typing import Any, cast
8from unittest.mock import AsyncMock, MagicMock
9
10import pytest
11from music_assistant_models.enums import ContentType, CrossfadeMode, MediaType, StreamType
12from music_assistant_models.media_items import AudioFormat
13from music_assistant_models.streamdetails import StreamDetails
14
15from music_assistant.controllers.streams.audio import (
16    MIN_CROSSFADE_DURATION,
17    StreamsAudio,
18)
19from music_assistant.controllers.streams.audio_buffer import AudioBuffer
20from music_assistant.controllers.streams.smart_fades.helpers import SMART_CROSSFADE_DURATION
21
22
23def _streamdetails(audio_buffer: AudioBuffer | None) -> StreamDetails:
24    """Build incoming track details with an optional prepared buffer."""
25    streamdetails = StreamDetails(
26        provider="test--1",
27        item_id="track-1",
28        audio_format=AudioFormat(content_type=ContentType.FLAC),
29        media_type=MediaType.TRACK,
30        stream_type=StreamType.HTTP,
31        path="http://test.invalid/track.flac",
32        duration=180,
33    )
34    streamdetails.buffer = audio_buffer
35    return streamdetails
36
37
38def _buffer(duration_available: float, ready: bool, eof: bool = False) -> AudioBuffer:
39    """Build a valid buffer with the requested resident duration."""
40    audio_buffer = MagicMock(spec=AudioBuffer)
41    audio_buffer.has_error = False
42    audio_buffer.is_valid.return_value = True
43    audio_buffer.duration_available = duration_available
44    audio_buffer.eof = eof
45    audio_buffer.ready = MagicMock()
46    audio_buffer.ready.is_set.return_value = ready
47    return audio_buffer
48
49
50def _delivered_buffer() -> SimpleNamespace:
51    """Build the outgoing track's buffer, with its source done delivering."""
52    return SimpleNamespace(eof=True, cancelled=False, has_error=False, max_size_seconds=300)
53
54
55def test_ready_incoming_buffer_keeps_smart_crossfade() -> None:
56    """A tail that carries the full smart window keeps the requested Smart Fade."""
57    audio = StreamsAudio(MagicMock())
58
59    mode, duration = audio._select_buffered_crossfade(
60        _streamdetails(_buffer(SMART_CROSSFADE_DURATION, ready=True)),
61        CrossfadeMode.SMART_CROSSFADE,
62        standard_crossfade_duration=8,
63        fade_out_seconds=SMART_CROSSFADE_DURATION,
64    )
65
66    assert mode == CrossfadeMode.SMART_CROSSFADE
67    assert duration == SMART_CROSSFADE_DURATION
68
69
70def test_a_partly_resident_incoming_buffer_keeps_the_full_window() -> None:
71    """The incoming side streams in while the blend plays, so residency does not cap it."""
72    audio = StreamsAudio(MagicMock())
73
74    mode, duration = audio._select_buffered_crossfade(
75        _streamdetails(_buffer(2, ready=True)),
76        CrossfadeMode.SMART_CROSSFADE,
77        standard_crossfade_duration=8,
78        fade_out_seconds=SMART_CROSSFADE_DURATION,
79    )
80
81    assert mode == CrossfadeMode.SMART_CROSSFADE
82    assert duration == SMART_CROSSFADE_DURATION
83
84
85@pytest.mark.parametrize(
86    "audio_buffer",
87    [
88        None,
89        _buffer(30, ready=False),
90    ],
91)
92def test_unprepared_incoming_buffer_disables_crossfade(
93    audio_buffer: AudioBuffer | None,
94) -> None:
95    """An incoming source that is not delivering yet falls back to playback without a fade."""
96    audio = StreamsAudio(MagicMock())
97
98    mode, duration = audio._select_buffered_crossfade(
99        _streamdetails(audio_buffer),
100        CrossfadeMode.SMART_CROSSFADE,
101        standard_crossfade_duration=8,
102        fade_out_seconds=SMART_CROSSFADE_DURATION,
103    )
104
105    assert mode == CrossfadeMode.DISABLED
106    assert duration == 0
107
108
109def test_a_tail_below_the_minimum_disables_the_crossfade() -> None:
110    """Too short a held tail is played out instead of blended."""
111    audio = StreamsAudio(MagicMock())
112
113    mode, duration = audio._select_buffered_crossfade(
114        _streamdetails(_buffer(30, ready=True)),
115        CrossfadeMode.SMART_CROSSFADE,
116        standard_crossfade_duration=8,
117        fade_out_seconds=MIN_CROSSFADE_DURATION - 0.5,
118    )
119
120    assert mode == CrossfadeMode.DISABLED
121    assert duration == 0
122
123
124async def test_unprepared_next_track_flushes_outgoing_tail_without_opening_source(
125    monkeypatch: pytest.MonkeyPatch,
126) -> None:
127    """Missing incoming PCM emits the complete outgoing track without a blocking fade fetch."""
128    pcm_format = AudioFormat(
129        content_type=ContentType.PCM_S16LE,
130        sample_rate=8000,
131        bit_depth=16,
132        channels=2,
133    )
134    current_details = SimpleNamespace(
135        duration=16,
136        seek_position=0,
137        seconds_streamed=0,
138        uri="test://current",
139        buffer=_delivered_buffer(),
140        is_realtime=False,
141    )
142    next_details = SimpleNamespace(
143        audio_format=pcm_format,
144        buffer=None,
145        duration=16,
146        seek_position=0,
147        uri="test://next",
148        is_realtime=False,
149    )
150    current_item = SimpleNamespace(
151        queue_id="queue-1",
152        queue_item_id="current",
153        name="Current",
154        streamdetails=current_details,
155        extra_attributes={},
156    )
157    next_item = SimpleNamespace(
158        queue_id="queue-1",
159        queue_item_id="next",
160        name="Next",
161        streamdetails=next_details,
162        extra_attributes={},
163        available=True,
164    )
165    queue = SimpleNamespace(
166        queue_id="queue-1",
167        display_name="Queue",
168        index_in_buffer=0,
169    )
170    player = SimpleNamespace(player_id="player-1", name="Player")
171    mass = MagicMock()
172    mass.player_queues.get.return_value = queue
173    mass.player_queues.load_next_queue_item = AsyncMock(return_value=next_item)
174    mass.player_queues.index_by_id.return_value = 1
175    audio = StreamsAudio(cast("Any", mass))
176    audio.setup()
177    audio.select_pcm_format = AsyncMock(return_value=pcm_format)  # type: ignore[method-assign]
178    audio.crossfade_allowed = MagicMock(return_value=True)  # type: ignore[method-assign]
179    build = AsyncMock()
180    monkeypatch.setattr(audio.smart_fades_mixer, "build", build)
181
182    async def _current_stream(
183        queue_item: object,
184        *_args: object,
185        **_kwargs: object,
186    ) -> AsyncGenerator[bytes]:
187        if queue_item is not current_item:
188            pytest.fail("The incoming source was opened during crossfade fallback")
189        yield bytes(pcm_format.pcm_sample_size * 8)
190        yield bytes(pcm_format.pcm_sample_size * 8)
191
192    monkeypatch.setattr(audio, "get_queue_item_stream", _current_stream)
193    stream = audio.get_queue_item_stream_with_smartfade(
194        cast("Any", player),
195        cast("Any", current_item),
196        pcm_format,
197        crossfade_mode=CrossfadeMode.STANDARD_CROSSFADE,
198        standard_crossfade_duration=8,
199    )
200
201    output = b"".join([chunk async for chunk in stream])
202
203    assert len(output) == pcm_format.pcm_sample_size * 16
204    assert next_item.available
205    build.assert_not_awaited()
206
207
208@pytest.mark.parametrize("playback_speed", [0.5, 2.0])
209async def test_crossfade_reads_its_window_past_the_resident_buffer(
210    monkeypatch: pytest.MonkeyPatch,
211    playback_speed: float,
212) -> None:
213    """The blend consumes its whole window as it arrives, and hands on the media time used."""
214    pcm_format = AudioFormat(
215        content_type=ContentType.PCM_S16LE,
216        sample_rate=8000,
217        bit_depth=16,
218        channels=2,
219    )
220    resident_media_duration = 2.0
221    current_details = SimpleNamespace(
222        duration=16,
223        seek_position=0,
224        seconds_streamed=0,
225        uri="test://current",
226        buffer=_delivered_buffer(),
227        is_realtime=False,
228    )
229    next_details = SimpleNamespace(
230        audio_format=pcm_format,
231        buffer=_buffer(resident_media_duration, ready=True),
232        duration=16,
233        seek_position=0,
234        uri="test://next",
235        volume_normalization_mode=None,
236        is_realtime=False,
237    )
238    current_item = SimpleNamespace(
239        queue_id="queue-1",
240        queue_item_id="current",
241        name="Current",
242        streamdetails=current_details,
243        extra_attributes={},
244    )
245    next_item = SimpleNamespace(
246        queue_id="queue-1",
247        queue_item_id="next",
248        name="Next",
249        streamdetails=next_details,
250        extra_attributes={"playback_speed": playback_speed},
251        available=True,
252    )
253    queue = SimpleNamespace(
254        queue_id="queue-1",
255        display_name="Queue",
256        index_in_buffer=0,
257    )
258    player = SimpleNamespace(player_id="player-1", name="Player")
259    mass = MagicMock()
260    mass.player_queues.get.return_value = queue
261    mass.player_queues.load_next_queue_item = AsyncMock(return_value=next_item)
262    mass.player_queues.index_by_id.return_value = 1
263    audio = StreamsAudio(cast("Any", mass))
264    audio.setup()
265    audio.select_pcm_format = AsyncMock(return_value=pcm_format)  # type: ignore[method-assign]
266    audio.crossfade_allowed = MagicMock(return_value=True)  # type: ignore[method-assign]
267    crossfade_duration = 8
268    smart_fade = SimpleNamespace(
269        timing_info=SimpleNamespace(
270            pre_crossfade_duration=0,
271            crossfade_duration=crossfade_duration,
272            fadein_trimmed_duration=0,
273        )
274    )
275    monkeypatch.setattr(
276        audio.smart_fades_mixer,
277        "build",
278        AsyncMock(return_value=smart_fade),
279    )
280
281    async def _mix(
282        _smart_fade: object,
283        *,
284        fade_in_part: AsyncGenerator[bytes],
285        **_kwargs: object,
286    ) -> AsyncGenerator[bytes]:
287        async for chunk in fade_in_part:
288            yield chunk
289
290    monkeypatch.setattr(audio.smart_fades_mixer, "mix", _mix)
291    incoming_seconds_read = 0
292
293    async def _item_stream(
294        queue_item: object,
295        *_args: object,
296        **_kwargs: object,
297    ) -> AsyncGenerator[bytes]:
298        nonlocal incoming_seconds_read
299        if queue_item is current_item:
300            yield bytes(pcm_format.pcm_sample_size * 8)
301            yield bytes(pcm_format.pcm_sample_size * 8)
302            return
303        # the incoming source keeps delivering beyond what was resident at the boundary
304        for _ in range(20):
305            incoming_seconds_read += 1
306            yield bytes(pcm_format.pcm_sample_size)
307
308    monkeypatch.setattr(audio, "get_queue_item_stream", _item_stream)
309    stream = audio.get_queue_item_stream_with_smartfade(
310        cast("Any", player),
311        cast("Any", current_item),
312        pcm_format,
313        crossfade_mode=CrossfadeMode.STANDARD_CROSSFADE,
314        standard_crossfade_duration=crossfade_duration,
315    )
316
317    _ = [chunk async for chunk in stream]
318
319    assert incoming_seconds_read > resident_media_duration
320    crossfade_data = audio._crossfade_data["queue-1"]
321    assert crossfade_data.queue_item_id == "next"
322    # the window is stream time, so fast playback reaches the incoming track's
323    # half-duration cap sooner: at 2x an 8s overlap would eat this whole track
324    expected_window = min(crossfade_duration, next_details.duration / playback_speed / 2)
325    # the next track resumes at the media time the blend already played
326    assert crossfade_data.fade_in_media_duration == pytest.approx(expected_window * playback_speed)
327    assert crossfade_data.fade_in_media_duration <= next_details.duration / 2
328