/
/
/
1"""
2Smart Fades - Transition renderer.
3
4The renderer is the DJ's hands: it picks the tools from the filter toolset
5(``filters.py``) that realize a ``TransitionPlan``, and produces the
6``CrossfadeTimingInfo`` breakdown. This is the only place where bytes
7re-enter: the plan is sized in seconds, the renderer reconciles it with the
8actual buffer lengths and drives both the blend overlap and the timing
9bookkeeping from that one reconciled value.
10"""
11
12from __future__ import annotations
13
14import logging
15from typing import TYPE_CHECKING
16
17from music_assistant.controllers.streams.smart_fades.filters import (
18 FadeInTrimFilter,
19 FadeOutTrimFilter,
20 Filter,
21 GradualTimeStretchFilter,
22 PeakFilter,
23 ShelfFilter,
24 ShelfType,
25 StreamingCrossfadeFilter,
26)
27from music_assistant.controllers.streams.smart_fades.models import (
28 CrossfadeTimingInfo,
29 ShelfSchedule,
30 TransitionPlan,
31)
32
33if TYPE_CHECKING:
34 from music_assistant_models.media_items import AudioFormat
35
36
37class TransitionRenderer:
38 """Renders a ``TransitionPlan`` into an FFmpeg filter chain and timing info."""
39
40 def __init__(self, logger: logging.Logger) -> None:
41 """Initialize the renderer."""
42 self.logger = logger
43
44 def render(
45 self,
46 plan: TransitionPlan,
47 pcm_format: AudioFormat,
48 fade_in_bytes_len: int,
49 ) -> tuple[list[Filter], CrossfadeTimingInfo]:
50 """
51 Build the filter chain and timing info for a transition plan.
52
53 :param plan: The transition plan to render.
54 :param pcm_format: PCM format of both input buffers and the output.
55 :param fade_in_bytes_len: Length in bytes of the incoming track's head buffer.
56 """
57 fade_out_seconds = plan.fade_out_window - plan.tempo_plan.savings_until(
58 plan.fade_out_window
59 )
60 fade_in_seconds = fade_in_bytes_len / pcm_format.pcm_sample_size
61 fadein_trimmed = plan.fadein_trim_start or 0.0
62 # clamp CF to fit shorter inputs (defensive â normally full buffers)
63 crossfade_samples = int(
64 min(
65 plan.crossfade_duration,
66 fade_out_seconds,
67 max(0.0, fade_in_seconds - fadein_trimmed),
68 )
69 * pcm_format.sample_rate
70 )
71 crossfade_seconds = crossfade_samples / pcm_format.sample_rate
72 pre_crossfade_samples = int(
73 max(0.0, fade_out_seconds - crossfade_seconds) * pcm_format.sample_rate
74 )
75 filters = self._build_filters(plan, crossfade_samples, pre_crossfade_samples)
76 timing = CrossfadeTimingInfo(
77 pre_crossfade_duration=max(0.0, fade_out_seconds - crossfade_seconds),
78 crossfade_duration=crossfade_seconds,
79 fadein_trimmed_duration=fadein_trimmed,
80 post_crossfade_duration=max(0.0, fade_in_seconds - fadein_trimmed - crossfade_seconds),
81 )
82 return filters, timing
83
84 def _build_filters(
85 self, plan: TransitionPlan, crossfade_samples: int, pre_crossfade_samples: int
86 ) -> list[Filter]:
87 """Assemble the ordered filter chain from the plan."""
88 filters: list[Filter] = []
89 # FadeOutTrim first: its cut point is on the untrimmed input timeline
90 if plan.fadeout_trim is not None:
91 filters.append(
92 FadeOutTrimFilter(
93 logger=self.logger,
94 fadeout_end_pos=plan.fadeout_trim.end_pos,
95 trimmed_seconds=plan.fadeout_trim.trimmed_seconds,
96 )
97 )
98 # outgoing shelves before the stretch keep their schedules in musical input time
99 self._append_shelf(filters, plan.eq_plan.low_out, "fadeout")
100 self._append_shelf(filters, plan.eq_plan.high_out, "fadeout")
101 self._append_shelf(filters, plan.eq_plan.mid_out, "fadeout")
102 if plan.tempo_plan:
103 filters.append(GradualTimeStretchFilter(self.logger, plan.tempo_plan.steps))
104 if plan.fadein_trim_start is not None:
105 filters.append(
106 FadeInTrimFilter(logger=self.logger, fadein_start_pos=plan.fadein_trim_start)
107 )
108 self._append_shelf(filters, plan.eq_plan.low_in, "fadein")
109 self._append_shelf(filters, plan.eq_plan.high_in, "fadein")
110 self._append_shelf(filters, plan.eq_plan.mid_in, "fadein")
111 # the streaming blend is positioned at the planned pre-point, so it can
112 # emit while the incoming window is still arriving; the hard cut at the
113 # planned end keeps any time-stretch drift out of the incoming audio
114 filters.append(
115 StreamingCrossfadeFilter(
116 logger=self.logger,
117 crossfade_samples=crossfade_samples,
118 pre_crossfade_samples=pre_crossfade_samples,
119 fadeout_curve=plan.fadeout_curve,
120 )
121 )
122 return filters
123
124 def _append_shelf(
125 self, filters: list[Filter], schedule: ShelfSchedule | None, stream_type: str
126 ) -> None:
127 """Append the matching filter for ``schedule``, or nothing when bypassed (None)."""
128 if schedule is None:
129 return
130 if schedule.shelf_type is ShelfType.PEAK:
131 filters.append(
132 PeakFilter(
133 logger=self.logger,
134 frequency=schedule.frequency,
135 width_oct=schedule.width_oct,
136 gain_steps=schedule.steps,
137 stream_type=stream_type,
138 )
139 )
140 return
141 filters.append(
142 ShelfFilter(
143 logger=self.logger,
144 shelf_type=schedule.shelf_type,
145 frequency=schedule.frequency,
146 gain_steps=schedule.steps,
147 stream_type=stream_type,
148 )
149 )
150