lasprs 0.14.0

use crate::I;
use crate::common::{ThreadPriority, spawn};
use crate::config::*;
use crate::daq::{InStreamMsg, StreamHandler, StreamMetaData, StreamMgr};
use crate::ps::ApsSettings;
use crate::ps::{AvPowerSpectra, CPSResult};
use anyhow::Result;
use crossbeam::channel::{Receiver, Sender, unbounded};
use parking_lot::Mutex;
use std::ops::Deref;
use std::sync::Arc;
use std::time::Duration;

type SharedRtApsStatus = Arc<Mutex<Option<RtApsResult>>>;

#[derive(Debug)]
enum RtApsMessage {
    StopThread,
    ResetStatus,
}
/// Result type coming from Real time Averaged Power Spectra computation engine
#[derive(Debug)]
pub enum RtApsResult {
    /// New result
    NewResult(CPSResult),
    /// New metadata
    NewMeta(Arc<StreamMetaData>),
}

/// Real time power spectra viewer. Shows cross-power or auto-power signal 'time-dependent'
#[cfg_attr(feature = "python-bindings", gen_stub_pyclass, pyclass)]
#[derive(Debug)]
pub struct RtAps {
    /// Storage for optional last result
    status: SharedRtApsStatus,

    // For sending messages to the data processing thread
    sender: Sender<RtApsMessage>,
}

impl RtAps {
    /// Create new Real time power spectra computing engine.
    pub fn new(mgr: &mut StreamMgr, settings: ApsSettings) -> RtAps {
        // Handler needs to be created here.
        let status = Arc::new(Mutex::new(None));

        let (sender, rx) = unbounded();
        let aps = AvPowerSpectra::new(settings);
        Self::startThread(aps, status.clone(), mgr, rx);

        RtAps { status, sender }
    }

    fn startThread(
        mut aps: AvPowerSpectra,
        status: SharedRtApsStatus,
        smgr: &mut StreamMgr,
        rxmsg: Receiver<RtApsMessage>,
    ) {
        // Obtain messages from stream manager
        let (tx, rxstream) = unbounded();

        // Add queue sender part of queue to stream manager
        smgr.addInQueue(tx);

        spawn(
            move || {
                // What is running on the thread
                'mainloop: loop {
                    let mut last_cps: Option<CPSResult> = None;
                    let mut meta: Option<Arc<StreamMetaData>> = None;

                    if let Ok(msg) = rxstream.recv_timeout(Duration::from_millis(10)) {
                        match msg {
                            InStreamMsg::StreamStarted(new_meta, _) => {
                                aps.reset();
                                last_cps = None;
                                meta = Some(new_meta);
                            }
                            InStreamMsg::StreamStopped | InStreamMsg::StreamError(_) => {
                                debug_assert!(meta.is_none());
                                last_cps = None;
                            }
                            InStreamMsg::InStreamData(id) => {
                                debug_assert!(meta.is_none());
                                // Do apply sensitivity here
                                let flt = id.getFloatData(true);
                                if let Some(cpsresult) = aps.compute_last(flt.view()) {
                                    last_cps = Some(cpsresult.clone());
                                }
                            }
                        }
                    }

                    // Check for messages and act accordingly
                    if let Ok(msg) = rxmsg.recv_timeout(Duration::from_millis(1)) {
                        match msg {
                            RtApsMessage::StopThread => {
                                let mut status = status.lock();
                                *status = None;
                                break 'mainloop;
                            }
                            RtApsMessage::ResetStatus => {
                                aps.reset();
                            }
                        }
                    }

                    // Communicate last result, if any.
                    'commscope: {
                        let mut status = status.lock();

                        if let Some(newmeta) = meta.take() {
                            // New metadata has arrived. This is always the first
                            // thing to push. Only when it is read, we will start
                            // pushing actual data.
                            *status = Some(RtApsResult::NewMeta(newmeta));
                            break 'commscope;
                        }

                        if let Some(RtApsResult::NewMeta(_)) = status.deref() {
                            // New metadata is not yet read by reading thread. It
                            // basically means we are not yet ready to give actual
                            // data back.
                            break 'commscope;
                        }
                        // Move last_cps into mutex.
                        if let Some(last_cps) = last_cps.take() {
                            *status = Some(RtApsResult::NewResult(last_cps));
                        }
                    } // end of commscope
                } // End of mainloop
            },
            ThreadPriority::Low,
        );
    }

    /// Take last updated result.
    pub fn get_last(&self) -> Option<RtApsResult> {
        let mut lck = self.status.lock();
        lck.take()
    }

    /// Reset power spectra estimator, start with a clean sleeve
    pub fn reset(&self) {
        self.sender.send(RtApsMessage::ResetStatus).unwrap();
    }
}
impl Drop for RtAps {
    fn drop(&mut self) {
        self.sender.send(RtApsMessage::StopThread).unwrap();
    }
}
#[cfg(feature = "python-bindings")]
#[cfg_attr(feature = "python-bindings", gen_stub_pymethods, pymethods)]
impl RtAps {
    #[new]
    #[gen_stub(skip)]
    fn new_py(smgr: &mut StreamMgr, settings: ApsSettings) -> Self {
        RtAps::new(smgr, settings)
    }
    // This method does not forward the metadata. Should come from somewhere else
    #[pyo3(name = "get_last")]
    fn get_last_py(&self) -> Option<CPSResult> {
        let res = self.get_last();
        if let Some(RtApsResult::NewResult(res)) = res {
            return Some(res.clone());
        }
        None
    }

    #[pyo3(name = "reset")]
    fn reset_py(&self) {
        self.reset()
    }
}
cfg_select! {
    feature = "python-bindings" => {
pyo3_stub_gen::inventory::submit! {
    // Stub for the `#[new]` constructor, which is marked `#[gen_stub(skip)]`
    // in the `#[pymethods]` impl. See `rtview.rs` for the same pattern.
    gen_methods_from_python! {
        r#"
        class RtAps:
            def __new__(smgr: StreamMgr, settings: ApsSettings) -> RtAps:
                """Create new Real time power spectra computing engine. Raises when
                no input stream is running."""
        "#
    }
}
    },
    _ => {}
}

#[cfg(test)]
mod test {
    use std::time::Duration;

    use anyhow::{Result, anyhow, bail};

    use super::*;
    use crate::{
        daq::{
            DaqApiDescriptor, DaqChannel, DaqConfig, DataType, StreamDirection, StreamMgr,
            StreamType,
        },
        ps::ApsSettingsBuilder,
    };
    use std::thread;
    #[test]
    fn test_rtaps1() -> Result<(), anyhow::Error> {
        {
            let mut smgr = StreamMgr::new_with_devices();
            let cfg = DaqConfig {
                api: DaqApiDescriptor::Loopback,
                device_name: "Loopback".into(),
                inchannel_config: vec![
                    DaqChannel::defaultAudio("ch0"),
                    DaqChannel::defaultAudio("ch1"),
                ],
                outchannel_config: vec![],
                dtype: DataType::F32,
                sampleRateIndex: 0,
                framesPerBlockIndex: 0,
                monitorOutput: false,
            };
            smgr.startStream(StreamType::Input, &cfg)?;
            let meta = smgr
                .getStreamMetaData(StreamDirection::Input)
                .ok_or_else(|| anyhow!("Stream is not running"))?;

            let settings = ApsSettingsBuilder::default()
                .nfft(2048)
                .fs(meta.samplerate)
                .build()
                .unwrap();
            let rtaps = RtAps::new(&mut smgr, settings);
            thread::sleep(Duration::from_secs(1));
            drop(rtaps);
        }
        Ok(())
    }
}