lasprs 0.14.0

//! Real-time Sound Level Meter (RtSlm).
//!
//! Computes sound levels per (fractional) octave band for a single
//! channel of the input stream, on a separate thread. The results are
//! obtained by the main thread using a triple buffer.

use crate::config::*;
use crate::daq::{InStreamMsg, StreamDirection, StreamMetaData, StreamMgr};
use crate::slm::{SLM, SLMResult, SLMSettings};
use anyhow::{Result, anyhow, ensure};
use crossbeam::channel::{Receiver, Sender, unbounded};
use std::sync::Arc;
use std::time::Duration;
use triple_buffer::triple_buffer;

/// Messages to the RtSlm processing thread
#[derive(Debug)]
enum RtSlmMessage {
    StopThread,
    ResetStatus,
}

/// Result coming from the Real time SLM computation engine
#[derive(Debug, Clone)]
pub enum RtSlmResult {
    /// New SLM result: per-band statistics, plus decimated levels vs
    /// time for the latest block (when decimation is enabled in the
    /// settings)
    NewResult(SLMResult),
    /// New stream metadata
    NewMeta(Arc<StreamMetaData>),
    /// An error occurred (e.g. channel index out of bounds)
    Error(String),
}

/// Real time Sound Level Meter for a single channel of the input
/// stream. The SLM runs on a separate thread; the latest result can be
/// obtained with [RtSlm::get_last], using a triple buffer so that the
/// main thread never blocks.
#[cfg_attr(feature = "python-bindings", gen_stub_pyclass, pyclass)]
#[derive(Debug)]
pub struct RtSlm {
    /// Triple buffer output: latest result, read by the main thread
    status: triple_buffer::Output<Option<RtSlmResult>>,

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

impl RtSlm {
    /// Create a new Real time Sound Level Meter for a single channel of
    /// the input stream.
    ///
    /// # Arguments
    ///
    /// * `mgr` - Stream manager. An input stream must be running.
    /// * `settings` - SLM settings (band descriptors, weightings, Lref).
    ///   The sample rate is set to the stream's sample rate.
    /// * `channel` - Input channel to compute levels for.
    ///
    /// # Errors
    ///
    /// When no input stream is running, or when the channel index is
    /// out of bounds.
    pub fn new(mgr: &mut StreamMgr, settings: SLMSettings, channel: usize) -> Result<RtSlm> {
        let meta = mgr
            .getStreamMetaData(StreamDirection::Input)
            .ok_or_else(|| anyhow!("Cannot create RtSlm: no input stream is running."))?;
        ensure!(
            channel < meta.nchannels(),
            "Channel index {} out of bounds: input stream has {} channels.",
            channel,
            meta.nchannels()
        );

        // The SLM must be configured with the stream's sample rate.
        let mut settings = settings;
        settings.fs = meta.samplerate;

        // Create the SLM on this thread; it is moved to the worker
        // thread below. (In the future SLM::new may return a Result.)
        let slm = SLM::new(settings, None);

        let (status_tx, status_rx) = triple_buffer(&None);
        let (sender, rxmsg) = unbounded();
        Self::startThread(slm, channel, status_tx, mgr, rxmsg);

        Ok(RtSlm {
            status: status_rx,
            sender,
        })
    }

    fn startThread(
        mut slm: SLM,
        channel: usize,
        mut status_tx: triple_buffer::Input<Option<RtSlmResult>>,
        smgr: &mut StreamMgr,
        rxmsg: Receiver<RtSlmMessage>,
    ) {
        // Obtain messages from stream manager
        let (tx, rxstream) = unbounded();

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

        rayon::spawn(move || {
            // Whether the channel index is valid for the current stream.
            // Validated at construction; re-checked when metadata arrives.
            let mut channel_valid = true;
            let mut last_result: Option<SLMResult> = None;
            let mut pending_error: Option<String> = None;

            'mainloop: loop {
                if let Ok(msg) = rxstream.recv_timeout(Duration::from_millis(10)) {
                    match msg {
                        InStreamMsg::StreamStarted(new_meta, _) => {
                            // Ignore pre-capture data: just run on live
                            // data. The SLM warms up on its own; results
                            // are None during warm-up, so nothing is
                            // published in that period.
                            channel_valid = channel < new_meta.nchannels();
                            if channel_valid {
                                // A new stream has started: reset the
                                // SLM statistics and publish the new
                                // metadata.
                                slm.reset_stats();
                                // Input::write() already publishes the value
                                // to the consumer.
                                status_tx.write(Some(RtSlmResult::NewMeta(new_meta)));
                            } else {
                                pending_error = Some(format!(
                                    "Channel index {} out of bounds: input stream has {} channels.",
                                    channel,
                                    new_meta.nchannels()
                                ));
                            }
                        }
                        InStreamMsg::StreamStopped | InStreamMsg::StreamError(_) => {
                            // The stream stopped: stop the thread. It is
                            // up to the main thread to create a new RtSlm
                            // instance when the stream restarts.
                            break 'mainloop;
                        }
                        InStreamMsg::InStreamData(id) => {
                            if channel_valid {
                                // Do apply sensitivity here
                                let flt = id.getFloatData(true);
                                let td = flt.column(channel).to_owned();
                                last_result = slm.run(td.as_slice().unwrap(), true);
                            }
                        }
                    }
                }

                // Check for messages and act accordingly
                if let Ok(msg) = rxmsg.recv_timeout(Duration::from_millis(1)) {
                    match msg {
                        RtSlmMessage::StopThread => {
                            break 'mainloop;
                        }
                        RtSlmMessage::ResetStatus => {
                            slm.reset_stats();
                        }
                    }
                }

                // Communicate the latest result, if any.
                if let Some(result) = last_result.take() {
                    // Input::write() already publishes the value to the
                    // consumer.
                    status_tx.write(Some(RtSlmResult::NewResult(result)));
                }
                if let Some(error) = pending_error.take() {
                    status_tx.write(Some(RtSlmResult::Error(error)));
                }
            } // End of mainloop
        });
    }

    /// Take a copy of the last updated result, if any.
    pub fn get_last(&mut self) -> Option<RtSlmResult> {
        self.status.read().clone()
    }

    /// Reset the SLM statistics, start with a clean slate
    pub fn reset(&self) {
        let _ = self.sender.send(RtSlmMessage::ResetStatus);
    }
}
impl Drop for RtSlm {
    fn drop(&mut self) {
        // The worker thread may already have exited when the stream
        // stopped; ignore send errors.
        let _ = self.sender.send(RtSlmMessage::StopThread);
    }
}
#[cfg(feature = "python-bindings")]
#[cfg_attr(feature = "python-bindings", gen_stub_pymethods, pymethods)]
impl RtSlm {
    #[new]
    #[gen_stub(skip)]
    fn new_py(smgr: &mut StreamMgr, settings: SLMSettings, channel: usize) -> PyResult<Self> {
        Ok(RtSlm::new(smgr, settings, channel)?)
    }
    // This method does not forward the metadata. Should come from somewhere else
    #[pyo3(name = "get_last")]
    fn get_last_py(&mut self) -> Option<SLMResult> {
        match self.get_last() {
            Some(RtSlmResult::NewResult(res)) => Some(res),
            _ => 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 RtSlm:
            def __new__(smgr: StreamMgr, settings: SLMSettings, channel: int) -> RtSlm:
                """Create a new Real time Sound Level Meter for a single channel of the
                input stream. Raises when no input stream is running, or when the
                channel index is out of bounds."""
        "#
    }
}
    },
    _ => {}
}

#[cfg(test)]
mod test {
    use super::*;
    use crate::{
        FreqWeighting, StrictlyPositive, TimeWeighting,
        daq::{DaqApiDescriptor, DaqChannel, DaqConfig, DataType, StreamType},
        filter::StandardFilterDescriptor,
        slm::SLMSettingsBuilder,
    };
    use std::thread;

    /// Requires the `loopback-api` feature (enabled via `test_features`).
    #[cfg(feature = "loopback-api")]
    #[test]
    fn test_rtslm1() -> Result<()> {
        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 desc = StandardFilterDescriptor::Overall().unwrap();
        let settings = SLMSettingsBuilder::default()
            .fs(meta.samplerate)
            .timeWeighting(TimeWeighting::Fast {})
            .freqWeighting(FreqWeighting::Z)
            .filterDescriptors([desc])
            .Lref(StrictlyPositive::new(2e-5).unwrap())
            .build()
            .unwrap();

        // Out-of-bounds channel should error at construction
        assert!(
            RtSlm::new(&mut smgr, settings.clone(), 5).is_err(),
            "Out-of-bounds channel should error"
        );

        let mut rtslm = RtSlm::new(&mut smgr, settings, 0)?;

        // Wait for a result; the SLM needs ~0.4 s warm-up (Fast weighting).
        let mut got_result = false;
        for _ in 0..50 {
            if let Some(RtSlmResult::NewResult(_)) = rtslm.get_last() {
                got_result = true;
                break;
            }
            thread::sleep(Duration::from_millis(100));
        }
        assert!(got_result, "Should receive an SLM result within 5 seconds");
        drop(rtslm);
        Ok(())
    }
}