music-assistant-server

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 acrossfade 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    CrossfadeFilter,
19    FadeInTrimFilter,
20    FadeOutTrimFilter,
21    Filter,
22    GradualTimeStretchFilter,
23    PeakFilter,
24    ShelfFilter,
25    ShelfType,
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        filters = self._build_filters(plan, crossfade_samples)
73        timing = CrossfadeTimingInfo(
74            pre_crossfade_duration=max(0.0, fade_out_seconds - crossfade_seconds),
75            crossfade_duration=crossfade_seconds,
76            fadein_trimmed_duration=fadein_trimmed,
77            post_crossfade_duration=max(0.0, fade_in_seconds - fadein_trimmed - crossfade_seconds),
78        )
79        return filters, timing
80
81    def _build_filters(self, plan: TransitionPlan, crossfade_samples: int) -> list[Filter]:
82        """Assemble the ordered filter chain from the plan."""
83        filters: list[Filter] = []
84        # FadeOutTrim first: its cut point is on the untrimmed input timeline
85        if plan.fadeout_trim is not None:
86            filters.append(
87                FadeOutTrimFilter(
88                    logger=self.logger,
89                    fadeout_end_pos=plan.fadeout_trim.end_pos,
90                    trimmed_seconds=plan.fadeout_trim.trimmed_seconds,
91                )
92            )
93        # outgoing shelves before the stretch keep their schedules in musical input time
94        self._append_shelf(filters, plan.eq_plan.low_out, "fadeout")
95        self._append_shelf(filters, plan.eq_plan.high_out, "fadeout")
96        self._append_shelf(filters, plan.eq_plan.mid_out, "fadeout")
97        if plan.tempo_plan:
98            filters.append(GradualTimeStretchFilter(self.logger, plan.tempo_plan.steps))
99        if plan.fadein_trim_start is not None:
100            filters.append(
101                FadeInTrimFilter(logger=self.logger, fadein_start_pos=plan.fadein_trim_start)
102            )
103        self._append_shelf(filters, plan.eq_plan.low_in, "fadein")
104        self._append_shelf(filters, plan.eq_plan.high_in, "fadein")
105        self._append_shelf(filters, plan.eq_plan.mid_in, "fadein")
106        filters.append(
107            CrossfadeFilter(
108                logger=self.logger,
109                crossfade_samples=crossfade_samples,
110                fadeout_curve=plan.fadeout_curve,
111            )
112        )
113        return filters
114
115    def _append_shelf(
116        self, filters: list[Filter], schedule: ShelfSchedule | None, stream_type: str
117    ) -> None:
118        """Append the matching filter for ``schedule``, or nothing when bypassed (None)."""
119        if schedule is None:
120            return
121        if schedule.shelf_type is ShelfType.PEAK:
122            filters.append(
123                PeakFilter(
124                    logger=self.logger,
125                    frequency=schedule.frequency,
126                    width_oct=schedule.width_oct,
127                    gain_steps=schedule.steps,
128                    stream_type=stream_type,
129                )
130            )
131            return
132        filters.append(
133            ShelfFilter(
134                logger=self.logger,
135                shelf_type=schedule.shelf_type,
136                frequency=schedule.frequency,
137                gain_steps=schedule.steps,
138                stream_type=stream_type,
139            )
140        )
141