use std::panic::AssertUnwindSafe;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::Arc;
use std::thread::{self, JoinHandle};
use std::time::{Duration, Instant};
use flexaudio_core::backend::{CaptureBackend, RawSink};
use flexaudio_core::clock::monotonic_now_ns;
use flexaudio_core::normalizer::Normalizer;
use flexaudio_core::raw_ring::{raw_ring, RawConsumer};
use flexaudio_core::types::{Error, OutputFormat, Result, CHANNELS, SAMPLE_RATE};
use crate::stream::RAW_RING_SAMPLES;
const STARVATION_FILL_THRESHOLD: Duration = Duration::from_millis(60);
const FIFO_MAX_SAMPLES: usize = 48_000;
const IDLE_SLEEP: Duration = Duration::from_millis(2);
const DRIFT_RATIO_LIMIT: f64 = 500e-6;
const DRIFT_UPDATE_INTERVAL_SAMPLES: usize = (SAMPLE_RATE as usize / 10) * CHANNELS as usize;
const DRIFT_EMA_ALPHA: f64 = 0.1;
const DRIFT_GAIN: f64 = DRIFT_RATIO_LIMIT / 19_200.0;
const DRIFT_SLEW_PER_UPDATE: f64 = 20e-6;
pub(crate) struct CompositeBackend {
mic: Box<dyn CaptureBackend>,
system: Box<dyn CaptureBackend>,
mic_gain: f32,
system_gain: f32,
stopping: Arc<AtomicBool>,
mixer: Option<JoinHandle<()>>,
}
impl CompositeBackend {
pub(crate) fn new(
mic: Box<dyn CaptureBackend>,
system: Box<dyn CaptureBackend>,
mic_gain: f32,
system_gain: f32,
) -> Self {
Self {
mic,
system,
mic_gain,
system_gain,
stopping: Arc::new(AtomicBool::new(false)),
mixer: None,
}
}
}
impl CaptureBackend for CompositeBackend {
fn native_format(&self) -> (u32, u16) {
(SAMPLE_RATE, CHANNELS)
}
fn start(&mut self, sink: RawSink) -> Result<()> {
if self.mixer.is_some() {
return Ok(());
}
let mic_lane = start_child(&mut self.mic)?;
let system_lane = match start_child(&mut self.system) {
Ok(lane) => lane,
Err(e) => {
stop_child(&mut self.mic);
return Err(e);
}
};
self.stopping = Arc::new(AtomicBool::new(false));
let stopping = self.stopping.clone();
let mic_gain = self.mic_gain;
let system_gain = self.system_gain;
let mixer = thread::Builder::new()
.name("flexaudio-mix".into())
.spawn(move || {
run_mixer(mic_lane, system_lane, mic_gain, system_gain, sink, stopping);
})
.map_err(|e| Error::Backend(format!("spawn mix thread: {e}")));
match mixer {
Ok(handle) => {
self.mixer = Some(handle);
Ok(())
}
Err(e) => {
stop_child(&mut self.mic);
stop_child(&mut self.system);
Err(e)
}
}
}
fn stop(&mut self) {
self.stopping.store(true, Ordering::SeqCst);
if let Some(h) = self.mixer.take() {
let _ = h.join();
}
stop_child(&mut self.mic);
stop_child(&mut self.system);
}
}
impl Drop for CompositeBackend {
fn drop(&mut self) {
self.stop();
}
}
struct ChildLane {
consumer: RawConsumer,
normalizer: Normalizer,
fifo: Vec<f32>,
last_supply: Instant,
}
impl ChildLane {
fn ingest(&mut self, scratch: &mut [f32]) -> Result<()> {
let got = self.consumer.pop_slice(scratch);
if got == 0 {
return Ok(());
}
self.normalizer.push(&scratch[..got], monotonic_now_ns())?;
let mut supplied = false;
while let Some((chunk, _pts)) = self.normalizer.pop_chunk() {
self.fifo.extend_from_slice(&chunk);
supplied = true;
}
if supplied {
self.last_supply = Instant::now();
if self.fifo.len() > FIFO_MAX_SAMPLES {
let excess = self.fifo.len() - FIFO_MAX_SAMPLES;
self.fifo.drain(..excess);
}
}
Ok(())
}
fn is_starved(&self, now: Instant) -> bool {
now.duration_since(self.last_supply) >= STARVATION_FILL_THRESHOLD
}
}
struct LinearStitcher {
frac: f64,
}
impl LinearStitcher {
fn new() -> Self {
Self { frac: 0.0 }
}
fn producible(&self, fifo_frames: usize, ratio: f64) -> usize {
if fifo_frames == 0 {
return 0;
}
let span = (fifo_frames - 1) as f64 - self.frac;
if span < 0.0 {
return 0;
}
(span / ratio) as usize + 1
}
fn pull(&mut self, fifo: &mut Vec<f32>, ratio: f64, out_frames: usize, out: &mut Vec<f32>) {
let ch = CHANNELS as usize;
let frames = fifo.len() / ch;
debug_assert!(out_frames <= self.producible(frames, ratio));
for k in 0..out_frames {
let pos = self.frac + k as f64 * ratio;
let left = pos as usize;
let right = (left + 1).min(frames - 1);
let w = (pos - left as f64) as f32;
for c in 0..ch {
let a = fifo[left * ch + c];
let b = fifo[right * ch + c];
out.push(a + w * (b - a));
}
}
let end = self.frac + out_frames as f64 * ratio;
let consumed = (end as usize).min(frames);
self.frac = end - consumed as f64;
fifo.drain(..consumed * ch);
}
fn reset(&mut self) {
self.frac = 0.0;
}
}
struct DriftController {
ratio: f64,
ema_diff: f64,
pending_samples: usize,
}
impl DriftController {
fn new() -> Self {
Self {
ratio: 1.0,
ema_diff: 0.0,
pending_samples: 0,
}
}
fn on_output(&mut self, samples: usize, mic_len: usize, system_len: usize) {
self.pending_samples += samples;
if self.pending_samples >= DRIFT_UPDATE_INTERVAL_SAMPLES {
self.pending_samples = 0;
self.update(mic_len, system_len);
}
}
fn update(&mut self, mic_len: usize, system_len: usize) {
let diff = mic_len as f64 - system_len as f64;
self.ema_diff += DRIFT_EMA_ALPHA * (diff - self.ema_diff);
let target = (1.0 - DRIFT_GAIN * self.ema_diff)
.clamp(1.0 - DRIFT_RATIO_LIMIT, 1.0 + DRIFT_RATIO_LIMIT);
let step = (target - self.ratio).clamp(-DRIFT_SLEW_PER_UPDATE, DRIFT_SLEW_PER_UPDATE);
self.ratio += step;
}
}
struct DriftCorrection {
stitcher: LinearStitcher,
controller: DriftController,
}
impl DriftCorrection {
fn new() -> Self {
Self {
stitcher: LinearStitcher::new(),
controller: DriftController::new(),
}
}
}
fn start_child(child: &mut Box<dyn CaptureBackend>) -> Result<ChildLane> {
let (rate, channels) = child.native_format();
if rate == 0 || channels == 0 {
return Err(Error::InvalidArg(
"mix child native_format must have non-zero rate and channels".into(),
));
}
let (producer, consumer) = raw_ring(RAW_RING_SAMPLES);
let sink = RawSink::new(producer, rate, channels);
match std::panic::catch_unwind(AssertUnwindSafe(|| child.start(sink))) {
Ok(Ok(())) => {}
Ok(Err(e)) => return Err(e),
Err(_) => return Err(Error::Backend("mix child panicked during start()".into())),
}
let normalizer = Normalizer::new(
rate,
channels,
OutputFormat {
sample_rate: SAMPLE_RATE,
channels: CHANNELS,
},
)
.inspect_err(|_| {
stop_child(child);
})?;
Ok(ChildLane {
consumer,
normalizer,
fifo: Vec::with_capacity(FIFO_MAX_SAMPLES),
last_supply: Instant::now(),
})
}
fn stop_child(child: &mut Box<dyn CaptureBackend>) {
let _ = std::panic::catch_unwind(AssertUnwindSafe(|| child.stop()));
}
fn run_mixer(
mut mic: ChildLane,
mut system: ChildLane,
mic_gain: f32,
system_gain: f32,
mut sink: RawSink,
stopping: Arc<AtomicBool>,
) {
let mut scratch = vec![0.0f32; RAW_RING_SAMPLES];
let mut mixed: Vec<f32> = Vec::with_capacity(FIFO_MAX_SAMPLES);
let mut drift = DriftCorrection::new();
if !prime_lanes(&mut mic, &mut system, &mut scratch, &stopping) {
return;
}
loop {
if stopping.load(Ordering::SeqCst) {
break;
}
if mic.ingest(&mut scratch).is_err() || system.ingest(&mut scratch).is_err() {
return;
}
let pushed = mix_and_push(
&mut mic,
&mut system,
mic_gain,
system_gain,
&mut drift,
&mut sink,
&mut mixed,
);
if !pushed {
thread::sleep(IDLE_SLEEP);
}
}
}
fn prime_lanes(
mic: &mut ChildLane,
system: &mut ChildLane,
scratch: &mut [f32],
stopping: &AtomicBool,
) -> bool {
let start = Instant::now();
while !stopping.load(Ordering::SeqCst) {
if mic.ingest(scratch).is_err() || system.ingest(scratch).is_err() {
return false;
}
if (!mic.fifo.is_empty() && !system.fifo.is_empty())
|| start.elapsed() >= STARVATION_FILL_THRESHOLD
{
break;
}
thread::sleep(IDLE_SLEEP);
}
true
}
fn mix_and_push(
mic: &mut ChildLane,
system: &mut ChildLane,
mic_gain: f32,
system_gain: f32,
drift: &mut DriftCorrection,
sink: &mut RawSink,
mixed: &mut Vec<f32>,
) -> bool {
let ch = CHANNELS as usize;
let ratio = drift.controller.ratio;
let steady_frames =
(mic.fifo.len() / ch).min(drift.stitcher.producible(system.fifo.len() / ch, ratio));
if steady_frames > 0 {
mixed.clear();
drift
.stitcher
.pull(&mut system.fifo, ratio, steady_frames, mixed);
let count = steady_frames * ch;
for (i, out) in mixed.iter_mut().enumerate() {
*out = (mic.fifo[i] * mic_gain + *out * system_gain).clamp(-1.0, 1.0);
}
mic.fifo.drain(..count);
drift
.controller
.on_output(count, mic.fifo.len(), system.fifo.len());
sink.push(mixed, monotonic_now_ns());
return true;
}
let now = Instant::now();
let (mic_take, system_take) = if !mic.fifo.is_empty() && system.is_starved(now) {
drift.stitcher.reset();
(mic.fifo.len(), system.fifo.len())
} else if mic.fifo.is_empty() && !system.fifo.is_empty() && mic.is_starved(now) {
drift.stitcher.reset();
(0, system.fifo.len())
} else {
return false;
};
let count = mic_take.max(system_take);
mixed.clear();
for i in 0..count {
let m = if i < mic_take { mic.fifo[i] } else { 0.0 };
let s = if i < system_take { system.fifo[i] } else { 0.0 };
mixed.push((m * mic_gain + s * system_gain).clamp(-1.0, 1.0));
}
mic.fifo.drain(..mic_take);
system.fifo.drain(..system_take);
sink.push(mixed, monotonic_now_ns());
true
}
#[cfg(test)]
mod tests {
use super::*;
use flexaudio_core::raw_ring::raw_ring;
use std::sync::atomic::AtomicU32;
struct ConstBackend {
sample_rate: u32,
channels: u16,
value: f32,
feed_for: Option<Duration>,
running: Arc<AtomicBool>,
handle: Option<JoinHandle<()>>,
}
impl ConstBackend {
fn new(value: f32, feed_for: Option<Duration>) -> Self {
Self {
sample_rate: 48_000,
channels: 2,
value,
feed_for,
running: Arc::new(AtomicBool::new(false)),
handle: None,
}
}
}
impl CaptureBackend for ConstBackend {
fn native_format(&self) -> (u32, u16) {
(self.sample_rate, self.channels)
}
fn start(&mut self, mut sink: RawSink) -> Result<()> {
if self.running.load(Ordering::SeqCst) {
return Ok(());
}
self.running.store(true, Ordering::SeqCst);
let running = self.running.clone();
let sample_rate = self.sample_rate;
let channels = self.channels as usize;
let value = self.value;
let feed_for = self.feed_for;
let handle = thread::Builder::new()
.name("flexaudio-const-gen".into())
.spawn(move || {
let frames_per_block = (sample_rate as usize / 100).max(1); let block = vec![value; frames_per_block * channels];
let start = Instant::now();
while running.load(Ordering::SeqCst) {
let feeding = feed_for.is_none_or(|d| start.elapsed() < d);
if !feeding {
thread::sleep(Duration::from_millis(5));
continue;
}
let accepted = sink.push(&block, start.elapsed().as_nanos() as i64);
if accepted < block.len() {
thread::sleep(Duration::from_millis(1));
}
}
})
.map_err(|e| Error::Backend(format!("spawn const gen thread: {e}")))?;
self.handle = Some(handle);
Ok(())
}
fn stop(&mut self) {
self.running.store(false, Ordering::SeqCst);
if let Some(h) = self.handle.take() {
let _ = h.join();
}
}
}
impl Drop for ConstBackend {
fn drop(&mut self) {
self.stop();
}
}
struct FailingStartBackend;
impl CaptureBackend for FailingStartBackend {
fn native_format(&self) -> (u32, u16) {
(48_000, 2)
}
fn start(&mut self, _sink: RawSink) -> Result<()> {
Err(Error::Backend("intentional start failure".into()))
}
fn stop(&mut self) {}
}
struct TrackingBackend {
starts: Arc<AtomicU32>,
stops: Arc<AtomicU32>,
}
impl CaptureBackend for TrackingBackend {
fn native_format(&self) -> (u32, u16) {
(48_000, 2)
}
fn start(&mut self, _sink: RawSink) -> Result<()> {
self.starts.fetch_add(1, Ordering::SeqCst);
Ok(())
}
fn stop(&mut self) {
self.stops.fetch_add(1, Ordering::SeqCst);
}
}
fn start_composite(
mic: Box<dyn CaptureBackend>,
system: Box<dyn CaptureBackend>,
mic_gain: f32,
system_gain: f32,
) -> (CompositeBackend, RawConsumer) {
let mut be = CompositeBackend::new(mic, system, mic_gain, system_gain);
assert_eq!(be.native_format(), (48_000, 2), "内部正規形を名乗るはず");
let (producer, consumer) = raw_ring(RAW_RING_SAMPLES);
let sink = RawSink::new(producer, 48_000, 2);
be.start(sink).expect("composite start");
(be, consumer)
}
const VALUE_TOL: f32 = 1e-4;
const ONE_CHUNK_SAMPLES: usize = 1_920;
fn collect_until(
consumer: &mut RawConsumer,
max_wait: Duration,
mut done: impl FnMut(&[f32]) -> bool,
) -> Vec<f32> {
let mut out = Vec::new();
let mut scratch = vec![0.0f32; RAW_RING_SAMPLES];
let start = Instant::now();
loop {
let got = consumer.pop_slice(&mut scratch);
out.extend_from_slice(&scratch[..got]);
if done(&scratch[..got]) || start.elapsed() >= max_wait {
return out;
}
thread::sleep(Duration::from_millis(5));
}
}
const COLLECT_MAX_WAIT: Duration = Duration::from_secs(30);
fn count_near(samples: &[f32], v: f32) -> usize {
samples
.iter()
.filter(|&&s| (s - v).abs() < VALUE_TOL)
.count()
}
fn assert_only_allowed_values(
samples: &[f32],
mixed: f32,
mic_only: f32,
system_only: f32,
) -> usize {
let allowed = [mixed, mic_only, system_only, 0.0];
let mut mixed_count = 0usize;
for (i, &s) in samples.iter().enumerate() {
if (s - mixed).abs() < VALUE_TOL {
mixed_count += 1;
} else {
assert!(
allowed.iter().any(|&a| (s - a).abs() < VALUE_TOL),
"許容集合 {allowed:?} の外の値(合成の数学の誤り): samples[{i}] = {s}"
);
}
}
mixed_count
}
#[test]
fn mix_sums_two_sources_with_gains() {
let mic = Box::new(ConstBackend::new(0.2, None));
let system = Box::new(ConstBackend::new(0.3, None));
let (mut be, mut consumer) = start_composite(mic, system, 1.0, 2.0);
let (mut total, mut mixed) = (0usize, 0usize);
let samples = collect_until(&mut consumer, COLLECT_MAX_WAIT, |new| {
total += new.len();
mixed += count_near(new, 0.8);
total >= 10_000 && mixed >= ONE_CHUNK_SAMPLES
});
be.stop();
assert!(
samples.len() >= 10_000,
"相応のサンプルが出るはず: {}",
samples.len()
);
let mixed_count = assert_only_allowed_values(&samples, 0.8, 0.2, 0.6);
assert!(
mixed_count >= ONE_CHUNK_SAMPLES,
"合成値 0.8 が相応に出現するはず: {mixed_count}/{}",
samples.len()
);
}
#[test]
fn mix_clamps_sum() {
let mic = Box::new(ConstBackend::new(0.8, None));
let system = Box::new(ConstBackend::new(0.8, None));
let (mut be, mut consumer) = start_composite(mic, system, 1.0, 1.0);
let mut clamped = 0usize;
let samples = collect_until(&mut consumer, COLLECT_MAX_WAIT, |new| {
clamped += count_near(new, 1.0);
clamped >= ONE_CHUNK_SAMPLES
});
be.stop();
for (i, &s) in samples.iter().enumerate() {
assert!(
s <= 1.0,
"クランプ後は 1.0 を超えないはず: samples[{i}] = {s}"
);
}
let clamped_count = assert_only_allowed_values(&samples, 1.0, 0.8, 0.8);
assert!(
clamped_count >= ONE_CHUNK_SAMPLES,
"クランプ値 1.0 が相応に出現するはず: {clamped_count}/{}",
samples.len()
);
}
#[test]
fn mix_survives_one_side_starvation() {
let mic = Box::new(ConstBackend::new(0.2, None));
let system = Box::new(ConstBackend::new(0.3, Some(Duration::from_millis(150))));
let (mut be, mut consumer) = start_composite(mic, system, 1.0, 1.0);
let mut mic_only = 0usize;
let samples = collect_until(&mut consumer, COLLECT_MAX_WAIT, |new| {
mic_only += count_near(new, 0.2);
mic_only >= ONE_CHUNK_SAMPLES
});
be.stop();
assert_only_allowed_values(&samples, 0.5, 0.2, 0.3);
let mic_only_count = count_near(&samples, 0.2);
assert!(
mic_only_count >= ONE_CHUNK_SAMPLES,
"飢餓後は mic 単独の値 0.2 が流れ続けるはず: {mic_only_count}/{}",
samples.len()
);
}
#[test]
fn mix_start_failure_cleans_up() {
let starts = Arc::new(AtomicU32::new(0));
let stops = Arc::new(AtomicU32::new(0));
let mic = Box::new(TrackingBackend {
starts: starts.clone(),
stops: stops.clone(),
});
let system = Box::new(FailingStartBackend);
let mut be = CompositeBackend::new(mic, system, 1.0, 1.0);
let (producer, _consumer) = raw_ring(RAW_RING_SAMPLES);
let sink = RawSink::new(producer, 48_000, 2);
let err = be
.start(sink)
.expect_err("system 起動失敗で全体も Err のはず");
assert!(
matches!(err, Error::Backend(_)),
"system 子の Err が伝播するはず: {err:?}"
);
assert_eq!(starts.load(Ordering::SeqCst), 1, "mic は一度起動される");
assert_eq!(
stops.load(Ordering::SeqCst),
1,
"system 失敗時に mic が stop されるはず"
);
}
#[test]
fn mix_mic_start_failure_is_immediate() {
let starts = Arc::new(AtomicU32::new(0));
let stops = Arc::new(AtomicU32::new(0));
let mic = Box::new(FailingStartBackend);
let system = Box::new(TrackingBackend {
starts: starts.clone(),
stops: stops.clone(),
});
let mut be = CompositeBackend::new(mic, system, 1.0, 1.0);
let (producer, _consumer) = raw_ring(RAW_RING_SAMPLES);
let sink = RawSink::new(producer, 48_000, 2);
assert!(be.start(sink).is_err(), "mic 起動失敗で即 Err のはず");
assert_eq!(
starts.load(Ordering::SeqCst),
0,
"mic 失敗なら system は起動されない"
);
be.stop();
be.stop();
}
#[test]
fn stream_delivers_mixed_chunks_end_to_end() {
use flexaudio_core::types::StreamConfig;
let mic = Box::new(ConstBackend::new(0.2, None));
let system = Box::new(ConstBackend::new(0.3, None));
let backend = Box::new(CompositeBackend::new(mic, system, 1.0, 1.0));
let mut stream = crate::Stream::open(StreamConfig::default(), backend).expect("open");
stream.start().expect("start");
let mut chunks = Vec::new();
let mut mixed = 0usize;
let deadline = Instant::now() + COLLECT_MAX_WAIT;
while Instant::now() < deadline && mixed < ONE_CHUNK_SAMPLES {
while let Some(c) = stream.poll_chunk() {
mixed += count_near(&c.data, 0.5);
chunks.push(c);
}
thread::sleep(Duration::from_millis(5));
}
stream.stop();
assert!(!chunks.is_empty(), "チャンクが届くはず");
for (i, c) in chunks.iter().enumerate() {
assert_eq!(c.frames, 960, "20ms@48k = 960 frame");
assert_eq!(c.data.len(), 960 * 2, "stereo interleaved");
if i > 0 {
assert!(c.seq > chunks[i - 1].seq, "seq は単調増加");
}
}
let all: Vec<f32> = chunks.iter().flat_map(|c| c.data.iter().copied()).collect();
let mixed_count = assert_only_allowed_values(&all, 0.5, 0.2, 0.3);
assert!(
mixed_count >= ONE_CHUNK_SAMPLES,
"合成値 0.5 が相応に出現するはず: {mixed_count}/{}",
all.len()
);
}
#[test]
fn drift_controller_moves_toward_lagging_side() {
let mut c = DriftController::new();
for _ in 0..50 {
c.update(0, 9_600);
}
assert!(
c.ratio > 1.0,
"system 側が溜まると r > 1.0 のはず: {}",
c.ratio
);
let mut c = DriftController::new();
for _ in 0..50 {
c.update(9_600, 0);
}
assert!(
c.ratio < 1.0,
"mic 側が溜まると r < 1.0 のはず: {}",
c.ratio
);
}
#[test]
fn drift_controller_clamps_at_ratio_limit() {
let mut c = DriftController::new();
for _ in 0..1_000 {
c.update(0, 10_000_000);
}
assert!(
(c.ratio - (1.0 + DRIFT_RATIO_LIMIT)).abs() < 1e-12,
"上側クランプちょうどで止まるはず: {}",
c.ratio
);
let mut c = DriftController::new();
for _ in 0..1_000 {
c.update(10_000_000, 0);
}
assert!(
(c.ratio - (1.0 - DRIFT_RATIO_LIMIT)).abs() < 1e-12,
"下側クランプちょうどで止まるはず: {}",
c.ratio
);
}
#[test]
fn drift_controller_slew_limits_change_per_update() {
let mut c = DriftController::new();
c.update(0, 10_000_000);
assert!(
(c.ratio - (1.0 + DRIFT_SLEW_PER_UPDATE)).abs() < 1e-12,
"1 回目の更新は slew 幅ちょうどで頭打ちのはず: {}",
c.ratio
);
c.update(0, 10_000_000);
assert!(
(c.ratio - (1.0 + 2.0 * DRIFT_SLEW_PER_UPDATE)).abs() < 1e-12,
"2 回目も 1 段階ずつ: {}",
c.ratio
);
let mut c = DriftController::new();
c.update(10_000_000, 0);
assert!(
(c.ratio - (1.0 - DRIFT_SLEW_PER_UPDATE)).abs() < 1e-12,
"反対向きも slew 幅ちょうど: {}",
c.ratio
);
}
#[test]
fn drift_controller_updates_only_at_interval() {
let mut c = DriftController::new();
c.on_output(DRIFT_UPDATE_INTERVAL_SAMPLES - 1, 0, 10_000_000);
assert!(
(c.ratio - 1.0).abs() < 1e-15,
"間隔未満では動かないはず: {}",
c.ratio
);
c.on_output(1, 0, 10_000_000);
assert!(c.ratio > 1.0, "間隔に達したら見直すはず: {}", c.ratio);
}
#[test]
fn stitcher_unity_ratio_is_passthrough() {
let mut st = LinearStitcher::new();
let src: Vec<f32> = (0..10).flat_map(|f| [f as f32, -(f as f32)]).collect();
let mut fifo = src.clone();
assert_eq!(st.producible(10, 1.0), 10);
let mut out = Vec::new();
st.pull(&mut fifo, 1.0, 10, &mut out);
assert_eq!(out, src, "r=1.0 はパススルーのはず");
assert!(fifo.is_empty(), "全量消費されるはず: {} 残り", fifo.len());
assert!(st.frac.abs() < 1e-12, "位相は 0 のまま: {}", st.frac);
}
#[test]
fn stitcher_interpolates_between_frames() {
let mut st = LinearStitcher::new();
let mut fifo: Vec<f32> = (0..5).flat_map(|f| [f as f32, f as f32 * 10.0]).collect();
assert_eq!(st.producible(5, 1.25), 4);
let mut out = Vec::new();
st.pull(&mut fifo, 1.25, 4, &mut out);
let expect = [0.0f32, 1.25, 2.5, 3.75];
for (k, &e) in expect.iter().enumerate() {
assert!(
(out[k * 2] - e).abs() < 1e-6,
"位置 {e} の補間値: {}",
out[k * 2]
);
assert!(
(out[k * 2 + 1] - e * 10.0).abs() < 1e-5,
"ch2 も同じ位置で補間: {}",
out[k * 2 + 1]
);
}
assert!(fifo.is_empty(), "全量消費されるはず: {} 残り", fifo.len());
assert!(st.frac.abs() < 1e-12, "位相: {}", st.frac);
}
fn sim_lane() -> ChildLane {
let (_producer, consumer) = raw_ring(16);
let normalizer = Normalizer::new(
SAMPLE_RATE,
CHANNELS,
OutputFormat {
sample_rate: SAMPLE_RATE,
channels: CHANNELS,
},
)
.expect("パススルー正規化器");
ChildLane {
consumer,
normalizer,
fifo: Vec::new(),
last_supply: Instant::now(),
}
}
fn sim_feed(lane: &mut ChildLane, value: f32, frames: usize) {
let new_len = lane.fifo.len() + frames * CHANNELS as usize;
lane.fifo.resize(new_len, value);
if lane.fifo.len() > FIFO_MAX_SAMPLES {
let excess = lane.fifo.len() - FIFO_MAX_SAMPLES;
lane.fifo.drain(..excess);
}
lane.last_supply = Instant::now();
}
struct DriftSimOutcome {
system_backlog: Vec<usize>,
mic_backlog: Vec<usize>,
final_ratio: f64,
}
fn run_drift_sim(ppm: f64, seconds: usize, fixed_unity_ratio: bool) -> DriftSimOutcome {
const TICK_FRAMES: usize = 960; const TICKS_PER_SEC: usize = 50;
let mut mic = sim_lane();
let mut system = sim_lane();
let mut drift = DriftCorrection::new();
let (producer, mut consumer) = raw_ring(RAW_RING_SAMPLES);
let mut sink = RawSink::new(producer, SAMPLE_RATE, CHANNELS);
let mut mixed: Vec<f32> = Vec::with_capacity(FIFO_MAX_SAMPLES);
let mut scratch = vec![0.0f32; RAW_RING_SAMPLES];
let mut system_carry = 0.0f64;
let mut outcome = DriftSimOutcome {
system_backlog: Vec::new(),
mic_backlog: Vec::new(),
final_ratio: 1.0,
};
for tick in 0..seconds * TICKS_PER_SEC {
sim_feed(&mut mic, 0.2, TICK_FRAMES);
system_carry += TICK_FRAMES as f64 * (1.0 + ppm * 1e-6);
let system_frames = system_carry as usize;
system_carry -= system_frames as f64;
sim_feed(&mut system, 0.3, system_frames);
mix_and_push(
&mut mic,
&mut system,
1.0,
1.0,
&mut drift,
&mut sink,
&mut mixed,
);
if fixed_unity_ratio {
drift.controller.ratio = 1.0;
drift.controller.ema_diff = 0.0;
}
let got = consumer.pop_slice(&mut scratch);
for &s in &scratch[..got] {
assert!(
(s - 0.5).abs() < VALUE_TOL,
"定常シミュレーションの出力は合成値 0.5 だけのはず(ゼロ埋め・片側値は \
出ない): {s}"
);
}
if (tick + 1) % TICKS_PER_SEC == 0 {
outcome.system_backlog.push(system.fifo.len());
outcome.mic_backlog.push(mic.fifo.len());
}
}
outcome.final_ratio = drift.controller.ratio;
outcome
}
#[test]
fn mix_drift_sim_plus_300ppm_stays_bounded() {
let corrected = run_drift_sim(300.0, 60, false);
let uncorrected = run_drift_sim(300.0, 60, true);
for (i, w) in uncorrected.system_backlog.windows(2).enumerate() {
assert!(
w[1] > w[0],
"補正なしでは残量が単調増大するはず: {i} 秒目 {} → {}",
w[0],
w[1]
);
}
let max_corrected = corrected.system_backlog.iter().copied().max().unwrap();
assert!(
max_corrected < FIFO_MAX_SAMPLES,
"補正ありなら安全弁に届かないはず: 最大 {max_corrected}"
);
let last_c = *corrected.system_backlog.last().unwrap();
let last_u = *uncorrected.system_backlog.last().unwrap();
assert!(
last_c < last_u,
"補正ありの残量 {last_c} < 補正なし {last_u} のはず"
);
let sb = &corrected.system_backlog;
let early = sb[9] - sb[0];
let late = sb[59] - sb[50];
assert!(
late < early,
"比率が追い付くにつれ伸びが減速するはず: 序盤 +{early} vs 終盤 +{late}"
);
assert!(
corrected.final_ratio > 1.0 + 2e-5,
"r は 1.0 より上へ動くはず: {}",
corrected.final_ratio
);
assert!(
corrected.final_ratio <= 1.0 + DRIFT_RATIO_LIMIT + 1e-12,
"r はクランプ内: {}",
corrected.final_ratio
);
}
#[test]
fn mix_drift_sim_minus_300ppm_no_steady_zero_fill() {
let corrected = run_drift_sim(-300.0, 60, false);
let uncorrected = run_drift_sim(-300.0, 60, true);
for (i, w) in uncorrected.mic_backlog.windows(2).enumerate() {
assert!(
w[1] > w[0],
"補正なしでは mic 残量が単調増大するはず: {i} 秒目 {} → {}",
w[0],
w[1]
);
}
let max_corrected = corrected.mic_backlog.iter().copied().max().unwrap();
assert!(
max_corrected < FIFO_MAX_SAMPLES,
"補正ありなら安全弁に届かないはず: 最大 {max_corrected}"
);
let last_c = *corrected.mic_backlog.last().unwrap();
let last_u = *uncorrected.mic_backlog.last().unwrap();
assert!(
last_c < last_u,
"補正ありの mic 残量 {last_c} < 補正なし {last_u} のはず"
);
let max_sys = corrected.system_backlog.iter().copied().max().unwrap();
assert!(
max_sys < ONE_CHUNK_SAMPLES,
"system 側は溜まらないはず: 最大 {max_sys}"
);
assert!(
corrected.final_ratio < 1.0 - 2e-5,
"r は 1.0 より下へ動くはず: {}",
corrected.final_ratio
);
assert!(
corrected.final_ratio >= 1.0 - DRIFT_RATIO_LIMIT - 1e-12,
"r はクランプ内: {}",
corrected.final_ratio
);
}
}