From 7353e63f53488739bf5fea30ef868a21bafc997a Mon Sep 17 00:00:00 2001 From: Manuel Alcocer J Date: Sat, 7 Nov 2026 12:28:20 +0100 Subject: [PATCH 1/3] Send DSD files to Sendspin players as DoP (DSD over PCM) Sendspin only carries PCM, FLAC and Opus, so a DSD file is always decoded to PCM. DoP packs the raw DSD bits into 24-bit PCM frames, which a DAC that supports DoP plays back as native DSD, as long as every sample arrives untouched. Add an advanced per-player option, "Send DSD as DoP" (Sendspin, automatic format). When it is on and the player plays on its own: - a flow that starts with a local DSF file runs at its DoP rate (176.4 kHz for DSD64, 352.8 kHz for DSD128) in F32, which holds every 24-bit DoP word exactly - the flow restarts whenever a DSD track needs a different DoP rate - DSD tracks are streamed straight from the file as DoP, bypassing the audio buffer, volume normalization, fades and the crossfade prefetch - crossfades into or out of a DSD track are disabled numpy is only imported when a DoP stream starts, so start-up is unaffected. --- music_assistant/constants.py | 1 + music_assistant/controllers/streams/audio.py | 210 ++++++++++++++- music_assistant/helpers/dop.py | 244 ++++++++++++++++++ music_assistant/providers/sendspin/player.py | 13 +- .../providers/sendspin/strings.json | 4 + music_assistant/translations/en.json | 2 + tests/controllers/streams/test_flow_dop.py | 184 +++++++++++++ tests/helpers/test_dop.py | 182 +++++++++++++ 8 files changed, 832 insertions(+), 8 deletions(-) create mode 100644 music_assistant/helpers/dop.py create mode 100644 tests/controllers/streams/test_flow_dop.py create mode 100644 tests/helpers/test_dop.py diff --git a/music_assistant/constants.py b/music_assistant/constants.py index c2bed89ee1e..c6562821631 100644 --- a/music_assistant/constants.py +++ b/music_assistant/constants.py @@ -156,6 +156,7 @@ def _get_own_version() -> str: # validating against this keeps a caller-supplied id from escaping the storage dir DSP_IR_ID_RE: Final = re.compile(r"^[a-z0-9]+$") CONF_OUTPUT_CHANNELS: Final[str] = "output_channels" +CONF_DSD_DOP: Final[str] = "dsd_dop" CONF_FLOW_MODE: Final[str] = "flow_mode" CONF_FLOW_MODE_SAMPLE_RATE: Final[str] = "flow_mode_sample_rate" CONF_LOG_LEVEL: Final[str] = "log_level" diff --git a/music_assistant/controllers/streams/audio.py b/music_assistant/controllers/streams/audio.py index 214a824a051..d8279c31445 100644 --- a/music_assistant/controllers/streams/audio.py +++ b/music_assistant/controllers/streams/audio.py @@ -65,6 +65,7 @@ from music_assistant.constants import ( CONF_CROSSFADE_DURATION, + CONF_DSD_DOP, CONF_ENTRY_CROSSFADE_DIFFERENT_SAMPLE_RATES, CONF_ENTRY_VOLUME_NORMALIZATION_TARGET, CONF_FLOW_MODE_SAMPLE_RATE, @@ -153,6 +154,7 @@ resolve_output_player_ids, ) from music_assistant.helpers.compare import compare_item_ids +from music_assistant.helpers.dop import get_dop_sample_rate, get_dop_stream, is_dop_source from music_assistant.helpers.dsp import ComplexFilter, filter_to_ffmpeg_params from music_assistant.helpers.ffmpeg import ( FFMpeg, @@ -1541,6 +1543,17 @@ async def select_flow_pcm_format( start_streamdetails, supported_sample_rates=supported_sample_rates, ) + if ( + dop_rate := self._get_dop_rate(start_streamdetails, supported_sample_rates) + ) and self.dop_enabled_for_players(players): + # DoP has to reach the player bit-exact: no resampling, and F32 holds + # every 24-bit DoP word exactly + return AudioFormat( + content_type=INTERNAL_PCM_FORMAT.content_type, + sample_rate=dop_rate, + bit_depth=INTERNAL_PCM_FORMAT.bit_depth, + channels=2, + ) flow_mode_conf = cast( "str", player.config.get_value(CONF_FLOW_MODE_SAMPLE_RATE, FLOW_MODE_SAMPLE_RATE_SMART), @@ -1643,7 +1656,7 @@ async def get_audio_source_stream( finally: streamdetails.seconds_streamed = bytes_received / pcm_format.pcm_sample_size - async def get_queue_item_stream( # noqa: C901, PLR0912 + async def get_queue_item_stream( # noqa: C901, PLR0912, PLR0913 self, queue_item: QueueItem, pcm_format: AudioFormat, @@ -1655,6 +1668,7 @@ async def get_queue_item_stream( # noqa: C901, PLR0912 prepared_buffer: AudioBuffer | None = None, exact_seek: bool = False, consumer_connected: Callable[[], bool] | None = None, + dop: bool = False, ) -> AsyncGenerator[bytes]: """ Get the (PCM) audio stream for a single queue item. @@ -1675,6 +1689,7 @@ async def get_queue_item_stream( # noqa: C901, PLR0912 :param consumer_connected: Reports whether the consumer of this stream is still connected; once it reports False before the source is opened, the stream ends without audio and without marking the item. + :param dop: Send a DSD source as untouched DoP instead of decoding it to PCM. """ streamdetails = queue_item.streamdetails assert streamdetails @@ -1697,6 +1712,12 @@ def _mark_item_unavailable() -> None: ): yield chunk return + if dop and is_dop_source(streamdetails): + async for chunk in self._get_dop_queue_item_stream( + queue_item, pcm_format, seek_position, raise_on_error + ): + yield chunk + return filter_params: list[str] = [] logger = self.logger.getChild("queue_item_stream") @@ -2586,6 +2607,8 @@ def _consumer_left() -> bool: flow_mode_sample_rate_conf, flow_supported_sample_rates = self._flow_restart_context( queue.queue_id, protocol_player ) + flow_player = protocol_player or self.mass.players.get_player(queue.queue_id) + flow_dop = flow_player is not None and self.dop_enabled_for_players([flow_player]) # note: get_crossfade_mode() already falls back to standard when smart fades aren't # available (no analysis provider / minimal buffer), so crossfade_mode is safe to use. self.logger.info( @@ -2643,6 +2666,7 @@ def _superseded() -> bool: flow_supported_sample_rates, flow_mode_sample_rate_conf, is_first_track=queue_track is start_queue_item, + dop_enabled=flow_dop, ): break @@ -2665,6 +2689,9 @@ def _superseded() -> bool: standard_crossfade_duration = self.mass.config.get_raw_core_config_value( CONF_PLAYER_QUEUES, CONF_CROSSFADE_DURATION, 8 ) + # a fade mixes two tracks, which destroys DoP on either side of it + if flow_dop and self._dop_transition(queue.queue_id, queue_track): + item_crossfade_mode = CrossfadeMode.DISABLED # the item before fades into this one as the queue is set; a declared # overlap only concerns this item's own tail, into the item after it incoming_mode = item_crossfade_mode @@ -2864,7 +2891,18 @@ def _superseded() -> bool: source_buffer: AudioBuffer | None = None holding_back = item_crossfade_mode != CrossfadeMode.DISABLED - item_stream = await incoming_prefetcher.take(queue_track, int(raw_seek_position)) + item_dop = flow_dop and ( + self._get_dop_rate(queue_track.streamdetails, [pcm_format.sample_rate]) + is not None + ) + if item_dop: + # a prefetch holds decoded PCM, never the raw DSD that DoP needs + await incoming_prefetcher.close() + item_stream = None + else: + item_stream = await incoming_prefetcher.take( + queue_track, int(raw_seek_position) + ) prefetched_size = incoming_prefetcher.collected_at_handover if item_stream else 0 if item_stream is None: item_stream = self.get_queue_item_stream( @@ -2877,6 +2915,7 @@ def _superseded() -> bool: raise_on_error=False, session_id=flow_session_id, prepared_buffer=incoming_audio_buffer, + dop=item_dop, ) # closing here releases the decoders on an early exit, @@ -3171,11 +3210,16 @@ def _note_overlap_bytes(count: int) -> None: else MIN_CROSSFADE_DURATION ) ) - if len(crossfade_buffer) >= min_fade_out_size and self.crossfade_allowed( - queue_track, - crossfade_mode=item_crossfade_mode, - player_id=queue.queue_id, - flow_mode=True, + if ( + len(crossfade_buffer) >= min_fade_out_size + # the next item may have changed into a DSD track since this one started + and not (flow_dop and self._dop_transition(queue.queue_id, queue_track)) + and self.crossfade_allowed( + queue_track, + crossfade_mode=item_crossfade_mode, + player_id=queue.queue_id, + flow_mode=True, + ) ): last_fadeout_part = bytes(crossfade_buffer[-crossfade_buffer_size:]) last_streamdetails = queue_track.streamdetails @@ -3409,6 +3453,23 @@ def fade_into_pending(self, queue_id: str, queue_item_id: str) -> bool: pending = self._crossfade_pending.get(queue_id) return pending is not None and pending[0] == queue_item_id + def dop_enabled_for_players(self, players: Iterable[Player]) -> bool: + """ + Return whether every given player takes DSD as DoP. + + DoP is only sent to a player that plays on its own: a group mixes, resamples or + re-encodes the shared stream for its members, which turns DoP into noise. + + :param players: The players consuming the stream. + """ + players = tuple(players) + return bool(players) and all( + self.mass.config.get_raw_player_config_value(player.player_id, CONF_DSD_DOP, False) + is True + and len(player.group_members) <= 1 + for player in players + ) + def clear_crossfade_handover(self, queue_id: str) -> None: """ Clear any pending crossfade data for a queue. @@ -5676,6 +5737,7 @@ def _flow_stream_needs_restart( supported_sample_rates: list[int], flow_mode_sample_rate_conf: str, is_first_track: bool, + dop_enabled: bool = False, ) -> bool: """ Return True if the upcoming queue track requires exiting the flow stream. @@ -5698,6 +5760,7 @@ def _flow_stream_needs_restart( :param supported_sample_rates: Sorted list of the player's supported rates. :param flow_mode_sample_rate_conf: The flow mode sample rate config value. :param is_first_track: Whether this is the first track of the flow stream. + :param dop_enabled: Whether the consuming player takes DSD as DoP. """ # live audio (radio, plugin or audio source) cannot be flowed; let the # queue controller fall back to single-item streaming for this item @@ -5713,6 +5776,21 @@ def _flow_stream_needs_restart( if is_first_track or queue_track.streamdetails is None: return False + if dop_enabled and ( + dop_rate := self._get_dop_rate(queue_track.streamdetails, supported_sample_rates) + ): + # DoP only survives at exactly its own rate, whatever the flow mode + if dop_rate != pcm_format.sample_rate: + self.logger.info( + "DSD track %s (%s) needs DoP rate %s, flow runs at %s " + "- breaking out to restart flow stream", + queue_track.queue_item_id, + queue_track.name, + dop_rate, + pcm_format.sample_rate, + ) + return True + return False raw_next_rate = queue_track.streamdetails.audio_format.sample_rate if not raw_next_rate or not supported_sample_rates: return False @@ -6000,3 +6078,121 @@ async def _add_remote_mp3_input_args( # seeks by bitrate or the Xing TOC instead of decoding up to the position: exact for # CBR, while VBR lands seconds off with a TOC and can be minutes off without one add_input_fflag(extra_input_args, "+fastseek") + + def _get_dop_rate( + self, + streamdetails: StreamDetails | None, + supported_sample_rates: Iterable[int], + ) -> int | None: + """ + Return the DoP rate a stream would play at, or None if it cannot play as DoP. + + :param streamdetails: Stream details of the item, if resolved. + :param supported_sample_rates: Sample rates the consuming player takes. + """ + if streamdetails is None or not is_dop_source(streamdetails): + return None + dop_rate = get_dop_sample_rate(streamdetails.audio_format) + return dop_rate if dop_rate in set(supported_sample_rates) else None + + def _dop_transition(self, queue_id: str, queue_item: QueueItem) -> bool: + """ + Return whether a queue item or the one after it is a DSD track. + + :param queue_id: The queue the items belong to. + :param queue_item: The item about to play (or playing). + """ + if self._is_dsd_queue_item(queue_item): + return True + next_item = self.mass.player_queues.get_next_item(queue_id, queue_item.queue_item_id) + return next_item is not None and self._is_dsd_queue_item(next_item) + + @staticmethod + def _is_dsd_queue_item(queue_item: QueueItem) -> bool: + """Return whether a queue item plays a DSD source, also before it is resolved.""" + if queue_item.streamdetails is not None: + return is_dop_source(queue_item.streamdetails) + media_item = queue_item.media_item + return isinstance(media_item, Track) and any( + mapping.available and get_dop_sample_rate(mapping.audio_format) is not None + for mapping in media_item.provider_mappings + ) + + async def _get_dop_queue_item_stream( + self, + queue_item: QueueItem, + pcm_format: AudioFormat, + seek_position: float, + raise_on_error: bool, + ) -> AsyncGenerator[bytes]: + """ + Stream a DSD queue item as untouched DoP. + + Volume normalization, fades, speed changes and the audio buffer are all + bypassed: anything that alters the samples turns DoP into noise. + + :param queue_item: The DSD queue item. + :param pcm_format: Flow PCM format, running at the DoP rate of the item. + :param seek_position: Start position in seconds. + :param raise_on_error: Raise on stream errors instead of logging them. + """ + streamdetails = queue_item.streamdetails + assert streamdetails + assert isinstance(streamdetails.path, str) + logger = self.logger.getChild("queue_item_stream") + streamdetails.volume_normalization_mode = VolumeNormalizationMode.DISABLED + streamdetails.volume_normalization_gain_correct = None + logger.info( + "Streaming %s (%s) as DoP at %s Hz", + queue_item.name, + streamdetails.uri, + pcm_format.sample_rate, + ) + bytes_received = 0 + finished = False + start_reported = False + next_buffer_triggered = False + try: + async for chunk in get_dop_stream(streamdetails.path, pcm_format, seek_position): + if bytes_received == 0 and streamdetails.seconds_streamed is None: + start_reported = self._notify_provider_stream_started(streamdetails) + bytes_received += len(chunk) + if ( + not next_buffer_triggered + and streamdetails.duration + and (bytes_received / pcm_format.pcm_sample_size + seek_position) + >= streamdetails.duration - 60 + and ( + next_item := self.mass.player_queues.get_next_item( + queue_item.queue_id, queue_item.queue_item_id + ) + ) + and next_item.queue_item_id != queue_item.queue_item_id + and next_item.media_type in (MediaType.TRACK, MediaType.SOUND_EFFECT) + ): + next_buffer_triggered = True + self.mass.player_queues.prepare_next_audio_buffer( + queue_item.queue_id, queue_item.queue_item_id + ) + yield chunk + finished = True + except asyncio.CancelledError: + raise + except Exception as err: + streamdetails.stream_error = True + if bytes_received == 0: + queue_item.available = False + if raise_on_error: + raise AudioError(f"DoP stream failed: {err}") from err + logger.error( + "Error while streaming %s (%s) as DoP: %s", + queue_item.name, + streamdetails.uri, + err, + ) + finally: + seconds_streamed = bytes_received / pcm_format.pcm_sample_size + streamdetails.seconds_streamed = seconds_streamed + self._notify_provider_streamed( + streamdetails, finished, seconds_streamed, start_reported + ) diff --git a/music_assistant/helpers/dop.py b/music_assistant/helpers/dop.py new file mode 100644 index 00000000000..33fa10d2b86 --- /dev/null +++ b/music_assistant/helpers/dop.py @@ -0,0 +1,244 @@ +""" +DSD over PCM (DoP) packing for local DSD files. + +DoP (DoP open standard v1.1) carries 1-bit DSD inside 24-bit PCM frames so it can +travel over any bit-perfect PCM path to a DAC that recognises it: each 24-bit word +holds a marker byte (alternating 0x05 / 0xFA per frame) followed by 16 DSD bits +of one channel, oldest bit first. The PCM rate is the DSD rate / 16, so DSD64 +(2.8224 MHz) travels as 176.4 kHz PCM, DSD128 as 352.8 kHz and DSD256 as 705.6 kHz. + +Only an untouched stream is DoP: any resampling, mixing, volume or other processing +on the way turns it into noise, which is why callers must bypass all of that. +""" + +from __future__ import annotations + +import asyncio +import struct +from dataclasses import dataclass +from functools import cache +from typing import TYPE_CHECKING + +from music_assistant_models.enums import ContentType, StreamType + +from music_assistant.helpers.audio import pcm_container_bit_depth + +if TYPE_CHECKING: + from collections.abc import AsyncGenerator + + import numpy as np + from music_assistant_models.media_items.audio_format import AudioFormat + from music_assistant_models.streamdetails import StreamDetails + +DOP_MARKERS = (0x05, 0xFA) +# DSD idle pattern, used to pad a trailing odd byte so the last frame stays valid DoP +DSD_SILENCE_BYTE = 0x69 +# ffprobe (and therefore the library) reports a DSD stream at DSD rate / 8 +_PROBED_DSD_RATE_DIVISOR = 8 +_DOP_BITS_PER_FRAME = 16 +# number of DSF blocks read per worker-thread call (~0.75 s of DSD64 audio) +_BLOCKS_PER_READ = 64 + +# numpy is imported on first use so it does not weigh on the server start-up + + +class DoPError(Exception): + """Raised when a file cannot be streamed as DoP.""" + + +@dataclass(frozen=True) +class DsfInfo: + """Layout details of a DSF file needed to stream it.""" + + channels: int + dsd_rate: int + lsb_first: bool + sample_count: int + block_size: int + data_offset: int + + @property + def dop_sample_rate(self) -> int: + """Return the PCM sample rate that carries this file as DoP.""" + return self.dsd_rate // _DOP_BITS_PER_FRAME + + @property + def bytes_per_channel(self) -> int: + """Return the number of valid DSD bytes per channel.""" + return self.sample_count // 8 + + +def get_dop_sample_rate(audio_format: AudioFormat) -> int | None: + """ + Return the PCM rate a DSD source travels at as DoP, or None if it is not DSD. + + :param audio_format: The (probed) audio format of the source. + """ + if audio_format.content_type != ContentType.DSF or not audio_format.sample_rate: + return None + dsd_rate = audio_format.sample_rate * _PROBED_DSD_RATE_DIVISOR + return dsd_rate // _DOP_BITS_PER_FRAME + + +def is_dop_source(streamdetails: StreamDetails) -> bool: + """ + Return whether a stream can be sent as DoP. + + Only local DSF files qualify: DoP needs the raw DSD bits, which only the file holds. + + :param streamdetails: The stream details of the source. + """ + return ( + streamdetails.stream_type == StreamType.LOCAL_FILE + and isinstance(streamdetails.path, str) + and streamdetails.audio_format.channels == 2 + and get_dop_sample_rate(streamdetails.audio_format) is not None + ) + + +def read_dsf_info(path: str) -> DsfInfo: + """ + Parse the header of a DSF file. + + :param path: Path to the DSF file. + """ + with open(path, "rb") as file: + header = file.read(92) + if len(header) < 92 or header[0:4] != b"DSD " or header[28:32] != b"fmt ": + raise DoPError(f"Not a DSF file: {path}") + fmt_chunk_size = struct.unpack_from(" tuple[np.ndarray, int]: + """ + Pack MSB-first DSD bytes into interleaved 24-bit DoP words. + + :param dsd: uint8 array of shape (channels, bytes), MSB-first, even byte count. + :param marker_phase: Index (0/1) of the marker the first frame must carry. + :return: The uint32 words (24-bit value in the low bits) interleaved per frame, + and the marker phase the next frame must carry. + """ + import numpy as np # noqa: PLC0415 + + channels = dsd.shape[0] + pairs = dsd.reshape(channels, -1, 2).astype(np.uint32) + frames = pairs.shape[1] + markers = np.where( + (np.arange(frames) + marker_phase) % 2 == 0, DOP_MARKERS[0], DOP_MARKERS[1] + ).astype(np.uint32) + words = (markers[None, :] << 16) | (pairs[:, :, 0] << 8) | pairs[:, :, 1] + return words.T.reshape(-1), (marker_phase + frames) % 2 + + +def dop_words_to_pcm(words: np.ndarray, pcm_format: AudioFormat) -> bytes: + """ + Encode 24-bit DoP words as raw PCM in the given format, bit-exactly. + + :param words: uint32 array holding the 24-bit DoP words. + :param pcm_format: The PCM format the stream consumer expects. + """ + import numpy as np # noqa: PLC0415 + + content_type = pcm_format.content_type + if content_type == ContentType.PCM_F32LE: + # every 24-bit integer is exactly representable in float32 + signed = words.astype(np.int32) + signed[signed >= 1 << 23] -= 1 << 24 + return (signed.astype(np.float32) / float(1 << 23)).astype(" AsyncGenerator[bytes]: + """ + Stream a DSF file as DoP, encoded in the given PCM format. + + :param path: Path to the DSF file. + :param pcm_format: PCM format of the consumer; its rate must be the DoP rate. + :param seek_position: Start position in seconds. + """ + import numpy as np # noqa: PLC0415 + + info = await asyncio.to_thread(read_dsf_info, path) + if info.channels != 2 or pcm_format.channels != 2: + raise DoPError("DoP streaming supports stereo DSD only") + if pcm_format.sample_rate != info.dop_sample_rate: + raise DoPError( + f"PCM rate {pcm_format.sample_rate} cannot carry DSD rate {info.dsd_rate} as DoP" + ) + if pcm_container_bit_depth(pcm_format.bit_depth) < 24: + raise DoPError(f"DoP needs a 24-bit or wider PCM format, not {pcm_format.bit_depth}") + channels = info.channels + block_size = info.block_size + total_blocks = -(-info.bytes_per_channel // block_size) + start_block = min(int(seek_position * info.dsd_rate / 8) // block_size, total_blocks) + marker_phase = 0 + + def _read_blocks(first_block: int, count: int) -> bytes: + with open(path, "rb") as file: + file.seek(info.data_offset + first_block * block_size * channels) + return file.read(count * block_size * channels) + + block = start_block + while block < total_blocks: + count = min(_BLOCKS_PER_READ, total_blocks - block) + raw = await asyncio.to_thread(_read_blocks, block, count) + whole_blocks = len(raw) // (block_size * channels) + if whole_blocks == 0: + break + dsd = ( + np.frombuffer(raw[: whole_blocks * block_size * channels], dtype=np.uint8) + .reshape(whole_blocks, channels, block_size) + .transpose(1, 0, 2) + .reshape(channels, -1) + ) + # drop the padding of the last block of the file + valid = info.bytes_per_channel - block * block_size + if valid < dsd.shape[1]: + dsd = dsd[:, :valid] + if info.lsb_first: + dsd = _bit_reverse_table()[dsd] + if dsd.shape[1] % 2: + dsd = np.concatenate( + [dsd, np.full((channels, 1), DSD_SILENCE_BYTE, dtype=np.uint8)], axis=1 + ) + if dsd.shape[1]: + words, marker_phase = pack_dop(dsd, marker_phase) + yield dop_words_to_pcm(words, pcm_format) + block += whole_blocks + + +@cache +def _bit_reverse_table() -> np.ndarray: + """Return the table that reverses the bit order of a byte.""" + import numpy as np # noqa: PLC0415 + + return np.array([int(f"{i:08b}"[::-1], 2) for i in range(256)], dtype=np.uint8) diff --git a/music_assistant/providers/sendspin/player.py b/music_assistant/providers/sendspin/player.py index cfcac4c1a5a..9a7c4e4d475 100644 --- a/music_assistant/providers/sendspin/player.py +++ b/music_assistant/providers/sendspin/player.py @@ -87,7 +87,7 @@ from music_assistant_models.player import DeviceInfo from PIL import Image -from music_assistant.constants import CONF_ENTRY_FLOW_MODE_SAMPLE_RATE +from music_assistant.constants import CONF_DSD_DOP, CONF_ENTRY_FLOW_MODE_SAMPLE_RATE from music_assistant.controllers.streams.audio_analysis import SMART_FADES_ANALYSIS_DOMAIN from music_assistant.helpers.util import is_valid_mac_address from music_assistant.models.player import ( @@ -2053,6 +2053,17 @@ async def get_config_entries(self) -> list[ConfigEntry]: sample_rate_entry.depends_on = CONF_PREFERRED_SENDSPIN_FORMAT sample_rate_entry.depends_on_value = SENDSPIN_FORMAT_AUTOMATIC entries.append(sample_rate_entry) + entries.append( + ConfigEntry( + key=CONF_DSD_DOP, + type=ConfigEntryType.BOOLEAN, + category="protocol_generic", + default_value=False, + depends_on=CONF_PREFERRED_SENDSPIN_FORMAT, + depends_on_value=SENDSPIN_FORMAT_AUTOMATIC, + advanced=True, + ) + ) if player_role is not None and not _OUTPUT_DELAY_COMMANDS.isdisjoint( player_role.state_supported_commands diff --git a/music_assistant/providers/sendspin/strings.json b/music_assistant/providers/sendspin/strings.json index a7c706c9c4d..a7539b01c79 100644 --- a/music_assistant/providers/sendspin/strings.json +++ b/music_assistant/providers/sendspin/strings.json @@ -225,6 +225,10 @@ }, "dynamic_pin_channel_display_speaker": { "label": "Enter the code the device shows on screen or speaks aloud" + }, + "dsd_dop": { + "label": "Send DSD as DoP (DSD over PCM)", + "description": "Play local DSD files (.dsf) bit-perfect by packing them as DoP for a DAC that decodes it, instead of converting them to PCM. Only enable this when the DAC detects DoP and nothing in between alters the audio: no DSP, no volume normalization and the player volume at 100% (DSD tracks skip crossfades; set the volume on your amplifier). The DAC must support the DoP rate (176.4 kHz for DSD64, 352.8 kHz for DSD128). It only applies while the player plays on its own, not in a group." } }, "setup_flow": { diff --git a/music_assistant/translations/en.json b/music_assistant/translations/en.json index 3f36d32e71f..a72ecef53d7 100644 --- a/music_assistant/translations/en.json +++ b/music_assistant/translations/en.json @@ -2395,6 +2395,8 @@ "provider.sendspin.config_entries.connect_method.options.unpaired": "Connect unpaired", "provider.sendspin.config_entries.connect_method.options.unpaired_with_input": "Connect unpaired with audio input", "provider.sendspin.config_entries.connect_method.options.unpaired_without_input": "Connect unpaired without audio input", + "provider.sendspin.config_entries.dsd_dop.description": "Play local DSD files (.dsf) bit-perfect by packing them as DoP for a DAC that decodes it, instead of converting them to PCM. Only enable this when the DAC detects DoP and nothing in between alters the audio: no DSP, no volume normalization and the player volume at 100% (DSD tracks skip crossfades; set the volume on your amplifier). The DAC must support the DoP rate (176.4 kHz for DSD64, 352.8 kHz for DSD128). It only applies while the player plays on its own, not in a group.", + "provider.sendspin.config_entries.dsd_dop.label": "Send DSD as DoP (DSD over PCM)", "provider.sendspin.config_entries.dynamic_pin_channel_display.label": "Enter the code shown on the device's screen", "provider.sendspin.config_entries.dynamic_pin_channel_display_speaker.label": "Enter the code the device shows on screen or speaks aloud", "provider.sendspin.config_entries.dynamic_pin_channel_speaker.label": "Enter the code the device speaks aloud", diff --git a/tests/controllers/streams/test_flow_dop.py b/tests/controllers/streams/test_flow_dop.py new file mode 100644 index 00000000000..fc164b0827c --- /dev/null +++ b/tests/controllers/streams/test_flow_dop.py @@ -0,0 +1,184 @@ +"""Tests for DSD over PCM (DoP) handling in the flow stream format and restart logic.""" + +from __future__ import annotations + +from unittest.mock import MagicMock, patch + +import pytest +from music_assistant_models.enums import ContentType, MediaType, StreamType +from music_assistant_models.media_items import AudioFormat + +from music_assistant.constants import ( + CONF_DSD_DOP, + CONF_FLOW_MODE_SAMPLE_RATE, + FLOW_MODE_SAMPLE_RATE_SMART, +) +from music_assistant.controllers.streams.audio import StreamsAudio + +DOP_RATES = [(44100, 24), (48000, 24), (96000, 24), (176400, 24), (192000, 24), (352800, 24)] + + +def _make_streams_audio(*, dop: bool) -> StreamsAudio: + """Return a StreamsAudio whose players have DoP switched on or off.""" + audio = StreamsAudio(MagicMock()) + audio.mass.config.get_player_dsp_config = MagicMock( # type: ignore[method-assign] + return_value=MagicMock(enabled=False) + ) + audio.mass.config.get_raw_player_config_value = MagicMock( # type: ignore[method-assign] + side_effect=lambda _player_id, key, default=None: dop if key == CONF_DSD_DOP else default + ) + return audio + + +def _make_player(*, supported: list[tuple[int, int]] = DOP_RATES, members: int = 1) -> MagicMock: + player = MagicMock() + player.player_id = "player" + player.group_members = [f"member-{index}" for index in range(members)] + player.get_supported_sample_rates = MagicMock(return_value=supported) + player.config.get_value = MagicMock( + side_effect=lambda key, default=None: ( + FLOW_MODE_SAMPLE_RATE_SMART if key == CONF_FLOW_MODE_SAMPLE_RATE else default + ) + ) + return player + + +def _dsd_streamdetails(probed_rate: int = 352800) -> MagicMock: + """Stream details of a local DSF file, as the library reports it (DSD rate / 8).""" + streamdetails = MagicMock() + streamdetails.audio_format = AudioFormat( + content_type=ContentType.DSF, sample_rate=probed_rate, bit_depth=8, channels=2 + ) + streamdetails.stream_type = StreamType.LOCAL_FILE + streamdetails.path = "/music/track.dsf" + streamdetails.decoded_audio_format = None + streamdetails.media_type = MediaType.TRACK + return streamdetails + + +def _queue_track(streamdetails: MagicMock) -> MagicMock: + track = MagicMock() + track.queue_item_id = "item-2" + track.name = "DSD Track" + track.media_type = MediaType.TRACK + track.streamdetails = streamdetails + return track + + +def _pcm_format(rate: int) -> AudioFormat: + return AudioFormat( + content_type=ContentType.PCM_F32LE, sample_rate=rate, bit_depth=32, channels=2 + ) + + +@pytest.mark.asyncio +async def test_flow_format_runs_at_dop_rate_for_dsd_start() -> None: + """A DSD64 first track anchors the flow at its DoP rate (176.4 kHz), in F32.""" + audio = _make_streams_audio(dop=True) + fmt = await audio.select_flow_pcm_format( + _make_player(), start_streamdetails=_dsd_streamdetails() + ) + assert fmt.sample_rate == 176400 + assert fmt.content_type == ContentType.PCM_F32LE + + +@pytest.mark.asyncio +async def test_flow_format_ignores_dop_when_disabled() -> None: + """Without the DoP setting a DSD track is decoded to PCM at the snapped source rate.""" + audio = _make_streams_audio(dop=False) + fmt = await audio.select_flow_pcm_format( + _make_player(), start_streamdetails=_dsd_streamdetails() + ) + assert fmt.sample_rate == 352800 + + +@pytest.mark.asyncio +async def test_flow_format_ignores_dop_for_unsupported_rate() -> None: + """ + A player that does not take the DoP rate gets the regular PCM conversion. + + The regular conversion keeps the 44.1 kHz family of the source. + """ + audio = _make_streams_audio(dop=True) + player = _make_player(supported=[(44100, 24), (96000, 24), (192000, 24)]) + fmt = await audio.select_flow_pcm_format(player, start_streamdetails=_dsd_streamdetails()) + assert fmt.sample_rate == 44100 + + +@pytest.mark.asyncio +async def test_flow_format_ignores_dop_for_groups() -> None: + """A grouped player never gets DoP: the group would alter the shared stream.""" + audio = _make_streams_audio(dop=True) + fmt = await audio.select_flow_pcm_format( + _make_player(members=2), start_streamdetails=_dsd_streamdetails() + ) + assert fmt.sample_rate == 352800 + + +@pytest.mark.parametrize( + ("flow_rate", "probed_rate", "expected"), + [ + (44100, 352800, True), # PCM flow -> DSD64 needs the 176.4 kHz DoP rate + (176400, 352800, False), # already at the DSD64 DoP rate + (176400, 705600, True), # DSD128 needs 352.8 kHz + ], +) +def test_restart_on_dop_rate_change(flow_rate: int, probed_rate: int, expected: bool) -> None: + """The flow restarts whenever a DSD track needs a different DoP rate.""" + audio = _make_streams_audio(dop=True) + assert ( + audio._flow_stream_needs_restart( + _queue_track(_dsd_streamdetails(probed_rate)), + _pcm_format(flow_rate), + [rate for rate, _ in DOP_RATES], + FLOW_MODE_SAMPLE_RATE_SMART, + is_first_track=False, + dop_enabled=True, + ) + is expected + ) + + +def test_no_dop_restart_when_disabled() -> None: + """Without DoP the regular smart-mode rule applies (352.8 kHz > 176.4 kHz restarts).""" + audio = _make_streams_audio(dop=False) + assert audio._flow_stream_needs_restart( + _queue_track(_dsd_streamdetails()), + _pcm_format(176400), + [rate for rate, _ in DOP_RATES], + FLOW_MODE_SAMPLE_RATE_SMART, + is_first_track=False, + dop_enabled=False, + ) + + +def test_dop_transition_detects_dsd_before_resolving() -> None: + """A DSD track is recognised from its provider mapping before streamdetails exist.""" + audio = _make_streams_audio(dop=True) + pcm_item = MagicMock() + pcm_item.queue_item_id = "item-1" + pcm_item.streamdetails = None + pcm_item.media_item = None + from music_assistant_models.media_items import ProviderMapping, Track # noqa: PLC0415 + + dsd_item = MagicMock() + dsd_item.streamdetails = None + dsd_item.media_item = Track( + item_id="1", + provider="filesystem", + name="DSD", + provider_mappings={ + ProviderMapping( + item_id="1", + provider_domain="filesystem", + provider_instance="filesystem", + audio_format=AudioFormat( + content_type=ContentType.DSF, sample_rate=352800, bit_depth=8 + ), + ) + }, + ) + with patch.object(audio.mass.player_queues, "get_next_item", return_value=dsd_item): + assert audio._dop_transition("queue", pcm_item) + with patch.object(audio.mass.player_queues, "get_next_item", return_value=None): + assert not audio._dop_transition("queue", pcm_item) diff --git a/tests/helpers/test_dop.py b/tests/helpers/test_dop.py new file mode 100644 index 00000000000..839be7f6799 --- /dev/null +++ b/tests/helpers/test_dop.py @@ -0,0 +1,182 @@ +"""Tests for DSD over PCM (DoP) packing.""" + +from __future__ import annotations + +import struct +from pathlib import Path + +import numpy as np +import pytest +from music_assistant_models.enums import ContentType, StreamType +from music_assistant_models.media_items.audio_format import AudioFormat +from music_assistant_models.streamdetails import StreamDetails + +from music_assistant.helpers.dop import ( + DoPError, + get_dop_sample_rate, + get_dop_stream, + is_dop_source, + read_dsf_info, +) + +DSD64_RATE = 2822400 +BLOCK_SIZE = 4096 + + +def _write_dsf(path: Path, left: bytes, right: bytes, *, lsb_first: bool = True) -> None: + """Write a minimal stereo DSF file whose channels hold the given DSD bytes.""" + assert len(left) == len(right) + valid = len(left) + blocks = -(-valid // BLOCK_SIZE) + data = bytearray() + for index in range(blocks): + for channel in (left, right): + chunk = channel[index * BLOCK_SIZE : (index + 1) * BLOCK_SIZE] + data += chunk + bytes(BLOCK_SIZE - len(chunk)) + fmt_chunk = b"fmt " + struct.pack( + " int: + return int(f"{value:08b}"[::-1], 2) + + +def _pcm_format(content_type: ContentType, bit_depth: int, rate: int = 176400) -> AudioFormat: + return AudioFormat(content_type=content_type, sample_rate=rate, bit_depth=bit_depth, channels=2) + + +async def _collect(path: Path, pcm_format: AudioFormat, seek: float = 0) -> bytes: + return b"".join([chunk async for chunk in get_dop_stream(str(path), pcm_format, seek)]) + + +def _words_from_s32(raw: bytes) -> np.ndarray: + return np.frombuffer(raw, dtype="> 8 + + +def test_get_dop_sample_rate() -> None: + """The library reports DSD at DSD rate / 8; DoP runs at DSD rate / 16.""" + dsd64 = AudioFormat(content_type=ContentType.DSF, sample_rate=352800, bit_depth=8) + dsd128 = AudioFormat(content_type=ContentType.DSF, sample_rate=705600, bit_depth=8) + flac = AudioFormat(content_type=ContentType.FLAC, sample_rate=176400, bit_depth=24) + assert get_dop_sample_rate(dsd64) == 176400 + assert get_dop_sample_rate(dsd128) == 352800 + assert get_dop_sample_rate(flac) is None + + +def test_is_dop_source_needs_local_stereo_dsf() -> None: + """Only a local stereo DSF file can be streamed as DoP.""" + dsd = AudioFormat(content_type=ContentType.DSF, sample_rate=352800, bit_depth=8, channels=2) + local = StreamDetails( + provider="test", + item_id="1", + audio_format=dsd, + stream_type=StreamType.LOCAL_FILE, + path="/music/a.dsf", + ) + remote = StreamDetails( + provider="test", + item_id="2", + audio_format=dsd, + stream_type=StreamType.HTTP, + path="http://host/a.dsf", + ) + assert is_dop_source(local) + assert not is_dop_source(remote) + + +async def test_dop_stream_markers_and_bit_order(tmp_path: Path) -> None: + """Frames alternate 0x05/0xFA on both channels and carry MSB-first DSD bytes in order.""" + rng = np.random.default_rng(1) + left = rng.integers(0, 256, BLOCK_SIZE * 3, dtype=np.uint8).tobytes() + right = rng.integers(0, 256, BLOCK_SIZE * 3, dtype=np.uint8).tobytes() + dsf = tmp_path / "test.dsf" + _write_dsf(dsf, left, right) + + words = _words_from_s32(await _collect(dsf, _pcm_format(ContentType.PCM_S32LE, 32))) + + frames = words.reshape(-1, 2) + assert len(frames) == len(left) // 2 + markers = frames >> 16 + assert (markers[0::2] == 0x05).all() + assert (markers[1::2] == 0xFA).all() + for channel, source in ((0, left), (1, right)): + expected = [_reverse_bits(b) for b in source] + got = np.stack([(frames[:, channel] >> 8) & 0xFF, frames[:, channel] & 0xFF], axis=1) + assert got.reshape(-1).tolist() == expected + + +async def test_dop_stream_trims_padding_and_pads_odd_byte(tmp_path: Path) -> None: + """The padding of the last DSF block is dropped; an odd last byte gets DSD silence.""" + left = bytes(range(256)) * 4 + b"\x01" + right = bytes(reversed(range(256))) * 4 + b"\x02" + dsf = tmp_path / "odd.dsf" + _write_dsf(dsf, left, right, lsb_first=False) + + words = _words_from_s32(await _collect(dsf, _pcm_format(ContentType.PCM_S32LE, 32))) + + frames = words.reshape(-1, 2) + assert len(frames) == (len(left) + 1) // 2 + assert frames[-1, 0] & 0xFFFF == 0x0169 + assert frames[-1, 1] & 0xFFFF == 0x0269 + + +async def test_dop_stream_formats_are_bit_exact(tmp_path: Path) -> None: + """F32, S32 and S24 output all carry the identical 24-bit DoP words.""" + rng = np.random.default_rng(2) + left = rng.integers(0, 256, BLOCK_SIZE, dtype=np.uint8).tobytes() + right = rng.integers(0, 256, BLOCK_SIZE, dtype=np.uint8).tobytes() + dsf = tmp_path / "fmt.dsf" + _write_dsf(dsf, left, right) + + s32 = _words_from_s32(await _collect(dsf, _pcm_format(ContentType.PCM_S32LE, 32))) + f32 = np.frombuffer(await _collect(dsf, _pcm_format(ContentType.PCM_F32LE, 32)), dtype=" None: + """Seeking starts at the DSF block holding the position, keeping markers valid.""" + left = bytes([0x11]) * BLOCK_SIZE + bytes([0x22]) * BLOCK_SIZE + right = bytes([0x33]) * BLOCK_SIZE + bytes([0x44]) * BLOCK_SIZE + dsf = tmp_path / "seek.dsf" + _write_dsf(dsf, left, right, lsb_first=False) + one_block_seconds = BLOCK_SIZE * 8 / DSD64_RATE + + words = _words_from_s32( + await _collect(dsf, _pcm_format(ContentType.PCM_S32LE, 32), seek=one_block_seconds * 1.5) + ) + + frames = words.reshape(-1, 2) + assert len(frames) == BLOCK_SIZE // 2 + assert (frames[:, 0] & 0xFFFF == 0x2222).all() + assert frames[0, 0] >> 16 == 0x05 + + +async def test_dop_stream_rejects_wrong_rate(tmp_path: Path) -> None: + """A PCM rate that is not the DoP rate of the file cannot carry it.""" + dsf = tmp_path / "rate.dsf" + _write_dsf(dsf, bytes(BLOCK_SIZE), bytes(BLOCK_SIZE)) + with pytest.raises(DoPError): + await _collect(dsf, _pcm_format(ContentType.PCM_S32LE, 32, rate=192000)) + + +def test_read_dsf_info(tmp_path: Path) -> None: + """The DSF header is parsed into the layout needed for streaming.""" + dsf = tmp_path / "info.dsf" + _write_dsf(dsf, bytes(10), bytes(10)) + info = read_dsf_info(str(dsf)) + assert info.channels == 2 + assert info.dsd_rate == DSD64_RATE + assert info.dop_sample_rate == 176400 + assert info.bytes_per_channel == 10 + assert info.data_offset == 92 From 0795c4c5322a376dbe5665853870f3806f76246a Mon Sep 17 00:00:00 2001 From: Manuel Alcocer J Date: Sun, 8 Nov 2026 07:42:33 +0100 Subject: [PATCH 2/3] Don't warm an audio buffer for tracks that play as DoP A DoP track is read straight from its DSF file, but the queue loader still created an AudioBuffer for the first track and the stream feeder warmed one for the next. That buffer holds decoded PCM at the probed DSD rate / 8 in 32-bit: 705.6 kHz for DSD128, about 5.6 MB/s, so up to ~1.7 GB for a five minute track that nothing ever reads. Two of them were enough to push the server past a 4 GiB memory limit. - add StreamsAudio.queue_item_plays_as_dop(), which checks the queue's active output protocol player (or a linked one before playback picks it) - skip the start buffer in the queue loader and the next-track warm-up in the stream feeder for DoP tracks; a track that ends up decoded creates its buffer on demand, as before - release any buffer already attached when the DoP stream starts --- .../controllers/player_queues/queue_loader.py | 7 +- .../player_queues/stream_feeder.py | 7 + music_assistant/controllers/streams/audio.py | 44 ++++++ .../player_queues/test_stream_feeder.py | 14 ++ tests/controllers/streams/test_flow_dop.py | 138 +++++++++++++++++- 5 files changed, 208 insertions(+), 2 deletions(-) diff --git a/music_assistant/controllers/player_queues/queue_loader.py b/music_assistant/controllers/player_queues/queue_loader.py index 5cb50e6f835..949a7d0ad01 100644 --- a/music_assistant/controllers/player_queues/queue_loader.py +++ b/music_assistant/controllers/player_queues/queue_loader.py @@ -415,7 +415,12 @@ async def _load_item( # immediate playback start. For preloaded next tracks we skip this and # initialize the buffer when the stream of the track before it nears its end. # AudioSource items are realtime/live and bypass the AudioBuffer. - if is_start and queue_item.streamdetails.media_type != MediaType.AUDIO_SOURCE: + # A DoP item is read straight from its file and never from the AudioBuffer. + if ( + is_start + and queue_item.streamdetails.media_type != MediaType.AUDIO_SOURCE + and not self.mass.streams.audio.queue_item_plays_as_dop(queue_item) + ): await self.mass.streams.audio.get_audio_buffer( queue_item, seek_position_ms=int(seek_position * 1000), diff --git a/music_assistant/controllers/player_queues/stream_feeder.py b/music_assistant/controllers/player_queues/stream_feeder.py index 1cb04331fce..7cda3c875fb 100644 --- a/music_assistant/controllers/player_queues/stream_feeder.py +++ b/music_assistant/controllers/player_queues/stream_feeder.py @@ -114,6 +114,13 @@ async def _do_prepare() -> None: streamed_item.streamdetails, prepared_item.streamdetails ) return + if self.mass.streams.audio.queue_item_plays_as_dop(prepared_item): + # streamed straight from its file, a buffer would never be read + self.logger.debug( + "Not preparing an audio buffer for %s: it streams as DoP", + prepared_item.name, + ) + return self.logger.debug( "Preparing audio buffer for next track %s on queue %s", prepared_item.name, diff --git a/music_assistant/controllers/streams/audio.py b/music_assistant/controllers/streams/audio.py index d8279c31445..bc148d96653 100644 --- a/music_assistant/controllers/streams/audio.py +++ b/music_assistant/controllers/streams/audio.py @@ -3470,6 +3470,45 @@ def dop_enabled_for_players(self, players: Iterable[Player]) -> bool: for player in players ) + def queue_item_plays_as_dop(self, queue_item: QueueItem) -> bool: + """ + Return whether a queue item will be streamed as DoP. + + A DoP item is read straight from its file, so an AudioBuffer for it would hold + decoded PCM nobody reads: for DSD128 that is 705.6 kHz / 32-bit, ~1.7 GB for a + five minute track. Callers use this to skip creating that buffer. When the item + ends up decoded after all (e.g. its DoP rate is not supported by the flow), the + decode path creates the buffer on demand, so only the warm-up is lost. + + The output player is the active protocol player of the queue player. Before + playback has picked one, any linked protocol player with DoP enabled counts. + + :param queue_item: The queue item, with its streamdetails resolved. + """ + streamdetails = queue_item.streamdetails + if streamdetails is None or not is_dop_source(streamdetails): + return False + if (queue_player := self.mass.players.get_player(queue_item.queue_id)) is None: + return False + if len(queue_player.group_members) > 1: + return False + output_player = queue_player.resolve_output_player() + candidates: list[Player] = [output_player] + if output_player is queue_player: + candidates.extend( + protocol_player + for linked in queue_player.linked_output_protocols + if (protocol_player := self.mass.players.get_player(linked.output_protocol_id)) + ) + return any( + self.dop_enabled_for_players([candidate]) + and self._get_dop_rate( + streamdetails, [rate for rate, _ in candidate.get_supported_sample_rates()] + ) + is not None + for candidate in candidates + ) + def clear_crossfade_handover(self, queue_id: str) -> None: """ Clear any pending crossfade data for a queue. @@ -6148,6 +6187,11 @@ async def _get_dop_queue_item_stream( streamdetails.uri, pcm_format.sample_rate, ) + if (unused_buffer := streamdetails.buffer) is not None: + # DoP never reads decoded PCM: release any buffer warmed for this item + # before DoP was decided, instead of holding it until its inactivity timeout + streamdetails.buffer = None + await unused_buffer.clear() bytes_received = 0 finished = False start_reported = False diff --git a/tests/controllers/player_queues/test_stream_feeder.py b/tests/controllers/player_queues/test_stream_feeder.py index dedf9201f8b..d762deb423e 100644 --- a/tests/controllers/player_queues/test_stream_feeder.py +++ b/tests/controllers/player_queues/test_stream_feeder.py @@ -294,6 +294,7 @@ async def _load_next(queue_id: str, item_id: str) -> Any: controller.load_next_queue_item = AsyncMock(side_effect=_load_next) # type: ignore[method-assign] mass = MagicMock() + mass.streams.audio.queue_item_plays_as_dop = MagicMock(return_value=False) controller.mass = mass return controller, next_item, mass @@ -860,3 +861,16 @@ async def _hang(*_args: object, **_kwargs: object) -> None: preparation.cancel() with pytest.raises(asyncio.CancelledError): await preparation + + +async def test_prepare_next_skips_the_buffer_of_a_dop_item() -> None: + """A DoP item streams straight from its file, so warming a decoded buffer only wastes RAM.""" + controller, next_item, mass = _controller_with_next_item() + mass.streams.audio.get_audio_buffer = AsyncMock() + mass.streams.audio.queue_item_plays_as_dop = MagicMock(return_value=True) + + controller.prepare_next_audio_buffer("queue-1", "current") + await mass.create_task.call_args.args[0] + + mass.streams.audio.queue_item_plays_as_dop.assert_called_once_with(next_item) + mass.streams.audio.get_audio_buffer.assert_not_awaited() diff --git a/tests/controllers/streams/test_flow_dop.py b/tests/controllers/streams/test_flow_dop.py index fc164b0827c..7b14c66336d 100644 --- a/tests/controllers/streams/test_flow_dop.py +++ b/tests/controllers/streams/test_flow_dop.py @@ -2,7 +2,8 @@ from __future__ import annotations -from unittest.mock import MagicMock, patch +from collections.abc import AsyncGenerator +from unittest.mock import AsyncMock, MagicMock, patch import pytest from music_assistant_models.enums import ContentType, MediaType, StreamType @@ -182,3 +183,138 @@ def test_dop_transition_detects_dsd_before_resolving() -> None: assert audio._dop_transition("queue", pcm_item) with patch.object(audio.mass.player_queues, "get_next_item", return_value=None): assert not audio._dop_transition("queue", pcm_item) + + +def _queue_item_audio( + *, + dop_player_ids: set[str], + queue_player: MagicMock, + protocol_players: dict[str, MagicMock] | None = None, +) -> StreamsAudio: + """Return a StreamsAudio resolving players by id, with DoP on for the given ids.""" + audio = StreamsAudio(MagicMock()) + players = {queue_player.player_id: queue_player, **(protocol_players or {})} + audio.mass.players.get_player = MagicMock(side_effect=players.get) # type: ignore[method-assign] + audio.mass.config.get_raw_player_config_value = MagicMock( # type: ignore[method-assign] + side_effect=lambda player_id, key, default=None: ( + player_id in dop_player_ids if key == CONF_DSD_DOP else default + ) + ) + return audio + + +def _universal_player( + output: MagicMock | None = None, linked_ids: tuple[str, ...] = () +) -> MagicMock: + """Return a queue player that renders through ``output`` (itself when None).""" + player = _make_player() + player.player_id = "queue" + player.resolve_output_player = MagicMock(return_value=output or player) + player.linked_output_protocols = [MagicMock(output_protocol_id=pid) for pid in linked_ids] + return player + + +def _dsd_queue_item(streamdetails: MagicMock | None = None) -> MagicMock: + item = _queue_track(streamdetails if streamdetails is not None else _dsd_streamdetails()) + item.queue_id = "queue" + return item + + +def _protocol_player(player_id: str, supported: list[tuple[int, int]] = DOP_RATES) -> MagicMock: + player = _make_player(supported=supported) + player.player_id = player_id + return player + + +def test_dop_item_on_active_dop_protocol_player_skips_buffer() -> None: + """The active Sendspin protocol player has DoP on: the item plays as DoP.""" + sendspin = _protocol_player("sendspin") + audio = _queue_item_audio( + dop_player_ids={"sendspin"}, + queue_player=_universal_player(output=sendspin, linked_ids=("sendspin",)), + protocol_players={"sendspin": sendspin}, + ) + assert audio.queue_item_plays_as_dop(_dsd_queue_item()) is True + + +def test_dop_item_before_protocol_selection_uses_linked_players() -> None: + """No active protocol yet: a linked DoP protocol player is enough.""" + sendspin = _protocol_player("sendspin") + audio = _queue_item_audio( + dop_player_ids={"sendspin"}, + queue_player=_universal_player(linked_ids=("airplay", "sendspin")), + protocol_players={"sendspin": sendspin, "airplay": _protocol_player("airplay")}, + ) + assert audio.queue_item_plays_as_dop(_dsd_queue_item()) is True + + +def test_active_non_dop_protocol_keeps_buffer() -> None: + """A different protocol is active: the item gets decoded, so it keeps its buffer.""" + airplay = _protocol_player("airplay") + audio = _queue_item_audio( + dop_player_ids={"sendspin"}, + queue_player=_universal_player(output=airplay, linked_ids=("sendspin", "airplay")), + protocol_players={"sendspin": _protocol_player("sendspin"), "airplay": airplay}, + ) + assert audio.queue_item_plays_as_dop(_dsd_queue_item()) is False + + +def test_unsupported_dop_rate_keeps_buffer() -> None: + """DSD128 needs 352.8 kHz: a player that tops out at 192 kHz decodes it.""" + sendspin = _protocol_player("sendspin", supported=[(176400, 24), (192000, 24)]) + audio = _queue_item_audio( + dop_player_ids={"sendspin"}, + queue_player=_universal_player(output=sendspin), + protocol_players={"sendspin": sendspin}, + ) + assert audio.queue_item_plays_as_dop(_dsd_queue_item(_dsd_streamdetails(705600))) is False + + +def test_grouped_queue_player_keeps_buffer() -> None: + """A group mixes the stream for its members, so DoP is never used.""" + queue_player = _universal_player() + queue_player.group_members = ["queue", "other"] + audio = _queue_item_audio(dop_player_ids={"queue"}, queue_player=queue_player) + assert audio.queue_item_plays_as_dop(_dsd_queue_item()) is False + + +def test_pcm_item_keeps_buffer() -> None: + """Only local DSF files play as DoP.""" + streamdetails = _dsd_streamdetails() + streamdetails.audio_format = AudioFormat( + content_type=ContentType.FLAC, sample_rate=44100, bit_depth=16, channels=2 + ) + audio = _queue_item_audio(dop_player_ids={"queue"}, queue_player=_universal_player()) + assert audio.queue_item_plays_as_dop(_dsd_queue_item(streamdetails)) is False + + +def test_unresolved_item_keeps_buffer() -> None: + """Without streamdetails nothing is known yet.""" + item = _dsd_queue_item() + item.streamdetails = None + audio = _queue_item_audio(dop_player_ids={"queue"}, queue_player=_universal_player()) + assert audio.queue_item_plays_as_dop(item) is False + + +@pytest.mark.asyncio +async def test_dop_stream_releases_a_warmed_buffer() -> None: + """A decoded buffer warmed before DoP was decided is released as DoP starts.""" + audio = StreamsAudio(MagicMock()) + warmed = MagicMock() + warmed.clear = AsyncMock() + streamdetails = _dsd_streamdetails() + streamdetails.buffer = warmed + streamdetails.duration = 300 + item = _dsd_queue_item(streamdetails) + + async def _no_audio(*_args: object, **_kwargs: object) -> AsyncGenerator[bytes]: + chunks: tuple[bytes, ...] = () + for chunk in chunks: + yield chunk + + with patch("music_assistant.controllers.streams.audio.get_dop_stream", _no_audio): + async for _ in audio._get_dop_queue_item_stream(item, _pcm_format(176400), 0, True): + pass + + warmed.clear.assert_awaited_once() + assert streamdetails.buffer is None From 46a511b56197d759b2dc132014a7a6ee1e5e7fd1 Mon Sep 17 00:00:00 2001 From: Manuel Alcocer J Date: Sun, 8 Nov 2026 08:02:18 +0100 Subject: [PATCH 3/3] Harden DoP against outputs and paths that would corrupt it Review follow-up for the DoP path: - Require a lossless stereo output of at least 24 bits at the DoP rate. New Player.supports_dop_output() (False by default); Sendspin answers it from the format a session at that rate actually negotiates (automatic format, top codec PCM/FLAC, stereo, >= 24 bits). A 16-bit-only or Opus client now gets decoded PCM instead of a truncated DoP payload. - Never use DoP when the samples get processed: DSP enabled on the player, a non-stereo output_channels mapping, or an active queue audio overlay. - Restart as decoded PCM when a member joins a player that is streaming DoP. The leader stops its session before the member is attached, so no DoP audio is backfilled to it, then resumes the queue at the same position; the grouped player no longer qualifies for DoP. - Read the DSF through the descriptor the filesystem provider opened and checked (open_local_files), with positional reads only, so a path swapped after stream details were resolved can not lead outside the source folder. - Keep the DoP markers alternating across gapless tracks: the marker phase is carried from one DoP item into the next within a flow and reset after decoded PCM. --- music_assistant/controllers/streams/audio.py | 109 ++++++++++++++-- music_assistant/helpers/dop.py | 96 +++++++++++--- music_assistant/models/player.py | 11 ++ music_assistant/providers/sendspin/player.py | 120 +++++++++++++++++- tests/controllers/streams/test_flow_dop.py | 88 +++++++++++++ tests/helpers/test_dop.py | 39 ++++++ tests/providers/sendspin/test_dop_grouping.py | 91 +++++++++++++ .../sendspin/test_session_sample_rate.py | 43 +++++++ 8 files changed, 562 insertions(+), 35 deletions(-) create mode 100644 tests/providers/sendspin/test_dop_grouping.py diff --git a/music_assistant/controllers/streams/audio.py b/music_assistant/controllers/streams/audio.py index bc148d96653..672f79a3121 100644 --- a/music_assistant/controllers/streams/audio.py +++ b/music_assistant/controllers/streams/audio.py @@ -154,7 +154,12 @@ resolve_output_player_ids, ) from music_assistant.helpers.compare import compare_item_ids -from music_assistant.helpers.dop import get_dop_sample_rate, get_dop_stream, is_dop_source +from music_assistant.helpers.dop import ( + DopMarkerState, + get_dop_sample_rate, + get_dop_stream, + is_dop_source, +) from music_assistant.helpers.dsp import ComplexFilter, filter_to_ffmpeg_params from music_assistant.helpers.ffmpeg import ( FFMpeg, @@ -638,6 +643,8 @@ def __init__(self, mass: MusicAssistant) -> None: # so without this the handoff is a race the fade loses. self._crossfade_pending: dict[str, tuple[str, asyncio.Event]] = {} self._smart_fades_mixer: SmartFadesMixer | None = None + # queue_id -> number of DoP item streams running for it (see is_streaming_dop) + self._dop_streams: dict[str, int] = {} # serializes buffer preparation per queue item, so concurrent callers share # the single source (and the single capacity reselection) instead of racing self._audio_buffer_locks: WeakValueDictionary[tuple[str, str], asyncio.Lock] = ( @@ -1544,8 +1551,12 @@ async def select_flow_pcm_format( supported_sample_rates=supported_sample_rates, ) if ( - dop_rate := self._get_dop_rate(start_streamdetails, supported_sample_rates) - ) and self.dop_enabled_for_players(players): + not overlay_active + and self.dop_enabled_for_players(players) + and ( + dop_rate := self._get_dop_rate(start_streamdetails, supported_sample_rates, players) + ) + ): # DoP has to reach the player bit-exact: no resampling, and F32 holds # every 24-bit DoP word exactly return AudioFormat( @@ -1669,6 +1680,7 @@ async def get_queue_item_stream( # noqa: C901, PLR0912, PLR0913 exact_seek: bool = False, consumer_connected: Callable[[], bool] | None = None, dop: bool = False, + dop_marker_state: DopMarkerState | None = None, ) -> AsyncGenerator[bytes]: """ Get the (PCM) audio stream for a single queue item. @@ -1714,7 +1726,7 @@ def _mark_item_unavailable() -> None: return if dop and is_dop_source(streamdetails): async for chunk in self._get_dop_queue_item_stream( - queue_item, pcm_format, seek_position, raise_on_error + queue_item, pcm_format, seek_position, raise_on_error, dop_marker_state ): yield chunk return @@ -2608,7 +2620,15 @@ def _consumer_left() -> bool: queue.queue_id, protocol_player ) flow_player = protocol_player or self.mass.players.get_player(queue.queue_id) - flow_dop = flow_player is not None and self.dop_enabled_for_players([flow_player]) + # an overlay is mixed into the flow after this generator, which DoP can not survive + flow_dop = ( + flow_player is not None + and not overlay_active(queue) + and self.dop_enabled_for_players([flow_player]) + ) + flow_dop_players = [flow_player] if flow_dop and flow_player is not None else [] + # markers keep alternating across gapless DoP tracks, see DopMarkerState + dop_marker_state = DopMarkerState() # note: get_crossfade_mode() already falls back to standard when smart fades aren't # available (no analysis provider / minimal buffer), so crossfade_mode is safe to use. self.logger.info( @@ -2667,6 +2687,7 @@ def _superseded() -> bool: flow_mode_sample_rate_conf, is_first_track=queue_track is start_queue_item, dop_enabled=flow_dop, + dop_players=flow_dop_players, ): break @@ -2892,9 +2913,14 @@ def _superseded() -> bool: holding_back = item_crossfade_mode != CrossfadeMode.DISABLED item_dop = flow_dop and ( - self._get_dop_rate(queue_track.streamdetails, [pcm_format.sample_rate]) + self._get_dop_rate( + queue_track.streamdetails, [pcm_format.sample_rate], flow_dop_players + ) is not None ) + if not item_dop: + # decoded PCM in between ends the DoP run: the next one starts afresh + dop_marker_state.phase = 0 if item_dop: # a prefetch holds decoded PCM, never the raw DSD that DoP needs await incoming_prefetcher.close() @@ -2916,6 +2942,7 @@ def _superseded() -> bool: session_id=flow_session_id, prepared_buffer=incoming_audio_buffer, dop=item_dop, + dop_marker_state=dop_marker_state, ) # closing here releases the decoders on an early exit, @@ -3457,8 +3484,9 @@ def dop_enabled_for_players(self, players: Iterable[Player]) -> bool: """ Return whether every given player takes DSD as DoP. - DoP is only sent to a player that plays on its own: a group mixes, resamples or - re-encodes the shared stream for its members, which turns DoP into noise. + DoP is only sent to a player that plays on its own and leaves the samples alone: a + group mixes, resamples or re-encodes the shared stream for its members, and DSP or + a non-stereo channel mapping filters it, all of which turn DoP into noise. :param players: The players consuming the stream. """ @@ -3467,9 +3495,21 @@ def dop_enabled_for_players(self, players: Iterable[Player]) -> bool: self.mass.config.get_raw_player_config_value(player.player_id, CONF_DSD_DOP, False) is True and len(player.group_members) <= 1 + and not self._player_processes_audio(player.player_id) for player in players ) + def is_streaming_dop(self, queue_id: str) -> bool: + """ + Return whether a queue is streaming a DSD item as DoP right now. + + Used to restart such a stream as decoded PCM when the player gets grouped, since + the running stream keeps its format for the members that join it. + + :param queue_id: The queue to check. + """ + return self._dop_streams.get(queue_id, 0) > 0 + def queue_item_plays_as_dop(self, queue_item: QueueItem) -> bool: """ Return whether a queue item will be streamed as DoP. @@ -3500,10 +3540,16 @@ def queue_item_plays_as_dop(self, queue_item: QueueItem) -> bool: for linked in queue_player.linked_output_protocols if (protocol_player := self.mass.players.get_player(linked.output_protocol_id)) ) + if (queue := self.mass.player_queues.get(queue_item.queue_id)) is not None and ( + overlay_active(queue) + ): + return False return any( self.dop_enabled_for_players([candidate]) and self._get_dop_rate( - streamdetails, [rate for rate, _ in candidate.get_supported_sample_rates()] + streamdetails, + [rate for rate, _ in candidate.get_supported_sample_rates()], + [candidate], ) is not None for candidate in candidates @@ -5777,6 +5823,7 @@ def _flow_stream_needs_restart( flow_mode_sample_rate_conf: str, is_first_track: bool, dop_enabled: bool = False, + dop_players: Iterable[Player] = (), ) -> bool: """ Return True if the upcoming queue track requires exiting the flow stream. @@ -5800,6 +5847,7 @@ def _flow_stream_needs_restart( :param flow_mode_sample_rate_conf: The flow mode sample rate config value. :param is_first_track: Whether this is the first track of the flow stream. :param dop_enabled: Whether the consuming player takes DSD as DoP. + :param dop_players: The players that must carry the DoP rate, see `_get_dop_rate`. """ # live audio (radio, plugin or audio source) cannot be flowed; let the # queue controller fall back to single-item streaming for this item @@ -5816,7 +5864,9 @@ def _flow_stream_needs_restart( if is_first_track or queue_track.streamdetails is None: return False if dop_enabled and ( - dop_rate := self._get_dop_rate(queue_track.streamdetails, supported_sample_rates) + dop_rate := self._get_dop_rate( + queue_track.streamdetails, supported_sample_rates, dop_players + ) ): # DoP only survives at exactly its own rate, whatever the flow mode if dop_rate != pcm_format.sample_rate: @@ -6118,21 +6168,41 @@ async def _add_remote_mp3_input_args( # CBR, while VBR lands seconds off with a TOC and can be minutes off without one add_input_fflag(extra_input_args, "+fastseek") + def _player_processes_audio(self, player_id: str) -> bool: + """ + Return whether a player's output path changes the samples (DSP or channel mapping). + + :param player_id: The player to check. + """ + if self.mass.config.get_player_dsp_config(player_id).enabled: + return True + output_channels = self.mass.config.get_raw_player_config_value( + player_id, CONF_OUTPUT_CHANNELS, "stereo" + ) + return str(output_channels or "stereo").strip().lower() in {"left", "right", "mono"} + def _get_dop_rate( self, streamdetails: StreamDetails | None, supported_sample_rates: Iterable[int], + players: Iterable[Player] = (), ) -> int | None: """ Return the DoP rate a stream would play at, or None if it cannot play as DoP. :param streamdetails: Stream details of the item, if resolved. :param supported_sample_rates: Sample rates the consuming player takes. + :param players: Players that must carry DoP at that rate untouched (lossless + stereo of at least 24 bits), see `Player.supports_dop_output`. """ if streamdetails is None or not is_dop_source(streamdetails): return None dop_rate = get_dop_sample_rate(streamdetails.audio_format) - return dop_rate if dop_rate in set(supported_sample_rates) else None + if dop_rate is None or dop_rate not in set(supported_sample_rates): + return None + if not all(player.supports_dop_output(dop_rate) for player in players): + return None + return dop_rate def _dop_transition(self, queue_id: str, queue_item: QueueItem) -> bool: """ @@ -6163,6 +6233,7 @@ async def _get_dop_queue_item_stream( pcm_format: AudioFormat, seek_position: float, raise_on_error: bool, + marker_state: DopMarkerState | None = None, ) -> AsyncGenerator[bytes]: """ Stream a DSD queue item as untouched DoP. @@ -6174,6 +6245,7 @@ async def _get_dop_queue_item_stream( :param pcm_format: Flow PCM format, running at the DoP rate of the item. :param seek_position: Start position in seconds. :param raise_on_error: Raise on stream errors instead of logging them. + :param marker_state: DoP marker phase shared with the gapless tracks around it. """ streamdetails = queue_item.streamdetails assert streamdetails @@ -6196,8 +6268,15 @@ async def _get_dop_queue_item_stream( finished = False start_reported = False next_buffer_triggered = False + fds: tuple[int, ...] = () + queue_id = queue_item.queue_id + self._dop_streams[queue_id] = self._dop_streams.get(queue_id, 0) + 1 try: - async for chunk in get_dop_stream(streamdetails.path, pcm_format, seek_position): + # read the file the provider checked, so a path swapped since the stream details + # were resolved can not lead outside the source folder + _inputs, fds = await self.open_local_files(streamdetails, [streamdetails.path]) + source: str | int = fds[0] if fds else streamdetails.path + async for chunk in get_dop_stream(source, pcm_format, seek_position, marker_state): if bytes_received == 0 and streamdetails.seconds_streamed is None: start_reported = self._notify_provider_stream_started(streamdetails) bytes_received += len(chunk) @@ -6235,6 +6314,12 @@ async def _get_dop_queue_item_stream( err, ) finally: + if (running := self._dop_streams.get(queue_id, 0) - 1) > 0: + self._dop_streams[queue_id] = running + else: + self._dop_streams.pop(queue_id, None) + for fd in fds: + os.close(fd) seconds_streamed = bytes_received / pcm_format.pcm_sample_size streamdetails.seconds_streamed = seconds_streamed self._notify_provider_streamed( diff --git a/music_assistant/helpers/dop.py b/music_assistant/helpers/dop.py index 33fa10d2b86..859e707075b 100644 --- a/music_assistant/helpers/dop.py +++ b/music_assistant/helpers/dop.py @@ -14,6 +14,7 @@ from __future__ import annotations import asyncio +import os import struct from dataclasses import dataclass from functools import cache @@ -46,6 +47,19 @@ class DoPError(Exception): """Raised when a file cannot be streamed as DoP.""" +@dataclass +class DopMarkerState: + """ + Marker phase carried from one DoP stream into the next. + + The markers have to keep alternating across a gapless track boundary: a file with an + odd number of DoP frames ends on the marker the next file would start with, and a DAC + that sees the same marker twice drops out of DSD mode for a moment. + """ + + phase: int = 0 + + @dataclass(frozen=True) class DsfInfo: """Layout details of a DSF file needed to stream it.""" @@ -102,24 +116,35 @@ def read_dsf_info(path: str) -> DsfInfo: :param path: Path to the DSF file. """ - with open(path, "rb") as file: - header = file.read(92) + fd = os.open(path, os.O_RDONLY) + try: + return read_dsf_info_fd(fd, path) + finally: + os.close(fd) + + +def read_dsf_info_fd(fd: int, name: str = "") -> DsfInfo: + """ + Parse the header of an open DSF file. + + :param fd: Descriptor of the DSF file, read with positional reads only. + :param name: Name of the file, for error messages. + """ + header = os.pread(fd, 92, 0) if len(header) < 92 or header[0:4] != b"DSD " or header[28:32] != b"fmt ": - raise DoPError(f"Not a DSF file: {path}") + raise DoPError(f"Not a DSF file: {name}") fmt_chunk_size = struct.unpack_from(" bytes: async def get_dop_stream( - path: str, + source: str | int, pcm_format: AudioFormat, seek_position: float = 0, + marker_state: DopMarkerState | None = None, ) -> AsyncGenerator[bytes]: """ Stream a DSF file as DoP, encoded in the given PCM format. - :param path: Path to the DSF file. + :param source: Path to the DSF file, or an open descriptor of it. A descriptor is only + read with positional reads and is left open for the caller to close. + :param pcm_format: PCM format of the consumer; its rate must be the DoP rate. + :param seek_position: Start position in seconds. + :param marker_state: Marker phase to start from and to leave for the next stream, so + the markers keep alternating across gapless tracks. A fresh stream starts at 0. + """ + if isinstance(source, int): + async for chunk in _get_dop_stream_fd( + source, str(source), pcm_format, seek_position, marker_state + ): + yield chunk + return + fd = await asyncio.to_thread(os.open, source, os.O_RDONLY) + try: + async for chunk in _get_dop_stream_fd(fd, source, pcm_format, seek_position, marker_state): + yield chunk + finally: + os.close(fd) + + +async def _get_dop_stream_fd( + fd: int, + name: str, + pcm_format: AudioFormat, + seek_position: float, + marker_state: DopMarkerState | None, +) -> AsyncGenerator[bytes]: + """ + Stream an open DSF file as DoP, see `get_dop_stream`. + + :param fd: Descriptor of the DSF file. + :param name: Name of the file, for error messages. :param pcm_format: PCM format of the consumer; its rate must be the DoP rate. :param seek_position: Start position in seconds. + :param marker_state: Marker phase carried across streams, if any. """ import numpy as np # noqa: PLC0415 - info = await asyncio.to_thread(read_dsf_info, path) + info = await asyncio.to_thread(read_dsf_info_fd, fd, name) if info.channels != 2 or pcm_format.channels != 2: raise DoPError("DoP streaming supports stereo DSD only") if pcm_format.sample_rate != info.dop_sample_rate: @@ -200,12 +259,15 @@ async def get_dop_stream( block_size = info.block_size total_blocks = -(-info.bytes_per_channel // block_size) start_block = min(int(seek_position * info.dsd_rate / 8) // block_size, total_blocks) - marker_phase = 0 + if marker_state is None: + marker_state = DopMarkerState() def _read_blocks(first_block: int, count: int) -> bytes: - with open(path, "rb") as file: - file.seek(info.data_offset + first_block * block_size * channels) - return file.read(count * block_size * channels) + return os.pread( + fd, + count * block_size * channels, + info.data_offset + first_block * block_size * channels, + ) block = start_block while block < total_blocks: @@ -231,7 +293,7 @@ def _read_blocks(first_block: int, count: int) -> bytes: [dsd, np.full((channels, 1), DSD_SILENCE_BYTE, dtype=np.uint8)], axis=1 ) if dsd.shape[1]: - words, marker_phase = pack_dop(dsd, marker_phase) + words, marker_state.phase = pack_dop(dsd, marker_state.phase) yield dop_words_to_pcm(words, pcm_format) block += whole_blocks diff --git a/music_assistant/models/player.py b/music_assistant/models/player.py index 36ec95ca40e..31a22f518ad 100644 --- a/music_assistant/models/player.py +++ b/music_assistant/models/player.py @@ -1501,6 +1501,17 @@ def enabled(self) -> bool: """Return if the player is enabled.""" return self._config.enabled + def supports_dop_output(self, sample_rate: int) -> bool: + """ + Return whether this player hands DoP at the given rate to its device untouched. + + DoP (DSD over PCM) only survives a lossless stereo output of at least 24 bits at + exactly its rate. Players that can not guarantee that keep this default: no DoP. + + :param sample_rate: The DoP rate (DSD rate / 16). + """ + return False + @final def get_supported_sample_rates(self) -> list[tuple[int, int]]: """ diff --git a/music_assistant/providers/sendspin/player.py b/music_assistant/providers/sendspin/player.py index 9a7c4e4d475..be651850f32 100644 --- a/music_assistant/providers/sendspin/player.py +++ b/music_assistant/providers/sendspin/player.py @@ -82,7 +82,12 @@ ProviderType, RepeatMode, ) -from music_assistant_models.errors import MediaNotFoundError, PlayerCommandFailed +from music_assistant_models.errors import ( + InvalidCommand, + MediaNotFoundError, + PlayerCommandFailed, + PlayerUnavailableError, +) from music_assistant_models.media_items import Album, Artist, is_track from music_assistant_models.player import DeviceInfo from PIL import Image @@ -413,7 +418,20 @@ def group_event_cb(self, group: SendspinGroup, event: GroupEvent) -> None: if client_id not in self._attr_group_members: self._attr_group_members.append(client_id) self.update_state() - self._schedule_membership_sync(group) + if ( + is_group_leader + and client_id != self.player_id + and (active_queue := self.mass.players.get_active_queue(self)) is not None + and self.mass.streams.audio.is_streaming_dop(active_queue.queue_id) + ): + # same task id: replaces any pending sync, which must not run first + self.mass.create_task( + self._restart_dop_stream_for_group(group, active_queue.queue_id), + task_id=f"sendspin_membership_sync_{self.player_id}", + abort_existing=True, + ) + else: + self._schedule_membership_sync(group) case GroupMemberRemovedEvent(client_id=client_id): self.mass.create_task(self._handle_group_member_removed(group, client_id)) self._schedule_membership_sync(group) @@ -861,6 +879,13 @@ def _on_group_changed(self, new_group: SendspinGroup) -> None: def _on_group_stopped(self) -> None: """Handle the group transitioning to STOPPED state.""" + async def _stop_playback_session(self, reason: str) -> None: + """ + Stop the playback session this player feeds, if it feeds one. + + :param reason: Why the session stops, for logging. + """ + async def _sync_membership_from_group(self, group: SendspinGroup) -> None: """ Sync MA player group membership from authoritative group state. @@ -877,6 +902,35 @@ async def _sync_membership_from_group(self, group: SendspinGroup) -> None: self._attr_group_members = desired_group_members self.update_state() + async def _restart_dop_stream_for_group(self, group: SendspinGroup, queue_id: str) -> None: + """ + Restart a DoP stream as decoded PCM now that this player leads a group. + + The running stream would otherwise carry on as DoP for the members that join it, + which resample or mix it, or can not decode it, and play it as full-scale noise. + The stream stops before the member is attached, so none of its DoP audio is + backfilled to the member. The queue then resumes at the same position, and the + new flow stream no longer qualifies for DoP because the player is grouped. + + :param group: The Sendspin group the member joined. + :param queue_id: The queue streaming DoP. + """ + queue = self.mass.player_queues.get(queue_id) + position = int(queue.corrected_elapsed_time) if queue is not None else 0 + self.logger.info( + "Player %s got grouped while streaming DSD as DoP, restarting as PCM at %ss", + self.display_name, + position, + ) + await self._stop_playback_session("grouped during DoP playback") + await self._sync_membership_from_group(group) + try: + await self.mass.player_queues.seek(queue_id, position) + except (InvalidCommand, PlayerUnavailableError) as err: + self.logger.warning( + "Could not resume %s as PCM after grouping: %s", self.display_name, err + ) + def _schedule_membership_sync(self, group: SendspinGroup) -> None: """Schedule a coalesced membership reconciliation task for this player.""" self.mass.create_task( @@ -1378,6 +1432,31 @@ def requires_flow_mode(self) -> bool: """Return if the player requires flow mode.""" return True + def supports_dop_output(self, sample_rate: int) -> bool: + """ + Return whether a session at the given rate reaches the client fit to carry DoP. + + DoP needs lossless stereo of at least 24 bits at exactly its rate: a 16-bit or lossy + output truncates the DSD payload. Only the automatic format follows the session rate. + + :param sample_rate: The DoP rate. + """ + player_role = self._player_role + if player_role is None: + return False + config_value = self.config.get_value( + CONF_PREFERRED_SENDSPIN_FORMAT, SENDSPIN_FORMAT_AUTOMATIC + ) + if config_value != SENDSPIN_FORMAT_AUTOMATIC: + return False + fmt = self._format_for_rate(self._top_codec_formats(player_role), sample_rate) + return ( + fmt is not None + and fmt.codec in (AudioCodec.PCM, AudioCodec.FLAC) + and fmt.channels == 2 + and fmt.bit_depth >= 24 + ) + def __init__( self, provider: SendspinProvider, @@ -2180,6 +2259,14 @@ def _on_group_stopped(self) -> None: return self.mass.create_task(self._cancel_stale_playback_session(stale_task)) + async def _stop_playback_session(self, reason: str) -> None: + """ + Stop the playback session this player feeds. + + :param reason: Why the session stops, for logging. + """ + await self.playback_session.cancel(reason) + async def _cancel_stale_playback_session(self, task: asyncio.Task[None]) -> None: """Cancel the playback session only if the given task is still the active one.""" if self.playback_session.playback_task is not task: @@ -2380,16 +2467,37 @@ def _session_format( formats = self._top_codec_formats(player_role) if not formats or self._session_sample_rate in (None, formats[0].sample_rate): return None, None - default = formats[0] - candidates = [fmt for fmt in formats if fmt.sample_rate == self._session_sample_rate] - if not candidates: + best = self._format_for_rate(formats, self._session_sample_rate) + if best is None: return None, None - best = max(candidates, key=lambda fmt: (fmt.channels == default.channels, fmt.bit_depth)) audio_format = SendspinAudioFormat( sample_rate=best.sample_rate, bit_depth=best.bit_depth, channels=best.channels ) return audio_format, best.codec + @staticmethod + def _format_for_rate( + formats: list[SupportedAudioFormat], sample_rate: int + ) -> SupportedAudioFormat | None: + """ + Return the format a session at the given rate negotiates, or None for no match. + + The client's own default is kept when it already runs at that rate. Otherwise the + deepest format at that rate wins, preferring the default's channel count. + + :param formats: The client's formats in the codec of its top choice, in priority order. + :param sample_rate: The session sample rate. + """ + if not formats: + return None + default = formats[0] + if default.sample_rate == sample_rate: + return default + candidates = [fmt for fmt in formats if fmt.sample_rate == sample_rate] + if not candidates: + return None + return max(candidates, key=lambda fmt: (fmt.channels == default.channels, fmt.bit_depth)) + @staticmethod def _top_codec_formats(player_role: PlayerRoleProtocol) -> list[SupportedAudioFormat]: """ diff --git a/tests/controllers/streams/test_flow_dop.py b/tests/controllers/streams/test_flow_dop.py index 7b14c66336d..0f65c194f14 100644 --- a/tests/controllers/streams/test_flow_dop.py +++ b/tests/controllers/streams/test_flow_dop.py @@ -2,7 +2,9 @@ from __future__ import annotations +import os from collections.abc import AsyncGenerator +from pathlib import Path from unittest.mock import AsyncMock, MagicMock, patch import pytest @@ -12,6 +14,7 @@ from music_assistant.constants import ( CONF_DSD_DOP, CONF_FLOW_MODE_SAMPLE_RATE, + CONF_OUTPUT_CHANNELS, FLOW_MODE_SAMPLE_RATE_SMART, ) from music_assistant.controllers.streams.audio import StreamsAudio @@ -116,6 +119,57 @@ async def test_flow_format_ignores_dop_for_groups() -> None: assert fmt.sample_rate == 352800 +@pytest.mark.asyncio +async def test_flow_format_ignores_dop_for_a_16_bit_output() -> None: + """A client that takes the DoP rate only at 16 bits would truncate the DSD payload.""" + audio = _make_streams_audio(dop=True) + player = _make_player() + player.supports_dop_output = MagicMock(return_value=False) + fmt = await audio.select_flow_pcm_format(player, start_streamdetails=_dsd_streamdetails()) + assert fmt.sample_rate == 352800 + player.supports_dop_output.assert_called_with(176400) + + +@pytest.mark.asyncio +async def test_flow_format_ignores_dop_with_dsp() -> None: + """DSP filters the stream, so a player with DSP on gets decoded PCM.""" + audio = _make_streams_audio(dop=True) + audio.mass.config.get_player_dsp_config = MagicMock( # type: ignore[method-assign] + return_value=MagicMock(enabled=True) + ) + fmt = await audio.select_flow_pcm_format( + _make_player(), start_streamdetails=_dsd_streamdetails() + ) + assert fmt.sample_rate == 352800 + + +@pytest.mark.asyncio +@pytest.mark.parametrize("output_channels", ["mono", "left", "right"]) +async def test_flow_format_ignores_dop_with_channel_mapping(output_channels: str) -> None: + """A non-stereo channel mapping mixes the channels, so DoP is not used.""" + audio = _make_streams_audio(dop=True) + audio.mass.config.get_raw_player_config_value = MagicMock( # type: ignore[method-assign] + side_effect=lambda _player_id, key, default=None: { + CONF_DSD_DOP: True, + CONF_OUTPUT_CHANNELS: output_channels, + }.get(key, default) + ) + fmt = await audio.select_flow_pcm_format( + _make_player(), start_streamdetails=_dsd_streamdetails() + ) + assert fmt.sample_rate == 352800 + + +@pytest.mark.asyncio +async def test_flow_format_ignores_dop_with_an_overlay() -> None: + """An audio overlay is mixed into the flow, so DoP is not used.""" + audio = _make_streams_audio(dop=True) + fmt = await audio.select_flow_pcm_format( + _make_player(), start_streamdetails=_dsd_streamdetails(), overlay_active=True + ) + assert fmt.sample_rate == 352800 + + @pytest.mark.parametrize( ("flow_rate", "probed_rate", "expected"), [ @@ -195,6 +249,10 @@ def _queue_item_audio( audio = StreamsAudio(MagicMock()) players = {queue_player.player_id: queue_player, **(protocol_players or {})} audio.mass.players.get_player = MagicMock(side_effect=players.get) # type: ignore[method-assign] + audio.mass.player_queues.get = MagicMock(return_value=None) # type: ignore[method-assign] + audio.mass.config.get_player_dsp_config = MagicMock( # type: ignore[method-assign] + return_value=MagicMock(enabled=False) + ) audio.mass.config.get_raw_player_config_value = MagicMock( # type: ignore[method-assign] side_effect=lambda player_id, key, default=None: ( player_id in dop_player_ids if key == CONF_DSD_DOP else default @@ -318,3 +376,33 @@ async def _no_audio(*_args: object, **_kwargs: object) -> AsyncGenerator[bytes]: warmed.clear.assert_awaited_once() assert streamdetails.buffer is None + + +@pytest.mark.asyncio +async def test_dop_stream_reads_the_validated_descriptor(tmp_path: Path) -> None: + """DoP reads the descriptor the provider opened and checked, and closes it afterwards.""" + audio = StreamsAudio(MagicMock()) + item = _dsd_queue_item() + item.streamdetails.buffer = None + item.streamdetails.duration = 300 + fd = os.open(tmp_path / "track.dsf", os.O_RDONLY | os.O_CREAT) + audio.open_local_files = AsyncMock( # type: ignore[method-assign] + return_value=([f"/proc/self/fd/{fd}"], (fd,)) + ) + sources: list[object] = [] + streaming: list[bool] = [] + + async def _capture(source: object, *_args: object, **_kwargs: object) -> AsyncGenerator[bytes]: + sources.append(source) + streaming.append(audio.is_streaming_dop(item.queue_id)) + yield b"\x00" * 8 + + with patch("music_assistant.controllers.streams.audio.get_dop_stream", _capture): + async for _ in audio._get_dop_queue_item_stream(item, _pcm_format(176400), 0, True): + pass + + assert sources == [fd] + assert streaming == [True] + assert not audio.is_streaming_dop(item.queue_id) + with pytest.raises(OSError, match="Bad file descriptor"): + os.fstat(fd) diff --git a/tests/helpers/test_dop.py b/tests/helpers/test_dop.py index 839be7f6799..7587d00a847 100644 --- a/tests/helpers/test_dop.py +++ b/tests/helpers/test_dop.py @@ -2,6 +2,7 @@ from __future__ import annotations +import os import struct from pathlib import Path @@ -13,6 +14,7 @@ from music_assistant.helpers.dop import ( DoPError, + DopMarkerState, get_dop_sample_rate, get_dop_stream, is_dop_source, @@ -180,3 +182,40 @@ def test_read_dsf_info(tmp_path: Path) -> None: assert info.dop_sample_rate == 176400 assert info.bytes_per_channel == 10 assert info.data_offset == 92 + + +async def test_dop_markers_alternate_across_gapless_tracks(tmp_path: Path) -> None: + """A track with an odd number of DoP frames hands the next track the other marker.""" + # 1026 bytes per channel = 513 DoP frames: the first track ends on marker 0x05 + first = tmp_path / "first.dsf" + second = tmp_path / "second.dsf" + _write_dsf(first, bytes(1026), bytes(1026)) + _write_dsf(second, bytes(1024), bytes(1024)) + pcm_format = _pcm_format(ContentType.PCM_S32LE, 32) + state = DopMarkerState() + + raw_first = b"".join( + [chunk async for chunk in get_dop_stream(str(first), pcm_format, marker_state=state)] + ) + raw_second = b"".join( + [chunk async for chunk in get_dop_stream(str(second), pcm_format, marker_state=state)] + ) + markers = np.concatenate([_words_from_s32(raw_first), _words_from_s32(raw_second)])[::2] >> 16 + + assert markers[512] == 0x05 + assert markers[513] == 0xFA + assert np.all(markers[1:] != markers[:-1]) + + +async def test_dop_stream_reads_an_open_descriptor(tmp_path: Path) -> None: + """A descriptor is read in place and left open for its owner to close.""" + path = tmp_path / "track.dsf" + _write_dsf(path, bytes(range(256)) * 4, bytes(range(256)) * 4) + pcm_format = _pcm_format(ContentType.PCM_S32LE, 32) + fd = os.open(path, os.O_RDONLY) + try: + from_fd = b"".join([chunk async for chunk in get_dop_stream(fd, pcm_format)]) + os.fstat(fd) # still open + finally: + os.close(fd) + assert from_fd == await _collect(path, pcm_format) diff --git a/tests/providers/sendspin/test_dop_grouping.py b/tests/providers/sendspin/test_dop_grouping.py new file mode 100644 index 00000000000..31da6bbd9ea --- /dev/null +++ b/tests/providers/sendspin/test_dop_grouping.py @@ -0,0 +1,91 @@ +"""Tests for restarting a DoP stream as PCM when a Sendspin player gets grouped.""" + +from __future__ import annotations + +from unittest.mock import AsyncMock, MagicMock, PropertyMock, patch + +from aiosendspin.server.events import GroupMemberAddedEvent + +from music_assistant.providers.sendspin.player import SendspinPlayer + +LEADER_ID = "leader" +MEMBER_ID = "member" + + +def _leader(*, streaming_dop: bool) -> tuple[SendspinPlayer, MagicMock]: + """Build an un-initialized SendspinPlayer leading a group, with its mocked mass.""" + player = SendspinPlayer.__new__(SendspinPlayer) + player._player_id = LEADER_ID + player._attr_group_members = [] + player.logger = MagicMock() + player._config = MagicMock(name="config") + mass = MagicMock() + mass.players.get_active_queue.return_value = MagicMock(queue_id="queue-1") + mass.streams.audio.is_streaming_dop.return_value = streaming_dop + player.mass = mass + return player, mass + + +def _group() -> MagicMock: + group = MagicMock() + group.clients = [MagicMock(client_id=LEADER_ID), MagicMock(client_id=MEMBER_ID)] + return group + + +def test_member_joining_a_dop_stream_restarts_it() -> None: + """A member joining while DoP plays replaces the membership sync with a restart.""" + player, mass = _leader(streaming_dop=True) + group = _group() + with ( + patch.object(SendspinPlayer, "synced_to", new_callable=PropertyMock, return_value=None), + patch.object(player, "update_state", MagicMock()), + patch.object(player, "_restart_dop_stream_for_group", MagicMock()) as restart, + patch.object(player, "_schedule_membership_sync", MagicMock()) as sync, + ): + player.group_event_cb(group, GroupMemberAddedEvent(client_id=MEMBER_ID)) + + restart.assert_called_once_with(group, "queue-1") + sync.assert_not_called() + assert mass.create_task.call_args.kwargs == { + "task_id": f"sendspin_membership_sync_{LEADER_ID}", + "abort_existing": True, + } + assert player._attr_group_members == [LEADER_ID, MEMBER_ID] + + +def test_member_joining_a_pcm_stream_only_syncs_membership() -> None: + """Without DoP the running stream is kept and the member is simply attached.""" + player, mass = _leader(streaming_dop=False) + group = _group() + with ( + patch.object(SendspinPlayer, "synced_to", new_callable=PropertyMock, return_value=None), + patch.object(player, "update_state", MagicMock()), + patch.object(player, "_schedule_membership_sync", MagicMock()) as sync, + ): + player.group_event_cb(group, GroupMemberAddedEvent(client_id=MEMBER_ID)) + + sync.assert_called_once_with(group) + mass.create_task.assert_not_called() + + +async def test_restart_stops_before_attaching_the_member_and_resumes_in_place() -> None: + """The DoP session stops before the member is attached, then the queue resumes there.""" + player, mass = _leader(streaming_dop=True) + group = _group() + calls: list[str] = [] + player.playback_session = MagicMock() + player.playback_session.cancel = AsyncMock(side_effect=lambda *_a, **_k: calls.append("stop")) + mass.player_queues.get.return_value = MagicMock(corrected_elapsed_time=42.7) + mass.player_queues.seek = AsyncMock(side_effect=lambda *_a: calls.append("seek")) + + async def _sync(_group: object) -> None: + calls.append("attach") + + with ( + patch.object(player, "_sync_membership_from_group", _sync), + patch.object(SendspinPlayer, "display_name", new_callable=PropertyMock, return_value="L"), + ): + await player._restart_dop_stream_for_group(group, "queue-1") + + assert calls == ["stop", "attach", "seek"] + mass.player_queues.seek.assert_awaited_once_with("queue-1", 42) diff --git a/tests/providers/sendspin/test_session_sample_rate.py b/tests/providers/sendspin/test_session_sample_rate.py index 86f86abda90..878e1712387 100644 --- a/tests/providers/sendspin/test_session_sample_rate.py +++ b/tests/providers/sendspin/test_session_sample_rate.py @@ -169,3 +169,46 @@ def _session(player: SendspinPlayer, *, sample_rate: int) -> SendspinPlaybackSes media = PlayerMedia(uri="track-1", source_id="queue-1", queue_item_id="item-1") session._start_streamdetails = session._get_start_streamdetails(media) return session + + +def _fmt(codec: AudioCodec, rate: int, bits: int, channels: int = 2) -> SupportedAudioFormat: + return SupportedAudioFormat(codec=codec, channels=channels, sample_rate=rate, bit_depth=bits) + + +def test_dop_output_needs_24_bit_lossless_stereo_at_the_rate() -> None: + """DoP needs the session at its rate to negotiate lossless stereo of 24 bits or more.""" + deep, _ = _player([_fmt(AudioCodec.FLAC, 48000, 16), _fmt(AudioCodec.FLAC, 352800, 24)]) + shallow, _ = _player([_fmt(AudioCodec.FLAC, 48000, 16), _fmt(AudioCodec.FLAC, 352800, 16)]) + pcm, _ = _player([_fmt(AudioCodec.PCM, 48000, 16), _fmt(AudioCodec.PCM, 352800, 32)]) + lossy, _ = _player([_fmt(AudioCodec.OPUS, 48000, 16), _fmt(AudioCodec.FLAC, 352800, 24)]) + + assert deep.supports_dop_output(352800) + assert not deep.supports_dop_output(176400) + assert not shallow.supports_dop_output(352800) + assert pcm.supports_dop_output(352800) + # the session stays in the client's top codec, which can not carry the rate + assert not lossy.supports_dop_output(352800) + + +def test_dop_output_follows_the_negotiated_format() -> None: + """The deepest stereo format at the rate is the one negotiated, so it decides.""" + player, _ = _player( + [ + _fmt(AudioCodec.PCM, 48000, 16), + _fmt(AudioCodec.PCM, 176400, 16), + _fmt(AudioCodec.PCM, 176400, 24), + ] + ) + assert player.supports_dop_output(176400) + # a client whose own default runs at the rate keeps that default + default_16, _ = _player([_fmt(AudioCodec.PCM, 176400, 16), _fmt(AudioCodec.PCM, 176400, 24)]) + assert not default_16.supports_dop_output(176400) + + +def test_dop_output_needs_the_automatic_format() -> None: + """An explicit Sendspin format never follows the DoP rate.""" + player, _ = _player( + [_fmt(AudioCodec.FLAC, 48000, 16), _fmt(AudioCodec.FLAC, 352800, 24)], + preferred_format="flac:48000:16:2", + ) + assert not player.supports_dop_output(352800)