#![allow(dead_code)]
use super::{DaqApiMethods, Stream, StreamMetaData};
use crate::config::{self, *};
use crate::daq::{self, *};
use crate::daq::{DaqApiDescriptor, streamdata::*};
use crate::tools::find_unused_buf;
use cpal::{
BufferSize, Device, Host, Sample, SampleFormat, SampleRate, SupportedBufferSize,
SupportedStreamConfig,
traits::{DeviceTrait, HostTrait, StreamTrait},
};
use crossbeam::channel::{Receiver, Sender};
use itertools::Itertools;
use num::ToPrimitive;
use parking_lot::RwLock;
use reinterpret::reinterpret_slice;
use snafu::prelude::*;
use std::collections::VecDeque;
use std::collections::btree_map::OccupiedEntry;
use std::fmt::Debug;
use std::sync::Arc;
type Result<T> = std::result::Result<T, StreamMgrError>;
impl From<DataType> for cpal::SampleFormat {
fn from(dt: DataType) -> cpal::SampleFormat {
match dt {
DataType::F64 => SampleFormat::F64,
DataType::F32 => SampleFormat::F32,
DataType::I8 => SampleFormat::I8,
DataType::I16 => SampleFormat::I16,
DataType::I32 => SampleFormat::I32,
DataType::I24 => SampleFormat::I24,
}
}
}
impl TryFrom<cpal::SampleFormat> for DataType {
type Error = ();
fn try_from(sf: cpal::SampleFormat) -> std::result::Result<DataType, ()> {
match sf {
SampleFormat::F64 => Ok(DataType::F64),
SampleFormat::F32 => Ok(DataType::F32),
SampleFormat::I8 => Ok(DataType::I8),
SampleFormat::I16 => Ok(DataType::I16),
SampleFormat::I32 => Ok(DataType::I32),
SampleFormat::I24 => Ok(DataType::I24),
SampleFormat::I64 => Err(()),
SampleFormat::U8 => Err(()),
SampleFormat::U16 => Err(()),
SampleFormat::U24 => Err(()),
SampleFormat::U32 => Err(()),
SampleFormat::U64 => Err(()),
SampleFormat::DsdU8 => Err(()),
SampleFormat::DsdU16 => Err(()),
SampleFormat::DsdU32 => Err(()),
_ => Err(()),
}
}
}
pub struct CpalApi {
host: cpal::Host,
}
impl Debug for CpalApi {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("CpalApi").finish()
}
}
pub struct CpalStream {
stream: cpal::Stream,
metadata: Arc<StreamMetaData>,
noutchannels: usize,
status: Arc<RwLock<StreamStatus>>,
}
impl Debug for CpalStream {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("CpalStream")
.field("metadata", &self.metadata)
.field("noutchannels", &self.noutchannels)
.field("status", &self.status)
.finish()
}
}
impl Stream for CpalStream {
fn inMetaData(&self) -> Option<Arc<StreamMetaData>> {
Some(self.metadata.clone())
}
fn status(&self, _dir: StreamDirection) -> StreamStatus {
self.status.read().clone()
}
fn outMetaData(&self) -> Option<Arc<StreamMetaData>> {
Some(self.metadata.clone())
}
}
impl DaqApiMethods for CpalApi {
fn getDeviceInfo(&self) -> Result<Vec<DeviceInfo>> {
let samplerates_set1 = [
1000, 2000, 4000, 8000, 12000, 16000, 24000, 48000, 96000, 192000, 384000,
];
let samplerates_set2 = [11025, 22050, 44100, 88200];
let mut samplerates_set =
Vec::from_iter(samplerates_set1.iter().chain(samplerates_set2.iter()));
samplerates_set.sort();
let samplerates_set = Vec::from_iter(samplerates_set.iter().map(|i| **i as Flt));
let mut devs = vec![];
for dev in self.host.devices()? {
let mut iChannelCount = 0;
let mut oChannelCount = 0;
let mut avSampleRates = samplerates_set.clone();
let mut avFramesPerBlock = vec![256_usize, 512, 1024, 2048, 4096, 8192];
let mut sample_formats = vec![];
if let Ok(icfg) = dev.supported_input_configs() {
for icfg in icfg {
let thissf = icfg.sample_format();
if thissf.is_uint() {
continue;
}
sample_formats.push(icfg.sample_format());
avSampleRates.retain(|sr| {
*sr >= (icfg.min_sample_rate() as Flt)
&& *sr <= (icfg.max_sample_rate() as Flt)
});
if let SupportedBufferSize::Range { min, max } = icfg.buffer_size() {
avFramesPerBlock.retain(|i| i >= &(*min as usize));
avFramesPerBlock.retain(|i| i <= &(*max as usize));
}
iChannelCount = icfg.channels() as u8;
}
}
if let Ok(ocfg) = dev.supported_output_configs() {
for ocfg in ocfg {
let thissf = ocfg.sample_format();
if thissf.is_uint() {
continue;
}
sample_formats.push(thissf);
avSampleRates.retain(|sr| *sr >= (ocfg.min_sample_rate() as Flt));
avSampleRates.retain(|sr| *sr <= (ocfg.max_sample_rate() as Flt));
if let SupportedBufferSize::Range { min, max } = ocfg.buffer_size() {
avFramesPerBlock.retain(|i| i >= &(*min as usize));
avFramesPerBlock.retain(|i| i <= &(*max as usize));
}
oChannelCount = ocfg.channels() as u8;
}
}
let dtypes: Vec<DataType> = sample_formats
.iter()
.unique()
.filter_map(|i| (*i).try_into().ok())
.collect();
if sample_formats.is_empty() || dtypes.is_empty() {
continue;
}
assert!(!dtypes.is_empty());
let prefDataType = match dtypes.iter().position(|d| d == &DataType::F32) {
Some(idx) => dtypes[idx],
None => dtypes[dtypes.len() - 1],
};
let prefSampleRate = 48000.;
if iChannelCount == oChannelCount && oChannelCount == 0 {
continue;
}
if avSampleRates.is_empty() {
continue;
}
devs.push(DeviceInfo {
api: DaqApiDescriptor::Cpal,
device_name: dev.description()?.name().into(),
avDataTypes: dtypes,
prefDataType,
avSampleRates,
prefSampleRate,
avFramesPerBlock,
prefFramesPerBlock: 2048,
iChannelCount,
oChannelCount,
avInputRanges: vec![(-1., 1.)],
avOutputRanges: vec![(-1., 1.)],
hasInputIEPE: false,
hasDuplexMode: false,
hasInputACCouplingSwitch: false,
hasInputTrigger: false,
hasInternalOutputMonitor: false,
duplexModeForced: false,
physicalIOQty: Qty::Number,
});
}
Ok(devs)
}
fn startInputOrDuplexStream(
&self,
stype: StreamType,
devinfo: &DeviceInfo,
conf: &DaqConfig,
sender: Sender<InStreamMsg>,
_receiver: Option<Receiver<Arc<RawStreamData>>>,
) -> Result<Box<dyn Stream>> {
ensure!(
stype != StreamType::Duplex,
DAQConfigSnafu {
msg: "Duplex mode not supported for CPAL api"
}
);
for cpaldev in self.host.devices()? {
if cpaldev.description()?.name() == conf.device_name {
let supported_config = (match stype {
StreamType::Duplex => DAQConfigSnafu {
msg: "Duplex mode not supported for CPAL api",
}
.fail(),
StreamType::Input => CpalApi::create_cpal_config(
stype,
devinfo,
conf,
&cpaldev,
cpaldev.supported_input_configs()?,
),
StreamType::Output => CpalApi::create_cpal_config(
stype,
devinfo,
conf,
&cpaldev,
cpaldev.supported_output_configs()?,
),
})?;
let framesPerBlock = conf.framesPerBlock(devinfo);
let sf = supported_config.sample_format();
let mut config: cpal::StreamConfig = supported_config.config();
config.buffer_size = BufferSize::Fixed(framesPerBlock as u32);
let meta = StreamMetaData::new(
&conf.enabledInChannels(),
conf.dtype,
(supported_config.sample_rate() as Flt)
.try_into()
.context(DAQConfigValidationSnafu {})?,
framesPerBlock,
Qty::Number,
);
let meta = Arc::new(meta);
let (stream, status) = CpalApi::build_input_stream(
meta.clone(),
sf,
&config,
&cpaldev,
sender,
conf.enabledInchannelsList(),
framesPerBlock,
)?;
stream.play()?;
*status.write() = StreamStatus::newRunning();
return Ok(Box::new(CpalStream {
stream,
metadata: meta,
noutchannels: 0,
status,
}));
}
}
DeviceNotAvailableSnafu {
device_name: &devinfo.device_name,
}
.fail()
}
fn startOutputStream(
&self,
dev: &DeviceInfo,
cfg: &DaqConfig,
receiver: Receiver<Arc<RawStreamData>>,
) -> Result<Box<dyn Stream>> {
let (device, cpalconfig, sampleformat, framesPerBlock) =
self.getCPALOutputConfig(dev, cfg)?;
let (stream, status) = Self::build_output_stream(
sampleformat,
&cpalconfig,
&device,
receiver,
&cfg.outchannel_config,
framesPerBlock,
)?;
stream.play()?;
*status.write() = StreamStatus::newRunning();
let dtype = DataType::try_from(sampleformat).map_err(|_| {
DataTypeNotSupportedSnafu {
dtype: format!("{sampleformat:?}"),
}
.build()
})?;
let md = StreamMetaData::new(
&cfg.enabledOutChannels(),
dtype,
(cpalconfig.sample_rate as Flt)
.try_into()
.context(DAQConfigValidationSnafu {})?,
framesPerBlock,
Qty::Number,
);
let md = Arc::new(md);
let str = Box::new(CpalStream {
stream,
metadata: md,
noutchannels: cpalconfig.channels as usize,
status,
});
Ok(str)
}
fn as_any(&self) -> &dyn std::any::Any {
self
}
}
impl CpalApi {
pub fn new() -> CpalApi {
CpalApi {
host: cpal::default_host(),
}
}
fn create_errfcn(
send_ch: Option<Sender<InStreamMsg>>,
status: Arc<RwLock<StreamStatus>>,
device_name: String,
is_input: bool,
) -> impl FnMut(cpal::Error) {
move |err: cpal::Error| {
let serr = match err.kind() {
cpal::ErrorKind::DeviceNotAvailable => StreamError::DeviceNotAvailable {
device_name: device_name.clone(),
},
cpal::ErrorKind::StreamInvalidated => StreamError::SystemError {
msg: "stream invalidated".into(),
},
cpal::ErrorKind::Xrun => {
if is_input {
StreamError::InputOverrun {}
} else {
StreamError::OutputUnderrun {}
}
}
_ => StreamError::SystemError {
msg: err.to_string(),
},
};
if let Some(sender) = &send_ch {
sender.send(InStreamMsg::StreamError(serr.clone())).unwrap();
}
*status.write() = StreamStatus::Error { e: serr };
}
}
fn create_incallback<T>(
meta: Arc<StreamMetaData>,
config: &cpal::StreamConfig,
sender: Sender<InStreamMsg>,
framesPerBlock: usize,
en_inchannels: Vec<usize>,
) -> impl FnMut(&[T], &cpal::InputCallbackInfo) + use<T>
where
T: 'static + Sample + ToPrimitive,
{
let tot_inch = config.channels as usize;
let mut q = VecDeque::<T>::with_capacity(2 * tot_inch * framesPerBlock);
let mut ctr = 0;
let mut bufs: VecDeque<Arc<RawStreamData>> = VecDeque::with_capacity(10);
move |input: &[T], _: &cpal::InputCallbackInfo| {
q.extend(input);
while q.len() > tot_inch * framesPerBlock {
let mut buf = find_unused_buf(&mut bufs).unwrap_or_else(|| {
let buf: Vec<T> =
vec![Sample::EQUILIBRIUM; en_inchannels.len() * framesPerBlock];
Arc::new(RawStreamData::new(buf))
});
let out = Arc::get_mut(&mut buf)
.expect("Buf should be mutable")
.get_mut();
for (i, ch) in en_inchannels.iter().enumerate() {
let in_iterator = q.iter().skip(*ch).step_by(tot_inch);
let out_iterator = out.iter_mut().skip(i).step_by(en_inchannels.len());
out_iterator.zip(in_iterator).for_each(|(o, i)| {
*o = *i;
});
}
q.drain(0..framesPerBlock * tot_inch);
bufs.push_front(buf.clone());
let streamdata = Arc::new(InStreamData::newFromRaw(ctr, meta.clone(), buf));
sender.send(InStreamMsg::InStreamData(streamdata)).unwrap();
ctr += 1;
}
}
}
fn build_input_stream(
meta: Arc<StreamMetaData>,
sf: cpal::SampleFormat,
config: &cpal::StreamConfig,
device: &cpal::Device,
sender: Sender<InStreamMsg>,
en_inchannels: Vec<usize>,
framesPerBlock: usize,
) -> Result<(cpal::Stream, Arc<RwLock<StreamStatus>>)> {
let status = Arc::new(RwLock::new(StreamStatus::NotRunning {}));
let device_name = device
.description()
.map(|d| d.name().to_string())
.unwrap_or_else(|_| "unknown device".to_string());
let errfcn =
CpalApi::create_errfcn(Some(sender.clone()), status.clone(), device_name, true);
macro_rules! build_stream {
($($cpaltype:pat => $rtype:ty),*) => {
match sf {
$(
$cpaltype => {
let icb = CpalApi::create_incallback::<$rtype>(
meta,
&config, sender, framesPerBlock, en_inchannels);
device.build_input_stream(
*config,
icb,
errfcn,
None)?
}),*,
_ => {return DAQConfigSnafu {msg: format!("Unsupported sample format: {}", sf)}.fail();}
}
};
}
let stream: cpal::Stream = build_stream!(
SampleFormat::I8 => i8,
SampleFormat::I16 => i16,
SampleFormat::I32 => i32,
SampleFormat::F32 => f32
);
Ok((stream, status))
}
fn create_outcallback<T>(
config: &cpal::StreamConfig,
streamstatus: Arc<RwLock<StreamStatus>>,
receiver: Receiver<Arc<RawStreamData>>,
ch_config: &[DaqChannel],
framesPerBlock: usize,
) -> impl FnMut(&mut [T], &cpal::OutputCallbackInfo) + use<T>
where
T: 'static + Sample + Debug,
{
let number_total_out_channels: usize = config.channels as usize;
let number_enabled_out_channels = ch_config.iter().filter(|ch| ch.enabled).count();
let disabled_ch = DaqChannel::default();
let disabled_repeater = std::iter::repeat(&disabled_ch);
let enabled_outch = ch_config.iter().chain(disabled_repeater);
let enabled_outch: Vec<bool> = (0..number_total_out_channels)
.zip(enabled_outch)
.map(|(_, b)| b.enabled)
.collect();
assert_eq!(enabled_outch.len(), number_total_out_channels);
let mut callback_ctr: usize = 0;
let mut q = VecDeque::<T>::with_capacity(2 * number_total_out_channels * framesPerBlock);
move |outdata, _info: &_| {
let nsamples_asked =
(outdata.len() / number_total_out_channels) * number_enabled_out_channels;
let status = streamstatus.read();
callback_ctr += 1;
let mut setToEquilibrium = || {
outdata.iter_mut().for_each(|v| {
*v = Sample::EQUILIBRIUM;
})
};
match *status {
StreamStatus::NotRunning {} | StreamStatus::Error { .. } => {
setToEquilibrium();
return;
}
_ => {}
}
if q.len() < nsamples_asked {
for dat in receiver.try_iter() {
let slice = dat.get_ref::<T>();
if let StreamStatus::Running { .. } = *status {
q.extend(slice);
}
}
}
if q.len() >= nsamples_asked {
let out_chunks = outdata.iter_mut().chunks(number_total_out_channels);
let siggen_chunks = q
.drain(..nsamples_asked)
.chunks(number_enabled_out_channels);
for (och, ich) in out_chunks.into_iter().zip(&siggen_chunks) {
let mut sig_frame_iter = ich.into_iter();
och.into_iter().zip(&enabled_outch).for_each(|(o, en)| {
if *en {
*o = sig_frame_iter.next().unwrap();
} else {
*o = Sample::EQUILIBRIUM;
}
});
}
} else if callback_ctr <= 2 {
setToEquilibrium();
} else {
*streamstatus.write() = StreamStatus::Error {
e: StreamError::OutputUnderrun {},
};
setToEquilibrium();
}
}
}
fn build_output_stream(
sf: cpal::SampleFormat,
config: &cpal::StreamConfig,
device: &cpal::Device,
receiver: Receiver<Arc<RawStreamData>>,
ch_config: &[DaqChannel],
framesPerBlock: usize,
) -> Result<(cpal::Stream, Arc<RwLock<StreamStatus>>)> {
let status = Arc::new(RwLock::new(StreamStatus::NotRunning {}));
let device_name = device
.description()
.map(|d| d.name().to_string())
.unwrap_or_else(|_| "unknown device".to_string());
let err_cb = CpalApi::create_errfcn(None, status.clone(), device_name, false);
macro_rules! build_stream {
($($cpaltype:pat => $rtype:ty),*) => {
match sf {
$(
$cpaltype => {
let outcallback = CpalApi::create_outcallback::<$rtype>(config, status.clone(), receiver, ch_config, framesPerBlock);
device.build_output_stream(
*config,
outcallback,
err_cb,
None)?
}),*,
_ => {
return DAQConfigSnafu{msg: format!("Unsupported sample format '{}'", sf)}.fail();}
}
};
}
let stream: cpal::Stream = build_stream!(
SampleFormat::I8 => i8,
SampleFormat::I16 => i16,
SampleFormat::I32 => i32,
SampleFormat::F32 => f32
);
Ok((stream, status))
}
fn create_cpal_config<T>(
st: StreamType,
devinfo: &DeviceInfo,
conf: &DaqConfig,
_dev: &cpal::Device,
conf_iterator: T,
) -> Result<cpal::SupportedStreamConfig>
where
T: Iterator<Item = cpal::SupportedStreamConfigRange>,
{
let nchannels = match st {
StreamType::Input => devinfo.iChannelCount,
StreamType::Output => devinfo.oChannelCount,
_ => unreachable!(),
};
for cpalconf in conf_iterator {
if cpalconf.sample_format() == conf.dtype.into() {
if cpalconf.channels() == (nchannels as u16) {
let requested_sr = conf.sampleRate(devinfo);
if (cpalconf.min_sample_rate() as Flt) <= requested_sr
&& (cpalconf.max_sample_rate() as Flt) >= requested_sr
{
let requested_fpb = conf.framesPerBlock(devinfo) as u32;
if let SupportedBufferSize::Range { min, max } = cpalconf.buffer_size()
&& (*min > requested_fpb || *max < requested_fpb)
{
return BufferSizeNotSupportedSnafu {
min: *min,
max: *max,
requested: requested_fpb,
}
.fail();
}
return Ok(cpalconf.with_sample_rate(requested_sr as u32));
}
}
}
}
APISpecificSnafu {
msg: "specified DAQ configuration is not available for device",
}
.fail()
}
pub fn startDefaultInputStream(&self, sender: Sender<InStreamMsg>) -> Result<Box<dyn Stream>> {
match self.host.default_input_device() {
Some(device) => {
if let Ok(config) = device.default_input_config() {
let framesPerBlock: usize = 4096;
let final_config = cpal::StreamConfig {
channels: config.channels(),
sample_rate: config.sample_rate(),
buffer_size: cpal::BufferSize::Fixed(framesPerBlock as u32),
};
let en_inchannels = Vec::from_iter((0..config.channels()).map(|i| i as usize));
let sf = config.sample_format();
let dtype = DataType::try_from(sf).map_err(|_| {
DataTypeNotSupportedSnafu {
dtype: format! {"{sf}"},
}
.build()
})?;
let daqchannels =
Vec::from_iter((0..final_config.channels).map(|i| {
DaqChannel::defaultAudio(format!("Unnamed input channel {i}"))
}));
let metadata = StreamMetaData::new(
&daqchannels,
dtype,
(config.sample_rate() as Flt)
.try_into()
.context(DAQConfigValidationSnafu {})?,
framesPerBlock,
Qty::Number,
);
let metadata = Arc::new(metadata);
let (stream, status) = CpalApi::build_input_stream(
metadata.clone(),
sf,
&final_config,
&device,
sender,
en_inchannels,
framesPerBlock,
)?;
stream.play()?;
*status.write() = StreamStatus::newRunning();
Ok(Box::new(CpalStream {
stream,
metadata,
noutchannels: 0,
status,
}))
} else {
NoDefaultDeviceFoundSnafu {}.fail()
}
}
_ => NoDefaultDeviceFoundSnafu {}.fail(),
}
}
fn getDefaultOutputConfig(&self) -> Result<(Device, cpal::StreamConfig, SampleFormat, usize)> {
if let Some(dev) = self.host.default_output_device() {
let cfg = dev.default_output_config()?;
let framesPerBlock: usize = cfg.sample_rate() as usize;
let final_config = cpal::StreamConfig {
channels: cfg.channels(),
sample_rate: cfg.sample_rate(),
buffer_size: cpal::BufferSize::Fixed(framesPerBlock as u32),
};
return Ok((dev, final_config, cfg.sample_format(), framesPerBlock));
}
DeviceNotAvailableSnafu {
device_name: "default output device".to_string(),
}
.fail()
}
pub fn startDefaultOutputStream(
&self,
receiver: Receiver<Arc<RawStreamData>>,
) -> Result<Box<dyn Stream>> {
let (device, config, sampleformat, framesPerBlock) = self.getDefaultOutputConfig()?;
let daqchannels = Vec::from_iter(
(0..config.channels)
.map(|i| DaqChannel::defaultAudio(format!("Unnamed output channel {i}"))),
);
let (stream, status) = CpalApi::build_output_stream(
sampleformat,
&config,
&device,
receiver,
&daqchannels,
framesPerBlock,
)?;
stream.play()?;
*status.write() = StreamStatus::newRunning();
let dtype = DataType::try_from(sampleformat).map_err(|_| {
DataTypeNotSupportedSnafu {
dtype: format! {"{sampleformat}"},
}
.build()
})?;
let md = StreamMetaData::new(
&daqchannels,
dtype,
(config.sample_rate as Flt)
.try_into()
.context(DAQConfigValidationSnafu {})?,
framesPerBlock,
Qty::Number,
);
let md = Arc::new(md);
let str = Box::new(CpalStream {
stream,
metadata: md,
noutchannels: daqchannels.len(),
status,
});
Ok(str)
}
fn getCPALOutputConfig(
&self,
dev: &DeviceInfo,
daqconfig: &DaqConfig,
) -> Result<(Device, cpal::StreamConfig, SampleFormat, usize)> {
let samplerate = dev.avSampleRates[daqconfig.sampleRateIndex] as u32;
let framesPerBlock = dev.avFramesPerBlock[daqconfig.framesPerBlockIndex];
let highest_ch: Result<usize> =
daqconfig
.highestEnabledOutChannel()
.with_context(|| DAQConfigSnafu {
msg: "Not enough enabled output channels",
});
let highest_ch = highest_ch? as u16;
for cpaldev in self.host.devices()? {
if cpaldev.description()?.name() == dev.device_name {
for cpalcfg in cpaldev.supported_output_configs()? {
let sf = cpalcfg.sample_format();
if sf == daqconfig.dtype.into() {
let max_sr = cpalcfg.max_sample_rate();
let min_sr = cpalcfg.min_sample_rate();
if samplerate <= max_sr && samplerate >= min_sr {
let cfg = cpalcfg.with_sample_rate(samplerate);
let mut cfg = cfg.config();
cfg.channels = highest_ch + 1;
cfg.buffer_size = cpal::BufferSize::Fixed(framesPerBlock as u32);
return Ok((cpaldev, cfg, sf, framesPerBlock));
}
}
}
}
}
DeviceNotAvailableSnafu {
device_name: &dev.device_name,
}
.fail()
}
}
impl From<cpal::Error> for StreamMgrError {
fn from(value: cpal::Error) -> Self {
match value.kind() {
cpal::ErrorKind::UnsupportedConfig | cpal::ErrorKind::InvalidInput => {
StreamMgrError::DAQConfigError {
msg: value.to_string(),
}
}
cpal::ErrorKind::HostUnavailable => StreamMgrError::ApiNotAvailable {
apiname: "CPAL".into(),
},
_ => StreamMgrError::BackendSpecificError {
msg: value.to_string(),
},
}
}
}