lasprs 0.14.1

Library for Acoustic Signal Processing (Rust edition, with optional Python bindings via pyo3)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101

//! Peak Programme Meter functionality
use crate::{config::*, daq::StreamMgr};
mod clipstate;
mod ppm;
pub mod ppmchannel;
mod ppmdropspeed;
pub use clipstate::ClipState;
pub use ppm::PPM;
pub use ppmchannel::PPMChannelStatus;
pub use ppmdropspeed::PPMDropSpeed;
use std::{sync::Arc, time::Duration};

/// If the signal level falls below this value, we indicate that the signal level is low.
pub const LEVEL_THRESHOLD_FOR_LOW_LEVEL: Flt = -60.;

/// If the signal level falls below this value, we indicate that the signal level is low.
pub const LEVEL_THRESHOLD_FOR_HIGH_LEVEL: Flt = -10.;

/// When the level reaches ALMOST_CLIPPED_REL_AMP in amplitude, w.r.t. to the
/// full scale, we mark a channel as almost clipped.
pub const ALMOST_CLIPPED_REL_AMP: Flt = 0.98;

/// If clipping occured, this is the time it keeps saying 'signal is clipped'
pub const CLIP_INDICATOR_WAIT_S: Flt = 2.;

/// Number of seconds the peak level stays at its position before slowly
/// decreasing.
pub const PEAK_LEVEL_HOLD_TIME_S: Flt = 1.;

/// Wrapper for PPM struct.
#[cfg(feature = "python-bindings")]
#[cfg_attr(feature = "python-bindings", gen_stub_pyclass, pyclass(name = "PPM"))]
pub struct PPMWrapper {
    PPM: Arc<PPM>,
}

#[cfg(feature = "python-bindings")]
#[cfg_attr(feature = "python-bindings", gen_stub_pymethods, pymethods)]
impl PPMWrapper {
    #[staticmethod]
    #[pyo3(name = "newMonitor")]
    #[gen_stub(skip)]
    fn newMonitor_py(smgr: &mut StreamMgr, dropSpeed: PPMDropSpeed) -> Self {
        Self {
            PPM: PPM::newMonitor(smgr, dropSpeed),
        }
    }
    #[staticmethod]
    #[pyo3(name = "newInput")]
    #[gen_stub(skip)]
    fn newInput_py(smgr: &mut StreamMgr, dropSpeed: PPMDropSpeed) -> Self {
        Self {
            PPM: PPM::newInput(smgr, dropSpeed),
        }
    }

    #[pyo3(name = "getState")]
    /// Get the current state of the PPM instance.
    ///
    /// Returns a tuple of (levels, peak_levels, clips).
    fn getState_py<'py>(
        &self,
        py: Python<'py>,
    ) -> (
        Bound<'py, PyArray1<Flt>>,
        Bound<'py, PyArray1<Flt>>,
        Vec<ClipState>,
    ) {
        let (levels, peak_levels, clips) = self.PPM.getState();
        let levels = levels.to_pyarray(py);
        let peak_levels = peak_levels.to_pyarray(py);
        (levels, peak_levels, clips)
    }
}

cfg_select! {
    feature = "python-bindings" => {
pyo3_stub_gen::inventory::submit! {
    // OK, one way or another pyo3_stub_gen translates PPMWrapper to PPM, so we
    // need to use PPMWrapper here
    gen_methods_from_python! {
        r#"
        import numpy
        import numpy.typing

        class PPMWrapper:
            @staticmethod
            def newMonitor(smgr: StreamMgr, dropSpeed: PPMDropSpeed) -> PPMWrapper:
               """Create a new monitor PPM instance"""
            @staticmethod
            def newInput(smgr: StreamMgr, dropSpeed: PPMDropSpeed) -> PPMWrapper:
                """Create a new input PPM instance"""
        "#
    }
}
    },
    _ => {}
}

//
// import numpy as np