music-assistant-server

1"""
2Smart Fades - data models.
3
4A ``TransitionPlan`` is the renderer-agnostic description of a transition: it
5captures every decision (where to cut, how long to blend, tempo ramp, shelf EQ)
6without owning a single audio byte or FFmpeg filter.  A ``TransitionPlanner``
7produces it from stored ``AudioAnalysisData``; a renderer turns it into the
8``Filter`` chain.  Keeping the plan free of bytes is what lets alternative
9planners be drop-in replacements and lets a plan be computed before any tail
10is buffered.
11"""
12
13from __future__ import annotations
14
15import itertools
16from dataclasses import dataclass, field, replace
17from enum import Enum, StrEnum
18from typing import TYPE_CHECKING
19
20from music_assistant.controllers.streams.smart_fades.filters import ShelfType
21from music_assistant.controllers.streams.smart_fades.helpers import db_ramp
22
23if TYPE_CHECKING:
24    import numpy as np
25    import numpy.typing as npt
26
27    from music_assistant.models.audio_analysis import AudioAnalysisData
28
29
30# Band edges (Hz) of the smart_fades ``extra_data["band_rms"]`` envelopes; None = up to Nyquist.
31# Stored rows keep whatever bands their analysis_version wrote — read historical rows by shape.
32BAND_RMS_BANDS: dict[str, tuple[float, float | None]] = {
33    "low": (20.0, 120.0),
34    "low_mid": (120.0, 400.0),
35    "mid": (400.0, 4000.0),
36    "high": (4000.0, None),
37}
38
39
40class SmartFadeNotApplicable(Exception):
41    """Raised when the tracks cannot yield a smart crossfade and the caller should fall back."""
42
43
44class TransitionTier(Enum):
45    """How ambitious the transition is, decided from tempo, key and blendability."""
46
47    # stretch range + key compatible: long, fully-featured DJ blend
48    FULL_BLEND = "full_blend"
49    # stretch range but keys clash: shorter energy-anchored blend
50    TEMPO_BLEND = "tempo_blend"
51    # tempos incompatible or material not blendable: short downbeat-snapped fade
52    QUICK_FADE = "quick_fade"
53
54
55@dataclass(slots=True)
56class Deck:
57    """
58    One track on the planner's (virtual) DJ deck.
59
60    Holds the track's stored analysis plus the beat grids that are usable for
61    this transition (masked/shifted to the relevant buffer window by the planner).
62    """
63
64    analysis: AudioAnalysisData
65    bpm: float
66    beats: npt.NDArray[np.float32]
67    downbeats: npt.NDArray[np.float32]
68    # time signature numerator; callers fall back to 4 when analysis omits it
69    beats_per_bar: int = 4
70
71
72@dataclass(slots=True)
73class BandProfile:
74    """Bar-level band-power view of one track, on its own downbeat grid."""
75
76    bar_starts: npt.NDArray[np.float64]
77    bar_power: dict[str, npt.NDArray[np.float64]]
78    total_power: npt.NDArray[np.float64]
79    active: npt.NDArray[np.bool_]
80    reference: dict[str, float]
81
82
83@dataclass(slots=True)
84class CrossfadeTimingInfo:
85    """Timing breakdown of a crossfade mix output: PRE | CF | POST."""
86
87    pre_crossfade_duration: float = 0.0
88    crossfade_duration: float = 0.0
89    fadein_trimmed_duration: float = 0.0
90    post_crossfade_duration: float = 0.0
91
92
93@dataclass(slots=True)
94class TempoPlan:
95    """
96    Tempo ramp schedule for the outgoing track.
97
98    ``steps`` is a list of ``(timestamp_seconds, tempo_ratio)`` points in the
99    outgoing track's buffer-local time; empty means no time-stretching.
100    """
101
102    steps: list[tuple[float, float]] = field(default_factory=list)
103
104    def __bool__(self) -> bool:
105        """Return True when the plan actually stretches time."""
106        return bool(self.steps)
107
108    def savings_until(self, t: float) -> float:
109        """
110        Seconds removed from the rendered stream by the stretch up to input time t.
111
112        Negative when the stretch slows the tail down (the rendered stream is
113        lengthened).
114
115        :param t: Input-time position (seconds) up to which to integrate savings.
116        """
117        savings = 0.0
118        # rubberband is initialized at the FIRST step's ratio from t=0, so the
119        # span before the first step already runs stretched (no-op for multi-step
120        # ramps, whose first step has ratio 1.0)
121        if self.steps and self.steps[0][0] > 0.0:
122            first_ts, first_ratio = self.steps[0]
123            span_end = min(first_ts, t)
124            savings += span_end * (1.0 - 1.0 / first_ratio)
125        for i, (ts, ratio) in enumerate(self.steps):
126            if ts >= t:
127                break
128            seg_end = min(self.steps[i + 1][0] if i + 1 < len(self.steps) else t, t)
129            savings += (seg_end - ts) * (1.0 - 1.0 / ratio)
130        return savings
131
132
133@dataclass(slots=True)
134class ShelfSchedule:
135    """One EQ gain schedule for a ShelfFilter (LOW/HIGH) or PeakFilter (PEAK)."""
136
137    shelf_type: ShelfType
138    frequency: int
139    # (time_seconds, gain_db); the step at t=0 sets the initial gain
140    steps: list[tuple[float, float]]
141    # PEAK bandwidth in octaves; unused for LOW/HIGH shelves
142    width_oct: float = 0.707
143
144    def gain_at(self, schedule_time: float) -> float:
145        """Interpolate the scheduled gain (dB) at a schedule-time position (clamped at the ends)."""
146        steps = self.steps
147        if schedule_time <= steps[0][0]:
148            return steps[0][1]
149        if schedule_time >= steps[-1][0]:
150            return steps[-1][1]
151        for (t0, g0), (t1, g1) in itertools.pairwise(steps):
152            if t0 <= schedule_time <= t1:
153                frac = (schedule_time - t0) / (t1 - t0) if t1 > t0 else 0.0
154                return g0 + frac * (g1 - g0)
155        return steps[-1][1]
156
157
158@dataclass(slots=True)
159class EqPlan:
160    """Bass-swap EQ across the transition: who owns the low end, and when it swaps."""
161
162    # seconds into the rendered crossfade
163    swap_at: float
164    # A-side schedules are in input time (pre-stretch), B-side in post-trim time;
165    # None means the schedule is bypassed (a shelf shallower than the bypass floor)
166    low_out: ShelfSchedule | None
167    low_in: ShelfSchedule | None
168    high_out: ShelfSchedule | None
169    high_in: ShelfSchedule | None
170    # measured mid-band (vocal) handover; None on either side means it is bypassed
171    mid_out: ShelfSchedule | None = None
172    mid_in: ShelfSchedule | None = None
173
174    @classmethod
175    def neutral(cls, swap_at: float = 0.0) -> EqPlan:
176        """Return an EqPlan that renders no shelves at all."""
177        return cls(swap_at=swap_at, low_out=None, low_in=None, high_out=None, high_in=None)
178
179    def with_mid_depth_scaled(self, shrink: float, *, bypass_below_db: float) -> EqPlan:
180        """
181        Return a copy with both mid schedules' depth scaled by ``shrink`` (fraction to KEEP).
182
183        Bypasses the mid swap entirely when scaling drops its depth below ``bypass_below_db``.
184        """
185        if self.mid_out is None and self.mid_in is None:
186            return self
187        source = self.mid_out or self.mid_in
188        assert source is not None  # narrowed by the check above
189        depth = source.steps[-1 if self.mid_out else 0][1]
190        if shrink <= 0.0 or abs(depth * shrink) < abs(bypass_below_db):
191            return replace(self, mid_out=None, mid_in=None)
192        mid_out = self.mid_out
193        mid_in = self.mid_in
194        if mid_out is not None:
195            mid_out = replace(mid_out, steps=[(t, g * shrink) for t, g in mid_out.steps])
196        if mid_in is not None:
197            mid_in = replace(mid_in, steps=[(t, g * shrink) for t, g in mid_in.steps])
198        return replace(self, mid_out=mid_out, mid_in=mid_in)
199
200    def with_low_ramps_steepened(
201        self, swap_at: float, new_len: float, ratio: float, cf_start_input: float
202    ) -> EqPlan:
203        """
204        Return a copy whose low ramps span ``new_len`` centered on ``swap_at``.
205
206        Only the ramp span is tightened; the endpoint gains stay put.
207        """
208        if self.low_out is None and self.low_in is None:
209            return self
210        start_in = max(0.0, swap_at - new_len / 2)
211        new_len_input = new_len * ratio
212        swap_at_input = cf_start_input + swap_at * ratio
213        start_out = max(cf_start_input, swap_at_input - new_len_input / 2)
214
215        low_out = self.low_out
216        if low_out is not None:
217            depth_a = low_out.steps[-1][1]
218            low_out = replace(
219                low_out, steps=[(0.0, 0.0), *db_ramp(start_out, new_len_input, 0.0, depth_a)]
220            )
221        low_in = self.low_in
222        if low_in is not None:
223            depth_b = low_in.steps[0][1]
224            low_in = replace(
225                low_in, steps=[(0.0, depth_b), *db_ramp(start_in, new_len, depth_b, 0.0)]
226            )
227        return replace(self, low_out=low_out, low_in=low_in)
228
229
230@dataclass(slots=True)
231class FadeOutTrim:
232    """Where the outgoing track's audible content ends and how much was dropped."""
233
234    end_pos: float
235    trimmed_seconds: float
236
237
238class TransitionStrategy(StrEnum):
239    """How a vocal-aware plan's final overlap was ultimately decided."""
240
241    # a normal phrased candidate cleared the vocal-collision guard
242    ENERGY_ALIGNED = "energy_aligned"
243    # every phrased candidate collided; shipped the click-free equal-power fallback
244    SHORT_VOCAL_HANDOFF = "short_vocal_handoff"
245    # grid unusable but both decks ambient: long unphrased equal-power overlay
246    LAZY_OVERLAY = "lazy_overlay"
247
248
249@dataclass(frozen=True, slots=True)
250class PlanMetrics:
251    """
252    Per-plan telemetry; every field defaults to its energy-only value.
253
254    The trim and downbeat facts are populated on every plan; the vocal fields
255    (collision, retained vocal time) additionally need both tracks to carry a
256    validated FireRed vocal-activity timeline and keep their defaults on an
257    energy-only plan.
258    """
259
260    strategy: TransitionStrategy = TransitionStrategy.ENERGY_ALIGNED
261    # seconds of RMS-audible outgoing material dropped before the crossfade start
262    audible_outgoing_trim: float = 0.0
263    # seconds of outgoing vocal activity that fall inside the rendered crossfade
264    outgoing_vocal_fade_seconds: float = 0.0
265    # whether the final fade_out_window sits on an outgoing downbeat
266    anchor_on_downbeat: bool = False
267    # simultaneous outgoing/incoming vocal overlap in rendered-crossfade seconds
268    collision_seconds: float = 0.0
269    # gain-weighted overlap (the acrossfade curve's simultaneous-power integral)
270    weighted_collision_seconds: float = 0.0
271
272
273@dataclass(slots=True)
274class TransitionPlan:
275    """
276    Renderer-agnostic description of how two tracks are joined.
277
278    All times are in the outgoing track's buffer-local seconds.
279    """
280
281    # how ambitious the transition is; drives overlap length, tempo and EQ
282    tier: TransitionTier
283    # audible end of the fade-out tail (buffer-local seconds)
284    fade_out_window: float
285    crossfade_duration: float
286    eq_plan: EqPlan = field(default_factory=EqPlan.neutral)
287    tempo_plan: TempoPlan = field(default_factory=TempoPlan)
288    fadeout_trim: FadeOutTrim | None = None
289    # seconds trimmed off the incoming head for beat alignment
290    fadein_trim_start: float | None = None
291    fadeout_curve: str = "qsin"
292    metrics: PlanMetrics = field(default_factory=PlanMetrics)
293