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