/
/
/
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 that owns the output, or the player holding a live source.
255 :param session_id: Queue session or source playback session that owns the output.
256 :param queue_item_id: Queue item for single-item output, or None for flow output
257 and for a live source.
258 :return: Whether the effective output changed.
259 """
260 source_session = (
261 self._get_source_session(queue_id, session_id, create=True)
262 if queue_item_id is None
263 else None
264 )
265 queue_session: _AudioProcessingSession | None = None
266 if source_session is not None:
267 session: _AudioProcessingSession | _AudioSourceProcessingSession = source_session
268 else:
269 queue_session = self._get_session(queue_id, session_id)
270 if queue_session is None:
271 return False
272 self._prune_played_items(queue_id, queue_session)
273 if queue_item_id is not None and self._is_played_item(queue_id, queue_item_id):
274 return False
275 session = queue_session
276
277 destination_player_ids = {player_id}
278 if shared_player_ids is not None:
279 destination_player_ids.update(shared_player_ids)
280
281 entries: dict[str, _AudioOutputEntry] = {}
282 for destination_player_id in destination_player_ids:
283 entry = _AudioOutputEntry(
284 details=deepcopy(output_plan.output_details),
285 input_format=deepcopy(output_plan.input_format),
286 handoff_format=deepcopy(output_plan.handoff_format),
287 dsp_config_id=output_plan.dsp_config_id,
288 )
289 entry.details.player_ids = [destination_player_id]
290 entries[destination_player_id] = entry
291
292 if shared_player_ids is None:
293 session.shared_output_templates.pop(queue_item_id, None)
294 else:
295 session.shared_output_templates[queue_item_id] = deepcopy(entries[player_id])
296
297 item_outputs = session.outputs.setdefault(queue_item_id, {})
298 changed_entries = {
299 destination_player_id: entry
300 for destination_player_id, entry in entries.items()
301 if item_outputs.get(destination_player_id) != entry
302 }
303 if not changed_entries:
304 return False
305 item_outputs.update(changed_entries)
306 if source_session is not None:
307 self._publish_source(queue_id, source_session)
308 return True
309 assert queue_session is not None
310 current_changed = self._publish_all(queue_id, queue_session)
311 queue = self.mass.player_queues.get(queue_id)
312 if current_changed or (
313 queue
314 and queue.current_item
315 and queue_item_id in (None, queue.current_item.queue_item_id)
316 ):
317 self.mass.player_queues.signal_update(queue_id)
318 return True
319
320 def retain_outputs(self, queue_id: str, player_ids: set[str]) -> bool:
321 """
322 Reconcile processing outputs with the current playback destinations.
323
324 :param queue_id: Queue identifier.
325 :param player_ids: Player identifiers that belong to the output.
326 :return: Whether reconciliation published an updated current chain.
327 """
328 queue_session = self._sessions.get(queue_id)
329 queue_changed = queue_session is not None and self._retain_session_outputs(
330 queue_session,
331 queue_id,
332 player_ids,
333 )
334 source_session = self._source_sessions.get(queue_id)
335 source_changed = source_session is not None and self._retain_session_outputs(
336 source_session,
337 queue_id,
338 player_ids,
339 )
340 current_changed = False
341 if queue_changed:
342 assert queue_session is not None
343 current_changed = self._publish_all(queue_id, queue_session)
344 if current_changed:
345 self.mass.player_queues.signal_update(queue_id)
346 if source_changed:
347 assert source_session is not None
348 self._publish_source(queue_id, source_session)
349 return current_changed
350
351 def clear_source(
352 self,
353 player_id: str,
354 session_id: str | None = None,
355 *,
356 preserve_details: bool = False,
357 ) -> None:
358 """
359 Clear processing details for a live AudioSource.
360
361 :param player_id: Player that owns the source selection.
362 :param session_id: Only clear when this playback session is still active.
363 :param preserve_details: Keep the last published snapshot until its replacement arrives.
364 """
365 processing = self._source_sessions.get(player_id)
366 if processing is None or (session_id is not None and processing.session_id != session_id):
367 return
368 del self._source_sessions[player_id]
369 if preserve_details:
370 return
371 source_session = self.mass.players.get_audio_source_session(player_id)
372 if (
373 source_session is None
374 or source_session.active_source_audio is None
375 or (session_id is not None and source_session.playback_session_id != session_id)
376 ):
377 return
378 source_session.active_source_audio = None
379 self.mass.players.trigger_player_update(player_id)
380
381 def update_player_dsp_preset(self, player_id: str, preset_id: str | None) -> None:
382 """
383 Update preset identity where the effective DSP config remains unchanged.
384
385 :param player_id: Player whose persisted DSP config changed.
386 :param preset_id: Selected preset identifier, or None when cleared.
387 """
388 for queue_id, queue_session in tuple(self._sessions.items()):
389 if self._update_session_dsp_preset(queue_session, player_id, preset_id) and (
390 self._publish_all(queue_id, queue_session)
391 ):
392 self.mass.player_queues.signal_update(queue_id)
393 for source_player_id, source_session in tuple(self._source_sessions.items()):
394 if self._update_session_dsp_preset(source_session, player_id, preset_id):
395 self._publish_source(source_player_id, source_session)
396
397 def clear(self, queue_id: str, session_id: str | None = None) -> None:
398 """
399 Clear processing details for a queue.
400
401 :param queue_id: Queue identifier.
402 :param session_id: Only clear when this playback session is still active.
403 """
404 session = self._sessions.get(queue_id)
405 if session is None or (session_id is not None and session.session_id != session_id):
406 return
407 del self._sessions[queue_id]
408 if self._clear_streamdetails(queue_id):
409 self.mass.player_queues.signal_update(queue_id)
410
411 def prune(self, queue_id: str) -> None:
412 """
413 Drop processing state for completed queue items.
414
415 :param queue_id: Queue identifier.
416 """
417 if session := self._sessions.get(queue_id):
418 self._prune_played_items(queue_id, session)
419
420 def _get_session(self, queue_id: str, session_id: str) -> _AudioProcessingSession | None:
421 """Return a session only when the producer still owns the queue."""
422 queue_data = self.mass.player_queues.queue_data_or_none(queue_id)
423 if queue_data is None or queue_data.session_id != session_id:
424 return None
425 session = self._sessions.get(queue_id)
426 if session is None or session.session_id != session_id:
427 return None
428 return session
429
430 def _get_source_session(
431 self,
432 player_id: str,
433 session_id: str,
434 *,
435 create: bool,
436 ) -> _AudioSourceProcessingSession | None:
437 """Return processing state only when the producer owns the source selection."""
438 source_session = self.mass.players.get_audio_source_session(player_id)
439 if source_session is None or source_session.playback_session_id != session_id:
440 return None
441 processing = self._source_sessions.get(player_id)
442 if processing is not None and processing.session_id == session_id:
443 return processing
444 if not create:
445 return None
446 processing = _AudioSourceProcessingSession(session_id=session_id)
447 self._source_sessions[player_id] = processing
448 return processing
449
450 @staticmethod
451 def _retain_session_outputs(
452 session: _AudioProcessingSession | _AudioSourceProcessingSession,
453 owner_id: str,
454 player_ids: set[str],
455 ) -> bool:
456 """Reconcile one processing session with its current destinations."""
457 changed = False
458 for queue_item_id, outputs in list(session.outputs.items()):
459 retained = {
460 player_id: output
461 for player_id, output in outputs.items()
462 if player_id in player_ids
463 }
464 if template := session.shared_output_templates.get(queue_item_id):
465 added_player_ids = player_ids - retained.keys()
466 if owner_id not in outputs:
467 added_player_ids.discard(owner_id)
468 for player_id in sorted(added_player_ids):
469 retained[player_id] = deepcopy(template)
470 retained[player_id].details.player_ids = [player_id]
471 if retained == outputs:
472 continue
473 changed = True
474 if retained:
475 session.outputs[queue_item_id] = retained
476 else:
477 del session.outputs[queue_item_id]
478 session.shared_output_templates.pop(queue_item_id, None)
479 return changed
480
481 @staticmethod
482 def _update_session_dsp_preset(
483 session: _AudioProcessingSession | _AudioSourceProcessingSession,
484 player_id: str,
485 preset_id: str | None,
486 ) -> bool:
487 """Update a player's preset identity in cached output details."""
488 changed = False
489 for outputs in session.outputs.values():
490 for entry in outputs.values():
491 if entry.dsp_config_id == player_id and entry.details.dsp.preset_id != preset_id:
492 entry.details.dsp.preset_id = preset_id
493 changed = True
494 for entry in session.shared_output_templates.values():
495 if entry.dsp_config_id == player_id:
496 entry.details.dsp.preset_id = preset_id
497 return changed
498
499 def _publish_all(self, queue_id: str, session: _AudioProcessingSession) -> bool:
500 """Attach complete chains for every prepared item."""
501 queue = self.mass.player_queues.get(queue_id)
502 self._prune_played_items(queue_id, session)
503 current_item_id = queue.current_item.queue_item_id if queue and queue.current_item else None
504 current_changed = False
505 for queue_item_id in tuple(session.items):
506 if self._publish_item(queue_id, queue_item_id, session, signal_update=False):
507 current_changed |= queue_item_id == current_item_id
508 return current_changed
509
510 def _publish_source(
511 self,
512 player_id: str,
513 processing: _AudioSourceProcessingSession,
514 ) -> bool:
515 """Attach live source audio details to the active source session."""
516 source_session = self.mass.players.get_audio_source_session(player_id)
517 if (
518 source_session is None
519 or source_session.playback_session_id != processing.session_id
520 or source_session.streamdetails is None
521 or not processing.context_ready
522 ):
523 return False
524 streamdetails = source_session.streamdetails
525 details = ActiveSourceAudioDetails(
526 input_format=deepcopy(streamdetails.audio_format),
527 input_fidelity=AudioFidelity(quality=get_audio_quality(streamdetails.audio_format)),
528 crossfade_mode=processing.crossfade_mode,
529 volume_normalization_mode=processing.volume_normalization_mode,
530 outputs=_group_outputs(
531 _player_output_details(
532 streamdetails,
533 _source_processing_item(processing, entry),
534 consumer_player_id,
535 entry,
536 )
537 for consumer_player_id, entry in sorted(self._get_outputs(processing, None).items())
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=_group_outputs(
568 _player_output_details(queue_item.streamdetails, item, player_id, entry)
569 for player_id, entry in sorted(output_entries.items())
570 ),
571 )
572 previous = queue_item.streamdetails.audio_processing
573 if previous == chain:
574 return False
575 queue_item.streamdetails.audio_processing = chain
576 queue = self.mass.player_queues.get(queue_id)
577 if (
578 signal_update
579 and queue
580 and queue.current_item
581 and queue.current_item.queue_item_id == queue_item_id
582 ):
583 self.mass.player_queues.signal_update(queue_id)
584 return True
585
586 @staticmethod
587 def _get_outputs(
588 session: _AudioProcessingSession | _AudioSourceProcessingSession,
589 queue_item_id: str | None,
590 ) -> dict[str, _AudioOutputEntry]:
591 """Return shared outputs overlaid with queue-item-specific outputs."""
592 outputs = dict(session.outputs.get(None, {}))
593 if queue_item_id is not None:
594 outputs.update(session.outputs.get(queue_item_id, {}))
595 return outputs
596
597 def _prune_played_items(
598 self,
599 queue_id: str,
600 session: _AudioProcessingSession,
601 ) -> None:
602 """Drop processing state for items before the current queue index."""
603 queue_data = self.mass.player_queues.queue_data_or_none(queue_id)
604 if queue_data is None or queue_data.queue.current_index is None:
605 return
606 for queue_item in queue_data.items[: queue_data.queue.current_index]:
607 session.items.pop(queue_item.queue_item_id, None)
608 session.outputs.pop(queue_item.queue_item_id, None)
609 session.shared_output_templates.pop(queue_item.queue_item_id, None)
610 if queue_item.streamdetails:
611 queue_item.streamdetails.audio_processing = None
612
613 def _is_played_item(self, queue_id: str, queue_item_id: str) -> bool:
614 """Return whether an item precedes the queue's current index."""
615 queue_data = self.mass.player_queues.queue_data_or_none(queue_id)
616 if queue_data is None or queue_data.queue.current_index is None:
617 return False
618 return any(
619 item.queue_item_id == queue_item_id
620 for item in queue_data.items[: queue_data.queue.current_index]
621 )
622
623 def _clear_streamdetails(self, queue_id: str) -> bool:
624 """Clear attached chains and return whether the current item changed."""
625 queue_data = self.mass.player_queues.queue_data_or_none(queue_id)
626 if queue_data is None:
627 return False
628 current_item = queue_data.queue.current_item
629 current_changed = bool(
630 current_item
631 and current_item.streamdetails
632 and current_item.streamdetails.audio_processing
633 )
634 for queue_item in queue_data.items:
635 if queue_item.streamdetails:
636 queue_item.streamdetails.audio_processing = None
637 return current_changed
638
639
640def get_audio_quality(audio_format: AudioFormat | None) -> AudioQuality:
641 """
642 Classify an audio format using server-owned codec semantics.
643
644 :param audio_format: Audio format to classify.
645 """
646 if audio_format is None:
647 return AudioQuality.UNKNOWN
648 content_type = (
649 audio_format.codec_type
650 if audio_format.codec_type != ContentType.UNKNOWN
651 else audio_format.content_type
652 )
653 if content_type == ContentType.UNKNOWN:
654 return AudioQuality.UNKNOWN
655 if content_type.is_lossless():
656 if audio_format.bit_depth > 16 or audio_format.sample_rate > 48000:
657 return AudioQuality.HI_RES
658 return AudioQuality.LOSSLESS
659 if not audio_format.bit_rate:
660 return AudioQuality.UNKNOWN
661 return AudioQuality.STANDARD if get_bit_rate(audio_format) >= 256 else AudioQuality.LOW
662
663
664def get_media_session_id(media: PlayerMedia) -> str | None:
665 """
666 Return the queue session carried by player media.
667
668 :param media: Player media that started the stream.
669 """
670 return media.queue_session_id
671
672
673def get_normalization_details(
674 streamdetails: StreamDetails,
675 applied_gain_db: float | None,
676) -> AudioNormalizationDetails | None:
677 """
678 Return the effective normalization applied to a queue item.
679
680 :param streamdetails: Effective stream details for a queue item.
681 :param applied_gain_db: Static gain applied by the selected mode.
682 """
683 mode = streamdetails.volume_normalization_mode
684 if mode in (None, VolumeNormalizationMode.DISABLED, VolumeNormalizationMode.UNKNOWN):
685 return None
686 assert mode is not None
687 if mode == VolumeNormalizationMode.SOURCE:
688 # the source set the level without saying to what, and a measurement of our
689 # own would describe audio it already levelled, so only the mode is known
690 return AudioNormalizationDetails(mode=mode)
691 measurement_source = AudioNormalizationMeasurementSource.UNKNOWN
692 measured_lufs: float | None = None
693 if mode == VolumeNormalizationMode.DYNAMIC:
694 measurement_source = AudioNormalizationMeasurementSource.LIVE
695 elif mode == VolumeNormalizationMode.FIXED_GAIN:
696 measurement_source = AudioNormalizationMeasurementSource.FALLBACK
697 elif streamdetails.prefer_album_loudness and streamdetails.loudness_album is not None:
698 measurement_source = AudioNormalizationMeasurementSource.ALBUM
699 measured_lufs = streamdetails.loudness_album
700 elif streamdetails.loudness is not None:
701 measurement_source = AudioNormalizationMeasurementSource.TRACK
702 measured_lufs = streamdetails.loudness
703 else:
704 measurement_source = AudioNormalizationMeasurementSource.FALLBACK
705 return AudioNormalizationDetails(
706 mode=mode,
707 measurement_source=measurement_source,
708 target_lufs=streamdetails.target_loudness,
709 measured_lufs=measured_lufs,
710 applied_gain_db=applied_gain_db,
711 )
712
713
714def _source_processing_item(
715 processing: _AudioSourceProcessingSession,
716 output: _AudioOutputEntry,
717) -> _AudioProcessingItem:
718 """
719 Describe a live source's audio path to one consumer in the shared processing shape.
720
721 :param processing: Runtime processing state of the live source selection.
722 :param output: Output entry of the consumer being described.
723 """
724 # Music Assistant mixes nothing into a live source: whatever the source
725 # reported applying reaches us already mixed, and a step it did not report
726 # counts as none. Both are recorded as what they are rather than flattened,
727 # so the shared check stays the one deciding what a source-applied step costs.
728 return _AudioProcessingItem(
729 queue_processing=AudioQueueProcessing(
730 # every consumer decodes the live source for itself, so the format
731 # entering its output is the one the source was decoded to for it
732 pcm_format=output.input_format,
733 normalization=(
734 AudioNormalizationDetails(mode=VolumeNormalizationMode.SOURCE)
735 if processing.volume_normalization_mode == VolumeNormalizationMode.SOURCE
736 else None
737 ),
738 crossfade_mode=(
739 CrossfadeMode.SOURCE
740 if processing.crossfade_mode == CrossfadeMode.SOURCE
741 else CrossfadeMode.DISABLED
742 ),
743 ),
744 )
745
746
747def _player_output_details(
748 streamdetails: StreamDetails,
749 item: _AudioProcessingItem,
750 player_id: str,
751 output: _AudioOutputEntry,
752) -> AudioOutputDetails:
753 """
754 Describe the effective output of one player.
755
756 :param streamdetails: Stream details of the audio being played.
757 :param item: Processing shape of the audio entering the player's own output.
758 :param player_id: Player receiving this output.
759 :param output: Prepared output of that player.
760 """
761 details = deepcopy(output.details)
762 details.player_ids = [player_id]
763 details.fidelity = _get_output_fidelity(streamdetails, item, output)
764 return details
765
766
767def _group_outputs(outputs: Iterable[AudioOutputDetails]) -> list[AudioOutputDetails]:
768 """
769 Merge players whose effective output processing is identical.
770
771 :param outputs: Effective output of each single player.
772 """
773 grouped: list[AudioOutputDetails] = []
774 for output in outputs:
775 for existing in grouped:
776 if _output_details_equal_ignoring_players(existing, output):
777 existing.player_ids.extend(output.player_ids)
778 break
779 else:
780 grouped.append(output)
781 return grouped
782
783
784def _get_output_fidelity(
785 streamdetails: StreamDetails,
786 item: _AudioProcessingItem,
787 output: _AudioOutputEntry,
788) -> AudioFidelity:
789 """Return effective quality and bit-perfect state for an output."""
790 return AudioFidelity(
791 quality=_get_effective_quality(
792 streamdetails.audio_format,
793 output.details.output_format,
794 ),
795 bit_perfect=_is_bit_perfect(streamdetails, item, output),
796 )
797
798
799def _get_effective_quality(
800 input_format: AudioFormat,
801 output_format: AudioFormat | None,
802) -> AudioQuality:
803 """Return the effective quality after the known output format."""
804 input_quality = get_audio_quality(input_format)
805 output_quality = get_audio_quality(output_format)
806 if AudioQuality.UNKNOWN in (input_quality, output_quality):
807 return AudioQuality.UNKNOWN
808 return min((input_quality, output_quality), key=_QUALITY_RANK.__getitem__)
809
810
811def _is_bit_perfect(
812 streamdetails: StreamDetails,
813 item: _AudioProcessingItem,
814 output: _AudioOutputEntry,
815) -> bool | None:
816 """Return whether an output preserves the decoded source samples."""
817 source_format = streamdetails.audio_format
818 queue_processing = item.queue_processing
819 output_format = output.details.output_format
820 if output_format is None or queue_processing is None or queue_processing.pcm_format is None:
821 return None
822 if item.alters_audio:
823 return False
824 if get_audio_quality(output_format) not in (AudioQuality.LOSSLESS, AudioQuality.HI_RES):
825 return False
826 formats = [source_format]
827 if streamdetails.decoded_audio_format:
828 formats.append(streamdetails.decoded_audio_format)
829 formats.extend(
830 (
831 item.input_format or queue_processing.pcm_format,
832 queue_processing.pcm_format,
833 output.input_format,
834 output.handoff_format or output_format,
835 output_format,
836 )
837 )
838 reference = source_format
839 # a wider container carries the source samples untouched â F32 processing
840 # headroom, or a provider that decoded upstream and hands over PCM wider than
841 # the tier it advertises. Only a stage that narrows below the source loses bits.
842 if any(
843 audio_format.sample_rate != reference.sample_rate
844 or audio_format.bit_depth < reference.bit_depth
845 or audio_format.channels != reference.channels
846 for audio_format in formats
847 ):
848 return False
849 # a step the source performed is reported for context but leaves our path untouched
850 if (
851 (
852 queue_processing.normalization is not None
853 and queue_processing.normalization.mode != VolumeNormalizationMode.SOURCE
854 )
855 or queue_processing.playback_speed != 1.0
856 or queue_processing.crossfade_mode not in (CrossfadeMode.DISABLED, CrossfadeMode.SOURCE)
857 or queue_processing.overlay_active
858 ):
859 return False
860 details = output.details
861 # an enabled DSP with no effective filters and no gain leaves samples untouched
862 dsp_alters_audio = details.dsp.state == DSPState.ENABLED and (
863 bool(details.dsp.filters) or details.dsp.input_gain != 0 or details.dsp.output_gain != 0
864 )
865 return not (dsp_alters_audio or details.source_channel is not None)
866
867
868def _output_details_equal_ignoring_players(
869 left: AudioOutputDetails,
870 right: AudioOutputDetails,
871) -> bool:
872 """Compare output details without their destination players."""
873 return replace(left, player_ids=[]) == replace(right, player_ids=[])
874