Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "lasprs"
-version = "0.6.3"
+version = "0.6.2"
 edition = "2021"
 authors = ["J.A. de Jong <j.a.dejong@ascee.nl>"]
 description = "Library for Acoustic Signal Processing (Rust edition, with optional Python bindings via pyo3)"

src/bin/lasp_inputdefault.rs

@@ -41,7 +41,7 @@ fn main() -> Result<()> {
                         println!("Stream started metadata: {meta:#?}");
                     },
                     InStreamMsg::InStreamData(_) => {}
-                    _ => { println!("Instreamdata arrived...");}
+                    _ => { println!("Other msg...");}
                 }
             }
             Err(e) => match e {

src/daq/streammetadata.rs

@@ -1,10 +1,8 @@
 use super::*;
 use crate::config::Flt;
 use anyhow::Result;
-
 /// Stream metadata. All information required for properly interpreting the raw
 /// data that is coming from the stream.
-#[cfg_attr(feature = "python-bindings", pyclass)]
 #[derive(Clone, Debug)]
 pub struct StreamMetaData {
     /// Information for each channel in the stream
@@ -60,26 +58,3 @@ impl StreamMetaData {
         self.channelInfo.len()
     }
 }
-
-/// Simple getters for all sub-attributes of the stream metadata.
-#[cfg(feature = "python-bindings")]
-#[cfg_attr(feature = "python-bindings", pymethods)]
-impl StreamMetaData {
-    #[getter]
-    pub fn channelInfo(&self) -> Vec<DaqChannel> {
-        self.channelInfo.clone()
-    }
-    #[getter]
-    fn rawDatatype(&self) -> DataType {
-        self.rawDatatype
-    }
-
-    #[getter]
-    fn samplerate(&self) -> Flt {
-        self.samplerate
-    }
-    #[getter]
-    fn framesPerBlock(&self) -> usize {
-        self.framesPerBlock
-    }
-}

src/daq/streammgr.rs

@@ -111,12 +111,6 @@ impl StreamMgr {
     fn setSiggen_py(&mut self, siggen: Siggen) {
         self.setSiggen(siggen)
     }
-    #[pyo3(name = "getStreamMetaData")]
-    fn getStreamMetaData_py(&self, st: StreamType) -> Option<StreamMetaData> {
-        // Unfortunately (but not really, only cosmetically), the underlying
-        // value (not the Arc) has to be cloned.
-        self.getStreamMetaData(st).map(|b| (*b).clone())
-    }
 }
 impl Default for StreamMgr {
     fn default() -> Self {
@@ -280,6 +274,7 @@ impl StreamMgr {
                     }
                 }
                 if let Ok(msg) = rx.recv_timeout(time::Duration::from_millis(10)) {
+
                     sendMsgToAllQueuesRemoveUnused(&mut iqueues, msg);
                 }
             }

src/lib.rs

@@ -55,31 +55,22 @@ fn lasprs(m: &Bound<'_, PyModule>) -> PyResult<()> {
     m.add_class::<filter::Biquad>()?;
     m.add_class::<filter::SeriesBiquad>()?;
     m.add_class::<filter::BiquadBank>()?;
+    m.add_class::<siggen::Siggen>()?;
     m.add_class::<filter::FilterSpec>()?;
     m.add_class::<filter::ZPKModel>()?;
     m.add_class::<filter::StandardFilterDescriptor>()?;
-
-    // Signal generator
-    m.add_class::<siggen::Siggen>()?;
-
-    // SLM
     m.add_class::<slm::TimeWeighting>()?;
+    m.add_class::<ps::FreqWeighting>()?;
     m.add_class::<slm::SLMSettings>()?;
     m.add_class::<slm::SLM>()?;
     m.add_class::<slm::TimeWeighting>()?;
-
-    // Power spectra classes
-    m.add_class::<ps::FreqWeighting>()?;
     m.add_class::<ps::WindowType>()?;
     m.add_class::<ps::Overlap>()?;
     m.add_class::<ps::ApsMode>()?;
     m.add_class::<ps::ApsSettings>()?;
     m.add_class::<ps::AvPowerSpectra>()?;
-    
-    // Real time classes
     m.add_class::<rt::PPM>()?;
     m.add_class::<rt::ClipState>()?;
-    m.add_class::<rt::RtAps>()?;
 
     Ok(())
 }

src/rt/ppm.rs

@@ -5,6 +5,7 @@ use crate::{config::*, FreqWeighting, StandardFilterDescriptor};
 use crate::{daq::StreamMgr, Dcol};
 use crossbeam::channel::{unbounded, Receiver, Sender};
 use crossbeam::utils::Backoff;
+use ndarray_stats::QuantileExt;
 use parking_lot::Mutex;
 use std::default;
 use std::ops::DerefMut;
@@ -13,10 +14,10 @@ use std::time::{Duration, Instant};
 
 /// When the level reaches ALMOST_CLIPPED_REL_AMP in amplitude, w.r.t. to the
 /// full scale, we mark a channel as almost clipped.
-const ALMOST_CLIPPED_REL_AMP: Flt = 0.98;
+const ALMOST_CLIPPED_REL_AMP: Flt = 0.95;
 
 /// If clipping occured, this is the time it keeps saying 'signal is clipped'
-const CLIP_INDICATOR_WAIT_S: Duration = Duration::from_secs(2);
+const CLIP_INDICATOR_WAIT_S: Duration = Duration::from_secs(4);
 
 /// If the signal level falls below this value, we indicate that the signal level is low.
 const LEVEL_THRESHOLD_FOR_LOW_LEVEL: Flt = -50.;
@@ -68,11 +69,9 @@ impl PPM {
                 *status = Default::default();
                 slms.clear();
             };
+            let mut sleep_dur = Duration::from_millis(50);
             loop {
-                if let Some(msg) = rxstream
+                for msg in rxstream.try_iter() {
-                    .recv_timeout(std::time::Duration::from_millis(10))
-                    .ok()
-                {
                     match msg {
                         InStreamMsg::InStreamData(d) => {
                             let mut status = status.lock();
@@ -121,6 +120,11 @@ impl PPM {
                             resetall(&mut slms);
                         }
                         InStreamMsg::StreamStarted(meta) => {
+                            // Compute 'good' sleep time, as half the approximate time between two blocks
+                            let block_time_us =
+                                ((1e6 * meta.framesPerBlock as Flt) / meta.samplerate) as u64;
+                            sleep_dur = Duration::from_micros(std::cmp::max(1, block_time_us / 2));
+
                             // Re-initalize sound level meters
                             slms.clear();
                             let mut s = status.lock();
@@ -132,8 +136,6 @@ impl PPM {
                                     .freqWeighting(FreqWeighting::Z)
                                     .Lref(1.0)
                                     .timeWeighting(TimeWeighting::Impulse {})
-                                    // .timeWeighting(TimeWeighting::CustomAsymmetric { tup: 0.01, tdown: 1.0 })
-                                    // .timeWeighting(TimeWeighting::Fast {})
                                     .filterDescriptors([
                                         StandardFilterDescriptor::Overall().unwrap()
                                     ])
@@ -150,7 +152,10 @@ impl PPM {
                                 });
                             });
                         }
-                        InStreamMsg::StreamStopped => {}
+                        InStreamMsg::StreamStopped => {
+                            // Restore sleep time to sensible polling default
+                            sleep_dur = Duration::from_millis(50);
+                        }
                     }
                     // Loop over any messages coming in from main thread
                     for msg in rxmsg.try_iter() {
@@ -166,6 +171,7 @@ impl PPM {
                             }
                         }
                     }
+                    std::thread::sleep(sleep_dur);
                 }
             }
         });

src/rt/rtaps.rs

@@ -1,26 +1,23 @@
+use std::ops::Deref;
 use crate::config::*;
+use std::thread::{self, JoinHandle};
+
 use crate::daq::{InStreamMsg, StreamHandler, StreamMetaData, StreamMgr};
 use crate::ps::ApsSettings;
 use crate::ps::{AvPowerSpectra, CPSResult};
 use crate::I;
 use anyhow::Result;
-use crossbeam::channel::{unbounded, Receiver, Sender};
 use parking_lot::Mutex;
 use rayon::ThreadPool;
-use std::ops::Deref;
 use std::sync::Arc;
-use std::thread::{self, JoinHandle};
-use std::time::Duration;
-
-type SharedRtApsStatus = Arc<Mutex<Option<RtApsResult>>>;
 
 #[derive(Debug)]
-enum RtApsMessage {
+enum RtApsComm {
-    StopThread,
+    CommStopThread,
-    ResetStatus,
+    NewResult(CPSResult),
+    NewMeta(Arc<StreamMetaData>),
 }
-/// Result type coming from Real time Averaged Power Spectra computation engine
+/// Result type coming from Real time Averaged Power Spectra computing engine
-#[derive(Debug)]
 pub enum RtApsResult {
     /// New result
     NewResult(CPSResult),
@@ -29,48 +26,34 @@ pub enum RtApsResult {
 }
 
 /// Real time power spectra viewer. Shows cross-power or auto-power signal 'time-dependent'
-#[cfg_attr(feature = "python-bindings", pyclass)]
-#[derive(Debug)]
 pub struct RtAps {
     /// Storage for optional last result
-    status: SharedRtApsStatus,
+    comm: Arc<Mutex<Option<RtApsComm>>>,
-
+    /// Settings used for real time power spectra.
-    // For sending messages to the data processing thread
+    pub settings: ApsSettings,
-    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 handler = StreamHandler::new(mgr);
+        let last_result = Arc::new(Mutex::new(None));
+        let last_result2 = last_result.clone();
+        let settings2 = settings.clone();
 
-        let (sender, rx) = unbounded();
+        let mut aps = AvPowerSpectra::new(settings);
-        let aps = AvPowerSpectra::new(settings);
-        Self::startThread(aps, status.clone(), mgr, rx);
 
-        RtAps { status, sender }
+        let thread = std::thread::spawn(move || {
-    }
+            // println!("Thread started...");
-
+            let rx = handler.rx;
-    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);
-
-        rayon::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 Some(msg) = rxstream.recv_timeout(Duration::from_millis(10)).ok() {
+                if let Some(msg) = rx.recv_timeout(std::time::Duration::from_millis(10)).ok() {
                     match msg {
                         InStreamMsg::StreamStarted(new_meta) => {
                             aps.reset();
@@ -91,33 +74,22 @@ impl RtAps {
                     }
                 }
 
-                // Check for messages and act accordingly
-                if let Some(msg) = rxmsg.recv_timeout(Duration::from_millis(10)).ok() {
-                    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();
+                    let mut last_result_lock = last_result.lock();
 
+                    if let Some(RtApsComm::CommStopThread) = *last_result_lock {
+                        break 'mainloop;
+                    }
                     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));
+                        *last_result_lock = Some(RtApsComm::NewMeta(newmeta));
                         break 'commscope;
                     }
 
-                    if let Some(RtApsResult::NewMeta(_)) = status.deref() {
+                    if let Some(RtApsComm::NewMeta(_)) = *last_result_lock {
                         // New metadata is not yet read by reading thread. It
                         // basically means we are not yet ready to give actual
                         // data back.
@@ -125,39 +97,36 @@ impl RtAps {
                     }
                     // Move last_cps into mutex.
                     if let Some(last_cps) = last_cps.take() {
-                        *status = Some(RtApsResult::NewResult(last_cps));
+                        *last_result_lock = Some(RtApsComm::NewResult(last_cps));
                     }
-                } // end of commscope
+                }
-            } // End of mainloop
+            } // End of loop
         });
-    }
+        assert!(!thread.is_finished());
 
-    /// Take last updated result.
+        RtAps {
+            comm: last_result2,
+            settings: settings2,
+        }
+    }
+    /// Get last computed value. When new stream metadata is
     pub fn get_last(&self) -> Option<RtApsResult> {
-        let mut lck = self.status.lock();
+        let mut lck = self.comm.lock();
-        lck.take()
+        let res = lck.take();
+        if let Some(res) = res {
+            match res {
+                RtApsComm::CommStopThread => panic!("BUG: CommStopThread should never be set!"),
+                RtApsComm::NewMeta(m) => return Some(RtApsResult::NewMeta(m)),
+                RtApsComm::NewResult(r) => return Some(RtApsResult::NewResult(r)),
+            }
+        }
+        None
     }
 }
 impl Drop for RtAps {
     fn drop(&mut self) {
-        self.sender.send(RtApsMessage::StopThread).unwrap();
+        let mut lck = self.comm.lock();
-    }
+        *lck = Some(RtApsComm::CommStopThread);
-}
-#[cfg(feature = "python-bindings")]
-#[cfg_attr(feature = "python-bindings", pymethods)]
-impl RtAps {
-    #[new]
-    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<'py>(&self, py: Python<'py>) -> Option<Bound<'py, PyArray3<Cflt>>> {
-        let res = self.get_last();
-        if let Some(RtApsResult::NewResult(res)) = res {
-            return Some(res.to_pyarray_bound(py));
-        }
-        None
     }
 }
 

src/slm/slm.rs

@@ -92,7 +92,7 @@ impl SLM {
         }
         let prefiltered = self.prefilter.filter(td);
 
-        let level_fun = |a| 10. * Flt::log10(a) / self.Lrefsq;
+        let level = |a| 10. * Flt::log10(a) / self.Lrefsq;
 
         let Lt_iter = self.channels.par_iter_mut().map(|ch| {
             let mut tmp = ch.bp.filter(&prefiltered);
@@ -135,7 +135,7 @@ impl SLM {
                         rectifier_down.setToDCValue(fup);
                         fup
                     } else {
-                        rectifier_up.setToDCValue(fdown);
+                        rectifier_up.setToDCValue(fup);
                         fdown
                     }
                 });
@@ -155,7 +155,7 @@ impl SLM {
             // Update last signal power coming from SLM
             ch.stat.Pt_last = *filtered_squared.last().unwrap();
             // Convert output to levels
-            filtered_squared.mapv_inplace(level_fun);
+            filtered_squared.mapv_inplace(level);
             tmp
         });
         if provide_output {

src/slm/tw.rs

@@ -49,10 +49,6 @@ impl TimeWeighting {
         use TimeWeighting::*;
         vec![Slow {}, Fast {}, Impulse {}]
     }
-    #[pyo3(name = "getLowpassPoles")]
-    fn getLowpassPoles_py(&self) -> (Flt, Option<Flt>) {
-        self.getLowpassPoles()
-    }
 }
 
 impl Default for TimeWeighting {
@@ -95,7 +91,7 @@ impl TimeWeighting {
             CustomAsymmetric { tup, tdown } => {
                 assert!(*tup > 0.);
                 assert!(*tdown > 0.);
-                (-1. / (*tup), Some(-1. / (*tdown)))
+                (-*tup, Some(-*tdown))
             }
         }
     }