music-assistant-server

1"""Runtime audio processing details for queue 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    AudioFidelity,
12    AudioNormalizationDetails,
13    AudioNormalizationMeasurementSource,
14    AudioOutputDetails,
15    AudioProcessingChain,
16    AudioQuality,
17    AudioQueueProcessing,
18)
19from music_assistant_models.dsp import DSPState
20from music_assistant_models.enums import ContentType, CrossfadeMode, VolumeNormalizationMode
21
22from music_assistant.helpers.audio import get_bit_rate
23
24if TYPE_CHECKING:
25    from music_assistant_models.media_items import AudioFormat
26    from music_assistant_models.streamdetails import StreamDetails
27
28    from music_assistant.helpers.dsp import ComplexFilter
29    from music_assistant.mass import MusicAssistant
30    from music_assistant.models.player import PlayerMedia
31
32
33_QUALITY_RANK = {
34    AudioQuality.UNKNOWN: 0,
35    AudioQuality.LOW: 1,
36    AudioQuality.STANDARD: 2,
37    AudioQuality.LOSSLESS: 3,
38    AudioQuality.HI_RES: 4,
39}
40
41
42@dataclass(slots=True)
43class AudioOutputPlan:
44    """Executable filters and matching client-facing output details."""
45
46    filter_params: list[str | ComplexFilter]
47    output_details: AudioOutputDetails
48    input_format: AudioFormat
49    handoff_format: AudioFormat | None = None
50    dsp_config_id: str | None = None
51
52
53@dataclass(slots=True)
54class _AudioProcessingItem:
55    """Processing details cached for one queue item."""
56
57    queue_processing: AudioQueueProcessing | None = None
58    input_format: AudioFormat | None = None
59    alters_audio: bool = False
60
61
62@dataclass(slots=True)
63class _AudioOutputEntry:
64    """Client-facing output details with private intermediate formats."""
65
66    details: AudioOutputDetails
67    input_format: AudioFormat
68    handoff_format: AudioFormat | None = None
69    dsp_config_id: str | None = None
70
71
72@dataclass(slots=True)
73class _AudioProcessingSession:
74    """Runtime processing state for one queue playback session."""
75
76    session_id: str
77    items: dict[str, _AudioProcessingItem] = field(default_factory=dict)
78    outputs: dict[str | None, dict[str, _AudioOutputEntry]] = field(default_factory=dict)
79    shared_output_templates: dict[str | None, _AudioOutputEntry] = field(default_factory=dict)
80
81
82class AudioProcessingManager:
83    """Build and attach effective audio processing chains to stream details."""
84
85    def __init__(self, mass: MusicAssistant) -> None:
86        """
87        Initialize the audio processing manager.
88
89        :param mass: Music Assistant instance.
90        """
91        self.mass = mass
92        self._sessions: dict[str, _AudioProcessingSession] = {}
93
94    def start_session(self, queue_id: str, session_id: str) -> None:
95        """
96        Start tracking a queue playback session.
97
98        :param queue_id: Queue identifier.
99        :param session_id: Internal queue playback session identifier.
100        """
101        current = self._sessions.get(queue_id)
102        if current and current.session_id == session_id:
103            return
104        if self._clear_streamdetails(queue_id):
105            self.mass.player_queues.signal_update(queue_id)
106        self._sessions[queue_id] = _AudioProcessingSession(session_id=session_id)
107
108    def update_item_context(
109        self,
110        queue_id: str,
111        session_id: str,
112        queue_item_id: str,
113        queue_processing: AudioQueueProcessing,
114        *,
115        alters_audio: bool = False,
116    ) -> None:
117        """
118        Store shared processing selected for a queue item.
119
120        :param queue_id: Queue identifier.
121        :param session_id: Internal queue playback session identifier.
122        :param queue_item_id: Queue item identifier.
123        :param queue_processing: Effective shared processing.
124        :param alters_audio: Whether an intentionally hidden transform alters samples.
125        """
126        session = self._get_session(queue_id, session_id)
127        if session is None:
128            return
129        self._prune_played_items(queue_id, session)
130        if self._is_played_item(queue_id, queue_item_id):
131            return
132        item = session.items.setdefault(queue_item_id, _AudioProcessingItem())
133        previous = item.queue_processing
134        item.queue_processing = AudioQueueProcessing(
135            pcm_format=deepcopy(queue_processing.pcm_format),
136            normalization=deepcopy(
137                previous.normalization if previous else queue_processing.normalization
138            ),
139            playback_speed=queue_processing.playback_speed,
140            crossfade_mode=queue_processing.crossfade_mode,
141            overlay_active=queue_processing.overlay_active,
142        )
143        item.alters_audio = alters_audio
144        self._publish_item(queue_id, queue_item_id, session)
145
146    def update_item_runtime(
147        self,
148        queue_id: str,
149        session_id: str,
150        queue_item_id: str,
151        input_format: AudioFormat,
152        pcm_format: AudioFormat,
153        normalization: AudioNormalizationDetails | None,
154        playback_speed: float,
155        *,
156        alters_audio: bool = False,
157    ) -> None:
158        """
159        Store shared processing constructed for a queue item.
160
161        :param queue_id: Queue identifier.
162        :param session_id: Internal queue playback session identifier.
163        :param queue_item_id: Queue item identifier.
164        :param input_format: PCM format entering shared queue processing.
165        :param pcm_format: Internal PCM format leaving shared processing.
166        :param normalization: Effective normalization details.
167        :param playback_speed: Effective playback-speed conversion.
168        :param alters_audio: Whether an intentionally hidden transform alters samples.
169        """
170        session = self._get_session(queue_id, session_id)
171        if session is None:
172            return
173        self._prune_played_items(queue_id, session)
174        if self._is_played_item(queue_id, queue_item_id):
175            return
176        item = session.items.setdefault(queue_item_id, _AudioProcessingItem())
177        previous = item.queue_processing or AudioQueueProcessing()
178        item.input_format = deepcopy(input_format)
179        item.queue_processing = AudioQueueProcessing(
180            pcm_format=deepcopy(pcm_format),
181            normalization=deepcopy(normalization),
182            playback_speed=playback_speed,
183            crossfade_mode=previous.crossfade_mode,
184            overlay_active=previous.overlay_active,
185        )
186        item.alters_audio = item.alters_audio or alters_audio
187        self._publish_item(queue_id, queue_item_id, session)
188
189    def update_output(
190        self,
191        player_id: str,
192        output_plan: AudioOutputPlan,
193        *,
194        shared_player_ids: Iterable[str] | None = None,
195        queue_id: str,
196        session_id: str,
197        queue_item_id: str | None = None,
198    ) -> bool:
199        """
200        Store an effective player output.
201
202        :param player_id: Destination player identifier.
203        :param output_plan: Effective output processing and private intermediate formats.
204        :param shared_player_ids: Additional players receiving this identical output path.
205            An empty iterable marks a path that can gain shared destinations later.
206        :param queue_id: Queue identifier that owns the output.
207        :param session_id: Queue session identifier that owns the output.
208        :param queue_item_id: Queue item for single-item output, or None for flow output.
209        :return: Whether the effective output changed.
210        """
211        session = self._get_session(queue_id, session_id)
212        if session is None:
213            return False
214        self._prune_played_items(queue_id, session)
215        if queue_item_id is not None and self._is_played_item(queue_id, queue_item_id):
216            return False
217
218        destination_player_ids = {player_id}
219        if shared_player_ids is not None:
220            destination_player_ids.update(shared_player_ids)
221
222        entries: dict[str, _AudioOutputEntry] = {}
223        for destination_player_id in destination_player_ids:
224            entry = _AudioOutputEntry(
225                details=deepcopy(output_plan.output_details),
226                input_format=deepcopy(output_plan.input_format),
227                handoff_format=deepcopy(output_plan.handoff_format),
228                dsp_config_id=output_plan.dsp_config_id,
229            )
230            entry.details.player_ids = [destination_player_id]
231            entries[destination_player_id] = entry
232
233        if shared_player_ids is None:
234            session.shared_output_templates.pop(queue_item_id, None)
235        else:
236            session.shared_output_templates[queue_item_id] = deepcopy(entries[player_id])
237
238        item_outputs = session.outputs.setdefault(queue_item_id, {})
239        changed_entries = {
240            destination_player_id: entry
241            for destination_player_id, entry in entries.items()
242            if item_outputs.get(destination_player_id) != entry
243        }
244        if not changed_entries:
245            return False
246        item_outputs.update(changed_entries)
247        current_changed = self._publish_all(queue_id, session)
248        queue = self.mass.player_queues.get(queue_id)
249        if current_changed or (
250            queue
251            and queue.current_item
252            and queue_item_id in (None, queue.current_item.queue_item_id)
253        ):
254            self.mass.player_queues.signal_update(queue_id)
255        return True
256
257    def retain_outputs(self, queue_id: str, player_ids: set[str]) -> bool:
258        """
259        Reconcile outputs with players attached to a queue.
260
261        :param queue_id: Queue identifier.
262        :param player_ids: Player identifiers that belong to the output.
263        :return: Whether reconciliation published an updated current chain.
264        """
265        session = self._sessions.get(queue_id)
266        if session is None:
267            return False
268        changed = False
269        for queue_item_id, outputs in list(session.outputs.items()):
270            retained = {
271                player_id: output
272                for player_id, output in outputs.items()
273                if player_id in player_ids
274            }
275            if template := session.shared_output_templates.get(queue_item_id):
276                added_player_ids = player_ids - retained.keys()
277                if queue_id not in outputs:
278                    added_player_ids.discard(queue_id)
279                for player_id in sorted(added_player_ids):
280                    retained[player_id] = deepcopy(template)
281                    retained[player_id].details.player_ids = [player_id]
282            if retained == outputs:
283                continue
284            changed = True
285            if retained:
286                session.outputs[queue_item_id] = retained
287            else:
288                del session.outputs[queue_item_id]
289                session.shared_output_templates.pop(queue_item_id, None)
290        if not changed:
291            return False
292        current_changed = self._publish_all(queue_id, session)
293        if current_changed:
294            self.mass.player_queues.signal_update(queue_id)
295        return current_changed
296
297    def update_player_dsp_preset(self, player_id: str, preset_id: str | None) -> None:
298        """
299        Update preset identity where the effective DSP config remains unchanged.
300
301        :param player_id: Player whose persisted DSP config changed.
302        :param preset_id: Selected preset identifier, or None when cleared.
303        """
304        for queue_id, session in tuple(self._sessions.items()):
305            changed = False
306            for outputs in session.outputs.values():
307                for entry in outputs.values():
308                    if (
309                        entry.dsp_config_id == player_id
310                        and entry.details.dsp.preset_id != preset_id
311                    ):
312                        entry.details.dsp.preset_id = preset_id
313                        changed = True
314            for entry in session.shared_output_templates.values():
315                if entry.dsp_config_id == player_id:
316                    entry.details.dsp.preset_id = preset_id
317            if changed and self._publish_all(queue_id, session):
318                self.mass.player_queues.signal_update(queue_id)
319
320    def clear(self, queue_id: str, session_id: str | None = None) -> None:
321        """
322        Clear processing details for a queue.
323
324        :param queue_id: Queue identifier.
325        :param session_id: Only clear when this playback session is still active.
326        """
327        session = self._sessions.get(queue_id)
328        if session is None or (session_id is not None and session.session_id != session_id):
329            return
330        del self._sessions[queue_id]
331        if self._clear_streamdetails(queue_id):
332            self.mass.player_queues.signal_update(queue_id)
333
334    def prune(self, queue_id: str) -> None:
335        """
336        Drop processing state for completed queue items.
337
338        :param queue_id: Queue identifier.
339        """
340        if session := self._sessions.get(queue_id):
341            self._prune_played_items(queue_id, session)
342
343    def _get_session(self, queue_id: str, session_id: str) -> _AudioProcessingSession | None:
344        """Return a session only when the producer still owns the queue."""
345        queue_data = self.mass.player_queues.queue_data_or_none(queue_id)
346        if queue_data is None or queue_data.session_id != session_id:
347            return None
348        session = self._sessions.get(queue_id)
349        if session is None or session.session_id != session_id:
350            return None
351        return session
352
353    def _publish_all(self, queue_id: str, session: _AudioProcessingSession) -> bool:
354        """Attach complete chains for every prepared item."""
355        queue = self.mass.player_queues.get(queue_id)
356        self._prune_played_items(queue_id, session)
357        current_item_id = queue.current_item.queue_item_id if queue and queue.current_item else None
358        current_changed = False
359        for queue_item_id in tuple(session.items):
360            if self._publish_item(queue_id, queue_item_id, session, signal_update=False):
361                current_changed |= queue_item_id == current_item_id
362        return current_changed
363
364    def _publish_item(
365        self,
366        queue_id: str,
367        queue_item_id: str,
368        session: _AudioProcessingSession,
369        *,
370        signal_update: bool = True,
371    ) -> bool:
372        """Attach one complete chain to its StreamDetails."""
373        queue_item = self.mass.player_queues.get_item(queue_id, queue_item_id)
374        if queue_item is None or queue_item.streamdetails is None:
375            return False
376        item = session.items.get(queue_item_id)
377        output_entries = self._get_outputs(session, queue_item_id)
378        chain = None
379        if item and item.queue_processing and output_entries:
380            chain = AudioProcessingChain(
381                input_fidelity=AudioFidelity(
382                    quality=get_audio_quality(queue_item.streamdetails.audio_format)
383                ),
384                queue_processing=deepcopy(item.queue_processing),
385                outputs=self._group_outputs(
386                    queue_item.streamdetails,
387                    item,
388                    output_entries,
389                ),
390            )
391        previous = queue_item.streamdetails.audio_processing
392        if previous == chain:
393            return False
394        queue_item.streamdetails.audio_processing = chain
395        queue = self.mass.player_queues.get(queue_id)
396        if (
397            signal_update
398            and queue
399            and queue.current_item
400            and queue.current_item.queue_item_id == queue_item_id
401        ):
402            self.mass.player_queues.signal_update(queue_id)
403        return True
404
405    def _group_outputs(
406        self,
407        streamdetails: StreamDetails,
408        item: _AudioProcessingItem,
409        player_outputs: dict[str, _AudioOutputEntry],
410    ) -> list[AudioOutputDetails]:
411        """Group players with identical effective output processing."""
412        grouped: list[AudioOutputDetails] = []
413        for player_id, entry in sorted(player_outputs.items()):
414            output = deepcopy(entry.details)
415            output.player_ids = [player_id]
416            output.fidelity = _get_output_fidelity(streamdetails, item, entry)
417            for existing in grouped:
418                if _output_details_equal_ignoring_players(existing, output):
419                    existing.player_ids.append(player_id)
420                    break
421            else:
422                grouped.append(output)
423        return grouped
424
425    @staticmethod
426    def _get_outputs(
427        session: _AudioProcessingSession,
428        queue_item_id: str | None,
429    ) -> dict[str, _AudioOutputEntry]:
430        """Return shared outputs overlaid with queue-item-specific outputs."""
431        outputs = dict(session.outputs.get(None, {}))
432        if queue_item_id is not None:
433            outputs.update(session.outputs.get(queue_item_id, {}))
434        return outputs
435
436    def _prune_played_items(
437        self,
438        queue_id: str,
439        session: _AudioProcessingSession,
440    ) -> None:
441        """Drop processing state for items before the current queue index."""
442        queue_data = self.mass.player_queues.queue_data_or_none(queue_id)
443        if queue_data is None or queue_data.queue.current_index is None:
444            return
445        for queue_item in queue_data.items[: queue_data.queue.current_index]:
446            session.items.pop(queue_item.queue_item_id, None)
447            session.outputs.pop(queue_item.queue_item_id, None)
448            session.shared_output_templates.pop(queue_item.queue_item_id, None)
449            if queue_item.streamdetails:
450                queue_item.streamdetails.audio_processing = None
451
452    def _is_played_item(self, queue_id: str, queue_item_id: str) -> bool:
453        """Return whether an item precedes the queue's current index."""
454        queue_data = self.mass.player_queues.queue_data_or_none(queue_id)
455        if queue_data is None or queue_data.queue.current_index is None:
456            return False
457        return any(
458            item.queue_item_id == queue_item_id
459            for item in queue_data.items[: queue_data.queue.current_index]
460        )
461
462    def _clear_streamdetails(self, queue_id: str) -> bool:
463        """Clear attached chains and return whether the current item changed."""
464        queue_data = self.mass.player_queues.queue_data_or_none(queue_id)
465        if queue_data is None:
466            return False
467        current_item = queue_data.queue.current_item
468        current_changed = bool(
469            current_item
470            and current_item.streamdetails
471            and current_item.streamdetails.audio_processing
472        )
473        for queue_item in queue_data.items:
474            if queue_item.streamdetails:
475                queue_item.streamdetails.audio_processing = None
476        return current_changed
477
478
479def get_audio_quality(audio_format: AudioFormat | None) -> AudioQuality:
480    """
481    Classify an audio format using server-owned codec semantics.
482
483    :param audio_format: Audio format to classify.
484    """
485    if audio_format is None:
486        return AudioQuality.UNKNOWN
487    content_type = (
488        audio_format.codec_type
489        if audio_format.codec_type != ContentType.UNKNOWN
490        else audio_format.content_type
491    )
492    if content_type == ContentType.UNKNOWN:
493        return AudioQuality.UNKNOWN
494    if content_type.is_lossless():
495        if audio_format.bit_depth > 16 or audio_format.sample_rate > 48000:
496            return AudioQuality.HI_RES
497        return AudioQuality.LOSSLESS
498    if not audio_format.bit_rate:
499        return AudioQuality.UNKNOWN
500    return AudioQuality.STANDARD if get_bit_rate(audio_format) >= 256 else AudioQuality.LOW
501
502
503def get_media_session_id(media: PlayerMedia) -> str | None:
504    """
505    Return the queue session carried by player media.
506
507    :param media: Player media that started the stream.
508    """
509    return media.queue_session_id
510
511
512def get_normalization_details(
513    streamdetails: StreamDetails,
514    applied_gain_db: float | None,
515) -> AudioNormalizationDetails | None:
516    """
517    Return the effective normalization applied to a queue item.
518
519    :param streamdetails: Effective stream details for a queue item.
520    :param applied_gain_db: Static gain applied by the selected mode.
521    """
522    mode = streamdetails.volume_normalization_mode
523    if mode in (None, VolumeNormalizationMode.DISABLED, VolumeNormalizationMode.UNKNOWN):
524        return None
525    assert mode is not None
526    if mode == VolumeNormalizationMode.SOURCE:
527        # the source set the level without saying to what, and a measurement of our
528        # own would describe audio it already levelled, so only the mode is known
529        return AudioNormalizationDetails(mode=mode)
530    measurement_source = AudioNormalizationMeasurementSource.UNKNOWN
531    measured_lufs: float | None = None
532    if mode == VolumeNormalizationMode.DYNAMIC:
533        measurement_source = AudioNormalizationMeasurementSource.LIVE
534    elif mode == VolumeNormalizationMode.FIXED_GAIN:
535        measurement_source = AudioNormalizationMeasurementSource.FALLBACK
536    elif streamdetails.prefer_album_loudness and streamdetails.loudness_album is not None:
537        measurement_source = AudioNormalizationMeasurementSource.ALBUM
538        measured_lufs = streamdetails.loudness_album
539    elif streamdetails.loudness is not None:
540        measurement_source = AudioNormalizationMeasurementSource.TRACK
541        measured_lufs = streamdetails.loudness
542    else:
543        measurement_source = AudioNormalizationMeasurementSource.FALLBACK
544    return AudioNormalizationDetails(
545        mode=mode,
546        measurement_source=measurement_source,
547        target_lufs=streamdetails.target_loudness,
548        measured_lufs=measured_lufs,
549        applied_gain_db=applied_gain_db,
550    )
551
552
553def _get_output_fidelity(
554    streamdetails: StreamDetails,
555    item: _AudioProcessingItem,
556    output: _AudioOutputEntry,
557) -> AudioFidelity:
558    """Return effective quality and bit-perfect state for an output."""
559    input_quality = get_audio_quality(streamdetails.audio_format)
560    output_quality = get_audio_quality(output.details.output_format)
561    if AudioQuality.UNKNOWN in (input_quality, output_quality):
562        quality = AudioQuality.UNKNOWN
563    else:
564        quality = min((input_quality, output_quality), key=_QUALITY_RANK.__getitem__)
565    return AudioFidelity(
566        quality=quality,
567        bit_perfect=_is_bit_perfect(streamdetails, item, output),
568    )
569
570
571def _is_bit_perfect(
572    streamdetails: StreamDetails,
573    item: _AudioProcessingItem,
574    output: _AudioOutputEntry,
575) -> bool | None:
576    """Return whether an output preserves the decoded source samples."""
577    source_format = streamdetails.audio_format
578    queue_processing = item.queue_processing
579    output_format = output.details.output_format
580    if output_format is None or queue_processing is None or queue_processing.pcm_format is None:
581        return None
582    if item.alters_audio:
583        return False
584    if get_audio_quality(output_format) not in (AudioQuality.LOSSLESS, AudioQuality.HI_RES):
585        return False
586    formats = [source_format]
587    if streamdetails.decoded_audio_format:
588        formats.append(streamdetails.decoded_audio_format)
589    formats.extend(
590        (
591            item.input_format or queue_processing.pcm_format,
592            queue_processing.pcm_format,
593            output.input_format,
594            output.handoff_format or output_format,
595            output_format,
596        )
597    )
598    reference = source_format
599    # a wider container carries the source samples untouched — F32 processing
600    # headroom, or a provider that decoded upstream and hands over PCM wider than
601    # the tier it advertises. Only a stage that narrows below the source loses bits.
602    if any(
603        audio_format.sample_rate != reference.sample_rate
604        or audio_format.bit_depth < reference.bit_depth
605        or audio_format.channels != reference.channels
606        for audio_format in formats
607    ):
608        return False
609    # a step the source performed is reported for context but leaves our path untouched
610    if (
611        (
612            queue_processing.normalization is not None
613            and queue_processing.normalization.mode != VolumeNormalizationMode.SOURCE
614        )
615        or queue_processing.playback_speed != 1.0
616        or queue_processing.crossfade_mode not in (CrossfadeMode.DISABLED, CrossfadeMode.SOURCE)
617        or queue_processing.overlay_active
618    ):
619        return False
620    details = output.details
621    # an enabled DSP with no effective filters and no gain leaves samples untouched
622    dsp_alters_audio = details.dsp.state == DSPState.ENABLED and (
623        bool(details.dsp.filters) or details.dsp.input_gain != 0 or details.dsp.output_gain != 0
624    )
625    return not (dsp_alters_audio or details.source_channel is not None)
626
627
628def _output_details_equal_ignoring_players(
629    left: AudioOutputDetails,
630    right: AudioOutputDetails,
631) -> bool:
632    """Compare output details without their destination players."""
633    return replace(left, player_ids=[]) == replace(right, player_ids=[])
634