use rubato::audioadapter_buffers::direct::InterleavedSlice;
use rubato::{
Async, FixedAsync, Indexing, Resampler, SincInterpolationParameters, SincInterpolationType,
WindowFunction,
};
use crate::types::{Error, OutputFormat, Result, CHANNELS, SAMPLE_RATE};
pub const CHUNK_FRAMES: usize = 960;
const INNER_CH: usize = CHANNELS as usize;
pub struct Normalizer {
in_sample_rate: u32,
in_channels: usize,
output: OutputFormat,
stage1_resampler: Option<ResamplerState>,
inner_buf: Vec<f32>,
stage2: Option<OutputStage>,
out_buf: Vec<f32>,
out_chunk_frames: usize,
out_channels: usize,
out_frame_origin: u64,
pts_anchor: Option<PtsAnchor>,
total_out_frames: u64,
total_inner_frames: u64,
}
#[derive(Clone, Copy)]
struct PtsAnchor {
out_frame: u64,
pts_ns: i64,
}
struct ResamplerState {
inner: Async<f32>,
channels: usize,
chunk_in_frames: usize,
max_out_frames: usize,
in_accum: Vec<f32>,
out_scratch: Vec<f32>,
}
struct OutputStage {
out_channels: usize,
resampler: Option<ResamplerState>,
ch_scratch: Vec<f32>,
}
impl Normalizer {
pub fn new(in_sample_rate: u32, in_channels: u16, output: OutputFormat) -> Result<Self> {
let in_channels = in_channels.max(1) as usize;
let stage1_resampler = if in_sample_rate == SAMPLE_RATE {
None
} else {
Some(ResamplerState::new(in_sample_rate, SAMPLE_RATE, INNER_CH)?)
};
let out_channels = (output.channels.max(1)) as usize;
let out_chunk_frames = output.chunk_frames().max(1);
let stage2 = if output.sample_rate == SAMPLE_RATE && out_channels == INNER_CH {
None
} else {
Some(OutputStage::new(output.sample_rate, out_channels)?)
};
Ok(Self {
in_sample_rate,
in_channels,
output,
stage1_resampler,
inner_buf: Vec::with_capacity(CHUNK_FRAMES * INNER_CH * 4),
stage2,
out_buf: Vec::with_capacity(out_chunk_frames * out_channels * 4),
out_chunk_frames,
out_channels,
out_frame_origin: 0,
pts_anchor: None,
total_out_frames: 0,
total_inner_frames: 0,
})
}
pub fn in_sample_rate(&self) -> u32 {
self.in_sample_rate
}
pub fn output(&self) -> OutputFormat {
self.output
}
pub fn is_passthrough(&self) -> bool {
self.stage1_resampler.is_none()
}
pub fn is_output_passthrough(&self) -> bool {
self.stage2.is_none()
}
pub fn push(&mut self, interleaved: &[f32], device_pts_ns: i64) -> Result<()> {
if interleaved.is_empty() {
return Ok(());
}
let in_frames = interleaved.len() / self.in_channels;
if in_frames == 0 {
return Ok(());
}
self.update_pts_anchor(device_pts_ns);
let mut stereo = Vec::with_capacity(in_frames * INNER_CH);
Self::mix_to_stereo(interleaved, self.in_channels, in_frames, &mut stereo);
match &mut self.stage1_resampler {
None => {
self.total_inner_frames += in_frames as u64;
self.inner_buf.extend_from_slice(&stereo);
}
Some(rs) => {
rs.in_accum.extend_from_slice(&stereo);
rs.drain_into(&mut self.inner_buf, &mut self.total_inner_frames)?;
}
}
self.pump_stage2()
}
pub fn pop_chunk(&mut self) -> Option<(Vec<f32>, i64)> {
let need = self.out_chunk_frames * self.out_channels;
if self.out_buf.len() < need {
return None;
}
let pts = self.pts_for_out_frame(self.out_frame_origin);
let chunk: Vec<f32> = self.out_buf.drain(..need).collect();
self.out_frame_origin += self.out_chunk_frames as u64;
Some((chunk, pts))
}
pub fn buffered_out_frames(&self) -> usize {
self.out_buf.len() / self.out_channels
}
fn pump_stage2(&mut self) -> Result<()> {
let inner_chunk = CHUNK_FRAMES * INNER_CH;
match &mut self.stage2 {
None => {
if !self.inner_buf.is_empty() {
let n_frames = (self.inner_buf.len() / INNER_CH) as u64;
self.total_out_frames += n_frames;
self.out_buf.append(&mut self.inner_buf);
}
}
Some(stage) => {
while self.inner_buf.len() >= inner_chunk {
let chunk: Vec<f32> = self.inner_buf.drain(..inner_chunk).collect();
stage.process_inner_chunk(
&chunk,
&mut self.out_buf,
&mut self.total_out_frames,
)?;
}
}
}
Ok(())
}
fn mix_to_stereo(src: &[f32], in_ch: usize, in_frames: usize, dst: &mut Vec<f32>) {
match in_ch {
1 => {
for &s in &src[..in_frames] {
dst.push(s);
dst.push(s);
}
}
2 => {
dst.extend_from_slice(&src[..in_frames * 2]);
}
_ => {
for f in 0..in_frames {
let base = f * in_ch;
dst.push(src[base]);
dst.push(src[base + 1]);
}
}
}
}
fn update_pts_anchor(&mut self, device_pts_ns: i64) {
let ratio = self.output.sample_rate as f64 / self.in_sample_rate as f64;
let projected_out_frame = (self.total_in_frames_estimate() as f64 * ratio) as u64;
self.pts_anchor = Some(PtsAnchor {
out_frame: projected_out_frame,
pts_ns: device_pts_ns,
});
}
fn total_in_frames_estimate(&self) -> u64 {
let inv = self.in_sample_rate as f64 / SAMPLE_RATE as f64;
(self.total_inner_frames as f64 * inv) as u64
}
fn pts_for_out_frame(&self, out_frame: u64) -> i64 {
match self.pts_anchor {
None => crate::clock::monotonic_now_ns(),
Some(anchor) => {
let frame_delta = out_frame as i64 - anchor.out_frame as i64;
let ns_per_out_frame = 1_000_000_000_i64 / self.output.sample_rate as i64;
anchor.pts_ns + frame_delta * ns_per_out_frame
}
}
}
}
impl ResamplerState {
fn new(in_sr: u32, out_sr: u32, channels: usize) -> Result<Self> {
let ratio = out_sr as f64 / in_sr as f64;
let chunk_in_frames = (in_sr as usize / 50).max(64);
let params = SincInterpolationParameters {
sinc_len: 128,
f_cutoff: 0.95,
interpolation: SincInterpolationType::Linear,
oversampling_factor: 128,
window: WindowFunction::BlackmanHarris2,
};
let inner = Async::<f32>::new_sinc(
ratio,
1.0, ¶ms,
chunk_in_frames,
channels,
FixedAsync::Input,
)
.map_err(|e| Error::Backend(format!("rubato sinc resampler construction failed: {e}")))?;
let max_out_frames = inner.output_frames_max();
Ok(Self {
inner,
channels,
chunk_in_frames,
max_out_frames,
in_accum: Vec::with_capacity(chunk_in_frames * channels * 4),
out_scratch: vec![0.0; max_out_frames * channels],
})
}
fn drain_into(&mut self, out_buf: &mut Vec<f32>, total_out_frames: &mut u64) -> Result<()> {
let step = self.chunk_in_frames * self.channels;
while self.in_accum.len() >= step {
let in_adapter =
InterleavedSlice::new(&self.in_accum[..step], self.channels, self.chunk_in_frames)
.map_err(|e| {
Error::Backend(format!("rubato interleaved input adapter failed: {e}"))
})?;
let mut out_adapter = InterleavedSlice::new_mut(
&mut self.out_scratch[..],
self.channels,
self.max_out_frames,
)
.map_err(|e| {
Error::Backend(format!("rubato interleaved output adapter failed: {e}"))
})?;
let indexing = Indexing {
input_offset: 0,
output_offset: 0,
partial_len: None,
active_channels_mask: None,
};
let (_in_used, out_written) = self
.inner
.process_into_buffer(&in_adapter, &mut out_adapter, Some(&indexing))
.map_err(|e| Error::Backend(format!("rubato process_into_buffer failed: {e}")))?;
let n_samples = out_written * self.channels;
out_buf.extend_from_slice(&self.out_scratch[..n_samples]);
*total_out_frames += out_written as u64;
self.in_accum.drain(..step);
}
Ok(())
}
}
impl OutputStage {
fn new(out_sample_rate: u32, out_channels: usize) -> Result<Self> {
let resampler = if out_sample_rate == SAMPLE_RATE {
None
} else {
Some(ResamplerState::new(
SAMPLE_RATE,
out_sample_rate,
out_channels,
)?)
};
Ok(Self {
out_channels,
resampler,
ch_scratch: Vec::with_capacity(CHUNK_FRAMES * out_channels),
})
}
fn process_inner_chunk(
&mut self,
inner_stereo: &[f32],
out_buf: &mut Vec<f32>,
total_out_frames: &mut u64,
) -> Result<()> {
debug_assert_eq!(inner_stereo.len(), CHUNK_FRAMES * INNER_CH);
self.ch_scratch.clear();
match self.out_channels {
1 => {
for f in 0..CHUNK_FRAMES {
let l = inner_stereo[f * 2];
let r = inner_stereo[f * 2 + 1];
self.ch_scratch.push((l + r) * 0.5);
}
}
2 => {
self.ch_scratch.extend_from_slice(inner_stereo);
}
_ => {
for f in 0..CHUNK_FRAMES {
let l = inner_stereo[f * 2];
for _ in 0..self.out_channels {
self.ch_scratch.push(l);
}
}
}
}
match &mut self.resampler {
None => {
let n_frames = (self.ch_scratch.len() / self.out_channels) as u64;
*total_out_frames += n_frames;
out_buf.extend_from_slice(&self.ch_scratch);
}
Some(rs) => {
rs.in_accum.extend_from_slice(&self.ch_scratch);
rs.drain_into(out_buf, total_out_frames)?;
}
}
Ok(())
}
}
#[cfg(test)]
mod tests {
use super::*;
use std::f32::consts::PI;
fn default_out() -> OutputFormat {
OutputFormat::default()
}
#[test]
fn mono_48k_to_stereo_duplicates_channels() {
let mut n = Normalizer::new(48_000, 1, default_out()).expect("normalizer");
assert!(n.is_passthrough());
assert!(n.is_output_passthrough());
let mono: Vec<f32> = (0..CHUNK_FRAMES).map(|i| (i as f32) * 0.001).collect();
n.push(&mono, 0).expect("push");
let (chunk, _pts) = n.pop_chunk().expect("one chunk");
assert_eq!(chunk.len(), CHUNK_FRAMES * 2);
for f in 0..CHUNK_FRAMES {
assert_eq!(chunk[f * 2], chunk[f * 2 + 1], "L==R at frame {f}");
assert_eq!(chunk[f * 2], mono[f]);
}
}
#[test]
fn passthrough_preserves_frame_count() {
let mut n = Normalizer::new(48_000, 2, default_out()).expect("normalizer");
assert!(n.is_passthrough());
assert!(n.is_output_passthrough());
let frames = CHUNK_FRAMES * 2 + 100;
let stereo: Vec<f32> = (0..frames * 2).map(|i| (i as f32) * 1e-4).collect();
n.push(&stereo, 0).expect("push");
let mut got_frames = 0usize;
while let Some((c, _)) = n.pop_chunk() {
assert_eq!(c.len(), CHUNK_FRAMES * 2);
got_frames += CHUNK_FRAMES;
}
assert_eq!(got_frames, CHUNK_FRAMES * 2);
assert_eq!(n.buffered_out_frames(), 100);
}
#[test]
fn stereo_44100_to_48000_yields_about_50_chunks_per_second() {
let mut n = Normalizer::new(44_100, 2, default_out()).expect("normalizer");
assert!(!n.is_passthrough());
let in_frames = 44_100;
let freq = 440.0_f32;
let mut interleaved = Vec::with_capacity(in_frames * 2);
for i in 0..in_frames {
let s = (2.0 * PI * freq * (i as f32) / 44_100.0).sin() * 0.5;
interleaved.push(s); interleaved.push(s); }
let mut pts = 0i64;
for block in interleaved.chunks(441 * 2) {
n.push(block, pts).expect("push");
pts += (block.len() as i64 / 2) * 1_000_000_000 / 44_100;
}
let mut chunks = 0usize;
while let Some((c, _pts)) = n.pop_chunk() {
assert_eq!(c.len(), CHUNK_FRAMES * 2);
chunks += 1;
}
assert!(
(47..=50).contains(&chunks),
"expected ~50 chunks, got {chunks}"
);
}
#[test]
fn pts_increases_monotonically_across_chunks() {
let mut n = Normalizer::new(48_000, 2, default_out()).expect("normalizer");
let frames = CHUNK_FRAMES * 3;
let stereo = vec![0.0f32; frames * 2];
n.push(&stereo, 100_000_000).expect("push");
let mut last = i64::MIN;
let mut count = 0;
while let Some((_, pts)) = n.pop_chunk() {
assert!(pts >= last, "pts must be non-decreasing");
last = pts;
count += 1;
}
assert_eq!(count, 3);
}
#[test]
fn output_16k_mono_yields_320_frame_mono_chunks() {
let out = OutputFormat {
sample_rate: 16_000,
channels: 1,
};
let mut n = Normalizer::new(48_000, 2, out).expect("normalizer");
assert!(n.is_passthrough()); assert!(!n.is_output_passthrough());
let in_frames = 48_000;
let freq = 440.0_f32;
let mut pts = 0i64;
for blk in 0..(in_frames / 480) {
let mut block = Vec::with_capacity(480 * 2);
for j in 0..480 {
let i = blk * 480 + j;
let s = (2.0 * PI * freq * (i as f32) / 48_000.0).sin() * 0.5;
block.push(s);
block.push(s);
}
n.push(&block, pts).expect("push");
pts += 480 * 1_000_000_000 / 48_000;
}
let mut chunks = 0usize;
while let Some((c, _)) = n.pop_chunk() {
assert_eq!(c.len(), 320, "16k mono 20ms = 320 sample (mono)");
chunks += 1;
}
assert!(
(47..=50).contains(&chunks),
"expected ~50 chunks, got {chunks}"
);
}
#[test]
fn output_16k_stereo_yields_320_frame_640_sample_chunks() {
let out = OutputFormat {
sample_rate: 16_000,
channels: 2,
};
let mut n = Normalizer::new(48_000, 2, out).expect("normalizer");
let in_frames = 48_000;
let stereo: Vec<f32> = (0..in_frames * 2)
.map(|i| ((i / 2) as f32 * 0.0001).sin() * 0.3)
.collect();
for block in stereo.chunks(480 * 2) {
n.push(block, 0).expect("push");
}
let mut chunks = 0usize;
while let Some((c, _)) = n.pop_chunk() {
assert_eq!(c.len(), 640, "16k stereo 20ms = 320 frame * 2 = 640 sample");
chunks += 1;
}
assert!(
(47..=50).contains(&chunks),
"expected ~50 chunks, got {chunks}"
);
}
#[test]
fn output_8k_stereo_yields_160_frame_chunks() {
let out = OutputFormat {
sample_rate: 8_000,
channels: 2,
};
let mut n = Normalizer::new(48_000, 2, out).expect("normalizer");
let stereo: Vec<f32> = (0..48_000 * 2)
.map(|i| (i as f32 * 1e-5).sin() * 0.2)
.collect();
for block in stereo.chunks(480 * 2) {
n.push(block, 0).expect("push");
}
let mut chunks = 0usize;
while let Some((c, _)) = n.pop_chunk() {
assert_eq!(c.len(), 320, "8k stereo 20ms = 160 frame * 2 = 320 sample");
chunks += 1;
}
assert!(
(47..=50).contains(&chunks),
"expected ~50 chunks, got {chunks}"
);
}
#[test]
fn stereo_to_mono_is_lr_average() {
let out = OutputFormat {
sample_rate: 48_000,
channels: 1,
};
let mut n = Normalizer::new(48_000, 2, out).expect("normalizer");
let mut stereo = Vec::with_capacity(CHUNK_FRAMES * 2);
for _ in 0..CHUNK_FRAMES {
stereo.push(0.5);
stereo.push(-0.5);
}
n.push(&stereo, 0).expect("push");
let (chunk, _) = n.pop_chunk().expect("one mono chunk");
assert_eq!(chunk.len(), CHUNK_FRAMES); for &s in &chunk {
assert!(s.abs() < 1e-6, "逆相の平均は 0 付近のはず: {s}");
}
}
fn rms(samples: &[f32]) -> f32 {
if samples.is_empty() {
return 0.0;
}
let sum_sq: f64 = samples.iter().map(|&x| (x as f64) * (x as f64)).sum();
(sum_sq / samples.len() as f64).sqrt() as f32
}
fn zero_crossings(samples: &[f32]) -> usize {
let mut crossings = 0;
for w in samples.windows(2) {
if (w[0] < 0.0 && w[1] >= 0.0) || (w[0] >= 0.0 && w[1] < 0.0) {
crossings += 1;
}
}
crossings
}
#[test]
fn resample_44100_to_48000_preserves_amplitude_and_frequency() {
let mut n = Normalizer::new(44_100, 1, default_out()).expect("normalizer");
let freq = 440.0_f32;
let amp = 0.5_f32;
let in_rate = 44_100usize;
let total_frames = in_rate * 2;
let mut pts = 0i64;
for blk in 0..(total_frames / 441) {
let mut block = Vec::with_capacity(441);
for j in 0..441 {
let i = blk * 441 + j;
block.push((2.0 * PI * freq * (i as f32) / in_rate as f32).sin() * amp);
}
n.push(&block, pts).expect("push");
pts += 441 * 1_000_000_000 / in_rate as i64;
}
let mut left: Vec<f32> = Vec::new();
while let Some((c, _)) = n.pop_chunk() {
assert_eq!(c.len(), CHUNK_FRAMES * 2);
for f in 0..CHUNK_FRAMES {
assert_eq!(c[f * 2], c[f * 2 + 1], "mono 入力なので L==R");
left.push(c[f * 2]);
}
}
assert!(left.len() >= 48_000, "1 秒以上の出力が必要: {}", left.len());
let start = 12_000;
let mid = &left[start..start + 48_000];
let got_rms = rms(mid);
let expect_rms = amp / std::f32::consts::SQRT_2;
let rms_err = ((got_rms - expect_rms) / expect_rms).abs();
assert!(
rms_err < 0.05,
"RMS 保存誤差が大きい: got={got_rms} expect={expect_rms} err={rms_err}"
);
let crossings = zero_crossings(mid);
let est_freq = crossings as f32 / 2.0; let freq_err = ((est_freq - freq) / freq).abs();
assert!(
freq_err < 0.02,
"周波数 保存誤差が大きい: 交差={crossings} 推定={est_freq}Hz err={freq_err}"
);
}
#[test]
fn output_16k_mono_preserves_values() {
let out = OutputFormat {
sample_rate: 16_000,
channels: 1,
};
let mut n = Normalizer::new(48_000, 2, out).expect("normalizer");
let freq = 440.0_f32;
let amp = 0.5_f32;
let in_rate = 48_000usize;
let total_frames = in_rate * 2;
let mut pts = 0i64;
for blk in 0..(total_frames / 480) {
let mut block = Vec::with_capacity(480 * 2);
for j in 0..480 {
let i = blk * 480 + j;
let s = (2.0 * PI * freq * (i as f32) / in_rate as f32).sin() * amp;
block.push(s); block.push(s); }
n.push(&block, pts).expect("push");
pts += 480 * 1_000_000_000 / in_rate as i64;
}
let mut mono: Vec<f32> = Vec::new();
while let Some((c, _)) = n.pop_chunk() {
assert_eq!(c.len(), 320, "16k/mono 20ms = 320 sample(1ch)");
mono.extend_from_slice(&c);
}
assert!(mono.len() >= 16_000, "1 秒以上必要: {}", mono.len());
let start = 4_000;
let mid = &mono[start..start + 16_000];
let got_rms = rms(mid);
let expect_rms = amp / std::f32::consts::SQRT_2;
let rms_err = ((got_rms - expect_rms) / expect_rms).abs();
assert!(
rms_err < 0.05,
"16k/mono RMS 保存誤差: got={got_rms} expect={expect_rms} err={rms_err}"
);
let est_freq = zero_crossings(mid) as f32 / 2.0;
let freq_err = ((est_freq - freq) / freq).abs();
assert!(
freq_err < 0.02,
"16k/mono 周波数 保存誤差: 推定={est_freq}Hz err={freq_err}"
);
}
#[test]
fn pts_delta_is_about_20ms_between_chunks() {
let mut n = Normalizer::new(48_000, 2, default_out()).expect("normalizer");
let mut device_pts = 1_000_000_000i64; let block_frames = 480usize;
for _ in 0..20 {
let stereo = vec![0.1f32; block_frames * 2];
n.push(&stereo, device_pts).expect("push");
device_pts += block_frames as i64 * 1_000_000_000 / 48_000;
}
let mut pts_list = Vec::new();
while let Some((_, pts)) = n.pop_chunk() {
pts_list.push(pts);
}
assert!(
pts_list.len() >= 5,
"十分なチャンク数が必要: {}",
pts_list.len()
);
for w in pts_list.windows(2) {
let delta = w[1] - w[0];
assert!(delta > 0, "PTS は厳密に増加: {} -> {}", w[0], w[1]);
assert!(
(delta - 20_000_000).abs() <= 1_000_000,
"隣接 PTS delta が ~20ms でない: {delta} ns"
);
}
}
#[test]
fn push_empty_and_subframe_are_noops() {
let mut n = Normalizer::new(48_000, 2, default_out()).expect("normalizer");
n.push(&[], 0).expect("empty push ok");
n.push(&[0.5], 0).expect("subframe push ok");
assert!(n.pop_chunk().is_none(), "端数だけでは 1 チャンクも出ない");
assert_eq!(n.buffered_out_frames(), 0);
}
#[test]
fn silence_input_yields_zero_output() {
let mut n = Normalizer::new(48_000, 2, default_out()).expect("normalizer");
let stereo = vec![0.0f32; CHUNK_FRAMES * 2];
n.push(&stereo, 0).expect("push");
let (chunk, _) = n.pop_chunk().expect("one chunk");
assert!(chunk.iter().all(|&s| s == 0.0), "無音入力は無音出力");
}
}