xil_audio/
pcm.rs

1//! Interleaved 16-bit PCM with the measurements pydub exposes.
2//!
3//! Every formula here is pydub's, which is `audioop`'s underneath, and the
4//! integer truncation is load-bearing: `rms` is an int, so `dBFS` is the
5//! log of a rounded-down value, not of the true RMS. Reproducing the
6//! rounding matters more than being more accurate.
7
8/// Sample width is always 2 in this pipeline; pydub's
9/// `max_possible_amplitude` is `2^16 / 2`.
10pub const MAX_POSSIBLE_AMPLITUDE: f64 = 32768.0;
11
12/// Interleaved `i16` samples plus the format they were decoded at.
13#[derive(Clone, Debug, PartialEq)]
14pub struct Pcm {
15    pub samples: Vec<i16>,
16    pub channels: u16,
17    pub frame_rate: u32,
18}
19
20impl Pcm {
21    pub fn new(samples: Vec<i16>, channels: u16, frame_rate: u32) -> Pcm {
22        Pcm {
23            samples,
24            channels,
25            frame_rate,
26        }
27    }
28
29    /// Decode raw little-endian `s16le` bytes. A trailing odd byte is
30    /// dropped, as reading 16-bit frames necessarily does.
31    pub fn from_s16le(bytes: &[u8], channels: u16, frame_rate: u32) -> Pcm {
32        let samples = bytes
33            .chunks_exact(2)
34            .map(|c| i16::from_le_bytes([c[0], c[1]]))
35            .collect();
36        Pcm::new(samples, channels, frame_rate)
37    }
38
39    /// Frames, i.e. samples divided by channels (`len(_data) // frame_width`).
40    pub fn frame_count(&self) -> usize {
41        if self.channels == 0 {
42            return 0;
43        }
44        self.samples.len() / self.channels as usize
45    }
46
47    /// `len(segment)` — duration in whole milliseconds, rounded.
48    pub fn len_ms(&self) -> i64 {
49        if self.frame_rate == 0 {
50            return 0;
51        }
52        py_round(1000.0 * self.frame_count() as f64 / self.frame_rate as f64)
53    }
54
55    /// `audioop.rms` — `int(sqrt(sum(v²)/n))`, accumulated in f64 and
56    /// truncated toward zero.
57    pub fn rms(&self) -> i64 {
58        if self.samples.is_empty() {
59            return 0;
60        }
61        let sum_squares: f64 = self.samples.iter().map(|&v| (v as f64) * (v as f64)).sum();
62        (sum_squares / self.samples.len() as f64).sqrt() as i64
63    }
64
65    /// `audioop.max` — the largest absolute sample. `-32768` yields 32768,
66    /// which is why this is not an `i16`.
67    pub fn max_abs(&self) -> i64 {
68        self.samples
69            .iter()
70            .map(|&v| (v as i64).abs())
71            .max()
72            .unwrap_or(0)
73    }
74
75    /// `segment.dBFS` — RMS relative to full scale. `-inf` for silence.
76    pub fn dbfs(&self) -> f64 {
77        ratio_to_db(self.rms() as f64 / MAX_POSSIBLE_AMPLITUDE)
78    }
79
80    /// `segment.max_dBFS` — peak sample relative to full scale.
81    pub fn max_dbfs(&self) -> f64 {
82        ratio_to_db(self.max_abs() as f64 / MAX_POSSIBLE_AMPLITUDE)
83    }
84
85    /// `segment[start_ms:end_ms]`. Both ends clamp to the segment length,
86    /// and the frame index is `int(ms * rate / 1000)`.
87    ///
88    /// pydub pads a short tail with silence so the slice is the length the
89    /// caller asked for; that only happens mid-buffer, never at the end,
90    /// so slicing the final partial chunk returns what is actually there.
91    pub fn slice_ms(&self, start_ms: i64, end_ms: i64) -> Pcm {
92        let total_ms = self.len_ms();
93        let start = self.frame_for_ms(start_ms.min(total_ms));
94        let end = self.frame_for_ms(end_ms.min(total_ms));
95        let ch = self.channels.max(1) as usize;
96        let lo = (start * ch).min(self.samples.len());
97        let hi = (end * ch).min(self.samples.len()).max(lo);
98        Pcm::new(
99            self.samples[lo..hi].to_vec(),
100            self.channels,
101            self.frame_rate,
102        )
103    }
104
105    /// `_parse_position` — `int(ms * frame_rate / 1000.0)`.
106    fn frame_for_ms(&self, ms: i64) -> usize {
107        if ms <= 0 {
108            return 0;
109        }
110        (ms as f64 * (self.frame_rate as f64 / 1000.0)) as usize
111    }
112}
113
114/// `pydub.utils.ratio_to_db` in amplitude mode, `-inf` at zero. See
115/// [`crate::segment::ratio_to_db`] for why this is not `log10`.
116pub fn ratio_to_db(ratio: f64) -> f64 {
117    crate::segment::ratio_to_db(ratio)
118}
119
120/// Python's `round()` — half away from zero is *not* what it does; it is
121/// banker's rounding, so 0.5 goes to 0 and 1.5 goes to 2.
122fn py_round(x: f64) -> i64 {
123    let r = x.round();
124    if (x - x.trunc()).abs() == 0.5 && r % 2.0 != 0.0 {
125        (r - x.signum()) as i64
126    } else {
127        r as i64
128    }
129}
130
131#[cfg(test)]
132mod tests {
133    use super::*;
134
135    fn tone(n: usize, amp: i16) -> Pcm {
136        Pcm::new(
137            (0..n)
138                .map(|i| if i % 2 == 0 { amp } else { -amp })
139                .collect(),
140            1,
141            1000,
142        )
143    }
144
145    #[test]
146    fn silence_reads_as_negative_infinity() {
147        let p = Pcm::new(vec![0; 100], 1, 1000);
148        assert_eq!(p.rms(), 0);
149        assert_eq!(p.max_abs(), 0);
150        assert_eq!(p.dbfs(), f64::NEG_INFINITY);
151        assert_eq!(p.max_dbfs(), f64::NEG_INFINITY);
152    }
153
154    #[test]
155    fn full_scale_square_wave_is_zero_dbfs() {
156        let p = tone(100, 32767);
157        assert_eq!(p.rms(), 32767);
158        assert_eq!(p.max_abs(), 32767);
159        // 32767/32768 is a hair under full scale.
160        assert!((p.dbfs() - -0.000265).abs() < 1e-5, "{}", p.dbfs());
161    }
162
163    #[test]
164    fn rms_truncates_like_audioop() {
165        // sum of squares = 2*(100^2) = 20000, mean 10000, sqrt exactly 100.
166        assert_eq!(Pcm::new(vec![100, -100], 1, 1000).rms(), 100);
167        // mean 16250, sqrt 127.475… → truncated to 127, matching audioop.
168        assert_eq!(Pcm::new(vec![100, -150], 1, 1000).rms(), 127);
169    }
170
171    #[test]
172    fn most_negative_sample_does_not_overflow() {
173        let p = Pcm::new(vec![-32768, 0], 1, 1000);
174        assert_eq!(p.max_abs(), 32768, "abs(-32768) does not fit in an i16");
175        assert_eq!(p.max_dbfs(), 0.0, "exactly full scale");
176    }
177
178    #[test]
179    fn duration_counts_frames_not_samples() {
180        let stereo = Pcm::new(vec![0; 2000], 2, 1000);
181        assert_eq!(stereo.frame_count(), 1000);
182        assert_eq!(stereo.len_ms(), 1000);
183        let mono = Pcm::new(vec![0; 2000], 1, 1000);
184        assert_eq!(mono.len_ms(), 2000);
185        assert_eq!(Pcm::new(vec![], 2, 44100).len_ms(), 0);
186    }
187
188    #[test]
189    fn slices_clamp_and_index_by_frame() {
190        let p = Pcm::new((0..1000i16).collect(), 1, 1000); // 1000 frames, 1000 ms
191        let s = p.slice_ms(0, 500);
192        assert_eq!(s.samples.len(), 500);
193        assert_eq!(s.samples[0], 0);
194        let tail = p.slice_ms(900, 1400);
195        assert_eq!(tail.samples.len(), 100, "the end clamps, it does not pad");
196        assert_eq!(tail.samples[0], 900);
197        assert!(p.slice_ms(2000, 3000).samples.is_empty());
198    }
199
200    #[test]
201    fn s16le_round_trip() {
202        let bytes = [0x01, 0x00, 0xff, 0xff, 0x00, 0x80];
203        let p = Pcm::from_s16le(&bytes, 1, 8000);
204        assert_eq!(p.samples, vec![1, -1, -32768]);
205        // A dangling byte cannot form a sample and is dropped.
206        assert_eq!(
207            Pcm::from_s16le(&[0x01, 0x00, 0x02], 1, 8000).samples,
208            vec![1]
209        );
210    }
211
212    #[test]
213    fn python_round_is_banker_s() {
214        assert_eq!(py_round(0.5), 0);
215        assert_eq!(py_round(1.5), 2);
216        assert_eq!(py_round(2.5), 2);
217        assert_eq!(py_round(-0.5), 0);
218        assert_eq!(py_round(-1.5), -2);
219        assert_eq!(py_round(2.4), 2);
220        assert_eq!(py_round(2.6), 3);
221    }
222}
xil_audio/pcm.rs

xil_audio/
pcm.rs
