/
/
/
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 self.logger.debug(
148 "Built standard fade: tail=%.2fs silence=%.2fs usable=%.2fs fade_in=%.2fs "
149 "requested=%.2fs applied=%.2fs",
150 len(fade_out_data) / pcm_format.pcm_sample_size,
151 trailing_silence_bytes / pcm_format.pcm_sample_size,
152 (len(fade_out_data) - trailing_silence_bytes) / pcm_format.pcm_sample_size,
153 fade_in_bytes_len / pcm_format.pcm_sample_size,
154 float(standard_crossfade_duration),
155 smart_fade.timing_info.crossfade_duration,
156 )
157 return smart_fade
158
159 async def _build_smart_crossfade(
160 self,
161 fade_in_streamdetails: StreamDetails,
162 fade_out_streamdetails: StreamDetails,
163 fade_out_bytes_len: int,
164 fade_in_bytes_len: int,
165 pcm_format: AudioFormat,
166 ) -> tuple[SmartFade | None, AudioAnalysisData | None]:
167 """
168 Attempt to build a SmartCrossFade and retain outgoing analysis for fallback.
169
170 Returns the built fade (or ``None`` when fallback is needed) together
171 with the outgoing analysis row, when available.
172 """
173 analyses = await self._load_analyses(fade_out_streamdetails, fade_in_streamdetails)
174 fade_out_analysis, fade_in_analysis = analyses
175 if not (
176 fade_out_analysis
177 and fade_in_analysis
178 and fade_out_analysis.bpm
179 and fade_in_analysis.bpm
180 and fade_out_analysis.beats is not None
181 and fade_in_analysis.beats is not None
182 ):
183 self.logger.debug(
184 "Smart fade lacks analysis: out_row=%s out_bpm=%s out_beats=%s | "
185 "in_row=%s in_bpm=%s in_beats=%s",
186 fade_out_analysis is not None,
187 fade_out_analysis.bpm if fade_out_analysis else None,
188 fade_out_analysis.beats is not None if fade_out_analysis else None,
189 fade_in_analysis is not None,
190 fade_in_analysis.bpm if fade_in_analysis else None,
191 fade_in_analysis.beats is not None if fade_in_analysis else None,
192 )
193 return None, fade_out_analysis
194 try:
195 smart_fade = SmartCrossFade(
196 logger=self.logger,
197 fade_out_analysis=fade_out_analysis,
198 fade_in_analysis=fade_in_analysis,
199 )
200 smart_fade.build(fade_out_bytes_len, fade_in_bytes_len, pcm_format)
201 except SmartFadeNotApplicable as e:
202 self.logger.debug("Smart crossfade not applicable: %s - using standard crossfade", e)
203 return None, fade_out_analysis
204 except Exception as e:
205 self.logger.warning(
206 "Smart crossfade build failed: %s, falling back to standard crossfade", e
207 )
208 return None, fade_out_analysis
209 return smart_fade, fade_out_analysis
210
211 async def _load_analyses(
212 self,
213 fade_out_streamdetails: StreamDetails,
214 fade_in_streamdetails: StreamDetails,
215 ) -> tuple[AudioAnalysisData | None, AudioAnalysisData | None]:
216 """
217 Load both tracks' analysis rows for a planned fade.
218
219 Rows are returned independently so a failed smart build can still use a
220 valid outgoing row to protect vocals in the standard fallback.
221 """
222 fade_out_analysis = await self.streams.audio_analysis.get_audio_analysis(
223 fade_out_streamdetails.item_id,
224 fade_out_streamdetails.provider,
225 priority=(SMART_FADES_ANALYSIS_DOMAIN,),
226 )
227 fade_in_analysis = await self.streams.audio_analysis.get_audio_analysis(
228 fade_in_streamdetails.item_id,
229 fade_in_streamdetails.provider,
230 priority=(SMART_FADES_ANALYSIS_DOMAIN,),
231 )
232 return fade_out_analysis, fade_in_analysis
233
234 @staticmethod
235 def _get_vocal_retention_bytes(
236 analysis: AudioAnalysisData | None,
237 fade_out_bytes_len: int,
238 pcm_format: AudioFormat,
239 ) -> int:
240 """
241 Return the frame-aligned outgoing length needed to retain an audible vocal.
242
243 Invalid or stale vocal data, missing/invalid RMS data, and vocals beyond
244 the RMS-audible boundary return zero so standard silence removal keeps its
245 existing behavior.
246
247 :param analysis: Outgoing track analysis retained from the smart path.
248 :param fade_out_bytes_len: Full outgoing PCM buffer length in bytes.
249 :param pcm_format: Audio format of the outgoing buffer.
250 """
251 if analysis is None or analysis.rms_energy is None:
252 return 0
253 rms_energy = analysis.rms_energy
254 if len(rms_energy) < 2 or any(
255 isinstance(value, bool)
256 or not isinstance(value, (int, float))
257 or not math.isfinite(value)
258 or value < 0.0
259 for value in rms_energy
260 ):
261 return 0
262 timeline = parse_vocal_probabilities(analysis)
263 if timeline is None:
264 return 0
265
266 buffer_duration = fade_out_bytes_len / pcm_format.pcm_sample_size
267 track_duration = analysis.duration
268 assert track_duration is not None # guaranteed by the validated timeline
269 buffer_offset = max(0.0, track_duration - buffer_duration)
270 vocal_mask = build_vocal_windows(
271 timeline.probabilities,
272 timeline.frame_duration,
273 buffer_offset,
274 track_duration,
275 beat_duration=60.0 / analysis.bpm if analysis.bpm and analysis.bpm > 0.0 else None,
276 config=PROTECTIVE_VOCAL_CONFIG,
277 )
278 if not vocal_mask.windows:
279 return 0
280
281 audio_end = detect_effective_audio_end(
282 rms_energy,
283 track_duration,
284 buffer_duration,
285 )
286 buffer_local_mask = VocalMask(
287 windows=[
288 (left - buffer_offset, right - buffer_offset) for left, right in vocal_mask.windows
289 ]
290 ).clamped_to(audio_end)
291 retained_seconds = buffer_local_mask.last_end()
292 frame_size = pcm_format.channels * pcm_format.bit_depth // 8
293 retained_frames = math.ceil(retained_seconds * pcm_format.pcm_sample_size / frame_size)
294 return min(fade_out_bytes_len, retained_frames * frame_size)
295