lasprs 0.14.1

Library for Acoustic Signal Processing (Rust edition, with optional Python bindings via pyo3)
use super::*;
use crate::filter::StandardFilterDescriptor;
use crate::*;
use std::collections::BTreeMap;
use std::sync::Arc;

#[cfg(feature = "python-bindings")]
use pyo3::{Py, types::PyDict};

/// Result produced by [SLM::run] after warm-up is complete and time is
/// non-negative (when an initial time was set).
///
/// For accumulated results over a full measurement, `t` may carry the
/// time axis for the decimated levels-vs-time output.
#[cfg_attr(feature = "python-bindings", gen_stub_pyclass, pyclass(from_py_object))]
#[derive(Debug, Clone)]
pub struct SLMResult {
    /// Maximum level per band [dB ref Lref], keyed by filter descriptor.
    pub Lmax: BTreeMap<StandardFilterDescriptor, Flt>,
    /// Peak level per band [dB ref Lref], keyed by filter descriptor.
    pub Lpk: BTreeMap<StandardFilterDescriptor, Flt>,
    /// Equivalent level per band [dB ref Lref], keyed by filter descriptor.
    pub Leq: BTreeMap<StandardFilterDescriptor, Flt>,
    /// Levels vs time for each band within this block (or accumulated
    /// over a full measurement). Keyed by filter descriptor, each value
    /// is a per-sample (or per-decimated-sample) time series.
    /// Only present when `provide_output` was true in [SLM::run].
    pub Lt: Option<BTreeMap<StandardFilterDescriptor, Vec<Flt>>>,
    /// Optional time axis [s] for the levels-vs-time data. Set when
    /// this result represents an accumulated measurement output;
    /// `None` for when levels vs time are not available.
    pub t: Option<Vec<Flt>>,
}

#[cfg(feature = "python-bindings")]
impl SLMResult {
    /// Convert a statistics map (Lmax, Lpk, or Leq) to a Python dict,
    /// keyed by filter descriptor. The statistics map is selected with
    /// a closure, just as done in [SLM::levels_from] for the Rust API.
    fn stats_to_py<'py, F>(&self, py: Python<'py>, stat_selector: F) -> PyResult<Bound<'py, PyDict>>
    where
        F: Fn(&Self) -> &BTreeMap<StandardFilterDescriptor, Flt>,
    {
        let dict = PyDict::new(py);
        for (k, v) in stat_selector(self) {
            let py_key = Py::new(py, *k)?.into_bound(py);
            dict.set_item(py_key, *v)?;
        }
        Ok(dict)
    }
}

#[cfg(feature = "python-bindings")]
#[cfg_attr(feature = "python-bindings", pymethods)]
// The stubs are generated manually below, as otherwise the dictionary keys and values
// are not typed.
impl SLMResult {
    #[pyo3(name = "Lmax")]
    fn Lmax_py<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyDict>> {
        self.stats_to_py(py, |res| &res.Lmax)
    }
    #[pyo3(name = "Lpk")]
    fn Lpk_py<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyDict>> {
        self.stats_to_py(py, |res| &res.Lpk)
    }
    #[pyo3(name = "Leq")]
    fn Leq_py<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyDict>> {
        self.stats_to_py(py, |res| &res.Leq)
    }
    #[pyo3(name = "t")]
    fn t_py<'py>(&self, py: Python<'py>) -> Option<Bound<'py, PyArray1<Flt>>> {
        self.t.as_ref().map(|t| t.to_pyarray(py))
    }
    #[pyo3(name = "Lt")]
    fn Lt_py<'py>(&self, py: Python<'py>) -> PyResult<Option<Bound<'py, PyDict>>> {
        if let Some(ref lt) = self.Lt {
            let dict = PyDict::new(py);
            for (k, v) in lt {
                let py_key = Py::new(py, *k)?.into_bound(py);
                dict.set_item(py_key, PyArray1::from_vec(py, v.clone()))?;
            }
            Ok(Some(dict))
        } else {
            Ok(None)
        }
    }

    fn __repr__(&self) -> String {
        format!("{self:#?}")
    }
}

cfg_select! {
    feature = "python-bindings" => {
pyo3_stub_gen::inventory::submit! {
    gen_methods_from_python! {
        r#"
        import numpy
        import numpy.typing
        import typing

        class SLMResult:
            @property
            def Lmax(self) -> dict[StandardFilterDescriptor, float]:
                """ Maximum level per band [dB ref Lref], keyed by filter descriptor. """
            @property
            def Lpk(self) -> dict[StandardFilterDescriptor, float]:
                """ Peak level per band [dB ref Lref], keyed by filter descriptor. """
            @property
            def Leq(self) -> dict[StandardFilterDescriptor, float]:
                """ Equivalent level per band [dB ref Lref], keyed by filter descriptor. """
            @property
            def t(self) -> typing.Optional[numpy.typing.NDArray[numpy.float64]]:
                """ Optional time axis [s] for the Lt data (measurement result). """
            @property
            def Lt(self) -> typing.Optional[dict[StandardFilterDescriptor, numpy.typing.NDArray[numpy.float64]]]:
                """ Levels vs time keyed by filter descriptor (dict of str → numpy array).
                Only present when provide_output was true. """
        "#
    }
}
    },
    _ => {}
}