/
/
/
1"""Smart Fades Mixer - Mixes audio tracks using smart fades."""
2
3from __future__ import annotations
4
5from typing import TYPE_CHECKING
6
7from music_assistant.controllers.streams.smart_fades.fades import (
8 SMART_CROSSFADE_DURATION,
9 SmartCrossFade,
10 SmartFade,
11 StandardCrossFade,
12)
13from music_assistant.helpers.audio import (
14 align_audio_to_frame_boundary,
15 strip_silence,
16)
17from music_assistant.models.smart_fades import (
18 SmartFadesAnalysis,
19 SmartFadesAnalysisFragment,
20 SmartFadesMode,
21)
22
23if TYPE_CHECKING:
24 from music_assistant_models.media_items import AudioFormat
25 from music_assistant_models.streamdetails import StreamDetails
26
27 from music_assistant.controllers.streams.streams_controller import StreamsController
28
29
30class SmartFadesMixer:
31 """Smart fades mixer class that mixes tracks based on analysis data."""
32
33 def __init__(self, streams: StreamsController) -> None:
34 """Initialize smart fades mixer."""
35 self.streams = streams
36 self.logger = streams.logger.getChild("smart_fades_mixer")
37
38 async def mix(
39 self,
40 fade_in_part: bytes,
41 fade_out_part: bytes,
42 fade_in_streamdetails: StreamDetails,
43 fade_out_streamdetails: StreamDetails,
44 pcm_format: AudioFormat,
45 standard_crossfade_duration: int = 10,
46 mode: SmartFadesMode = SmartFadesMode.SMART_CROSSFADE,
47 ) -> bytes:
48 """Apply crossfade with internal state management and smart/standard fallback logic."""
49 if mode == SmartFadesMode.DISABLED:
50 # No crossfade, just concatenate
51 # Note that this should not happen since we check this before calling mix()
52 # but just to be sure...
53 return fade_out_part + fade_in_part
54
55 # strip silence from end of audio of fade_out_part
56 fade_out_part = await strip_silence(
57 self.streams.mass,
58 fade_out_part,
59 pcm_format=pcm_format,
60 reverse=True,
61 )
62 # Ensure frame alignment after silence stripping
63 fade_out_part = align_audio_to_frame_boundary(fade_out_part, pcm_format)
64
65 # strip silence from begin of audio of fade_in_part
66 fade_in_part = await strip_silence(
67 self.streams.mass,
68 fade_in_part,
69 pcm_format=pcm_format,
70 reverse=False,
71 )
72 # Ensure frame alignment after silence stripping
73 fade_in_part = align_audio_to_frame_boundary(fade_in_part, pcm_format)
74 fadeout_duration = len(fade_out_part) / pcm_format.pcm_sample_size
75 fadein_duration = len(fade_in_part) / pcm_format.pcm_sample_size
76 fadeout_stripped = SMART_CROSSFADE_DURATION - fadeout_duration
77 fadein_stripped = SMART_CROSSFADE_DURATION - fadein_duration
78 self.logger.debug(
79 "Buffer durations after silence stripping: "
80 "fade_out=%.2fs (%.2fs stripped), fade_in=%.2fs (%.2fs stripped)%s",
81 fadeout_duration,
82 fadeout_stripped,
83 fadein_duration,
84 fadein_stripped,
85 " *** WARNING: fade_in significantly shorter than"
86 f" SMART_CROSSFADE_DURATION ({SMART_CROSSFADE_DURATION}s)!"
87 if fadein_stripped > 2.0
88 else "",
89 )
90 if mode == SmartFadesMode.STANDARD_CROSSFADE:
91 smart_fade: SmartFade = StandardCrossFade(
92 logger=self.logger,
93 crossfade_duration=standard_crossfade_duration,
94 )
95 return await smart_fade.apply(
96 fade_out_part,
97 fade_in_part,
98 pcm_format,
99 )
100 # Attempt smart crossfade with analysis data
101 fade_out_analysis: SmartFadesAnalysis | None
102 if stored_analysis := await self.streams.mass.music.get_smart_fades_analysis(
103 fade_out_streamdetails.item_id,
104 fade_out_streamdetails.provider,
105 SmartFadesAnalysisFragment.OUTRO,
106 ):
107 fade_out_analysis = stored_analysis
108 else:
109 fade_out_analysis = await self.streams.mass.streams.smart_fades_analyzer.analyze(
110 fade_out_streamdetails.item_id,
111 fade_out_streamdetails.provider,
112 SmartFadesAnalysisFragment.OUTRO,
113 fade_out_part,
114 pcm_format,
115 )
116
117 fade_in_analysis: SmartFadesAnalysis | None
118 if stored_analysis := await self.streams.mass.music.get_smart_fades_analysis(
119 fade_in_streamdetails.item_id,
120 fade_in_streamdetails.provider,
121 SmartFadesAnalysisFragment.INTRO,
122 ):
123 fade_in_analysis = stored_analysis
124 else:
125 fade_in_analysis = await self.streams.mass.streams.smart_fades_analyzer.analyze(
126 fade_in_streamdetails.item_id,
127 fade_in_streamdetails.provider,
128 SmartFadesAnalysisFragment.INTRO,
129 fade_in_part,
130 pcm_format,
131 )
132 if (
133 fade_out_analysis
134 and fade_in_analysis
135 and fade_out_analysis.confidence > 0.3
136 and fade_in_analysis.confidence > 0.3
137 and mode == SmartFadesMode.SMART_CROSSFADE
138 ):
139 try:
140 smart_fade = SmartCrossFade(
141 logger=self.logger,
142 fade_out_analysis=fade_out_analysis,
143 fade_in_analysis=fade_in_analysis,
144 )
145 return await smart_fade.apply(
146 fade_out_part,
147 fade_in_part,
148 pcm_format,
149 )
150 except Exception as e:
151 self.logger.warning(
152 "Smart crossfade failed: %s, falling back to standard crossfade", e
153 )
154
155 # Always fallback to Standard Crossfade in case something goes wrong
156 smart_fade = StandardCrossFade(
157 logger=self.logger,
158 crossfade_duration=standard_crossfade_duration,
159 )
160 return await smart_fade.apply(
161 fade_out_part,
162 fade_in_part,
163 pcm_format,
164 )
165