music-assistant-server

1"""Runtime audio processing details for queue and live source streams."""
2
3from __future__ import annotations
4
5from collections.abc import Iterable
6from copy import deepcopy
7from dataclasses import dataclass, field, replace
8from typing import TYPE_CHECKING
9
10from music_assistant_models.audio_processing import (
11    ActiveSourceAudioDetails,
12    AudioFidelity,
13    AudioNormalizationDetails,
14    AudioNormalizationMeasurementSource,
15    AudioOutputDetails,
16    AudioProcessingChain,
17    AudioQuality,
18    AudioQueueProcessing,
19)
20from music_assistant_models.dsp import DSPState
21from music_assistant_models.enums import ContentType, CrossfadeMode, VolumeNormalizationMode
22
23from music_assistant.helpers.audio import get_bit_rate
24
25if TYPE_CHECKING:
26    from music_assistant_models.media_items import AudioFormat
27    from music_assistant_models.streamdetails import StreamDetails
28
29    from music_assistant.helpers.dsp import ComplexFilter
30    from music_assistant.mass import MusicAssistant
31    from music_assistant.models.player import PlayerMedia
32
33
34_QUALITY_RANK = {
35    AudioQuality.UNKNOWN: 0,
36    AudioQuality.LOW: 1,
37    AudioQuality.STANDARD: 2,
38    AudioQuality.LOSSLESS: 3,
39    AudioQuality.HI_RES: 4,
40}
41
42
43@dataclass(slots=True)
44class AudioOutputPlan:
45    """Executable filters and matching client-facing output details."""
46
47    filter_params: list[str | ComplexFilter]
48    output_details: AudioOutputDetails
49    input_format: AudioFormat
50    handoff_format: AudioFormat | None = None
51    dsp_config_id: str | None = None
52
53
54@dataclass(slots=True)
55class _AudioProcessingItem:
56    """Processing details cached for one queue item."""
57
58    queue_processing: AudioQueueProcessing | None = None
59    input_format: AudioFormat | None = None
60    alters_audio: bool = False
61
62
63@dataclass(slots=True)
64class _AudioOutputEntry:
65    """Client-facing output details with private intermediate formats."""
66
67    details: AudioOutputDetails
68    input_format: AudioFormat
69    handoff_format: AudioFormat | None = None
70    dsp_config_id: str | None = None
71
72
73@dataclass(slots=True)
74class _AudioProcessingSession:
75    """Runtime processing state for one queue playback session."""
76
77    session_id: str
78    items: dict[str, _AudioProcessingItem] = field(default_factory=dict)
79    outputs: dict[str | None, dict[str, _AudioOutputEntry]] = field(default_factory=dict)
80    shared_output_templates: dict[str | None, _AudioOutputEntry] = field(default_factory=dict)
81
82
83@dataclass(slots=True)
84class _AudioSourceProcessingSession:
85    """Runtime processing state for one live AudioSource selection."""
86
87    session_id: str
88    context_ready: bool = False
89    pcm_format: AudioFormat | None = None
90    crossfade_mode: CrossfadeMode = CrossfadeMode.UNKNOWN
91    volume_normalization_mode: VolumeNormalizationMode = VolumeNormalizationMode.UNKNOWN
92    outputs: dict[str | None, dict[str, _AudioOutputEntry]] = field(default_factory=dict)
93    shared_output_templates: dict[str | None, _AudioOutputEntry] = field(default_factory=dict)
94
95
96class AudioProcessingManager:
97    """Build effective audio processing details for active playback."""
98
99    def __init__(self, mass: MusicAssistant) -> None:
100        """
101        Initialize the audio processing manager.
102
103        :param mass: Music Assistant instance.
104        """
105        self.mass = mass
106        self._sessions: dict[str, _AudioProcessingSession] = {}
107        self._source_sessions: dict[str, _AudioSourceProcessingSession] = {}
108
109    def start_session(self, queue_id: str, session_id: str) -> None:
110        """
111        Start tracking a queue playback session.
112
113        :param queue_id: Queue identifier.
114        :param session_id: Internal queue playback session identifier.
115        """
116        current = self._sessions.get(queue_id)
117        if current and current.session_id == session_id:
118            return
119        if self._clear_streamdetails(queue_id):
120            self.mass.player_queues.signal_update(queue_id)
121        self._sessions[queue_id] = _AudioProcessingSession(session_id=session_id)
122
123    def update_item_context(
124        self,
125        queue_id: str,
126        session_id: str,
127        queue_item_id: str,
128        queue_processing: AudioQueueProcessing,
129        *,
130        alters_audio: bool = False,
131    ) -> None:
132        """
133        Store shared processing selected for a queue item.
134
135        :param queue_id: Queue identifier.
136        :param session_id: Internal queue playback session identifier.
137        :param queue_item_id: Queue item identifier.
138        :param queue_processing: Effective shared processing.
139        :param alters_audio: Whether an intentionally hidden transform alters samples.
140        """
141        session = self._get_session(queue_id, session_id)
142        if session is None:
143            return
144        self._prune_played_items(queue_id, session)
145        if self._is_played_item(queue_id, queue_item_id):
146            return
147        item = session.items.setdefault(queue_item_id, _AudioProcessingItem())
148        previous = item.queue_processing
149        item.queue_processing = AudioQueueProcessing(
150            pcm_format=deepcopy(queue_processing.pcm_format),
151            normalization=deepcopy(
152                previous.normalization if previous else queue_processing.normalization
153            ),
154            playback_speed=queue_processing.playback_speed,
155            crossfade_mode=queue_processing.crossfade_mode,
156            overlay_active=queue_processing.overlay_active,
157        )
158        item.alters_audio = alters_audio
159        self._publish_item(queue_id, queue_item_id, session)
160
161    def update_item_runtime(
162        self,
163        queue_id: str,
164        session_id: str,
165        queue_item_id: str,
166        input_format: AudioFormat,
167        pcm_format: AudioFormat,
168        normalization: AudioNormalizationDetails | None,
169        playback_speed: float,
170        *,
171        alters_audio: bool = False,
172    ) -> None:
173        """
174        Store shared processing constructed for a queue item.
175
176        :param queue_id: Queue identifier.
177        :param session_id: Internal queue playback session identifier.
178        :param queue_item_id: Queue item identifier.
179        :param input_format: PCM format entering shared queue processing.
180        :param pcm_format: Internal PCM format leaving shared processing.
181        :param normalization: Effective normalization details.
182        :param playback_speed: Effective playback-speed conversion.
183        :param alters_audio: Whether an intentionally hidden transform alters samples.
184        """
185        session = self._get_session(queue_id, session_id)
186        if session is None:
187            return
188        self._prune_played_items(queue_id, session)
189        if self._is_played_item(queue_id, queue_item_id):
190            return
191        item = session.items.setdefault(queue_item_id, _AudioProcessingItem())
192        previous = item.queue_processing or AudioQueueProcessing()
193        item.input_format = deepcopy(input_format)
194        item.queue_processing = AudioQueueProcessing(
195            pcm_format=deepcopy(pcm_format),
196            normalization=deepcopy(normalization),
197            playback_speed=playback_speed,
198            crossfade_mode=previous.crossfade_mode,
199            overlay_active=previous.overlay_active,
200        )
201        item.alters_audio = item.alters_audio or alters_audio
202        self._publish_item(queue_id, queue_item_id, session)
203
204    def update_source_context(
205        self,
206        player_id: str,
207        session_id: str,
208        *,
209        pcm_format: AudioFormat,
210        crossfade_enabled: bool | None,
211        volume_normalization_enabled: bool | None,
212    ) -> None:
213        """
214        Store the source-owned processing active for a live AudioSource.
215
216        :param player_id: Player that owns the source selection.
217        :param session_id: Source playback session that owns the update.
218        :param pcm_format: PCM format the source's audio is decoded to.
219        :param crossfade_enabled: Whether the source applies crossfade, or None if unknown.
220        :param volume_normalization_enabled: Whether the source normalizes, or None if unknown.
221        """
222        session = self._get_source_session(player_id, session_id, create=True)
223        if session is None:
224            return
225        session.context_ready = True
226        session.pcm_format = deepcopy(pcm_format)
227        if crossfade_enabled is None:
228            session.crossfade_mode = CrossfadeMode.UNKNOWN
229        elif crossfade_enabled:
230            session.crossfade_mode = CrossfadeMode.SOURCE
231        else:
232            session.crossfade_mode = CrossfadeMode.DISABLED
233        if volume_normalization_enabled is None:
234            session.volume_normalization_mode = VolumeNormalizationMode.UNKNOWN
235        elif volume_normalization_enabled:
236            session.volume_normalization_mode = VolumeNormalizationMode.SOURCE
237        else:
238            session.volume_normalization_mode = VolumeNormalizationMode.DISABLED
239        self._publish_source(player_id, session)
240
241    def update_output(
242        self,
243        player_id: str,
244        output_plan: AudioOutputPlan,
245        *,
246        shared_player_ids: Iterable[str] | None = None,
247        queue_id: str,
248        session_id: str,
249        queue_item_id: str | None = None,
250    ) -> bool:
251        """
252        Store an effective player output.
253
254        :param player_id: Destination player identifier.
255        :param output_plan: Effective output processing and private intermediate formats.
256        :param shared_player_ids: Additional players receiving this identical output path.
257            An empty iterable marks a path that can gain shared destinations later.
258        :param queue_id: Queue that owns the output, or the player holding a live source.
259        :param session_id: Queue session or source playback session that owns the output.
260        :param queue_item_id: Queue item for single-item output, or None for flow output
261            and for a live source.
262        :return: Whether the effective output changed.
263        """
264        source_session = (
265            self._get_source_session(queue_id, session_id, create=True)
266            if queue_item_id is None
267            else None
268        )
269        queue_session: _AudioProcessingSession | None = None
270        if source_session is not None:
271            session: _AudioProcessingSession | _AudioSourceProcessingSession = source_session
272        else:
273            queue_session = self._get_session(queue_id, session_id)
274            if queue_session is None:
275                return False
276            self._prune_played_items(queue_id, queue_session)
277            if queue_item_id is not None and self._is_played_item(queue_id, queue_item_id):
278                return False
279            session = queue_session
280
281        destination_player_ids = {player_id}
282        if shared_player_ids is not None:
283            destination_player_ids.update(shared_player_ids)
284
285        entries: dict[str, _AudioOutputEntry] = {}
286        for destination_player_id in destination_player_ids:
287            entry = _AudioOutputEntry(
288                details=deepcopy(output_plan.output_details),
289                input_format=deepcopy(output_plan.input_format),
290                handoff_format=deepcopy(output_plan.handoff_format),
291                dsp_config_id=output_plan.dsp_config_id,
292            )
293            entry.details.player_ids = [destination_player_id]
294            entries[destination_player_id] = entry
295
296        if shared_player_ids is None:
297            session.shared_output_templates.pop(queue_item_id, None)
298        else:
299            session.shared_output_templates[queue_item_id] = deepcopy(entries[player_id])
300
301        item_outputs = session.outputs.setdefault(queue_item_id, {})
302        changed_entries = {
303            destination_player_id: entry
304            for destination_player_id, entry in entries.items()
305            if item_outputs.get(destination_player_id) != entry
306        }
307        if not changed_entries:
308            return False
309        item_outputs.update(changed_entries)
310        if source_session is not None:
311            self._publish_source(queue_id, source_session)
312            return True
313        assert queue_session is not None
314        current_changed = self._publish_all(queue_id, queue_session)
315        queue = self.mass.player_queues.get(queue_id)
316        if current_changed or (
317            queue
318            and queue.current_item
319            and queue_item_id in (None, queue.current_item.queue_item_id)
320        ):
321            self.mass.player_queues.signal_update(queue_id)
322        return True
323
324    def retain_outputs(self, queue_id: str, player_ids: set[str]) -> bool:
325        """
326        Reconcile processing outputs with the current playback destinations.
327
328        :param queue_id: Queue identifier.
329        :param player_ids: Player identifiers that belong to the output.
330        :return: Whether reconciliation published an updated current chain.
331        """
332        queue_session = self._sessions.get(queue_id)
333        queue_changed = queue_session is not None and self._retain_session_outputs(
334            queue_session,
335            queue_id,
336            player_ids,
337        )
338        source_session = self._source_sessions.get(queue_id)
339        source_changed = source_session is not None and self._retain_session_outputs(
340            source_session,
341            queue_id,
342            player_ids,
343        )
344        current_changed = False
345        if queue_changed:
346            assert queue_session is not None
347            current_changed = self._publish_all(queue_id, queue_session)
348            if current_changed:
349                self.mass.player_queues.signal_update(queue_id)
350        if source_changed:
351            assert source_session is not None
352            self._publish_source(queue_id, source_session)
353        return current_changed
354
355    def clear_source(
356        self,
357        player_id: str,
358        session_id: str | None = None,
359        *,
360        preserve_details: bool = False,
361    ) -> None:
362        """
363        Clear processing details for a live AudioSource.
364
365        :param player_id: Player that owns the source selection.
366        :param session_id: Only clear when this playback session is still active.
367        :param preserve_details: Keep the last published snapshot until its replacement arrives.
368        """
369        processing = self._source_sessions.get(player_id)
370        if processing is None or (session_id is not None and processing.session_id != session_id):
371            return
372        del self._source_sessions[player_id]
373        if preserve_details:
374            return
375        source_session = self.mass.players.get_audio_source_session(player_id)
376        if (
377            source_session is None
378            or source_session.active_source_audio is None
379            or (session_id is not None and source_session.playback_session_id != session_id)
380        ):
381            return
382        source_session.active_source_audio = None
383        self.mass.players.trigger_player_update(player_id)
384
385    def update_player_dsp_preset(self, player_id: str, preset_id: str | None) -> None:
386        """
387        Update preset identity where the effective DSP config remains unchanged.
388
389        :param player_id: Player whose persisted DSP config changed.
390        :param preset_id: Selected preset identifier, or None when cleared.
391        """
392        for queue_id, queue_session in tuple(self._sessions.items()):
393            if self._update_session_dsp_preset(queue_session, player_id, preset_id) and (
394                self._publish_all(queue_id, queue_session)
395            ):
396                self.mass.player_queues.signal_update(queue_id)
397        for source_player_id, source_session in tuple(self._source_sessions.items()):
398            if self._update_session_dsp_preset(source_session, player_id, preset_id):
399                self._publish_source(source_player_id, source_session)
400
401    def clear(self, queue_id: str, session_id: str | None = None) -> None:
402        """
403        Clear processing details for a queue.
404
405        :param queue_id: Queue identifier.
406        :param session_id: Only clear when this playback session is still active.
407        """
408        session = self._sessions.get(queue_id)
409        if session is None or (session_id is not None and session.session_id != session_id):
410            return
411        del self._sessions[queue_id]
412        if self._clear_streamdetails(queue_id):
413            self.mass.player_queues.signal_update(queue_id)
414
415    def prune(self, queue_id: str) -> None:
416        """
417        Drop processing state for completed queue items.
418
419        :param queue_id: Queue identifier.
420        """
421        if session := self._sessions.get(queue_id):
422            self._prune_played_items(queue_id, session)
423
424    def _get_session(self, queue_id: str, session_id: str) -> _AudioProcessingSession | None:
425        """Return a session only when the producer still owns the queue."""
426        queue_data = self.mass.player_queues.queue_data_or_none(queue_id)
427        if queue_data is None or queue_data.session_id != session_id:
428            return None
429        session = self._sessions.get(queue_id)
430        if session is None or session.session_id != session_id:
431            return None
432        return session
433
434    def _get_source_session(
435        self,
436        player_id: str,
437        session_id: str,
438        *,
439        create: bool,
440    ) -> _AudioSourceProcessingSession | None:
441        """Return processing state only when the producer owns the source selection."""
442        source_session = self.mass.players.get_audio_source_session(player_id)
443        if source_session is None or source_session.playback_session_id != session_id:
444            return None
445        processing = self._source_sessions.get(player_id)
446        if processing is not None and processing.session_id == session_id:
447            return processing
448        if not create:
449            return None
450        processing = _AudioSourceProcessingSession(session_id=session_id)
451        self._source_sessions[player_id] = processing
452        return processing
453
454    @staticmethod
455    def _retain_session_outputs(
456        session: _AudioProcessingSession | _AudioSourceProcessingSession,
457        owner_id: str,
458        player_ids: set[str],
459    ) -> bool:
460        """Reconcile one processing session with its current destinations."""
461        changed = False
462        for queue_item_id, outputs in list(session.outputs.items()):
463            retained = {
464                player_id: output
465                for player_id, output in outputs.items()
466                if player_id in player_ids
467            }
468            if template := session.shared_output_templates.get(queue_item_id):
469                added_player_ids = player_ids - retained.keys()
470                if owner_id not in outputs:
471                    added_player_ids.discard(owner_id)
472                for player_id in sorted(added_player_ids):
473                    retained[player_id] = deepcopy(template)
474                    retained[player_id].details.player_ids = [player_id]
475            if retained == outputs:
476                continue
477            changed = True
478            if retained:
479                session.outputs[queue_item_id] = retained
480            else:
481                del session.outputs[queue_item_id]
482                session.shared_output_templates.pop(queue_item_id, None)
483        return changed
484
485    @staticmethod
486    def _update_session_dsp_preset(
487        session: _AudioProcessingSession | _AudioSourceProcessingSession,
488        player_id: str,
489        preset_id: str | None,
490    ) -> bool:
491        """Update a player's preset identity in cached output details."""
492        changed = False
493        for outputs in session.outputs.values():
494            for entry in outputs.values():
495                if entry.dsp_config_id == player_id and entry.details.dsp.preset_id != preset_id:
496                    entry.details.dsp.preset_id = preset_id
497                    changed = True
498        for entry in session.shared_output_templates.values():
499            if entry.dsp_config_id == player_id:
500                entry.details.dsp.preset_id = preset_id
501        return changed
502
503    def _publish_all(self, queue_id: str, session: _AudioProcessingSession) -> bool:
504        """Attach complete chains for every prepared item."""
505        queue = self.mass.player_queues.get(queue_id)
506        self._prune_played_items(queue_id, session)
507        current_item_id = queue.current_item.queue_item_id if queue and queue.current_item else None
508        current_changed = False
509        for queue_item_id in tuple(session.items):
510            if self._publish_item(queue_id, queue_item_id, session, signal_update=False):
511                current_changed |= queue_item_id == current_item_id
512        return current_changed
513
514    def _publish_source(
515        self,
516        player_id: str,
517        processing: _AudioSourceProcessingSession,
518    ) -> bool:
519        """Attach live source audio details to the active source session."""
520        source_session = self.mass.players.get_audio_source_session(player_id)
521        if (
522            source_session is None
523            or source_session.playback_session_id != processing.session_id
524            or source_session.streamdetails is None
525            or not processing.context_ready
526        ):
527            return False
528        streamdetails = source_session.streamdetails
529        details = ActiveSourceAudioDetails(
530            input_format=deepcopy(streamdetails.audio_format),
531            input_fidelity=AudioFidelity(quality=get_audio_quality(streamdetails.audio_format)),
532            crossfade_mode=processing.crossfade_mode,
533            volume_normalization_mode=processing.volume_normalization_mode,
534            outputs=self._group_outputs(
535                streamdetails,
536                self._get_outputs(processing, None),
537                item=_source_processing_item(processing),
538            ),
539        )
540        if source_session.active_source_audio == details:
541            return False
542        source_session.active_source_audio = details
543        self.mass.players.trigger_player_update(player_id)
544        return True
545
546    def _publish_item(
547        self,
548        queue_id: str,
549        queue_item_id: str,
550        session: _AudioProcessingSession,
551        *,
552        signal_update: bool = True,
553    ) -> bool:
554        """Attach one complete chain to its StreamDetails."""
555        queue_item = self.mass.player_queues.get_item(queue_id, queue_item_id)
556        if queue_item is None or queue_item.streamdetails is None:
557            return False
558        item = session.items.get(queue_item_id)
559        output_entries = self._get_outputs(session, queue_item_id)
560        chain = None
561        if item and item.queue_processing and output_entries:
562            chain = AudioProcessingChain(
563                input_fidelity=AudioFidelity(
564                    quality=get_audio_quality(queue_item.streamdetails.audio_format)
565                ),
566                queue_processing=deepcopy(item.queue_processing),
567                outputs=self._group_outputs(
568                    queue_item.streamdetails,
569                    output_entries,
570                    item=item,
571                ),
572            )
573        previous = queue_item.streamdetails.audio_processing
574        if previous == chain:
575            return False
576        queue_item.streamdetails.audio_processing = chain
577        queue = self.mass.player_queues.get(queue_id)
578        if (
579            signal_update
580            and queue
581            and queue.current_item
582            and queue.current_item.queue_item_id == queue_item_id
583        ):
584            self.mass.player_queues.signal_update(queue_id)
585        return True
586
587    def _group_outputs(
588        self,
589        streamdetails: StreamDetails,
590        player_outputs: dict[str, _AudioOutputEntry],
591        item: _AudioProcessingItem,
592    ) -> list[AudioOutputDetails]:
593        """Group players with identical effective output processing."""
594        grouped: list[AudioOutputDetails] = []
595        for player_id, entry in sorted(player_outputs.items()):
596            output = deepcopy(entry.details)
597            output.player_ids = [player_id]
598            output.fidelity = _get_output_fidelity(streamdetails, item, entry)
599            for existing in grouped:
600                if _output_details_equal_ignoring_players(existing, output):
601                    existing.player_ids.append(player_id)
602                    break
603            else:
604                grouped.append(output)
605        return grouped
606
607    @staticmethod
608    def _get_outputs(
609        session: _AudioProcessingSession | _AudioSourceProcessingSession,
610        queue_item_id: str | None,
611    ) -> dict[str, _AudioOutputEntry]:
612        """Return shared outputs overlaid with queue-item-specific outputs."""
613        outputs = dict(session.outputs.get(None, {}))
614        if queue_item_id is not None:
615            outputs.update(session.outputs.get(queue_item_id, {}))
616        return outputs
617
618    def _prune_played_items(
619        self,
620        queue_id: str,
621        session: _AudioProcessingSession,
622    ) -> None:
623        """Drop processing state for items before the current queue index."""
624        queue_data = self.mass.player_queues.queue_data_or_none(queue_id)
625        if queue_data is None or queue_data.queue.current_index is None:
626            return
627        for queue_item in queue_data.items[: queue_data.queue.current_index]:
628            session.items.pop(queue_item.queue_item_id, None)
629            session.outputs.pop(queue_item.queue_item_id, None)
630            session.shared_output_templates.pop(queue_item.queue_item_id, None)
631            if queue_item.streamdetails:
632                queue_item.streamdetails.audio_processing = None
633
634    def _is_played_item(self, queue_id: str, queue_item_id: str) -> bool:
635        """Return whether an item precedes the queue's current index."""
636        queue_data = self.mass.player_queues.queue_data_or_none(queue_id)
637        if queue_data is None or queue_data.queue.current_index is None:
638            return False
639        return any(
640            item.queue_item_id == queue_item_id
641            for item in queue_data.items[: queue_data.queue.current_index]
642        )
643
644    def _clear_streamdetails(self, queue_id: str) -> bool:
645        """Clear attached chains and return whether the current item changed."""
646        queue_data = self.mass.player_queues.queue_data_or_none(queue_id)
647        if queue_data is None:
648            return False
649        current_item = queue_data.queue.current_item
650        current_changed = bool(
651            current_item
652            and current_item.streamdetails
653            and current_item.streamdetails.audio_processing
654        )
655        for queue_item in queue_data.items:
656            if queue_item.streamdetails:
657                queue_item.streamdetails.audio_processing = None
658        return current_changed
659
660
661def get_audio_quality(audio_format: AudioFormat | None) -> AudioQuality:
662    """
663    Classify an audio format using server-owned codec semantics.
664
665    :param audio_format: Audio format to classify.
666    """
667    if audio_format is None:
668        return AudioQuality.UNKNOWN
669    content_type = (
670        audio_format.codec_type
671        if audio_format.codec_type != ContentType.UNKNOWN
672        else audio_format.content_type
673    )
674    if content_type == ContentType.UNKNOWN:
675        return AudioQuality.UNKNOWN
676    if content_type.is_lossless():
677        if audio_format.bit_depth > 16 or audio_format.sample_rate > 48000:
678            return AudioQuality.HI_RES
679        return AudioQuality.LOSSLESS
680    if not audio_format.bit_rate:
681        return AudioQuality.UNKNOWN
682    return AudioQuality.STANDARD if get_bit_rate(audio_format) >= 256 else AudioQuality.LOW
683
684
685def get_media_session_id(media: PlayerMedia) -> str | None:
686    """
687    Return the queue session carried by player media.
688
689    :param media: Player media that started the stream.
690    """
691    return media.queue_session_id
692
693
694def get_normalization_details(
695    streamdetails: StreamDetails,
696    applied_gain_db: float | None,
697) -> AudioNormalizationDetails | None:
698    """
699    Return the effective normalization applied to a queue item.
700
701    :param streamdetails: Effective stream details for a queue item.
702    :param applied_gain_db: Static gain applied by the selected mode.
703    """
704    mode = streamdetails.volume_normalization_mode
705    if mode in (None, VolumeNormalizationMode.DISABLED, VolumeNormalizationMode.UNKNOWN):
706        return None
707    assert mode is not None
708    if mode == VolumeNormalizationMode.SOURCE:
709        # the source set the level without saying to what, and a measurement of our
710        # own would describe audio it already levelled, so only the mode is known
711        return AudioNormalizationDetails(mode=mode)
712    measurement_source = AudioNormalizationMeasurementSource.UNKNOWN
713    measured_lufs: float | None = None
714    if mode == VolumeNormalizationMode.DYNAMIC:
715        measurement_source = AudioNormalizationMeasurementSource.LIVE
716    elif mode == VolumeNormalizationMode.FIXED_GAIN:
717        measurement_source = AudioNormalizationMeasurementSource.FALLBACK
718    elif streamdetails.prefer_album_loudness and streamdetails.loudness_album is not None:
719        measurement_source = AudioNormalizationMeasurementSource.ALBUM
720        measured_lufs = streamdetails.loudness_album
721    elif streamdetails.loudness is not None:
722        measurement_source = AudioNormalizationMeasurementSource.TRACK
723        measured_lufs = streamdetails.loudness
724    else:
725        measurement_source = AudioNormalizationMeasurementSource.FALLBACK
726    return AudioNormalizationDetails(
727        mode=mode,
728        measurement_source=measurement_source,
729        target_lufs=streamdetails.target_loudness,
730        measured_lufs=measured_lufs,
731        applied_gain_db=applied_gain_db,
732    )
733
734
735def _source_processing_item(
736    processing: _AudioSourceProcessingSession,
737) -> _AudioProcessingItem:
738    """
739    Describe a live source's audio path in the shared processing shape.
740
741    :param processing: Runtime processing state of the live source selection.
742    """
743    # Music Assistant mixes nothing into a live source: whatever the source
744    # reported applying reaches us already mixed, and a step it did not report
745    # counts as none. Both are recorded as what they are rather than flattened,
746    # so the shared check stays the one deciding what a source-applied step costs.
747    return _AudioProcessingItem(
748        queue_processing=AudioQueueProcessing(
749            pcm_format=processing.pcm_format,
750            normalization=(
751                AudioNormalizationDetails(mode=VolumeNormalizationMode.SOURCE)
752                if processing.volume_normalization_mode == VolumeNormalizationMode.SOURCE
753                else None
754            ),
755            crossfade_mode=(
756                CrossfadeMode.SOURCE
757                if processing.crossfade_mode == CrossfadeMode.SOURCE
758                else CrossfadeMode.DISABLED
759            ),
760        ),
761    )
762
763
764def _get_output_fidelity(
765    streamdetails: StreamDetails,
766    item: _AudioProcessingItem,
767    output: _AudioOutputEntry,
768) -> AudioFidelity:
769    """Return effective quality and bit-perfect state for an output."""
770    return AudioFidelity(
771        quality=_get_effective_quality(
772            streamdetails.audio_format,
773            output.details.output_format,
774        ),
775        bit_perfect=_is_bit_perfect(streamdetails, item, output),
776    )
777
778
779def _get_effective_quality(
780    input_format: AudioFormat,
781    output_format: AudioFormat | None,
782) -> AudioQuality:
783    """Return the effective quality after the known output format."""
784    input_quality = get_audio_quality(input_format)
785    output_quality = get_audio_quality(output_format)
786    if AudioQuality.UNKNOWN in (input_quality, output_quality):
787        return AudioQuality.UNKNOWN
788    return min((input_quality, output_quality), key=_QUALITY_RANK.__getitem__)
789
790
791def _is_bit_perfect(
792    streamdetails: StreamDetails,
793    item: _AudioProcessingItem,
794    output: _AudioOutputEntry,
795) -> bool | None:
796    """Return whether an output preserves the decoded source samples."""
797    source_format = streamdetails.audio_format
798    queue_processing = item.queue_processing
799    output_format = output.details.output_format
800    if output_format is None or queue_processing is None or queue_processing.pcm_format is None:
801        return None
802    if item.alters_audio:
803        return False
804    if get_audio_quality(output_format) not in (AudioQuality.LOSSLESS, AudioQuality.HI_RES):
805        return False
806    formats = [source_format]
807    if streamdetails.decoded_audio_format:
808        formats.append(streamdetails.decoded_audio_format)
809    formats.extend(
810        (
811            item.input_format or queue_processing.pcm_format,
812            queue_processing.pcm_format,
813            output.input_format,
814            output.handoff_format or output_format,
815            output_format,
816        )
817    )
818    reference = source_format
819    # a wider container carries the source samples untouched — F32 processing
820    # headroom, or a provider that decoded upstream and hands over PCM wider than
821    # the tier it advertises. Only a stage that narrows below the source loses bits.
822    if any(
823        audio_format.sample_rate != reference.sample_rate
824        or audio_format.bit_depth < reference.bit_depth
825        or audio_format.channels != reference.channels
826        for audio_format in formats
827    ):
828        return False
829    # a step the source performed is reported for context but leaves our path untouched
830    if (
831        (
832            queue_processing.normalization is not None
833            and queue_processing.normalization.mode != VolumeNormalizationMode.SOURCE
834        )
835        or queue_processing.playback_speed != 1.0
836        or queue_processing.crossfade_mode not in (CrossfadeMode.DISABLED, CrossfadeMode.SOURCE)
837        or queue_processing.overlay_active
838    ):
839        return False
840    details = output.details
841    # an enabled DSP with no effective filters and no gain leaves samples untouched
842    dsp_alters_audio = details.dsp.state == DSPState.ENABLED and (
843        bool(details.dsp.filters) or details.dsp.input_gain != 0 or details.dsp.output_gain != 0
844    )
845    return not (dsp_alters_audio or details.source_channel is not None)
846
847
848def _output_details_equal_ignoring_players(
849    left: AudioOutputDetails,
850    right: AudioOutputDetails,
851) -> bool:
852    """Compare output details without their destination players."""
853    return replace(left, player_ids=[]) == replace(right, player_ids=[])
854