/
/
/
1"""Smart Fades Mixer - Mixes audio tracks using smart fades."""
2
3from __future__ import annotations
4
5import math
6from collections.abc import AsyncGenerator
7from typing import TYPE_CHECKING
8
9from music_assistant_models.enums import CrossfadeMode
10
11from music_assistant.controllers.streams.audio_analysis import SMART_FADES_ANALYSIS_DOMAIN
12from music_assistant.controllers.streams.smart_fades.fades import (
13 SmartCrossFade,
14 SmartFade,
15 SmartFadeNotApplicable,
16 StandardCrossFade,
17)
18from music_assistant.controllers.streams.smart_fades.helpers import detect_effective_audio_end
19from music_assistant.controllers.streams.smart_fades.vocal import (
20 PROTECTIVE_VOCAL_CONFIG,
21 VocalMask,
22 build_vocal_windows,
23 parse_vocal_probabilities,
24)
25from music_assistant.helpers.audio import align_audio_to_frame_boundary, strip_silence
26from music_assistant.models.audio_analysis import AudioAnalysisData
27
28if TYPE_CHECKING:
29 from music_assistant_models.media_items import AudioFormat
30 from music_assistant_models.streamdetails import StreamDetails
31
32 from music_assistant.controllers.streams.controller import StreamsController
33
34
35class SmartFadesMixer:
36 """Smart fades mixer class that mixes tracks based on analysis data."""
37
38 def __init__(self, streams: StreamsController) -> None:
39 """Initialize smart fades mixer."""
40 self.streams = streams
41 self.logger = streams.logger.getChild("smart_fades_mixer")
42
43 async def build(
44 self,
45 fade_in_streamdetails: StreamDetails,
46 fade_out_streamdetails: StreamDetails,
47 pcm_format: AudioFormat,
48 standard_crossfade_duration: int,
49 mode: CrossfadeMode,
50 fade_out_data: bytes,
51 fade_in_bytes_len: int,
52 ) -> SmartFade:
53 """
54 Pick the SmartFade implementation, prime its filters, and return it.
55
56 For the standard crossfade path (explicit mode or smart-crossfade fallback)
57 the trailing silence in ``fade_out_data`` is measured so that ``timing_info``
58 reflects the audio that will actually be rendered. The trim itself is deferred
59 to ``apply()``, which executes it as a plain slice â no bytes are modified here.
60
61 :param fade_in_streamdetails: Stream details for the incoming track.
62 :param fade_out_streamdetails: Stream details for the outgoing track.
63 :param pcm_format: Audio format of both input buffers (and mix output).
64 :param standard_crossfade_duration: Duration in seconds for standard crossfade.
65 :param mode: Smart fades mode (SMART_CROSSFADE or STANDARD_CROSSFADE).
66 :param fade_out_data: PCM buffer of the outgoing track's tail.
67 :param fade_in_bytes_len: Expected length in bytes of the fade-in input.
68 """
69 # degradation chain: smart-crossfade â standard; richer modes prepend their builder
70 smart_fade: SmartFade | None = None
71 fade_out_analysis: AudioAnalysisData | None = None
72 if mode == CrossfadeMode.SMART_CROSSFADE:
73 smart_fade, fade_out_analysis = await self._build_smart_crossfade(
74 fade_in_streamdetails=fade_in_streamdetails,
75 fade_out_streamdetails=fade_out_streamdetails,
76 fade_out_bytes_len=len(fade_out_data),
77 fade_in_bytes_len=fade_in_bytes_len,
78 pcm_format=pcm_format,
79 )
80 if smart_fade is None:
81 smart_fade = await self._build_standard_crossfade(
82 fade_out_data=fade_out_data,
83 fade_in_bytes_len=fade_in_bytes_len,
84 pcm_format=pcm_format,
85 standard_crossfade_duration=standard_crossfade_duration,
86 fade_out_analysis=fade_out_analysis,
87 )
88 return smart_fade
89
90 async def mix(
91 self,
92 smart_fade: SmartFade,
93 fade_in_part: bytes | AsyncGenerator[bytes],
94 fade_out_part: bytes,
95 pcm_format: AudioFormat,
96 ) -> AsyncGenerator[bytes]:
97 """Run the already-built SmartFade and yield mixed PCM audio chunks."""
98 async for chunk in smart_fade.apply(fade_out_part, fade_in_part, pcm_format):
99 yield chunk
100
101 async def _build_standard_crossfade(
102 self,
103 fade_out_data: bytes,
104 fade_in_bytes_len: int,
105 pcm_format: AudioFormat,
106 standard_crossfade_duration: int,
107 fade_out_analysis: AudioAnalysisData | None = None,
108 ) -> StandardCrossFade:
109 """
110 Build a StandardCrossFade â the tail of the degradation chain, never fails.
111
112 Measures the trailing silence here so timing_info reflects the audio that
113 will actually be rendered; apply() executes the cut as a plain slice.
114
115 :param fade_out_data: PCM buffer of the outgoing track's tail.
116 :param fade_in_bytes_len: Expected length in bytes of the fade-in input.
117 :param pcm_format: Audio format of both input buffers.
118 :param standard_crossfade_duration: Duration in seconds for standard crossfade.
119 :param fade_out_analysis: Outgoing analysis retained from a failed smart
120 build, or ``None`` for the regular standard path.
121 """
122 trailing_silence_bytes = 0
123 try:
124 stripped = align_audio_to_frame_boundary(
125 await strip_silence(fade_out_data, pcm_format=pcm_format, reverse=True),
126 pcm_format,
127 )
128 retained_bytes = max(
129 len(stripped),
130 self._get_vocal_retention_bytes(
131 fade_out_analysis,
132 len(fade_out_data),
133 pcm_format,
134 ),
135 )
136 trailing_silence_bytes = max(0, len(fade_out_data) - retained_bytes)
137 except Exception as err:
138 # a failed measurement degrades to the old late-boundary bookkeeping
139 # instead of killing the stream
140 self.logger.warning("Measuring trailing silence failed: %s", err)
141 smart_fade = StandardCrossFade(
142 logger=self.logger,
143 crossfade_duration=standard_crossfade_duration,
144 trailing_silence_bytes=trailing_silence_bytes,
145 )
146 smart_fade.build(len(fade_out_data) - trailing_silence_bytes, fade_in_bytes_len, pcm_format)
147 return smart_fade
148
149 async def _build_smart_crossfade(
150 self,
151 fade_in_streamdetails: StreamDetails,
152 fade_out_streamdetails: StreamDetails,
153 fade_out_bytes_len: int,
154 fade_in_bytes_len: int,
155 pcm_format: AudioFormat,
156 ) -> tuple[SmartFade | None, AudioAnalysisData | None]:
157 """
158 Attempt to build a SmartCrossFade and retain outgoing analysis for fallback.
159
160 Returns the built fade (or ``None`` when fallback is needed) together
161 with the outgoing analysis row, when available.
162 """
163 analyses = await self._load_analyses(fade_out_streamdetails, fade_in_streamdetails)
164 fade_out_analysis, fade_in_analysis = analyses
165 if not (
166 fade_out_analysis
167 and fade_in_analysis
168 and fade_out_analysis.bpm
169 and fade_in_analysis.bpm
170 and fade_out_analysis.beats is not None
171 and fade_in_analysis.beats is not None
172 ):
173 return None, fade_out_analysis
174 try:
175 smart_fade = SmartCrossFade(
176 logger=self.logger,
177 fade_out_analysis=fade_out_analysis,
178 fade_in_analysis=fade_in_analysis,
179 )
180 smart_fade.build(fade_out_bytes_len, fade_in_bytes_len, pcm_format)
181 except SmartFadeNotApplicable as e:
182 self.logger.debug("Smart crossfade not applicable: %s - using standard crossfade", e)
183 return None, fade_out_analysis
184 except Exception as e:
185 self.logger.warning(
186 "Smart crossfade build failed: %s, falling back to standard crossfade", e
187 )
188 return None, fade_out_analysis
189 return smart_fade, fade_out_analysis
190
191 async def _load_analyses(
192 self,
193 fade_out_streamdetails: StreamDetails,
194 fade_in_streamdetails: StreamDetails,
195 ) -> tuple[AudioAnalysisData | None, AudioAnalysisData | None]:
196 """
197 Load both tracks' analysis rows for a planned fade.
198
199 Rows are returned independently so a failed smart build can still use a
200 valid outgoing row to protect vocals in the standard fallback.
201 """
202 fade_out_analysis = await self.streams.audio_analysis.get_audio_analysis(
203 fade_out_streamdetails.item_id,
204 fade_out_streamdetails.provider,
205 priority=(SMART_FADES_ANALYSIS_DOMAIN,),
206 )
207 fade_in_analysis = await self.streams.audio_analysis.get_audio_analysis(
208 fade_in_streamdetails.item_id,
209 fade_in_streamdetails.provider,
210 priority=(SMART_FADES_ANALYSIS_DOMAIN,),
211 )
212 return fade_out_analysis, fade_in_analysis
213
214 @staticmethod
215 def _get_vocal_retention_bytes(
216 analysis: AudioAnalysisData | None,
217 fade_out_bytes_len: int,
218 pcm_format: AudioFormat,
219 ) -> int:
220 """
221 Return the frame-aligned outgoing length needed to retain an audible vocal.
222
223 Invalid or stale vocal data, missing/invalid RMS data, and vocals beyond
224 the RMS-audible boundary return zero so standard silence removal keeps its
225 existing behavior.
226
227 :param analysis: Outgoing track analysis retained from the smart path.
228 :param fade_out_bytes_len: Full outgoing PCM buffer length in bytes.
229 :param pcm_format: Audio format of the outgoing buffer.
230 """
231 if analysis is None or analysis.rms_energy is None:
232 return 0
233 rms_energy = analysis.rms_energy
234 if len(rms_energy) < 2 or any(
235 isinstance(value, bool)
236 or not isinstance(value, (int, float))
237 or not math.isfinite(value)
238 or value < 0.0
239 for value in rms_energy
240 ):
241 return 0
242 timeline = parse_vocal_probabilities(analysis)
243 if timeline is None:
244 return 0
245
246 buffer_duration = fade_out_bytes_len / pcm_format.pcm_sample_size
247 track_duration = analysis.duration
248 assert track_duration is not None # guaranteed by the validated timeline
249 buffer_offset = max(0.0, track_duration - buffer_duration)
250 vocal_mask = build_vocal_windows(
251 timeline.probabilities,
252 timeline.frame_duration,
253 buffer_offset,
254 track_duration,
255 beat_duration=60.0 / analysis.bpm if analysis.bpm and analysis.bpm > 0.0 else None,
256 config=PROTECTIVE_VOCAL_CONFIG,
257 )
258 if not vocal_mask.windows:
259 return 0
260
261 audio_end = detect_effective_audio_end(
262 rms_energy,
263 track_duration,
264 buffer_duration,
265 )
266 buffer_local_mask = VocalMask(
267 windows=[
268 (left - buffer_offset, right - buffer_offset) for left, right in vocal_mask.windows
269 ]
270 ).clamped_to(audio_end)
271 retained_seconds = buffer_local_mask.last_end()
272 frame_size = pcm_format.channels * pcm_format.bit_depth // 8
273 retained_frames = math.ceil(retained_seconds * pcm_format.pcm_sample_size / frame_size)
274 return min(fade_out_bytes_len, retained_frames * frame_size)
275