lasprs 0.14.1

//! Loopback testing api. The architecture is as follows.
//!
//! It only allows for 2 channels, sample rate only 48k, data type only F32.
//! There is a global flag to indicate if the API is running. "The API running",
//! means either an input stream or an output stream is running.
//!
//! The output thread is leading, and generates data with a certain pace. If the
//! input thread is running, and the output thread, then the input thread waits
//! for data coming from the output thread. If the output thread is not running
//! and the input thread is running, it generates data at a pace determined by a
//! timer.
//!
use super::*;
use crate::daq::streamerror::{OutputUnderrunSnafu, StreamError, SystemSnafu};
use crate::tools::find_unused_buf;
use core::time;
use parking_lot::{Condvar, Mutex, RwLock};
use snafu::prelude::*;
use std::collections::VecDeque;
use std::sync::OnceLock;
use std::sync::atomic::{AtomicBool, Ordering};
use std::thread::{self, JoinHandle};
use std::time::{Duration, Instant};
use std::{any::Any, fmt::Debug, sync::Arc};

type Result<T> = std::result::Result<T, StreamMgrError>;

const SAMPLERATE: Flt = 48000.;

// Number of channels of the API
const NCHANNELS: usize = 2;

// Number of seconds after stream started, that it is allowed that the signal
// generator is not yet producing any data, without stopping due to buffer
// underrun.
const RUNINTIME: Flt = 2.;

#[derive(Debug, Clone)]
struct StreamChannels {
    /// Used for output thread, to send data to the input thread (the loopback
    /// audio). It sends a tuple of the audio buffer and a boolean array for the
    /// enabled channels.
    output_thread_sender: Sender<Arc<Vec<f32>>>,
    /// Flag indicating whether the input stream, and thus the input thread is
    /// running.
    instream_running: Arc<AtomicBool>,
    /// Flag indicating whether the output stream, and thus the output thread is
    /// running.
    outstream_running: Arc<AtomicBool>,
    /// Other end of the channel, used for input thread to receive data from the
    /// output thread. It receives a tuple of the audio buffer and a boolean
    /// array for the enabled channels.
    input_thread_receiver: Receiver<Arc<Vec<f32>>>,
}

/// Loopback testign api
#[derive(Debug)]
pub struct LoopbackApi {
    channels: StreamChannels,
}

impl LoopbackApi {
    pub fn new() -> Self {
        let (tx, rx) = bounded(1);
        Self {
            channels: StreamChannels {
                instream_running: Arc::new(AtomicBool::new(false)),
                outstream_running: Arc::new(AtomicBool::new(false)),
                output_thread_sender: tx,
                input_thread_receiver: rx,
            },
        }
    }
    /// General checks for loopback API. Check for sample rate, number of
    /// channels, data type. If anything is wrong, it errors. This is a helper
    /// function used both for LoopbackApi::startInputStream and
    /// LoopbackApi::startOutputStream.
    ///
    /// Returns the number of frames per block and the sample rate
    fn generalChecks(
        &self,
        dev: &DeviceInfo,
        cfg: &DaqConfig,
    ) -> Result<(usize, StrictlyPositive)> {
        ensure!(
            *dev == self.getDeviceInfo().expect("Should not give errors")[0],
            DeviceNotAvailableSnafu {
                device_name: dev.device_name.clone()
            }
        );
        // Check if only low channels are enabled - input
        for in_ch_unavailable in cfg.inchannel_config.iter().skip(dev.iChannelCount as usize) {
            ensure!(
                !in_ch_unavailable.enabled,
                DAQConfigSnafu {
                    msg: "Too high enabled input channels",
                }
            );
        }
        // Check if only low channels are enabled - output
        for out_ch_unavailable in cfg
            .outchannel_config
            .iter()
            .skip(dev.oChannelCount as usize)
        {
            ensure!(
                !out_ch_unavailable.enabled,
                DAQConfigSnafu {
                    msg: "Too high enabled output channels",
                }
            );
        }
        let framesPerBlock =
            *dev.avFramesPerBlock
                .get(cfg.framesPerBlockIndex)
                .context(DAQConfigSnafu {
                    msg: "Frames per block index out of range",
                })?;
        let sampleRate = *dev
            .avSampleRates
            .get(cfg.sampleRateIndex)
            .context(DAQConfigSnafu {
                msg: "Samplerate index out of range",
            })?;

        ensure!(
            dev.avDataTypes.contains(&cfg.dtype),
            DAQConfigSnafu {
                msg: "Datatype not available for device",
            }
        );

        Ok((
            framesPerBlock,
            sampleRate.try_into().context(DAQConfigValidationSnafu)?,
        ))
    }
}
impl DaqApiMethods for LoopbackApi {
    fn getDeviceInfo(&self) -> Result<Vec<DeviceInfo>> {
        Ok(vec![DeviceInfo {
            api: DaqApiDescriptor::Loopback,
            device_name: String::from("Loopback"),
            avDataTypes: vec![DataType::F32],
            prefDataType: DataType::F32,
            avFramesPerBlock: vec![512],
            prefFramesPerBlock: 512,
            avSampleRates: vec![SAMPLERATE],
            prefSampleRate: SAMPLERATE,
            iChannelCount: 2,
            oChannelCount: 2,
            hasInputIEPE: false,
            hasInputACCouplingSwitch: false,
            hasInputTrigger: false,
            avInputRanges: vec![(-1.0, 1.0)],
            avOutputRanges: vec![(-1.0, 1.0)],
            physicalIOQty: Qty::Number,
            hasDuplexMode: false,
            duplexModeForced: false,
            hasInternalOutputMonitor: false,
        }])
    }

    fn startInputOrDuplexStream(
        &self,
        stype: StreamType,
        devinfo: &DeviceInfo,
        conf: &DaqConfig,
        sender: Sender<InStreamMsg>,
        receiver: Option<Receiver<Arc<RawStreamData>>>,
    ) -> Result<Box<dyn Stream>> {
        let (framesPerBlock, sampleRate) = self.generalChecks(devinfo, conf)?;

        ensure!(
            matches!(stype, StreamType::Input),
            DAQConfigSnafu {
                msg: "Duplex stream not supported for loopback API",
            }
        );
        ensure!(
            receiver.is_none(),
            DAQConfigSnafu {
                msg: "No receiver should be applied",
            }
        );
        ensure!(
            conf.inchannel_config.len() >= NCHANNELS,
            DAQConfigSnafu {
                msg: format!(
                    "Too few input channels configurations specified. Device
                    requires at least {NCHANNELS} of input channels."
                )
            }
        );

        ensure!(
            conf.inchannel_config
                .iter()
                .skip(NCHANNELS)
                .filter(|ch| ch.enabled)
                .count()
                == 0,
            DAQConfigSnafu {
                msg: format!(
                    "Too high input channels enabled. Device only supports {NCHANNELS} of input channels."
                )
            }
        );

        ensure!(
            conf.inchannel_config
                .iter()
                .take(NCHANNELS)
                .filter(|ch| ch.enabled)
                .count()
                > 0,
            DAQConfigSnafu {
                msg: "No input channels enabled"
            }
        );

        let stopThread = Arc::new(AtomicBool::new(false));
        let stopThread2 = stopThread.clone();
        let status = Arc::new(RwLock::new(StreamStatus::NotRunning {}));
        let status2 = status.clone();
        let channelInfo = conf
            .inchannel_config
            .iter()
            .take(NCHANNELS)
            .filter(|ch| ch.enabled)
            .cloned()
            .collect::<Vec<_>>();
        let meta = Arc::new(StreamMetaData {
            channelInfo,
            rawDatatype: conf.dtype,
            samplerate: sampleRate,
            framesPerBlock,
            physicalIOQty: devinfo.physicalIOQty,
        });
        let ch2 = self.channels.clone();
        let devinfo2 = devinfo.clone();
        let conf2 = conf.clone();
        let meta2 = meta.clone();

        spawn(
            move || {
                loopbackInputThreadFcn(devinfo2, conf2, stopThread2, status2, ch2, sender, meta2);
            },
            ThreadPriority::High,
        );

        Ok(Box::new(InputLoopbackStream {
            stopThread,
            meta,
            status,
        }))
    }

    /// Start an output-only stream. Is a sink-only for data, unless an input
    /// stream is also running, in which case it loops back the data.
    fn startOutputStream(
        &self,
        dev: &DeviceInfo,
        cfg: &DaqConfig,
        receiver: Receiver<Arc<RawStreamData>>,
    ) -> Result<Box<dyn Stream>> {
        let (framesPerBlock, samplerate) = self.generalChecks(dev, cfg)?;
        ensure!(
            cfg.outchannel_config.len() >= NCHANNELS,
            DAQConfigSnafu {
                msg: "Too few out channel configurations provided"
            }
        );
        let out_cfg = &cfg.outchannel_config[..NCHANNELS];
        ensure!(
            out_cfg.iter().filter(|ch| ch.enabled).count() != 0,
            DAQConfigSnafu {
                msg: "No output channels enabled in stream"
            }
        );

        let channelInfo = cfg
            .outchannel_config
            .iter()
            .take(NCHANNELS)
            .filter(|ch| ch.enabled)
            .cloned()
            .collect::<Vec<_>>();

        let stopThread = Arc::new(AtomicBool::new(false));
        let stopThread2 = stopThread.clone();
        let status = Arc::new(RwLock::new(StreamStatus::NotRunning {}));
        let status2 = status.clone();
        let meta = Arc::new(StreamMetaData {
            channelInfo,
            rawDatatype: cfg.dtype,
            samplerate,
            framesPerBlock,
            physicalIOQty: dev.physicalIOQty,
        });

        let dev_clone = dev.clone();
        let cfg_clone = cfg.clone();
        let channels_clone = self.channels.clone();
        spawn(
            move || {
                loopbackOutputThreadFcn(
                    dev_clone,
                    cfg_clone,
                    stopThread2,
                    status2,
                    channels_clone,
                    receiver,
                );
            },
            ThreadPriority::High,
        );

        Ok(Box::new(OutputLoopbackStream {
            stopThread,
            meta,
            status,
        }))
    }

    fn as_any(&self) -> &dyn Any {
        self
    }
}

#[derive(Debug)]
struct OutputLoopbackStream {
    stopThread: Arc<AtomicBool>,
    meta: Arc<StreamMetaData>,
    status: Arc<RwLock<StreamStatus>>,
}
impl Drop for OutputLoopbackStream {
    fn drop(&mut self) {
        self.stopThread.store(true, Ordering::Relaxed);
    }
}
impl Stream for OutputLoopbackStream {
    /// Input stream metadata. Only available for input streams
    fn inMetaData(&self) -> Option<Arc<StreamMetaData>> {
        None
    }

    /// Output stream metadata. Only available for output / duplex streams
    fn outMetaData(&self) -> Option<Arc<StreamMetaData>> {
        Some(self.meta.clone())
    }

    /// Obtain stream status
    fn status(&self, dir: StreamDirection) -> StreamStatus {
        match dir {
            StreamDirection::Input => StreamStatus::NotRunning {},
            StreamDirection::Output => self.status.read().clone(),
        }
    }
}

#[derive(Debug)]
struct InputLoopbackStream {
    stopThread: Arc<AtomicBool>,
    meta: Arc<StreamMetaData>,
    status: Arc<RwLock<StreamStatus>>,
}
impl Drop for InputLoopbackStream {
    fn drop(&mut self) {
        self.stopThread.store(true, Ordering::Relaxed);
    }
}
impl Stream for InputLoopbackStream {
    /// Input stream metadata. Only available for input streams
    fn inMetaData(&self) -> Option<Arc<StreamMetaData>> {
        Some(self.meta.clone())
    }

    /// Output stream metadata. Only available for output / duplex streams
    fn outMetaData(&self) -> Option<Arc<StreamMetaData>> {
        None
    }

    /// Obtain stream status
    fn status(&self, dir: StreamDirection) -> StreamStatus {
        match dir {
            StreamDirection::Input => self.status.read().clone(),
            StreamDirection::Output => StreamStatus::NotRunning {},
        }
    }
}

// Runs on the loopback API thread
fn loopbackOutputThreadFcn(
    dev: DeviceInfo,
    cfg: DaqConfig,
    stopThread: Arc<AtomicBool>,
    status: Arc<RwLock<StreamStatus>>,
    ch: StreamChannels,
    siggenchannel: Receiver<Arc<RawStreamData>>,
) {
    let framesPerBlock = cfg.framesPerBlock(&dev);
    let fs = cfg.sampleRate(&dev);
    let StreamChannels {
        output_thread_sender,
        instream_running,
        outstream_running,
        ..
    } = ch;
    outstream_running.store(true, Ordering::Relaxed);

    // Number of blocks initially allowed to have no data yet
    let run_in_blocks = (RUNINTIME * SAMPLERATE / (framesPerBlock as Flt)) as u64;

    // Store which output channels are enabled
    let outch_enabled = [
        cfg.outchannel_config[0].enabled,
        cfg.outchannel_config[1].enabled,
    ];
    let noutch_enabled = outch_enabled.iter().copied().filter(|val| *val).count();

    // Start instance
    let mut ctr: u64 = 0;
    let mut curtime = Instant::now();

    {
        let mut s = status.write();
        *s = StreamStatus::newRunning();
    }
    // Buffers for sending data to the input stream
    let mut bufs: VecDeque<Arc<Vec<f32>>> = VecDeque::with_capacity(10);

    let interval: Duration = Duration::from_micros((1e6 * framesPerBlock as Flt / fs) as u64);
    let long_sleep = 9 * interval / 10;
    'threadloop: while !stopThread.load(Ordering::Relaxed) {
        // Large step to approximately the moment we need to generate data
        if curtime.elapsed() < interval - long_sleep {
            std::thread::sleep(long_sleep);
        }
        // Small steps for fine stepping
        while curtime.elapsed() <= interval {
            std::thread::sleep(interval / 100);
        }
        curtime += interval;

        // Try to see if the signal generator has produced data. If not, we
        // check whether the run-in time has elapsed. If that is the case, a
        // buffer underrun occured.
        match siggenchannel.try_recv() {
            Ok(dat) => match dat.as_ref() {
                RawStreamData::Dataf32(items) => {
                    // Check whether the right amount of samples is provided
                    if items.len() != framesPerBlock * noutch_enabled {
                        let mut w = status.write();
                        *w = StreamStatus::Error {
                            e: StreamError::FramesMismatchError {},
                        };
                        outstream_running.store(true, Ordering::Relaxed);
                        return;
                    }

                    // If the input stream is running, we pass forward the data.
                    // If not, we discard the data.
                    if instream_running.load(Ordering::Relaxed) {
                        // Find an unused buffer, or create a new one.
                        let mut buf = find_unused_buf(&mut bufs)
                            .unwrap_or_else(|| Arc::new(vec![0.; framesPerBlock * NCHANNELS]));
                        let bufmut = Arc::get_mut(&mut buf).expect("Buffer in use?");

                        assert!(noutch_enabled > 0);
                        if noutch_enabled == 2 {
                            // Copy over the buffer
                            bufmut.copy_from_slice(items);
                        } else {
                            // Below code only works for 2 channels
                            const {
                                assert!(NCHANNELS == 2);
                            }
                            assert!(noutch_enabled == 1);
                            // When outch_enabled[0] is true, the items are
                            // copied to the first channel. When outch_enabled[1]
                            // is true, the items are copied to the second
                            // channel
                            bufmut
                                .iter_mut()
                                .skip(outch_enabled[1] as usize)
                                .step_by(2)
                                .zip(items)
                                .for_each(|(b, i)| *b = *i);
                        }
                        // Push the buf to the back side of the ring buffer of buffers.
                        bufs.push_back(buf.clone());

                        // Try to loopback data to input stream
                        if let Err(e) = output_thread_sender.try_send(buf) {
                            match e {
                                TrySendError::Full(_) => {
                                    if instream_running.load(Ordering::Relaxed) {
                                        // Stream is stil running, but
                                        // the queue is full: input
                                        // overrun.

                                        let mut w = status.write();
                                        *w = StreamStatus::Error {
                                            e: StreamError::InputOverrun {},
                                        };
                                        outstream_running.store(false, Ordering::Relaxed);
                                        return;
                                    } else {
                                        continue 'threadloop;
                                    }
                                }
                                TrySendError::Disconnected(_) => {
                                    // Disconnected not possible as handle to
                                    // channel is still present in this thread
                                    unreachable!()
                                }
                            }
                        }
                    }
                }

                // Logic error: data type is not correct
                _ => {
                    let mut w = status.write();
                    *w = StreamStatus::Error {
                        e: StreamError::DTypeMismatchError {},
                    };
                    outstream_running.store(false, Ordering::Relaxed);
                    return;
                }
            },
            Err(e) => {
                match e {
                    TryRecvError::Empty => {
                        // If the number of run_in_blocks is not reached, we do
                        // not error with a buffer underrun
                        if ctr > run_in_blocks {
                            let mut w = status.write();
                            // e is either empty or disconnected. For both cases we set
                            *w = StreamStatus::Error {
                                e: OutputUnderrunSnafu {}.build(),
                            };
                            outstream_running.store(false, Ordering::Relaxed);
                            return;
                        }
                    }
                    TryRecvError::Disconnected => {
                        let mut w = status.write();
                        *w = StreamStatus::Error {
                            e: SystemSnafu {
                                msg: "signal generator channel closed",
                            }
                            .build(),
                        };
                        outstream_running.store(false, Ordering::Relaxed);
                        return;
                    }
                }
            }
        };
        ctr += 1;
    }
    let mut w = status.write();
    *w = StreamStatus::NotRunning {};
    outstream_running.store(false, Ordering::Relaxed);
}

/// The thread that is emulating an input stream.
///
/// # Args
///
/// - `dev`: DeviceInfo
/// - `cfg`: DaqConfig
/// - `stopThread`: Flag that is used to stop the thread that is executing this function
/// - `status`: To update the status of the stream
/// - `ch`: StreamChannels
/// - `sender`: Here, stream data is sent to the receiver(s)
/// - `meta`: Stream metadata
fn loopbackInputThreadFcn(
    dev: DeviceInfo,
    cfg: DaqConfig,
    stopThread: Arc<AtomicBool>,
    status: Arc<RwLock<StreamStatus>>,
    ch: StreamChannels,
    sender: Sender<InStreamMsg>,
    meta: Arc<StreamMetaData>,
) {
    // dbg!("loopbackInputThreadFcn");
    let StreamChannels {
        instream_running,
        outstream_running,
        // The output thread generates data and pushes to this channel
        input_thread_receiver,
        ..
    } = ch;

    let ninch_enabled = cfg.numberEnabledInChannels();
    let framesPerBlock = cfg.framesPerBlock(&dev);
    let fs = cfg.sampleRate(&dev);

    let inch_enabled: [bool; 2] = array_init::array_init(|i| cfg.inchannel_config[i].enabled);

    assert!(ninch_enabled > 0 && ninch_enabled <= 2);
    instream_running.store(true, Ordering::Relaxed);

    // Clear the existing messages
    while !input_thread_receiver.is_empty() {
        let _ = input_thread_receiver.recv();
    }

    let mut ctr = 0;
    let mut bufs: VecDeque<Arc<RawStreamData>> = VecDeque::with_capacity(10);
    {
        let mut s = status.write();
        *s = StreamStatus::newRunning();
    }

    while !stopThread.load(Ordering::Relaxed) {
        // Find an unused buffer, or create a new one
        let mut rawstreamdata = find_unused_buf(&mut bufs).unwrap_or_else(|| {
            Arc::new(RawStreamData::Dataf32(vec![
                0.;
                framesPerBlock * ninch_enabled
            ]))
        });

        let RawStreamData::Dataf32(bufmut) =
            Arc::get_mut(&mut rawstreamdata).expect("Should be mutable")
        else {
            unreachable!()
        };

        // The sampling period times the number of frames per block gives the
        // time between sample buffers.
        let interval: Duration = Duration::from_nanos((1e9 / fs * framesPerBlock as Flt) as u64);

        let receiver_timeout = 2 * interval;
        let long_sleep = 8 * interval / 10;
        if outstream_running.load(Ordering::Relaxed) {
            match input_thread_receiver.recv_timeout(receiver_timeout) {
                Ok(buf) => {
                    if ninch_enabled == 2 {
                        bufmut.copy_from_slice(&buf);
                    } else {
                        debug_assert!(ninch_enabled == 1);
                        // Only copy over one channel, to a RawStreamData buffer that
                        // only has space for one channel
                        bufmut
                            .iter_mut()
                            .zip(buf.iter().skip(inch_enabled[1] as usize).step_by(2))
                            .for_each(|(out, inn)| *out = *inn);
                    }
                }
                Err(e) => match e {
                    RecvTimeoutError::Timeout => {
                        // Timeout occured. See of output stream is turned off.
                        // If not, we have situation where the output thread did
                        // not provide data within a reasonable time.
                        if outstream_running.load(Ordering::Relaxed) {
                            let mut s = status.write();
                            *s = StreamStatus::Error {
                                e: StreamError::SystemError {
                                    msg: "output thread timeout".into(),
                                },
                            };
                            instream_running.store(false, Ordering::Relaxed);
                            return;
                        }
                    }
                    RecvTimeoutError::Disconnected => unreachable!(),
                },
            }
        } else {
            // Time-based returning of zeros, output stream does not put any data
            let curtime = Instant::now();
            while curtime.elapsed() < interval {
                // Large step
                if curtime.elapsed() < interval - long_sleep {
                    std::thread::sleep(long_sleep);
                }
                // Small steps
                std::thread::sleep(interval / 100);
            }

            // No output stream running, so we return run 0's.
            bufmut.fill(0.);
        }

        // Send the buffer over
        if sender
            .send(InStreamMsg::InStreamData(Arc::new(
                InStreamData::newFromRaw(ctr, meta.clone(), rawstreamdata),
            )))
            .is_err()
        {
            // Sending message failure: system error
            let mut w = status.write();
            *w = StreamStatus::Error {
                e: SystemSnafu {
                    msg: "sending message failure: system error",
                }
                .build(),
            };
            instream_running.store(false, Ordering::Relaxed);
            return;
        }
        ctr += 1;
    } // end of mainloop
    let mut w = status.write();
    *w = StreamStatus::NotRunning {};
    instream_running.store(false, Ordering::Relaxed);
}