/
/
/
1"""Smart Fades - Audio fade implementations."""
2
3from __future__ import annotations
4
5import asyncio
6import logging
7import os
8from abc import ABC, abstractmethod
9from collections.abc import AsyncGenerator
10from contextlib import suppress
11from typing import TYPE_CHECKING
12
13from music_assistant.constants import VERBOSE_LOG_LEVEL
14from music_assistant.controllers.streams.smart_fades.filters import (
15 Filter,
16 StreamingCrossfadeFilter,
17)
18from music_assistant.controllers.streams.smart_fades.helpers import SMART_CROSSFADE_DURATION
19from music_assistant.controllers.streams.smart_fades.models import (
20 CrossfadeTimingInfo,
21 SmartFadeNotApplicable,
22 TransitionPlan,
23)
24from music_assistant.controllers.streams.smart_fades.planner import SmartCrossFadePlanner
25from music_assistant.controllers.streams.smart_fades.renderer import TransitionRenderer
26from music_assistant.helpers.audio import iter_pcm_slices
27from music_assistant.helpers.ffmpeg import get_ffmpeg_channel_args
28from music_assistant.helpers.process import AsyncProcess
29
30if TYPE_CHECKING:
31 from music_assistant_models.media_items import AudioFormat
32
33 from music_assistant.models.audio_analysis import AudioAnalysisData
34
35__all__ = [
36 "CrossfadeTimingInfo",
37 "SmartCrossFade",
38 "SmartFade",
39 "SmartFadeNotApplicable",
40 "StandardCrossFade",
41]
42
43
44def _close_if_open(*fds: int) -> None:
45 """Close the given fds, ignoring ones already handed off (marked -1)."""
46 for fd in fds:
47 if fd != -1:
48 os.close(fd)
49
50
51def _feed_pipe_blocking(write_fd: int, payload: bytes) -> None:
52 """
53 Write a payload into a pipe fd with plain blocking writes, then close it.
54
55 Blocking on purpose (run in a thread): the pipe applies the backpressure,
56 and a consumer that went away surfaces as a broken pipe, which simply ends
57 the feed â the consumer's own exit status tells the story.
58
59 :param write_fd: Write end of the pipe; closed when done, whatever happens.
60 :param payload: The bytes to deliver.
61 """
62 try:
63 view = memoryview(payload)
64 while view:
65 written = os.write(write_fd, view[: 1024 * 1024])
66 view = view[written:]
67 except BrokenPipeError:
68 # ffmpeg stopped reading (its filter took all it needed, or it exited)
69 pass
70 finally:
71 os.close(write_fd)
72
73
74async def _feed_ffmpeg_stdin(
75 proc: AsyncProcess, fade_in_part: bytes | AsyncGenerator[bytes]
76) -> None:
77 """
78 Write the incoming track's head to the mixer, always ending with an EOF.
79
80 :param proc: The mixer process to feed.
81 :param fade_in_part: Raw PCM bytes, or a stream delivering them.
82 """
83 try:
84 if isinstance(fade_in_part, bytes):
85 await proc.write(fade_in_part)
86 else:
87 async for fade_chunk in fade_in_part:
88 await proc.write(fade_chunk)
89 finally:
90 # a feed that stops without an EOF leaves ffmpeg waiting for input
91 # while its consumer waits for output
92 await proc.write_eof()
93
94
95class SmartFade(ABC):
96 """Abstract base class for Smart Fades."""
97
98 filters: list[Filter]
99 timing_info: CrossfadeTimingInfo
100
101 def __init__(self, logger: logging.Logger) -> None:
102 """Initialize SmartFade base class."""
103 self.filters = []
104 self.logger = logger
105
106 @abstractmethod
107 def build(
108 self,
109 fade_out_bytes_len: int,
110 fade_in_bytes_len: int,
111 pcm_format: AudioFormat,
112 ) -> None:
113 """
114 Build the filter chain and assign ``self.timing_info``.
115
116 Must be called once before ``apply()``.
117
118 :param fade_out_bytes_len: Length in bytes of the outgoing track's tail buffer.
119 :param fade_in_bytes_len: Length in bytes of the incoming track's head buffer.
120 :param pcm_format: Audio format of both input buffers.
121 """
122 ...
123
124 async def apply(
125 self,
126 fade_out_part: bytes,
127 fade_in_part: bytes | AsyncGenerator[bytes],
128 pcm_format: AudioFormat,
129 ) -> AsyncGenerator[bytes]:
130 """
131 Apply the smart fade, yielding PCM audio chunks as they become available.
132
133 :param fade_out_part: Raw PCM bytes for the outgoing track's tail.
134 :param fade_in_part: Raw PCM bytes or async generator for the incoming track's head.
135 :param pcm_format: Audio format of both input parts and the output.
136 """
137 # The fade-out side goes in through its own pipe: ffmpeg takes any number
138 # of pipe:<fd> inputs, so no temp file has to touch the disk. The pipe far
139 # exceeds the kernel buffer, so it is fed alongside the stdin feeder below.
140 fadeout_read_fd, fadeout_write_fd = os.pipe()
141
142 self.logger.debug(
143 "Applying smartfade: %s",
144 self,
145 )
146 args = self._mix_ffmpeg_args(pcm_format, fadeout_read_fd)
147 self.logger.log(VERBOSE_LOG_LEVEL, "FFmpeg command args: %s", " ".join(args))
148
149 got_output = False
150 stderr_lines: list[str] = []
151 try:
152 proc = AsyncProcess(
153 args,
154 stdin=True,
155 stdout=True,
156 stderr=True,
157 name="smartfade",
158 pass_fds=(fadeout_read_fd,),
159 )
160 async with proc:
161 # the child holds its own copy of the read end now
162 os.close(fadeout_read_fd)
163 fadeout_read_fd = -1
164
165 async def _drain_stderr() -> None:
166 """Read stderr to prevent pipe deadlock."""
167 async for line in proc.iter_stderr():
168 stderr_lines.append(line)
169
170 fadeout_task = asyncio.create_task(
171 asyncio.to_thread(_feed_pipe_blocking, fadeout_write_fd, fade_out_part)
172 )
173 fadeout_write_fd = -1 # the feeder owns and closes it now
174 feed_task = asyncio.create_task(_feed_ffmpeg_stdin(proc, fade_in_part))
175 stderr_task = asyncio.create_task(_drain_stderr())
176 try:
177 async for chunk in proc.iter_any():
178 got_output = True
179 yield chunk
180 finally:
181 if not feed_task.done():
182 feed_task.cancel()
183 with suppress(asyncio.CancelledError):
184 await feed_task
185 # Bounded wait: on consumer abort a paused-but-alive ffmpeg can
186 # leave the writer blocked on a full pipe until proc.close()
187 # (in __aexit__, after this finally) breaks it â the orphaned
188 # thread then ends on its own, a completed write closed already.
189 with suppress(TimeoutError, asyncio.CancelledError):
190 await asyncio.wait_for(fadeout_task, timeout=2)
191 # Bounded wait on stderr_task so its output is still captured
192 # for error reporting on the happy/error paths, but we don't
193 # hang on consumer abort â ffmpeg is still alive then and
194 # stderr won't EOF until proc.close() closes stdin, which
195 # only runs via the async-with __aexit__ *after* this finally.
196 # wait_for cancels stderr_task on timeout so cleanup proceeds.
197 with suppress(TimeoutError, asyncio.CancelledError):
198 await asyncio.wait_for(stderr_task, timeout=2)
199
200 if proc.returncode != 0:
201 stderr_msg = "; ".join(stderr_lines) if stderr_lines else "(no stderr)"
202 raise RuntimeError(f"Crossfade FFmpeg failed (rc={proc.returncode}): {stderr_msg}")
203 if not got_output:
204 msg = "Crossfade FFmpeg produced no output"
205 if stderr_lines:
206 msg += f": {'; '.join(stderr_lines)}"
207 raise RuntimeError(msg)
208 finally:
209 # close whichever pipe ends this coroutine still owns (spawn failures)
210 _close_if_open(fadeout_read_fd, fadeout_write_fd)
211
212 def __repr__(self) -> str:
213 """Return string representation of SmartFade showing the filter chain."""
214 if not self.filters:
215 return f"<{self.__class__.__name__}: 0 filters>"
216
217 chain = " â ".join(repr(f) for f in self.filters)
218 return f"<{self.__class__.__name__}: {len(self.filters)} filters> {chain}"
219
220 def _mix_ffmpeg_args(self, pcm_format: AudioFormat, fadeout_read_fd: int) -> list[str]:
221 """
222 Build the mix's ffmpeg argv: fade-out on its own pipe, fade-in on stdin.
223
224 Both inputs are fully specified raw PCM, so the demuxer's probe buffer is
225 disabled â it would otherwise swallow seconds of a streamed fade-in
226 before the filter graph produces its first frame.
227
228 :param pcm_format: Audio format of both inputs and the output.
229 :param fadeout_read_fd: Read end of the fade-out pipe (passed to the child).
230 """
231 input_format = [
232 "-probesize",
233 "32",
234 "-analyzeduration",
235 "0",
236 "-acodec",
237 pcm_format.content_type.name.lower(), # e.g., "pcm_f32le" not just "f32le"
238 *get_ffmpeg_channel_args(pcm_format),
239 "-ar",
240 str(pcm_format.sample_rate),
241 "-f",
242 pcm_format.content_type.value,
243 ]
244 return [
245 "ffmpeg",
246 "-hide_banner",
247 "-loglevel",
248 "error",
249 *input_format,
250 "-i",
251 f"pipe:{fadeout_read_fd}",
252 *input_format,
253 "-i",
254 "-",
255 "-filter_complex",
256 ";".join(self._get_ffmpeg_filters()),
257 # output format matches the input codec format
258 "-acodec",
259 pcm_format.content_type.name.lower(),
260 *get_ffmpeg_channel_args(pcm_format),
261 "-ar",
262 str(pcm_format.sample_rate),
263 "-f",
264 pcm_format.content_type.value,
265 "-",
266 ]
267
268 def _get_ffmpeg_filters(
269 self,
270 input_fadein_label: str = "[1]",
271 input_fadeout_label: str = "[0]",
272 ) -> list[str]:
273 """Get FFmpeg filters for smart fades."""
274 if not self.filters:
275 raise RuntimeError("SmartFade not built â call Mixer.build() first")
276 filters = []
277 _cur_fadein_label = input_fadein_label
278 _cur_fadeout_label = input_fadeout_label
279 for audio_filter in self.filters:
280 filter_strings = audio_filter.apply(_cur_fadein_label, _cur_fadeout_label)
281 filters.extend(filter_strings)
282 _cur_fadein_label = f"[{audio_filter.output_fadein_label}]"
283 _cur_fadeout_label = f"[{audio_filter.output_fadeout_label}]"
284 return filters
285
286
287class SmartCrossFade(SmartFade):
288 """
289 Smart fades class that implements a Smart Fade mode.
290
291 Delegates the decision-making to a ``SmartCrossFadePlanner`` (pure, over the
292 stored analysis) and the filter/timing construction to a ``TransitionRenderer``.
293 Alternative transition strategies are siblings that swap in their own planner.
294 """
295
296 def __init__(
297 self,
298 logger: logging.Logger,
299 fade_out_analysis: AudioAnalysisData,
300 fade_in_analysis: AudioAnalysisData,
301 ) -> None:
302 """
303 Initialize SmartCrossFade with analysis data.
304
305 :param logger: Logger for debug output.
306 :param fade_out_analysis: Analysis data for the outgoing track.
307 :param fade_in_analysis: Analysis data for the incoming track.
308 """
309 super().__init__(logger)
310 self.fade_out_analysis = fade_out_analysis
311 self.fade_in_analysis = fade_in_analysis
312 self.planner = SmartCrossFadePlanner(logger)
313 self.renderer = TransitionRenderer(logger)
314 self.plan: TransitionPlan | None = None
315 # populated by build(); read by the timing/lyrics-sync tests
316 self.effective_end: float = float(SMART_CROSSFADE_DURATION)
317 self.tempo_steps: list[tuple[float, float]] = []
318
319 def build(
320 self,
321 fade_out_bytes_len: int,
322 fade_in_bytes_len: int,
323 pcm_format: AudioFormat,
324 ) -> None:
325 """Plan the transition, then render its filter chain and ``timing_info``."""
326 buffer_duration = min(
327 float(SMART_CROSSFADE_DURATION),
328 fade_out_bytes_len / pcm_format.pcm_sample_size,
329 )
330 self.plan = self.planner.plan(
331 self.fade_out_analysis, self.fade_in_analysis, buffer_duration
332 )
333 self.filters, self.timing_info = self.renderer.render(
334 self.plan, pcm_format, fade_in_bytes_len
335 )
336 # convenience copies for the timing/lyrics-sync tests
337 self.effective_end = self.plan.fade_out_window
338 self.tempo_steps = self.plan.tempo_plan.steps
339 self.fade_out_beats = self.planner.outgoing.beats
340
341
342class StandardCrossFade(SmartFade):
343 """Standard crossfade class that implements a standard crossfade mode."""
344
345 def __init__(
346 self,
347 logger: logging.Logger,
348 crossfade_duration: float = 10.0,
349 trailing_silence_bytes: int = 0,
350 ) -> None:
351 """
352 Initialize StandardCrossFade.
353
354 :param logger: Logger for debug output.
355 :param crossfade_duration: Length of the crossfade overlap in seconds.
356 :param trailing_silence_bytes: Trailing silence in the outgoing tail that
357 ``apply()`` slices off before crossfading.
358 """
359 super().__init__(logger)
360 self.crossfade_duration = crossfade_duration
361 self.trailing_silence_bytes = trailing_silence_bytes
362 self.crossfade_size: int = 0
363
364 def build(
365 self,
366 fade_out_bytes_len: int,
367 fade_in_bytes_len: int,
368 pcm_format: AudioFormat,
369 ) -> None:
370 """Build the standard crossfade filter chain and assign ``self.timing_info``."""
371 fade_out_seconds = fade_out_bytes_len / pcm_format.pcm_sample_size
372 fade_in_seconds = fade_in_bytes_len / pcm_format.pcm_sample_size
373 # clamp CF to fit shorter inputs (defensive â normally full buffers)
374 effective_cf = min(self.crossfade_duration, fade_out_seconds, fade_in_seconds)
375 # Quantize the overlap to a whole number of PCM frames and drive both the
376 # byte slice (in apply) and the acrossfade length from this one integer.
377 # apply slices the buffers on frame boundaries, so a fractional effective_cf
378 # leaves the rendered buffer a fraction of a sample short of the acrossfade
379 # duration â and acrossfade then silently produces no output at all.
380 frame_size = (pcm_format.bit_depth // 8) * pcm_format.channels
381 crossfade_bytes = int(pcm_format.pcm_sample_size * effective_cf)
382 self.crossfade_size = crossfade_bytes // frame_size * frame_size
383 crossfade_samples = self.crossfade_size // frame_size
384 effective_cf = self.crossfade_size / pcm_format.pcm_sample_size
385 self.timing_info = CrossfadeTimingInfo(
386 pre_crossfade_duration=max(0.0, fade_out_seconds - effective_cf),
387 crossfade_duration=effective_cf,
388 fadein_trimmed_duration=0.0,
389 post_crossfade_duration=max(0.0, fade_in_seconds - effective_cf),
390 )
391 # the streaming variant, so a fade-in that is still arriving (realtime
392 # source) is blended and delivered as it comes in
393 self.filters = [
394 StreamingCrossfadeFilter(logger=self.logger, crossfade_samples=crossfade_samples),
395 ]
396
397 async def apply(
398 self,
399 fade_out_part: bytes,
400 fade_in_part: bytes | AsyncGenerator[bytes],
401 pcm_format: AudioFormat,
402 ) -> AsyncGenerator[bytes]:
403 """
404 Apply standard crossfade, yielding PCM audio chunks.
405
406 Only the overlapping portions are crossfaded, not the full buffers.
407 """
408 # crossfade_size legitimately ends up 0 for a silent/tiny buffer, so guard on
409 # the filter chain (set in build) to still fail fast on apply-before-build,
410 # consistent with SmartFade._get_ffmpeg_filters()
411 if not self.filters:
412 raise RuntimeError("SmartFade not built â call Mixer.build() first")
413 if self.trailing_silence_bytes:
414 fade_out_part = fade_out_part[: len(fade_out_part) - self.trailing_silence_bytes]
415 # frame-aligned overlap computed once in build, so it exactly matches the
416 # acrossfade `ns=` length the filter was built with
417 crossfade_size = self.crossfade_size
418 if crossfade_size == 0:
419 # nothing to blend â concatenate without spawning ffmpeg
420 for pcm_slice in iter_pcm_slices(fade_out_part, pcm_format, 1000):
421 yield pcm_slice
422 if isinstance(fade_in_part, bytes):
423 for pcm_slice in iter_pcm_slices(fade_in_part, pcm_format, 1000):
424 yield pcm_slice
425 else:
426 async for chunk in fade_in_part:
427 for pcm_slice in iter_pcm_slices(chunk, pcm_format, 1000):
428 yield pcm_slice
429 return
430 # Pre-crossfade: outgoing track minus the crossfaded portion. Emitted
431 # before the incoming side is touched at all: with a streamed fade-in the
432 # overlap is still arriving, and the player keeps playing this meanwhile.
433 split = len(fade_out_part) - crossfade_size
434 pre_crossfade = fade_out_part[:split]
435 adjusted_fade_out_part = fade_out_part[split:]
436 for pcm_slice in iter_pcm_slices(pre_crossfade, pcm_format, 1000):
437 yield pcm_slice
438
439 if isinstance(fade_in_part, bytes):
440 async for chunk in super().apply(
441 adjusted_fade_out_part, fade_in_part[:crossfade_size], pcm_format
442 ):
443 yield chunk
444 for pcm_slice in iter_pcm_slices(fade_in_part[crossfade_size:], pcm_format, 1000):
445 yield pcm_slice
446 return
447
448 # Generator fade-in: hand exactly the overlap to the (streaming) blend as
449 # it arrives; whatever the last chunk carried beyond it opens the post part
450 overshoot = bytearray()
451
452 async def _overlap_stream() -> AsyncGenerator[bytes]:
453 taken = 0
454 async for chunk in fade_in_part:
455 remaining = crossfade_size - taken
456 if len(chunk) >= remaining:
457 taken += remaining
458 overshoot.extend(chunk[remaining:])
459 yield chunk[:remaining]
460 return
461 taken += len(chunk)
462 yield chunk
463
464 async for chunk in super().apply(adjusted_fade_out_part, _overlap_stream(), pcm_format):
465 yield chunk
466 if overshoot:
467 for pcm_slice in iter_pcm_slices(bytes(overshoot), pcm_format, 1000):
468 yield pcm_slice
469 async for remaining_chunk in fade_in_part:
470 for pcm_slice in iter_pcm_slices(remaining_chunk, pcm_format, 1000):
471 yield pcm_slice
472