music-assistant-server

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