use cpal::traits::DeviceTrait;
use cpal::{Device, Error, ErrorKind, SampleFormat, Stream, StreamConfig};
use ringbuf::traits::{Consumer, Producer, Split};
use ringbuf::{HeapCons, HeapProd, HeapRb};
use std::marker::PhantomData;
use std::sync::{Arc, Mutex, PoisonError};
pub mod clock;
pub mod mixer;
pub mod scheduler;
pub mod types;
pub mod utils;
pub use clock::PlayClock;
pub use mixer::{ClearSignal, MixCommand, Mixer, MixerHandle, Voices};
pub use scheduler::Scheduler;
pub use types::{OutputType, SampleRate, SampleType};
pub use utils::{
default_device, device_name, find_device, get_host_by_id, get_host_by_name, get_host_in_name,
list_device_names, list_devices, list_hosts,
};
pub(crate) const DEFAULT_BUFFER_FRAMES: i32 = 1024;
const COMMAND_QUEUE_LEN: usize = 256;
pub struct OutputClass<S: SampleType> {
channels: u16,
device: Device,
out_type: OutputType,
buffer_size: Option<i32>,
sample_rate: SampleRate,
stream: Option<Stream>,
stream_config: StreamConfig,
commands: Arc<Mutex<HeapProd<MixCommand<S>>>>,
consumer: Option<HeapCons<S>>,
mixer: MixerHandle,
clock: PlayClock,
sample_type: PhantomData<S>,
}
pub(crate) fn resample_into<S: SampleType>(
samples: &[S],
frames: usize,
channels: usize,
from: u32,
to: u32,
out: &mut Vec<S>,
) {
out.clear();
if frames == 0 || from == 0 || to == 0 {
return;
}
let ratio = to as f64 / from as f64;
let out_frames = ((frames as f64) * ratio).round().max(1.0) as usize;
let last = frames - 1;
out.reserve(out_frames * channels);
for frame in 0..out_frames {
let position = frame as f64 / ratio;
let left = (position.floor() as usize).min(last);
let right = (left + 1).min(last);
let t = (position - position.floor()) as f32;
for channel in 0..channels {
let a = samples[left * channels + channel].to_f32();
let b = samples[right * channels + channel].to_f32();
out.push(S::from_f32(a + (b - a) * t));
}
}
}
pub(crate) fn fallback_rates(wanted: SampleRate, native: Option<SampleRate>) -> Vec<SampleRate> {
let mut higher: Vec<SampleRate> = SampleRate::ALL
.into_iter()
.filter(|rate| rate.hz() > wanted.hz() && Some(*rate) != native)
.collect();
higher.sort_by_key(|rate| rate.hz() - wanted.hz());
higher
}
impl<S: SampleType> OutputClass<S> {
pub fn new(device: Option<Device>, out_type: OutputType, channels: u16, sample_rate:SampleRate, buffer_size: Option<i32>) -> Self {
let buffer_size_form = match buffer_size {
Some(frames) => cpal::BufferSize::Fixed(frames as u32),
None => cpal::BufferSize::Default,
};
let frames = buffer_size.unwrap_or(DEFAULT_BUFFER_FRAMES).max(1) as usize;
let stream_config = StreamConfig {
channels,
sample_rate: sample_rate as u32,
buffer_size: buffer_size_form,
};
let capacity = frames * channels.max(1) as usize;
let (audio_producer, consumer) = HeapRb::<S>::new(capacity).split();
let (commands, pending) = HeapRb::<MixCommand<S>>::new(COMMAND_QUEUE_LEN).split();
let mixer = MixerHandle::spawn(Mixer::new(pending, audio_producer));
OutputClass {
channels,
device: device
.unwrap_or_else(|| default_device().expect("no default output device available")),
out_type,
buffer_size,
sample_rate,
stream: None,
stream_config,
commands: Arc::new(Mutex::new(commands)),
consumer: Some(consumer),
mixer,
clock: PlayClock::new(sample_rate as u32),
sample_type: PhantomData,
}
}
pub fn channels(&self) -> u16 {
self.channels
}
pub fn device(&self) -> Device {
self.device.clone()
}
pub fn name(&self) -> String {
device_name(&self.device)
}
pub fn buffer_size(&self) -> Option<i32> {
self.buffer_size
}
pub fn sample_rate(&self) -> i32 {
self.sample_rate as i32
}
pub fn sample_format(&self) -> SampleFormat {
S::format()
}
pub fn sample_type(&self) -> SampleFormat {
S::format()
}
pub fn out_type(&self) -> OutputType {
self.out_type
}
fn data_callback(
data: &mut [S],
buffer: &mut HeapCons<S>,
clear: &ClearSignal,
cleared: &mut usize,
clock: &PlayClock,
channels: usize,
latency: std::time::Duration,
) {
let channels = channels.max(1);
let epoch = clear.epoch();
if epoch != *cleared {
let dropped = buffer.clear();
data.fill(S::SILENCE);
if clear.acked() == epoch {
*cleared = epoch;
}
clock.advance((dropped / channels) as u64, latency);
return;
}
let read = buffer.pop_slice(data);
if read < data.len() {
data[read..].fill(S::SILENCE);
}
clock.advance((read / channels) as u64, latency);
}
fn opens_at(&self, rate: SampleRate) -> bool {
let config = StreamConfig {
sample_rate: rate.hz(),
..self.stream_config
};
self.device
.build_output_stream(
config,
move |_data: &mut [S], _info: &cpal::OutputCallbackInfo| {},
|_error| {},
None,
)
.is_ok()
}
fn usable_rate(&self) -> SampleRate {
let mixing_at = self.device.default_output_config().ok().map(|config| config.sample_rate());
if mixing_at == Some(self.sample_rate.hz()) || self.opens_at(self.sample_rate) {
return self.sample_rate;
}
let native = mixing_at
.and_then(SampleRate::from_hz)
.filter(|rate| *rate != self.sample_rate);
if let Some(native) = native.filter(|rate| self.opens_at(*rate)) {
return native;
}
fallback_rates(self.sample_rate, native)
.into_iter()
.find(|rate| self.opens_at(*rate))
.unwrap_or(self.sample_rate)
}
fn retune(&mut self, rate: SampleRate) {
self.sample_rate = rate;
self.stream_config.sample_rate = rate.hz();
self.clock.retune(rate.hz());
}
pub fn build_stream(&mut self) -> Result<&Stream, Error> {
let usable = self.usable_rate();
if usable != self.sample_rate {
self.retune(usable);
}
let mut buffer = self.consumer.take().ok_or_else(|| {
Error::with_message(
ErrorKind::UnsupportedOperation,
"stream already built for this output",
)
})?;
let clear = self.mixer.clear_signal();
let mut cleared = clear.epoch();
let clock = self.clock.clone();
let channels = usize::from(self.channels);
let stream = self.device.build_output_stream(
self.stream_config,
move |data: &mut [S], info: &cpal::OutputCallbackInfo| {
let stamp = info.timestamp();
let latency = stamp.playback.duration_since(stamp.callback);
Self::data_callback(data, &mut buffer, &clear, &mut cleared, &clock, channels, latency)
},
Self::err_fn,
None,
)?;
self.clock.opened();
self.stream = Some(stream);
Ok(self.stream.as_ref().unwrap())
}
pub fn add_samples(&mut self, samples: &[S], index: usize) -> Result<usize, Error> {
let command = MixCommand {
index,
samples: samples.to_vec(),
voice: None,
};
self.commands.lock().unwrap_or_else(PoisonError::into_inner).try_push(command).map_err(|_| {
Error::with_message(ErrorKind::ResourceExhausted, "mixer command queue is full")
})?;
Ok(index + samples.len())
}
pub fn add_samples_time(&mut self, samples: &[S], time: usize) -> Result<usize, Error> {
let index = time * self.sample_rate as usize * self.channels.max(1) as usize;
let command = MixCommand {
index,
samples: samples.to_vec(),
voice: None,
};
self.commands.lock().unwrap_or_else(PoisonError::into_inner).try_push(command).map_err(|_| {
Error::with_message(ErrorKind::ResourceExhausted, "mixer command queue is full")
})?;
Ok(index + samples.len())
}
pub fn stop(&mut self) {
self.mixer.clear_samples();
}
pub fn close(&mut self) {
self.stream.take();
self.clock.closed();
}
pub fn clock(&self) -> PlayClock {
self.clock.clone()
}
pub fn scheduler(&self) -> Scheduler<S> {
Scheduler::new(
Arc::clone(&self.commands),
self.mixer.voices(),
self.clock.clone(),
self.channels,
self.sample_rate as u32,
)
}
pub fn mixer(&self) -> &MixerHandle {
&self.mixer
}
pub fn align_samples(&self, samples: &[S], sample_rate: SampleRate, channels: u16, interleaved: bool) -> Result<Vec<S>, Error> {
if channels == 0 {
return Err(Error::with_message(
ErrorKind::InvalidInput,
"sample data must have at least one channel",
));
}
let src_channels = channels as usize;
if samples.len() % src_channels != 0 {
return Err(Error::with_message(
ErrorKind::InvalidInput,
"sample count is not a whole number of frames",
));
}
if sample_rate == self.sample_rate && channels == self.channels && interleaved {
return Ok(samples.to_vec());
}
let dst_channels = self.channels.max(1) as usize;
let frames = samples.len() / src_channels;
let mut buffer = if interleaved || src_channels == 1 {
samples.to_vec()
} else {
Self::weave(samples, frames, src_channels)
};
if src_channels != dst_channels {
buffer = Self::map_channels(&buffer, frames, src_channels, dst_channels);
}
if sample_rate != self.sample_rate {
buffer = Self::resample(
&buffer,
frames,
dst_channels,
sample_rate as u32,
self.sample_rate as u32,
);
}
Ok(buffer)
}
fn weave(samples: &[S], frames: usize, channels: usize) -> Vec<S> {
let mut out = Vec::with_capacity(samples.len());
for frame in 0..frames {
for channel in 0..channels {
out.push(samples[(channel * frames) + frame]);
}
}
out
}
fn map_channels(samples: &[S], frames: usize, src_channels: usize, dst_channels: usize) -> Vec<S> {
let mut out = Vec::with_capacity(frames * dst_channels);
for frame in 0..frames {
let start = frame * src_channels;
if src_channels == 1 {
out.extend(std::iter::repeat(samples[start]).take(dst_channels));
} else if dst_channels == 1 {
let mut sum = S::SILENCE;
for sample in &samples[start..start + src_channels] {
sum = sum.mix(*sample);
}
out.push(sum);
} else {
for channel in 0..dst_channels {
out.push(if channel < src_channels {
samples[start + channel]
} else {
S::SILENCE
});
}
}
}
out
}
fn resample(samples: &[S], frames: usize, channels: usize, from: u32, to: u32) -> Vec<S> {
let mut out = Vec::new();
resample_into(samples, frames, channels, from, to, &mut out);
out
}
fn err_fn(err: cpal::Error) {
eprintln!("audio output stream error: {}", err);
}
}
#[cfg(test)]
mod rate_tests {
use super::*;
#[test]
fn a_fallback_only_reaches_for_higher_rates() {
let order = fallback_rates(SampleRate::Hz44_1k, None);
assert_eq!(order[0], SampleRate::Hz48k, "the next rate up, first");
assert_eq!(order[1], SampleRate::Hz88_2k);
assert!(order.iter().all(|rate| rate.hz() > SampleRate::Hz44_1k.hz()), "{order:?}");
assert!(!order.contains(&SampleRate::Hz44_1k), "the rate that failed is not retried");
}
#[test]
fn a_fallback_leaves_out_the_rate_already_tried() {
let order = fallback_rates(SampleRate::Hz44_1k, Some(SampleRate::Hz48k));
assert!(!order.contains(&SampleRate::Hz48k));
assert_eq!(order[0], SampleRate::Hz88_2k, "the next one up after it");
}
#[test]
fn the_top_rate_has_nowhere_to_fall_back_to() {
assert!(fallback_rates(SampleRate::Hz352_8, None).is_empty());
}
#[test]
fn an_output_opens_at_a_rate_its_device_will_take() {
let Some(device) = default_device() else { return };
let channels = device.default_output_config().map(|config| config.channels()).unwrap_or(2);
for asked in [SampleRate::Hz44_1k, SampleRate::Hz48k, SampleRate::Hz88_2k] {
let mut output =
OutputClass::<f32>::new(Some(device.clone()), OutputType::WASAPI, channels, asked, Some(512));
let built = output.build_stream().is_ok();
let got = output.sample_rate() as u32;
if built {
assert!(
SampleRate::from_hz(got).is_some(),
"opened at {got} Hz, which is not one of atome's rates"
);
} else {
assert_eq!(got, asked.hz(), "a refused output should keep the rate it was asked for");
}
}
}
}