/
/
/
1"""Helper functions for the Smart Fades audio analysis provider."""
2
3from __future__ import annotations
4
5from typing import TYPE_CHECKING
6
7import numpy as np
8import numpy.typing as npt
9import torch
10from music_assistant_models.enums import ContentType
11
12from music_assistant.controllers.streams.smart_fades.models import BAND_RMS_BANDS
13
14if TYPE_CHECKING:
15 from music_assistant_models.media_items import AudioFormat
16
17
18def calculate_overall_bpm(beats: np.ndarray, n_segments: int = 5) -> float:
19 """
20 Calculate overall BPM.
21
22 Splits the beat array into N segments, computes a BPM per segment, then
23 discards outlier segments (those deviating more than 3 BPM from the median)
24 before averaging the consistent remainder. This prevents a single poorly-tracked
25 section from pulling the final BPM away from the true value.
26
27 :param beats: Array of beat timestamps in seconds.
28 :param n_segments: Number of equal segments to split the beats into.
29 """
30 if len(beats) < n_segments * 2:
31 return float(60.0 / np.mean(np.diff(beats)))
32
33 segment_bpms = []
34 for idx in np.array_split(np.arange(len(beats)), n_segments):
35 if len(idx) < 2:
36 continue
37 segment_bpms.append(60.0 / float(np.mean(np.diff(beats[idx]))))
38
39 if len(segment_bpms) < 2:
40 return float(60.0 / np.mean(np.diff(beats)))
41
42 seg_arr = np.array(segment_bpms)
43 median_bpm = float(np.median(seg_arr))
44 consistent = seg_arr[np.abs(seg_arr - median_bpm) <= 3.0]
45
46 if len(consistent) < 2:
47 # All segments too spread out â fall back to unfiltered mean
48 return float(np.mean(seg_arr))
49
50 return float(np.mean(consistent))
51
52
53def aggregate_series_to_bins(
54 values: npt.NDArray[np.float32],
55 n_bins: int,
56 *,
57 power: bool = False,
58) -> npt.NDArray[np.float32]:
59 """
60 Resample a frame series to a fixed bin count by averaging over each bin's span.
61
62 Every bin is the exact fractional-overlap mean of the frames it covers
63 (mean of squares then square root when ``power`` is True, appropriate for
64 RMS/energy series). Unlike point sampling this acts as a boxcar low-pass,
65 so periodic frame-rate detail (e.g. beat-level energy ripple) cannot alias
66 into the bin grid.
67
68 :param values: Input frame series (uniform frame spacing).
69 :param n_bins: Number of output bins.
70 :param power: Average squared values and return the root (RMS-correct).
71 """
72 x = values.astype(np.float64)
73 if power:
74 x = x * x
75 # interpolating the cumulative sum at fractional bin edges = exact boxcar average in O(n)
76 cum = np.concatenate(([0.0], np.cumsum(x)))
77 edges = np.linspace(0.0, len(values), n_bins + 1)
78 cum_at_edges = np.interp(edges, np.arange(len(values) + 1, dtype=np.float64), cum)
79 binned = np.diff(cum_at_edges) / np.diff(edges)
80 if power:
81 binned = np.sqrt(binned)
82 return binned.astype(np.float32)
83
84
85def compute_band_rms_frames(
86 pcm: npt.NDArray[np.float32],
87 sample_rate: int,
88 window_samples: int,
89) -> dict[str, npt.NDArray[np.float32]]:
90 """
91 Compute per-band RMS frames over fixed windows of a mono PCM block.
92
93 Returns one RMS value per window (including a trailing partial window) per
94 band in ``BAND_RMS_BANDS``, at the same frame cadence as the full-band RMS
95 frames, so both series can be aggregated to the same bin grid.
96
97 :param pcm: Mono float32 PCM samples.
98 :param sample_rate: Sample rate of ``pcm`` in Hz.
99 :param window_samples: Frame length in samples (e.g. 100 ms worth).
100 """
101 out: dict[str, list[float]] = {name: [] for name in BAND_RMS_BANDS}
102 n_full = len(pcm) // window_samples
103 windows: list[npt.NDArray[np.float32]] = []
104 if n_full > 0:
105 windows.extend(pcm[: n_full * window_samples].reshape(n_full, window_samples))
106 if len(pcm) - n_full * window_samples > 0:
107 windows.append(pcm[n_full * window_samples :])
108 for window in windows:
109 spectrum = np.fft.rfft(window.astype(np.float64))
110 freqs = np.fft.rfftfreq(len(window), 1.0 / sample_rate)
111 # Parseval, single-sided: double every bin except DC (and Nyquist for even N),
112 # which are not mirrored in the negative-frequency half.
113 power = np.abs(spectrum) ** 2 / len(window) ** 2
114 power[1:] *= 2.0
115 if len(window) % 2 == 0:
116 power[-1] /= 2.0
117 for name, (lo, hi) in BAND_RMS_BANDS.items():
118 mask = (freqs >= lo) & (freqs < hi if hi is not None else np.ones_like(freqs, bool))
119 out[name].append(float(np.sqrt(power[mask].sum())))
120 return {name: np.array(vals, dtype=np.float32) for name, vals in out.items()}
121
122
123def decode_pcm_chunk_to_mono(audio_format: AudioFormat, pcm_chunk: bytes) -> np.ndarray:
124 """
125 Decode a raw PCM chunk to a mono float32 numpy array.
126
127 :param audio_format: The audio format describing the PCM data.
128 :param pcm_chunk: Raw PCM audio data.
129 """
130 content_type = audio_format.content_type
131 writable = bytearray(pcm_chunk)
132
133 if content_type == ContentType.PCM_F32LE:
134 audio = torch.frombuffer(writable, dtype=torch.float32).clone()
135 elif content_type == ContentType.PCM_F64LE:
136 audio = torch.frombuffer(writable, dtype=torch.float64).clone().to(torch.float32)
137 elif content_type == ContentType.PCM_S32LE:
138 audio = (
139 torch.frombuffer(writable, dtype=torch.int32).clone().to(torch.float32) / 2147483648.0
140 )
141 elif content_type == ContentType.PCM_S24LE:
142 raw = torch.frombuffer(writable, dtype=torch.uint8).clone()
143 raw = raw[: (raw.numel() // 3) * 3].reshape(-1, 3).to(torch.int32)
144 audio = raw[:, 0] | (raw[:, 1] << 8) | (raw[:, 2] << 16)
145 audio = torch.where(audio & 0x800000 != 0, audio - 0x1000000, audio)
146 audio = audio.to(torch.float32) / 8388608.0
147 else:
148 audio = torch.frombuffer(writable, dtype=torch.int16).clone().to(torch.float32) / 32768.0
149
150 channels = audio_format.channels
151 if channels > 1:
152 frame_samples = (audio.numel() // channels) * channels
153 audio = audio[:frame_samples].reshape(-1, channels).mean(dim=1)
154
155 return audio.numpy()
156