lasprs 0.14.2

//! Contains stream manager. The `StreamMgr` is responsible for managing the
//! input and output streams, and attaching queues that process input data. The
//! output stream can be used by the signal generator to send test signals.
use super::api::*;
use super::*;
use crate::{
    daq::error::SiggenSnafu,
    rt::PPM,
    siggen::{self, Siggen, SiggenCommand, SiggenError, SourceDescriptor},
    *,
};
use api::DaqApiMethods;
use api::*;
use array_init::from_iter;
use core::time;
use crossbeam::channel::{Receiver, Sender, TrySendError, bounded, unbounded};
use dasp_sample::Sample;
use snafu::prelude::*;
use std::{
    any::Any,
    collections::HashMap,
    mem::{replace, swap},
    sync::{
        Arc, LazyLock, Mutex, Weak,
        atomic::{AtomicBool, Ordering},
    },
    thread::sleep,
    time::Duration,
};
use streamcmd::InputStreamCommand;
use streamdata::*;
use streammetadata::*;
use streammgr_details::*;
use streammsg::*;
use thread_priority::{ThreadPriority, set_current_thread_priority};

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

/// Keep track of whether the stream has been created. To ensure singleton behaviour.
#[cfg(not(feature = "test_features"))]
static STREAMMGR_CREATED: AtomicBool = AtomicBool::new(false);

/// Store a queue in a shared pointer, to share sending
/// and receiving part of the queue.
pub type SharedInQueue = Sender<InStreamMsg>;

/// Vector of queues for stream messages
pub type InQueues = Vec<SharedInQueue>;
/// Thread-safe list of APIs
pub type ApiMap = HashMap<DaqApiDescriptor, Box<dyn DaqApiMethods>>;

/// Thread-safe list of devices
pub type DeviceList = Vec<DeviceInfo>;

#[derive(Debug)]
enum InputStreamState {
    Undefined,
    NotRunning {
        queues: InQueues,
    },
    Running {
        stream: Box<dyn Stream>,
        stream_thread: JoinHandle<InQueues>,
        // Stream thread communication channels
        commtx: Sender<InputStreamCommand>,
        commrx: Receiver<std::result::Result<(), StreamMgrError>>,
    },
    RunningInDuplex {
        stream: Box<dyn Stream>,
        stream_thread: JoinHandle<(InQueues, Siggen, InQueues)>,
        commtx: Sender<StreamCommand>,
        commrx: Receiver<std::result::Result<(), StreamMgrError>>,
    },
}

#[derive(Debug)]
// Low priority TODO: Put running variant in a Box.
#[allow(clippy::large_enum_variant)]
enum OutputStreamState {
    Undefined,
    NotRunning {
        monqueues: InQueues,
        siggen: Siggen,
    },
    RunningInDuplex,
    Running {
        stream: Box<dyn Stream>,
        siggen_thread: JoinHandle<(Siggen, InQueues)>,
        commtx: Sender<OutputStreamCommand>,
        commrx: Receiver<std::result::Result<(), StreamMgrError>>,
    },
}

/// Configure and manage input / output streams. This method is supposed to be a
/// SINGLETON. Runtime checks are performed to see whether this is true.
///
/// A stream manager provides the interaction layer for interacting with audio /
/// data streams.
///
/// * See [Recording] for an example of starting a recording on an input stream.
/// * See [Siggen] for an example of playing a signal to an output stream.
///
#[cfg_attr(feature = "python-bindings", gen_stub_pyclass, pyclass(unsendable))]
#[derive(Debug)]
pub struct StreamMgr {
    /// Local copy of the source descriptor, to be communicated when a recording
    /// is requested. This is a bit ugly, but otherwise we had to install
    /// bi-directional communication with the signal generator thread.
    srcdesc: SourceDescriptor,

    /// Input PPM detector, if running
    PPMmon: Weak<PPM>,

    /// Monitor PPM detector, if running
    PPMinp: Weak<PPM>,

    // List of available devices
    devs: DeviceList,

    // Input stream can be both input and duplex
    input_stream: InputStreamState,

    // Output only stream
    output_stream: OutputStreamState,

    // List of apis
    apis: ApiMap,

    // When a device scan is in progress, this attribute is Some. The first
    // element is a flag indicating whether the scan is done, and the second
    // element is the handle to the scan thread. When the first value is true, a
    // join() can happen without waiting.
    #[allow(clippy::type_complexity)]
    devices_scan: Option<(
        Arc<AtomicBool>,
        std::thread::JoinHandle<(ApiMap, DeviceList)>,
    )>,
}

#[cfg(feature = "python-bindings")]
#[cfg_attr(feature = "python-bindings", gen_stub_pymethods, pymethods)]
impl StreamMgr {
    /// See (StreamMgr::new())
    #[new]
    fn py_new() -> StreamMgr {
        StreamMgr::new()
    }

    fn __repr__(&self) -> String {
        format!("{self:#?}")
    }
    #[pyo3(name = "startDefaultInputStream")]
    fn startDefaultInputStream_py(&mut self) -> PyResult<()> {
        Ok(self.startDefaultInputStream()?)
    }
    #[pyo3(name = "startDefaultOutputStream")]
    fn startDefaultOutputStream_py(&mut self) -> PyResult<()> {
        Ok(self.startDefaultOutputStream()?)
    }
    #[pyo3(name = "startStream")]
    fn startStream_py(&mut self, st: StreamType, d: &DaqConfig) -> PyResult<()> {
        Ok(self.startStream(st, d)?)
    }
    #[pyo3(name = "stopStream")]
    fn stopStream_py(&mut self, st: StreamType) -> PyResult<()> {
        Ok(self.stopStream(st)?)
    }
    #[pyo3(name = "getDeviceInfo")]
    fn getDeviceInfo_py(&mut self) -> PyResult<Vec<DeviceInfo>> {
        Ok(self.getDeviceInfo())
    }
    #[pyo3(name = "getStatus")]
    fn getStatus_py(&self, dir: StreamDirection) -> StreamStatus {
        self.getStatus(dir)
    }
    #[pyo3(name = "getStreamMetaData")]
    fn getStreamMetaData_py(&self, dir: StreamDirection) -> Option<StreamMetaData> {
        // Unfortunately (but not really, only cosmetically), the underlying
        // value (not the Arc) has to be cloned.
        self.getStreamMetaData(dir).map(|b| (*b).clone())
    }
    #[pyo3(name = "reScanDevices")]
    fn reScanDevices_py(&mut self) -> PyResult<()> {
        self.reScanDevices()?;
        Ok(())
    }
    #[pyo3(name = "isSomeStreamRunning")]
    fn isSomeStreamRunning_py(&self) -> bool {
        self.isSomeStreamRunning()
    }
    #[pyo3(name = "isStreamRunning")]
    fn isStreamRunning_py(&self, dir: StreamDirection) -> bool {
        self.isStreamRunning(dir)
    }
    #[pyo3(name = "isStreamRunningOK")]
    fn isStreamRunningOK_py(&self, dir: StreamDirection) -> PyResult<bool> {
        Ok(self.isStreamRunningOK(dir))
    }

    #[pyo3(name = "siggenCommand")]
    fn siggenCommand_py(&mut self, cmd: SiggenCommand) -> PyResult<()> {
        Ok(self.siggenCommand(cmd)?)
    }
}

impl Default for StreamMgr {
    fn default() -> Self {
        Self::new()
    }
}

impl StreamMgr {
    /// Create new stream manager. A stream manager is supposed to be a
    /// singleton. Note that we let Rust's ownership model handle that there is
    /// only a single [StreamMgr].
    ///
    /// # Panics
    ///
    /// When a StreamMgr object is already alive.
    ///
    pub fn new() -> StreamMgr {
        cfg_select! {
            not(feature = "test_features")=> {
                if STREAMMGR_CREATED
                    .compare_exchange(false, true, Ordering::Acquire, Ordering::Relaxed)
                    .is_err()
                {
                    panic!("BUG: Stream manager is supposed to be a singleton");
                }
            },
            _ => {}
        }

        let mut apis: HashMap<_, Box<dyn DaqApiMethods>> = HashMap::new();

        #[cfg(feature = "cpal-api")]
        apis.insert(DaqApiDescriptor::Cpal, Box::new(CpalApi::new()));

        #[cfg(feature = "loopback-api")]
        apis.insert(DaqApiDescriptor::Loopback, Box::new(LoopbackApi::new()));

        #[cfg(feature = "uldaq-api")]
        apis.insert(DaqApiDescriptor::Uldaq, Box::new(UldaqApi::new()));

        let srcdesc = SourceDescriptor::Silence {};
        let mut smgr = StreamMgr {
            apis,
            devs: vec![],
            input_stream: InputStreamState::NotRunning { queues: vec![] },
            output_stream: OutputStreamState::NotRunning {
                monqueues: vec![],
                siggen: Siggen::new(1, srcdesc.clone()),
            },
            PPMmon: Weak::new(),
            PPMinp: Weak::new(),
            devices_scan: None,
            srcdesc,
        };

        smgr.reScanDevices().unwrap();
        smgr
    }

    /// Create a new stream manager, and wait till the device scan is complete.
    /// Avoids errors in code that tries to access devices before they are
    /// available.
    pub fn new_with_devices() -> Self {
        let mut smgr = Self::new();
        while smgr.isDeviceScanRunning() {
            std::thread::sleep(Duration::from_millis(100));
            // This updates the device list from the scan results, such that
            // isDeviceScanRunning eventually returns false, so it should stay
            // in the loop.
            smgr.updateDeviceListFromScan();
        }
        smgr
    }

    /// (re)scan for DAQ devices
    ///
    /// Errors when a scan operation is already running, or when a stream is running
    pub fn reScanDevices(&mut self) -> Result<()> {
        ensure!(
            self.devices_scan.is_none(),
            DeviceScanAlreadyInProgressSnafu
        );
        ensure!(
            matches!(self.input_stream, InputStreamState::NotRunning { .. }),
            InputStreamAlreadyRunningSnafu
        );
        ensure!(
            matches!(self.output_stream, OutputStreamState::NotRunning { .. }),
            OutputStreamAlreadyRunningSnafu
        );

        let mut apis = HashMap::new();
        swap(&mut self.apis, &mut apis);
        let scanfinished = Arc::new(AtomicBool::new(false));
        let scanfinished_clone = scanfinished.clone();
        self.devices_scan = Some((
            scanfinished_clone,
            std::thread::spawn(move || {
                // We initialize the Rayon thread pools here that have
                // different priorities.
                create_thread_pool_if_not_created();

                cfg_select! {
                    all(not(debug_assertions), target_os = "linux") => {
                        // Silence printing to stderr. This is what ALSA does when probing
                        // for devices, and the system has an improper ALSA configuration.
                        let print_gag = gag::Gag::stderr();
                        if let Err(e) = print_gag {
                            eprintln!("Unable to capture stderr: {e}");
                        }
                    },
                    target_os = "linux" => {
                        eprintln!("****** Any possible ALSA errors printed below are suppressed in release builds. ******");
                    },
                    _ => {}
                }
                let mut all_devices = vec![];
                for api in apis.values() {
                    let devs = api.getDeviceInfo();
                    if let Ok(devs) = devs {
                        all_devices.extend(devs);
                    }
                }
                #[cfg(target_os = "linux")]
                eprintln!("****** End of any possible ALSA errors. ******");

                scanfinished.store(true, Ordering::Relaxed);
                (apis, all_devices)
            }),
        ));

        Ok(())
    }

    /// Returns true if any stream is running.
    pub fn isSomeStreamRunning(&self) -> bool {
        !matches!(self.input_stream, InputStreamState::NotRunning { .. })
            || !matches!(self.output_stream, OutputStreamState::NotRunning { .. })
    }
    /// Check whether some stream is running, given the direction (input or
    /// output).
    pub fn isStreamRunning(&self, dir: StreamDirection) -> bool {
        match dir {
            StreamDirection::Input => match &self.input_stream {
                InputStreamState::NotRunning { .. } => false,
                InputStreamState::Running { .. } => true,
                InputStreamState::RunningInDuplex { .. } => true,
                InputStreamState::Undefined => unreachable!(),
            },
            StreamDirection::Output => match &self.output_stream {
                OutputStreamState::NotRunning { .. } => false,
                OutputStreamState::Running { .. } => true,
                OutputStreamState::RunningInDuplex => true,
                OutputStreamState::Undefined => unreachable!(),
            },
        }
    }

    /// Check whether some stream is running, and does not have any errors.
    pub fn isStreamRunningOK(&self, dir: StreamDirection) -> bool {
        match dir {
            StreamDirection::Input => match &self.input_stream {
                InputStreamState::NotRunning { .. } => false,
                InputStreamState::Running { stream, .. } => {
                    matches!(stream.status(dir), StreamStatus::Running { .. })
                }
                InputStreamState::RunningInDuplex { stream, .. } => {
                    matches!(stream.status(dir), StreamStatus::Running { .. })
                }
                InputStreamState::Undefined => unreachable!(),
            },
            StreamDirection::Output => match &self.output_stream {
                OutputStreamState::NotRunning { .. } => false,
                OutputStreamState::Running { stream, .. } => {
                    matches!(stream.status(dir), StreamStatus::Running { .. })
                }
                OutputStreamState::RunningInDuplex => {
                    let InputStreamState::RunningInDuplex { stream, .. } = &self.input_stream
                    else {
                        unreachable!(
                            "Invalid input stream state, does not match output stream state, which is in duplex mode"
                        )
                    };
                    matches!(stream.status(dir), StreamStatus::Running { .. })
                }
                OutputStreamState::Undefined => unreachable!(),
            },
        }
    }

    /// Returns the metadata for a given stream, when the stream type (see
    /// [StreamType]) is alive, i.e. (StreamMgr::getStatus) gives a 'Running'.
    ///
    pub fn getStreamMetaData(&self, dir: StreamDirection) -> Option<Arc<StreamMetaData>> {
        match dir {
            StreamDirection::Input => match &self.input_stream {
                InputStreamState::NotRunning { .. } => None,
                InputStreamState::Running { stream, .. } => stream.inMetaData(),
                InputStreamState::RunningInDuplex { stream, .. } => stream.inMetaData(),
                InputStreamState::Undefined => unreachable!(),
            },
            StreamDirection::Output => {
                if let InputStreamState::RunningInDuplex { stream, .. } = &self.input_stream {
                    stream.outMetaData()
                } else {
                    match &self.output_stream {
                        OutputStreamState::Undefined => unreachable!(),
                        OutputStreamState::NotRunning { .. } => None,
                        OutputStreamState::RunningInDuplex => unreachable!(),
                        OutputStreamState::Running { stream, .. } => stream.outMetaData(),
                    }
                }
            }
        }
    }

    /// Get stream status for given stream direction.
    pub fn getStatus(&self, dir: StreamDirection) -> StreamStatus {
        if let InputStreamState::RunningInDuplex { stream, .. } = &self.input_stream {
            return stream.status(dir);
        }
        match dir {
            StreamDirection::Input => {
                match &self.input_stream {
                    InputStreamState::NotRunning { .. } => StreamStatus::NotRunning {},
                    InputStreamState::Running { stream, .. } => {
                        // dbg!(dir);
                        // dbg!(stream.status(dir));
                        return stream.status(dir);
                    }
                    InputStreamState::RunningInDuplex { .. } => {
                        unreachable!()
                    }
                    InputStreamState::Undefined => unreachable!(),
                };
            }
            StreamDirection::Output => {
                match &self.output_stream {
                    OutputStreamState::NotRunning { .. } => StreamStatus::NotRunning {},
                    OutputStreamState::RunningInDuplex => unreachable!(),
                    OutputStreamState::Running { stream, .. } => {
                        return stream.status(dir);
                    }
                    OutputStreamState::Undefined => unreachable!(),
                };
            }
        }
        StreamStatus::NotRunning {}
    }

    /// Get the source descriptor of the signal generator.
    pub fn getSourceDescriptor(&self) -> SourceDescriptor {
        self.srcdesc.clone()
    }

    /// Set a new signal generator source. Returns an error if it is
    /// unapplicable.
    ///
    /// # Arguments
    /// * `src` - The new signal generator source.
    pub fn setSiggenSource(&mut self, src: SourceDescriptor) -> Result<()> {
        if let InputStreamState::RunningInDuplex { commtx, commrx, .. } = &self.input_stream {
            commtx
                .send(StreamCommand::OutputStreamCommand(
                    OutputStreamCommand::SiggenCommand(SiggenCommand::ChangeSource {
                        src: src.clone(),
                    }),
                ))
                .unwrap();
            match commrx.recv().unwrap() {
                Ok(()) => {
                    self.srcdesc = src;
                    return Ok(());
                }
                err @ Err(_) => {
                    return err;
                }
            }
        }

        // Current signal generator. Where to place it?
        match &mut self.output_stream {
            OutputStreamState::NotRunning {
                monqueues: _,
                siggen,
            } => {
                siggen
                    .applyCommand(SiggenCommand::ChangeSource { src: src.clone() })
                    .context(SiggenSnafu {})?;
                self.srcdesc = src;
                Ok(())
            }
            OutputStreamState::RunningInDuplex => {
                unreachable!("Cannot get here!")
            }
            OutputStreamState::Running { commtx, commrx, .. } => {
                commtx
                    .send(OutputStreamCommand::SiggenCommand(
                        SiggenCommand::ChangeSource { src: src.clone() },
                    ))
                    .unwrap();
                match commrx.recv().unwrap() {
                    Ok(_) => {
                        self.srcdesc = src;
                        Ok(())
                    }
                    err @ Err(_) => err,
                }
            }
            OutputStreamState::Undefined => unreachable!(),
        }
    }

    /// Update the device list from a scan that might be finished. Does nothing
    /// if no scan is running.
    fn updateDeviceListFromScan(&mut self) {
        if let Some((scan_done, joinhandle)) = self.devices_scan.take() {
            if scan_done.load(Ordering::Relaxed) {
                let (apilist, devlist) = joinhandle.join().expect("Device scan panicked");
                self.apis = apilist;
                self.devs = devlist;
                // Drops the receiver as well
                self.devices_scan = None;
            } else {
                // Put it back, not yet finished
                self.devices_scan = Some((scan_done, joinhandle));
            }
        }
    }

    /// Returns true when a device scan is currently running
    pub fn isDeviceScanRunning(&self) -> bool {
        self.devices_scan.is_some()
    }

    /// Obtain a list of devices that are available for each available API
    pub fn getDeviceInfo(&mut self) -> Vec<DeviceInfo> {
        self.updateDeviceListFromScan();
        self.devs.clone()
    }

    /// Add a new queue to the lists of queues. On the queue, input data is
    /// added.
    ///
    /// If the stream is unable to write data on the queue (which might
    /// happen when the handler is dropped), the queue is removed from the list
    /// of queues that get data from the stream.
    pub fn addInQueue(&mut self, tx: Sender<InStreamMsg>) {
        match &mut self.input_stream {
            InputStreamState::NotRunning { queues } => {
                queues.push(tx);
            }
            InputStreamState::Running { commtx, commrx, .. } => {
                commtx.send(InputStreamCommand::AddInQueue(tx)).unwrap();
                commrx
                    .recv()
                    .unwrap()
                    .expect("Adding a queue should never fail")
            }
            InputStreamState::RunningInDuplex { commtx, commrx, .. } => {
                commtx
                    .send(StreamCommand::InputStreamCommand(
                        InputStreamCommand::AddInQueue(tx),
                    ))
                    .unwrap();
                commrx
                    .recv()
                    .unwrap()
                    .expect("Adding a queue should never fail")
            }
            InputStreamState::Undefined => unreachable!(),
        }
    }

    /// Add a new monitor queue to the lists of queues. On the queue, monitor of
    /// output data is added.
    ///
    /// If the stream is unable to write data on the queue (which might happen
    /// when the handler is dropped), the queue is removed from the list of
    /// queues that get data from the signal generator.
    pub fn addMonitorQueue(&mut self, tx: SharedInQueue) {
        if let InputStreamState::RunningInDuplex { commtx, commrx, .. } = &self.input_stream {
            commtx
                .send(StreamCommand::OutputStreamCommand(
                    OutputStreamCommand::AddMonitorQueue(tx),
                ))
                .unwrap();
            commrx
                .recv()
                .unwrap()
                .expect("Adding a queue should never fail");
            return;
        };
        match &mut self.output_stream {
            OutputStreamState::NotRunning { monqueues, .. } => {
                monqueues.push(tx);
            }
            OutputStreamState::RunningInDuplex => unreachable!(),
            OutputStreamState::Running {
                stream: _,
                siggen_thread: _,
                commtx,
                commrx,
            } => {
                commtx
                    .send(OutputStreamCommand::AddMonitorQueue(tx))
                    .unwrap();
                commrx
                    .recv()
                    .unwrap()
                    .expect("Adding a queue should never fail");
            }
            OutputStreamState::Undefined => unreachable!(),
        }
    }

    // Match device info struct on given daq config.
    fn find_device(&self, cfg: &DaqConfig) -> Result<&DeviceInfo> {
        ensure!(
            self.devices_scan.is_none(),
            DeviceScanAlreadyInProgressSnafu
        );
        if let Some(matching_dev) = self
            .devs
            .iter()
            .find(|&d| d.device_name == cfg.device_name && d.api == cfg.api)
        {
            return Ok(matching_dev);
        }
        // Device not found
        DeviceNotAvailableSnafu {
            device_name: &cfg.device_name,
        }
        .fail()
    }

    /// Start a stream of certain type, using given configuration
    pub fn startStream(&mut self, stype: StreamType, cfg: &DaqConfig) -> Result<()> {
        self.updateDeviceListFromScan();
        ensure!(
            self.devices_scan.is_none(),
            DeviceScanAlreadyInProgressSnafu
        );
        match stype {
            StreamType::Input | StreamType::Duplex => {
                self.startInputOrDuplexStream(stype, cfg)?;
            }
            StreamType::Output => {
                self.startOutputStream(cfg)?;
            }
        }
        Ok(())
    }

    /// Start a stream for output only, using only the output channel
    /// configuration as given in the `cfg`.
    fn startOutputStream(&mut self, cfg: &DaqConfig) -> Result<()> {
        let stream = replace(&mut self.output_stream, OutputStreamState::Undefined);

        match stream {
            OutputStreamState::NotRunning { monqueues, siggen } => {
                let (tx, rx): (Sender<Arc<RawStreamData>>, Receiver<Arc<RawStreamData>>) =
                    unbounded();

                let startstream = |rx, tx, mut siggen: Siggen, mon_queues| -> Result<_> {
                    let api = self
                        .getDaqApi(&cfg.api)
                        .with_context(|| ApiNotAvailableSnafu {
                            apiname: cfg.api.name(),
                        })?;

                    let devinfo = self.find_device(cfg)?;
                    let stream = api.startOutputStream(devinfo, cfg, rx)?;
                    let meta = stream.outMetaData().expect("No stream metadata available");
                    // Reset the signal generator with the new sample rate if it
                    // fails, we error the stream start process.
                    siggen.reset(meta.samplerate).context(SiggenSnafu)?;

                    let (siggen_thread, commtx, commrx) =
                        startSiggenThread(meta, siggen, tx, mon_queues)?;
                    Ok((stream, siggen_thread, commtx, commrx))
                };

                match startstream(rx, tx, siggen.clone(), monqueues.clone()) {
                    Ok((stream, siggen_thread, commtx, commrx)) => {
                        self.output_stream = OutputStreamState::Running {
                            stream,
                            siggen_thread,
                            commtx,
                            commrx,
                        }
                    }
                    Err(e) => {
                        self.output_stream = OutputStreamState::NotRunning { monqueues, siggen };
                        return Err(e);
                    }
                }
            }
            _ => {
                let und = replace(&mut self.output_stream, stream);
                assert!(matches!(und, OutputStreamState::Undefined));
                return OutputStreamAlreadyRunningSnafu.fail();
            }
        }
        Ok(())
    }

    /// Start an input or duplex stream
    fn startInputOrDuplexStream(&mut self, stype: StreamType, cfg: &DaqConfig) -> Result<()> {
        assert!(!matches!(self.input_stream, InputStreamState::Undefined));
        // dbg!("startInputOrDuplexStream");
        ensure!(
            cfg.numberEnabledInChannels() > 0,
            DAQConfigSnafu {
                msg: "At least one input channel should be enabled \
             for an input stream"
            }
        );
        let duplex = matches!(stype, StreamType::Duplex);

        ensure!(
            !(stype == StreamType::Duplex && cfg.numberEnabledOutChannels() == 0),
            DAQConfigSnafu {
                msg: "At least one output channel should be enabled for a duplex stream"
            }
        );

        ensure!(
            !(matches!(self.output_stream, OutputStreamState::Running { .. }) && duplex),
            DAQConfigSnafu {
                msg: "An output stream is already running. Please first stop existing output stream."
            }
        );

        let instream = replace(&mut self.input_stream, InputStreamState::Undefined);
        match instream {
            InputStreamState::NotRunning { queues } => {
                if duplex {
                    let ostream = replace(&mut self.output_stream, OutputStreamState::Undefined);
                    let startduplexstream = |mut iqueues: InQueues,
                                             siggen,
                                             mut mon_queues: InQueues|
                     -> Result<_> {
                        assert!(!duplex);
                        let (tx, rx_in): (Sender<InStreamMsg>, Receiver<InStreamMsg>) = unbounded();
                        let (tx_out, rx_out) = unbounded();
                        let api = self.getDaqApi(&cfg.api).context(ApiNotAvailableSnafu {
                            apiname: cfg.api.name(),
                        })?;
                        let devinfo = self.find_device(cfg)?;
                        let stream =
                            api.startInputOrDuplexStream(stype, devinfo, cfg, tx, Some(rx_out))?;
                        let in_meta = stream
                            .inMetaData()
                            .expect("No input stream metadata available");
                        let out_meta = stream
                            .outMetaData()
                            .expect("No output stream metadata for duplex stream!");

                        sendMsgToAllQueuesRemoveUnused(
                            &mut iqueues,
                            InStreamMsg::StreamStarted(
                                in_meta.clone(),
                                PreCaptureBuffer::NotLoaded,
                            ),
                        );
                        sendMsgToAllQueuesRemoveUnused(
                            &mut mon_queues,
                            InStreamMsg::StreamStarted(
                                out_meta.clone(),
                                PreCaptureBuffer::NotLoaded,
                            ),
                        );

                        let (threadhandle, commtx, commrx) = startDuplexThread(
                            in_meta, rx_in, iqueues, out_meta, siggen, tx_out, mon_queues,
                        )?;
                        Ok((stream, threadhandle, commtx, commrx))
                    };
                    let (monqueues, siggen) =
                        if let OutputStreamState::NotRunning { monqueues, siggen } = ostream {
                            (monqueues, siggen)
                        } else {
                            unreachable!()
                        };

                    match startduplexstream(queues.clone(), siggen.clone(), monqueues.clone()) {
                        Ok((stream, stream_thread, commtx, commrx)) => {
                            let _ = replace(
                                &mut self.input_stream,
                                InputStreamState::RunningInDuplex {
                                    stream,
                                    stream_thread,
                                    commtx,
                                    commrx,
                                },
                            );
                            let _ = replace(
                                &mut self.output_stream,
                                OutputStreamState::RunningInDuplex {},
                            );
                            Ok(())
                        }
                        Err(e) => {
                            let _ = replace(
                                &mut self.input_stream,
                                InputStreamState::NotRunning { queues },
                            );
                            let _ = replace(
                                &mut self.output_stream,
                                OutputStreamState::NotRunning { monqueues, siggen },
                            );
                            Err(e)
                        }
                    }
                } else {
                    // dbg!("Start normal stream");
                    let startinstream = |mut iqueues: InQueues| -> Result<_> {
                        assert!(!duplex);
                        let (tx, rx): (Sender<InStreamMsg>, Receiver<InStreamMsg>) = unbounded();
                        let api = self.getDaqApi(&cfg.api).context(ApiNotAvailableSnafu {
                            apiname: cfg.api.name(),
                        })?;
                        let devinfo = self.find_device(cfg)?;

                        let stream = api.startInputOrDuplexStream(stype, devinfo, cfg, tx, None)?;
                        let meta = stream
                            .inMetaData()
                            .expect("No input stream metadata available");
                        sendMsgToAllQueuesRemoveUnused(
                            &mut iqueues,
                            InStreamMsg::StreamStarted(meta.clone(), PreCaptureBuffer::NotLoaded),
                        );

                        let (threadhandle, commtx, commrx) =
                            startInputStreamThread(meta, rx, iqueues);
                        Ok((stream, threadhandle, commtx, commrx))
                    };

                    // Not duplex case
                    match startinstream(queues.clone()) {
                        Ok((stream, stream_thread, commtx, commrx)) => {
                            self.input_stream = InputStreamState::Running {
                                stream,
                                stream_thread,
                                commtx,
                                commrx,
                            };
                            Ok(())
                        }
                        Err(e) => {
                            let und = replace(
                                &mut self.input_stream,
                                InputStreamState::NotRunning { queues },
                            );
                            assert!(matches!(und, InputStreamState::Undefined));
                            Err(e)
                        }
                    }
                }
            }
            _ => {
                let _ = replace(&mut self.input_stream, instream);
                InputStreamAlreadyRunningSnafu.fail()
            }
        }
    }

    /// Start a default input stream, using default settings on everything. This is only possible
    /// when the CPAL_api is available
    pub fn startDefaultInputStream(&mut self) -> Result<()> {
        self.updateDeviceListFromScan();
        while self.isDeviceScanRunning() {
            eprintln!("Cannot yet start stream: a device scan is still in progress.");
            sleep(Duration::from_millis(20));
            self.updateDeviceListFromScan();
        }
        // Only a default input stream when CPAL feature is enabled
        cfg_select! {
            feature = "cpal-api" => {
                let stream = replace(&mut self.input_stream, InputStreamState::Undefined);

                if let InputStreamState::NotRunning { queues } = stream {

                    let startstream = |mut iqueues: InQueues| -> Result<_> {

                        let (tx, rx): (Sender<InStreamMsg>, Receiver<InStreamMsg>) = unbounded();

                        let cpal_api: &CpalApi = self
                            .getDaqApiT::<CpalApi>()
                            .context(CPALNotAvailableSnafu)?;

                        let stream = cpal_api.startDefaultInputStream(tx)?;
                        let meta = stream
                            .inMetaData()
                            .expect("No input stream metadata available");
                        sendMsgToAllQueuesRemoveUnused(&mut iqueues, InStreamMsg::StreamStarted(meta.clone(), PreCaptureBuffer::NotLoaded));
                        let (threadhandle, commtx, commrx) = startInputStreamThread(meta, rx, iqueues);
                        Ok((stream, threadhandle, commtx, commrx))
                    };

                    match startstream(queues.clone()) {
                        Ok((stream, stream_thread, commtx, commrx)) => {
                            self.input_stream = InputStreamState::Running { stream, stream_thread, commtx, commrx };
                            Ok(())
                        }
                        Err(e) => {
                            let _ = replace(
                                &mut self.input_stream,
                                InputStreamState::NotRunning { queues },
                            );
                            Err(e)
                        }
                    }
                } else {
                    let _ = replace(&mut self.input_stream, stream);
                    InputStreamAlreadyRunningSnafu.fail()
                }
            },
            _ => {
                CPALNotAvailableSnafu.fail()
            }
        }
    }

    /// Start a default output stream. Only possible when CPAL Api is available.
    pub fn startDefaultOutputStream(&mut self) -> Result<()> {
        self.updateDeviceListFromScan();
        while self.isDeviceScanRunning() {
            eprintln!("Cannot yet start stream: a device scan is still in progress.");
            sleep(Duration::from_millis(20));
            self.updateDeviceListFromScan();
        }
        cfg_select! {
            feature = "cpal-api" => {
                let stream = replace(&mut self.output_stream, OutputStreamState::Undefined);
                if let OutputStreamState::NotRunning { monqueues, siggen } = stream {

                    let startstream = |mut mon_queues: InQueues, siggen| -> Result<_> {

                        let (tx, rx) = unbounded();

                        let cpal_api: &CpalApi =
                            self.getDaqApiT::<CpalApi>().expect("CPal API not present");
                        let stream = cpal_api.startDefaultOutputStream(rx)?;
                        let meta = stream.outMetaData().expect("Output metadata not available");
                        sendMsgToAllQueuesRemoveUnused(&mut mon_queues, InStreamMsg::StreamStarted(meta.clone(), PreCaptureBuffer::NotLoaded));

                        // Last step: reset the signal generator to actually
                        // generate output. If this fails here, we fail starting
                        // the stream. The rest is cleaned up with a drop on the
                        // handlees.
                        Ok((stream, startSiggenThread(meta, siggen, tx, mon_queues)?))
                    };

                    match startstream(monqueues.clone(), siggen.clone()) {
                        Ok((stream, (siggen_thread, commtx, commrx))) => {
                            let _ = replace(&mut self.output_stream, OutputStreamState::Running { stream, siggen_thread, commtx, commrx });
                            Ok(())
                        },
                        Err(e) => {
                                let _ = replace(
                                    &mut self.output_stream,
                                    OutputStreamState::NotRunning { monqueues, siggen }
                                );
                                Err(e)
                        },
                    }

                } else {
                    let _ = replace(&mut self.output_stream, stream);
                    OutputStreamAlreadyRunningSnafu.fail()
                }

            },  // end if cpal api available
            _ => {
                CPALNotAvailableSnafu.fail()
            }
        } // end of cfg_select
    }

    /// Stop existing input stream.
    pub fn stopInputStream(&mut self) -> Result<()> {
        assert!(!matches!(self.input_stream, InputStreamState::Undefined));
        let stream = replace(&mut self.input_stream, InputStreamState::Undefined);
        if let InputStreamState::Running {
            stream: _,
            stream_thread,
            commtx,
            commrx,
        } = stream
        {
            // dbg!("stopInputStream: normal input stream");
            commtx.send(InputStreamCommand::StopThread).unwrap();
            let _ = commrx.recv().unwrap();
            let queues = stream_thread.join();
            self.input_stream = InputStreamState::NotRunning { queues };
            Ok(())
        } else if let InputStreamState::RunningInDuplex {
            stream: _,
            stream_thread,
            commtx,
            commrx,
        } = stream
        {
            // dbg!("stopInputStream: duplex stream");
            commtx.send(StreamCommand::StopThread).unwrap();
            let _ = commrx.recv().unwrap();
            let (iqueues, siggen, monqueues) = stream_thread.join();

            self.input_stream = InputStreamState::NotRunning { queues: iqueues };
            self.output_stream = OutputStreamState::NotRunning { monqueues, siggen };

            Ok(())
        } else {
            self.input_stream = stream;
            InputStreamNotRunningSnafu.fail()
        }
    }

    /// Stop existing output stream
    pub fn stopOutputStream(&mut self) -> Result<()> {
        let stream = replace(&mut self.output_stream, OutputStreamState::Undefined);
        if let OutputStreamState::Running {
            stream: _,
            siggen_thread,
            commtx,
            commrx,
        } = stream
        {
            commtx.send(OutputStreamCommand::StopThread).unwrap();
            let _ = commrx.recv().unwrap();

            // eprintln!("Wainting for threadhandle to join...");
            let (siggen, monqueues) = siggen_thread.join();
            self.output_stream = OutputStreamState::NotRunning { monqueues, siggen };
            Ok(())
        } else {
            let _ = replace(&mut self.output_stream, stream);
            OutputStreamNotRunningSnafu.fail()
        }
    }

    /// Stop existing running stream.
    ///
    /// Args
    ///
    /// * st: The stream type.
    pub fn stopStream(&mut self, st: StreamType) -> Result<()> {
        assert!(!matches!(self.input_stream, InputStreamState::Undefined));
        assert!(!matches!(self.output_stream, OutputStreamState::Undefined));
        match st {
            StreamType::Input | StreamType::Duplex => self.stopInputStream(),
            StreamType::Output => self.stopOutputStream(),
        }
    }

    /// Apply a signal generator command to control the output stream's signal
    /// generator. see [SiggenCommand] for types of commands. Muting, setting
    /// gain etc. A result code is given back and should be checked for errors.
    pub fn siggenCommand(&mut self, cmd: SiggenCommand) -> Result<()> {
        if let InputStreamState::RunningInDuplex { commtx, commrx, .. } = &self.input_stream {
            commtx
                .send(StreamCommand::OutputStreamCommand(
                    OutputStreamCommand::SiggenCommand(cmd),
                ))
                .unwrap();
            commrx.recv().unwrap()
        } else if let OutputStreamState::Running { commtx, commrx, .. } = &self.output_stream {
            commtx
                .send(OutputStreamCommand::SiggenCommand(cmd))
                .unwrap();
            commrx.recv().unwrap()
        } else if let OutputStreamState::NotRunning {
            monqueues: _,
            siggen,
        } = &mut self.output_stream
        {
            siggen.applyCommand(cmd).context(SiggenSnafu)
        } else {
            unreachable!()
        }
    }

    /// Get an already running PPM for the input stream. This method should not
    /// be called directly by the user. Instead, the user should call
    /// [PPM::newInput] instead.
    pub fn getPPMInput(&self) -> Option<Arc<PPM>> {
        self.PPMinp.upgrade()
    }

    /// Called to update the weakref to the input PPM
    pub fn setPPMInput(&mut self, ppmMon: &Arc<PPM>) {
        if self.PPMinp.upgrade().is_some() {
            panic!("Input PPM is already running!")
        }
        self.PPMinp = Arc::downgrade(ppmMon);
    }

    /// Get reference to Daq api for certain API descriptor, if available.
    pub fn getDaqApi(&self, apidescr: &DaqApiDescriptor) -> Option<&dyn DaqApiMethods> {
        self.apis.get(apidescr).map(|v| &**v)
    }

    /// Get API by downcasting to the proper type. Returns Some() if API is available.
    pub fn getDaqApiT<T>(&self) -> Option<&T>
    where
        T: DaqApiMethods,
    {
        for val in self.apis.values() {
            // If you only would know how much time this has cost me to create
            // these couple of lines of code, you would kill me. A trait object
            // does not know how to upcast itself. You have to do that by using
            // a method in the trait itself, such that you can cast it to an
            // &dyn Any. Only after that, you can downcast_ref() it.
            //
            // Sometimes Rust is really hard.

            // Update: as of Rust
            let val: &dyn Any = val.as_any();
            let val = val.downcast_ref();
            if val.is_some() {
                return val;
            }
        }
        None
    }

    /// Get an already running PPM for the monitor stream. This method should not
    /// be called directly by the user. Instead, the user should call
    /// [PPM::newMonitor] instead.
    pub fn getPPMMon(&self) -> Option<Arc<PPM>> {
        self.PPMmon.upgrade()
    }

    /// Called to update the weakref to the monitor PPM
    pub fn setPPMMon(&mut self, ppmMon: &Arc<PPM>) {
        if self.PPMmon.upgrade().is_some() {
            panic!("Monitor PPM is already running!")
        }
        self.PPMmon = Arc::downgrade(ppmMon);
    }
} // impl StreamMgr
impl Drop for StreamMgr {
    fn drop(&mut self) {
        // Stop input or duplex stream, if any
        if matches!(
            self.input_stream,
            InputStreamState::Running { .. } | InputStreamState::RunningInDuplex { .. }
        ) {
            let _ = self.stopInputStream();
        }
        if matches!(self.output_stream, OutputStreamState::Running { .. }) {
            let _ = self.stopOutputStream();
        }
        // Wait until devices scan is done before shutting down stream manager
        while self.devices_scan.is_some() {
            sleep(Duration::from_millis(1));
            self.updateDeviceListFromScan();
        }

        // Decref the singleton, ordering release as it should be the last step
        // (first the streams should be stopped)
        cfg_select! {
            not(feature = "test_features") => {
                STREAMMGR_CREATED.store(false, Ordering::Release);
            },
            _ => {}
        }
    }
}