music-assistant-server

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