music-assistant-server

1"""
2Smart Fades - bar-level musical structure detectors over ``BandProfile``.
3
4Detects the energy structure the transition planner reasons about: mastered
5fadeouts and valid outro/coda zones.  Every energy feature is self-referenced
6against the track's own full-track active-bar medians, so nothing here compares
7material across tracks.  The vocal authority is the FireRed ``VocalMask``, which
8the coda detector consumes to keep sung material out of a candidate exit zone -
9this module never runs its own spectral vocal classifier.
10"""
11
12from __future__ import annotations
13
14from dataclasses import dataclass
15from typing import TYPE_CHECKING
16
17if TYPE_CHECKING:
18    import numpy as np
19    import numpy.typing as npt
20
21    from music_assistant.controllers.streams.smart_fades.models import BandProfile
22    from music_assistant.controllers.streams.smart_fades.vocal import VocalMask
23
24# below -10 dB of the peak-normalized scale a band reference is silence
25_REFERENCE_EPSILON = 1e-4
26# fade stationarity is referenced to the run's first bars: prefix-decidable,
27# so the earliest-start scan needs no self-referential run median
28_STATIONARITY_PREFIX_BARS = 3
29# kick-qualifying bar: low band at half the track's own kick reference
30_KICK_RATIO = 0.5
31
32
33@dataclass(slots=True)
34class CodaZone:
35    """Media-time extent of a valid outro/coda zone, on downbeat-aligned bar edges."""
36
37    start_s: float
38    end_s: float
39
40
41def mask_intersects(
42    mask: VocalMask, start_s: float, end_s: float, bar_s: float, min_fraction: float = 0.25
43) -> bool:
44    """
45    Whether a media-time interval meaningfully overlaps any vocal window.
46
47    :param mask: Vocal-activity mask for the track (windows in the same time origin).
48    :param start_s: Interval start in media seconds (inclusive).
49    :param end_s: Interval end in media seconds (exclusive).
50    :param bar_s: Bar duration in seconds; the overlap unit.
51    :param min_fraction: Bar fraction an overlap must cover (epsilon-touch guard).
52    """
53    required = min_fraction * bar_s
54    return any(
55        min(end_s, window_end) - max(start_s, window_start) >= required
56        for window_start, window_end in mask.windows
57    )
58
59
60def point_in_mask(mask: VocalMask, t_s: float) -> bool:
61    """
62    Whether a media-time instant falls inside any vocal window.
63
64    :param mask: Vocal-activity mask for the track.
65    :param t_s: Instant in media seconds.
66    """
67    return any(window_start <= t_s < window_end for window_start, window_end in mask.windows)
68
69
70def detect_mastered_fadeout(
71    profile: BandProfile,
72    start_s: float,
73    end_s: float,
74    *,
75    min_bars: int = 4,
76    drop_db: float = 10.0,
77    monotone_share: float = 0.8,
78    jitter_db: float = 0.5,
79    frac_drift: float = 0.15,
80    audible_floor: float = 0.01,
81    frac_floor: float = 0.10,
82) -> float | None:
83    """
84    Detect a mastered fadeout over a media-time range.
85
86    A mastering fade is a post-mix gain ramp: total power collapses
87    monotonically by ``drop_db`` or more while the band fractions stay frozen.
88    Only the run of consecutive audible bars ending at the range's last
89    audible bar is considered; the earliest qualifying run start wins.
90    Returns the downbeat-snapped fade onset in media seconds, or ``None``
91    when the tail holds power, re-orchestrates, or merely decrescendos.
92
93    :param profile: The track's band profile.
94    :param start_s: Range start in media seconds (inclusive).
95    :param end_s: Range end in media seconds (exclusive).
96    :param min_bars: Minimum run length in bars.
97    :param drop_db: Total-power drop the run must reach, first to last bar.
98    :param monotone_share: Share of run steps that must not rise past the jitter.
99    :param jitter_db: Per-step rise tolerated as monotone.
100    :param frac_drift: Band-fraction drift allowed from the run-prefix reference.
101    :param audible_floor: Total-power ratio under which a bar is inaudible.
102    :param frac_floor: Total-power ratio under which fractions are exempt from drift.
103    """
104    import numpy as np  # noqa: PLC0415
105
106    indices = _range_indices(profile, start_s, end_s)
107    if len(indices) == 0:
108        return None
109    r_total = _total_ratio(profile, indices)
110    audible = r_total >= audible_floor
111    if not audible.any():
112        return None
113    run_end = int(np.nonzero(audible)[0][-1])
114    run_start = run_end
115    while run_start > 0 and audible[run_start - 1]:
116        run_start -= 1
117    levels_db = np.zeros(len(indices))
118    levels_db[audible] = 10.0 * np.log10(r_total[audible])
119    fractions = {band: _band_fraction(profile, band, indices) for band in profile.bar_power}
120    for k0 in range(run_start, run_end - min_bars + 2):
121        steps = np.diff(levels_db[k0 : run_end + 1])
122        if float((steps <= jitter_db).mean()) < monotone_share:
123            continue
124        if levels_db[k0] - levels_db[run_end] < drop_db:
125            continue
126        if not _fractions_stationary(
127            fractions, r_total, k0, run_end, frac_drift=frac_drift, frac_floor=frac_floor
128        ):
129            continue
130        # a spectrally frozen flat bed before the ramp legally joins the run;
131        # the onset is where the descent actually starts, so skip flat prefix
132        # bars (and land on the ramp's first bar) while the remaining run
133        # still qualifies as a fade on its own
134        onset = k0
135        while (
136            onset + min_bars <= run_end
137            and levels_db[onset] - levels_db[onset + 1] < jitter_db
138            and levels_db[onset + 1] - levels_db[run_end] >= drop_db
139        ):
140            onset += 1
141        if (
142            onset > k0
143            and onset + min_bars <= run_end
144            and levels_db[onset + 1] - levels_db[run_end] >= drop_db
145        ):
146            onset += 1
147        return float(profile.bar_starts[indices[onset]])
148    return None
149
150
151def detect_coda_zone(
152    profile: BandProfile,
153    vocal_mask: VocalMask,
154    earliest_s: float,
155    fade_onset_s: float | None,
156    start_s: float,
157    end_s: float,
158    *,
159    total_floor: float = 0.15,
160    min_seconds: float = 4.0,
161    min_bars: int = 2,
162    level_hold: float = 0.5,
163) -> CodaZone | None:
164    """
165    Detect a valid outro/coda zone over a media-time range.
166
167    A zone is a run of consecutive bars at audible program level that carry no
168    vocal activity, starting at or after ``earliest_s`` (the later of the vocal
169    end and the kick end) and ahead of any detected fade onset.  The run must
170    be terminal - no vocal window and no kick-qualifying bar after it in the
171    range - because a breakdown before returning content is never an exit.
172    Validity needs one musical gesture (``min_bars`` bars AND ``min_seconds``
173    seconds) and a sustained level: designed outros hold, undetected fades
174    halve and fail.  Returns ``None`` when no valid zone exists.
175
176    :param profile: The track's band profile.
177    :param vocal_mask: The track's FireRed vocal mask (media time); its windows
178        keep sung bars out of the zone and terminate it when a vocal returns.
179    :param earliest_s: Earliest allowed bar start in media seconds.
180    :param fade_onset_s: Detected mastered-fade onset; zone bars must end at
181        or before it (``None`` when no fade was detected).
182    :param start_s: Range start in media seconds (inclusive).
183    :param end_s: Range end in media seconds (exclusive).
184    :param total_floor: Total-power ratio a zone bar must hold (the bed must
185        still carry audible program under an equal-power fade).
186    :param min_seconds: Minimum zone duration in seconds.
187    :param min_bars: Minimum zone length in bars.
188    :param level_hold: Minimum second-half/first-half mean total power ratio.
189    """
190    import numpy as np  # noqa: PLC0415
191
192    indices = _range_indices(profile, start_s, end_s)
193    if len(indices) == 0:
194        return None
195    starts = profile.bar_starts[indices]
196    ends = np.array([_bar_end(profile, int(index)) for index in indices])
197    bar_lengths = ends - starts
198    not_vocal = np.array(
199        [
200            not mask_intersects(vocal_mask, float(starts[i]), float(ends[i]), float(bar_lengths[i]))
201            for i in range(len(indices))
202        ]
203    )
204    qualifying = (
205        (starts >= earliest_s) & (_total_ratio(profile, indices) >= total_floor) & not_vocal
206    )
207    if fade_onset_s is not None:
208        qualifying &= ends <= fade_onset_s
209    kick = _band_ratio(profile, "low", indices) >= _KICK_RATIO
210    zone: CodaZone | None = None
211    for first, last in _true_runs(qualifying):
212        # terminal: no returning vocal and no kick-qualifying bar after the run
213        if (
214            mask_intersects(vocal_mask, float(ends[last]), end_s, float(bar_lengths[last]))
215            or kick[last + 1 :].any()
216        ):
217            continue
218        if last - first + 1 < min_bars or float(ends[last] - starts[first]) < min_seconds:
219            continue
220        totals = profile.total_power[indices[first : last + 1]]
221        half = (last - first + 1) // 2
222        if float(totals[half:].mean()) < level_hold * float(totals[:half].mean()):
223            continue
224        zone = CodaZone(start_s=float(starts[first]), end_s=float(ends[last]))
225    return zone
226
227
228def _range_indices(profile: BandProfile, start_s: float, end_s: float) -> npt.NDArray[np.intp]:
229    """Find the indices of the profile bars whose start lies in [start_s, end_s)."""
230    import numpy as np  # noqa: PLC0415
231
232    return np.nonzero((profile.bar_starts >= start_s) & (profile.bar_starts < end_s))[0]
233
234
235def _total_ratio(profile: BandProfile, indices: npt.NDArray[np.intp]) -> npt.NDArray[np.float64]:
236    """r_T per bar: total power over the full-track median active-bar total."""
237    import numpy as np  # noqa: PLC0415
238
239    reference = float(np.median(profile.total_power[profile.active]))
240    if reference < _REFERENCE_EPSILON:
241        return np.zeros(len(indices))
242    return profile.total_power[indices] / reference
243
244
245def _band_ratio(
246    profile: BandProfile, band: str, indices: npt.NDArray[np.intp]
247) -> npt.NDArray[np.float64]:
248    """r_b per bar: band power over the track reference (0 on a silent reference)."""
249    import numpy as np  # noqa: PLC0415
250
251    reference = profile.reference[band]
252    if reference < _REFERENCE_EPSILON:
253        return np.zeros(len(indices))
254    return profile.bar_power[band][indices] / reference
255
256
257def _band_fraction(
258    profile: BandProfile, band: str, indices: npt.NDArray[np.intp]
259) -> npt.NDArray[np.float64]:
260    """f_b per bar: band share of the bar's total power (0 on silent bars)."""
261    import numpy as np  # noqa: PLC0415
262
263    total = profile.total_power[indices]
264    return np.divide(
265        profile.bar_power[band][indices],
266        total,
267        out=np.zeros(len(indices)),
268        where=total > 0,
269    )
270
271
272def _fractions_stationary(
273    fractions: dict[str, npt.NDArray[np.float64]],
274    r_total: npt.NDArray[np.float64],
275    run_start: int,
276    run_end: int,
277    *,
278    frac_drift: float,
279    frac_floor: float,
280) -> bool:
281    """Check that every band's fraction stays within frac_drift of the run-prefix mean."""
282    import numpy as np  # noqa: PLC0415
283
284    prefix = slice(run_start, min(run_start + _STATIONARITY_PREFIX_BARS, run_end + 1))
285    tested = np.nonzero(r_total[run_start : run_end + 1] >= frac_floor)[0] + run_start
286    for values in fractions.values():
287        anchor = float(values[prefix].mean())
288        if np.any(np.abs(values[tested] - anchor) > frac_drift):
289            return False
290    return True
291
292
293def _true_runs(mask: npt.NDArray[np.bool_]) -> list[tuple[int, int]]:
294    """Find the (first, last) index pairs of the maximal True runs in a boolean mask."""
295    runs: list[tuple[int, int]] = []
296    run_start: int | None = None
297    for index, value in enumerate(mask):
298        if value and run_start is None:
299            run_start = index
300        elif not value and run_start is not None:
301            runs.append((run_start, index - 1))
302            run_start = None
303    if run_start is not None:
304        runs.append((run_start, len(mask) - 1))
305    return runs
306
307
308def _bar_end(profile: BandProfile, bar_index: int) -> float:
309    """Media end of a bar: the next downbeat, or +median bar for the last bar."""
310    import numpy as np  # noqa: PLC0415
311
312    starts = profile.bar_starts
313    if bar_index + 1 < len(starts):
314        return float(starts[bar_index + 1])
315    return float(starts[-1] + np.median(np.diff(starts)))
316