feat(pyargus): add interpolation method parameter for Signal construction

Author: anand-bala

pyargus/argus/__init__.py

@@ -2,10 +2,10 @@
 
 from typing import List, Optional, Tuple, Type, Union
 
-from argus import _argus
-from argus._argus import dtype
-from argus.exprs import ConstBool, ConstFloat, ConstInt, ConstUInt, VarBool, VarFloat, VarInt, VarUInt
-from argus.signals import BoolSignal, FloatSignal, IntSignal, Signal, UnsignedIntSignal
+from . import _argus
+from ._argus import dtype
+from .exprs import ConstBool, ConstFloat, ConstInt, ConstUInt, VarBool, VarFloat, VarInt, VarUInt
+from .signals import BoolSignal, FloatSignal, IntSignal, Signal, UnsignedIntSignal
 
 try:
     __doc__ = _argus.__doc__

pyargus/argus/_argus.pyi

@@ -1,4 +1,4 @@
-from typing import ClassVar, Protocol, final
+from typing import ClassVar, Literal, Protocol, TypeAlias, final
 
 from typing_extensions import Self
 
@@ -146,36 +146,46 @@ class Signal:
 @final
 class BoolSignal(Signal):
     @classmethod
-    def constant(cls, value: bool) -> Self: ...
+    def constant(cls, value: bool, *, interpolation_method: _InterpolationMethod = "linear") -> Self: ...
     @classmethod
-    def from_samples(cls, samples: list[tuple[float, bool]]) -> Self: ...
+    def from_samples(
+        cls, samples: list[tuple[float, bool]], *, interpolation_method: _InterpolationMethod = "linear"
+    ) -> Self: ...
     def push(self, time: float, value: bool) -> None: ...
     def at(self, time: float) -> bool | None: ...
 
+_InterpolationMethod: TypeAlias = Literal["linear", "constant"]
+
 @final
 class IntSignal(Signal):
     @classmethod
-    def constant(cls, value: int) -> Self: ...
+    def constant(cls, value: int, *, interpolation_method: _InterpolationMethod = "linear") -> Self: ...
     @classmethod
-    def from_samples(cls, samples: list[tuple[float, int]]) -> Self: ...
+    def from_samples(
+        cls, samples: list[tuple[float, int]], *, interpolation_method: _InterpolationMethod = "linear"
+    ) -> Self: ...
     def push(self, time: float, value: int) -> None: ...
     def at(self, time: float) -> int | None: ...
 
 @final
 class UnsignedIntSignal(Signal):
     @classmethod
-    def constant(cls, value: int) -> Self: ...
+    def constant(cls, value: int, *, interpolation_method: _InterpolationMethod = "linear") -> Self: ...
     @classmethod
-    def from_samples(cls, samples: list[tuple[float, int]]) -> Self: ...
+    def from_samples(
+        cls, samples: list[tuple[float, int]], *, interpolation_method: _InterpolationMethod = "linear"
+    ) -> Self: ...
     def push(self, time: float, value: int) -> None: ...
     def at(self, time: float) -> int | None: ...
 
 @final
 class FloatSignal(Signal):
     @classmethod
-    def constant(cls, value: float) -> Self: ...
+    def constant(cls, value: float, *, interpolation_method: _InterpolationMethod = "linear") -> Self: ...
     @classmethod
-    def from_samples(cls, samples: list[tuple[float, float]]) -> Self: ...
+    def from_samples(
+        cls, samples: list[tuple[float, float]], *, interpolation_method: _InterpolationMethod = "linear"
+    ) -> Self: ...
     def push(self, time: float, value: float) -> None: ...
     def at(self, time: float) -> float | None: ...
 

pyargus/src/signals.rs

@@ -1,5 +1,6 @@
-use argus::signals::interpolation::Linear;
+use argus::signals::interpolation::{Constant, Linear};
 use argus::signals::Signal;
+use pyo3::exceptions::PyValueError;
 use pyo3::prelude::*;
 use pyo3::types::PyType;
 
@@ -10,6 +11,7 @@ use crate::{DType, PyArgusError};
 pub enum PyInterp {
     #[default]
     Linear,
+    Constant,
 }
 
 #[derive(Debug, Clone, derive_more::From, derive_more::TryInto)]
@@ -122,31 +124,49 @@ macro_rules! impl_signals {
             impl [<$ty_name Signal>] {
                 /// Create a new empty signal
                 #[new]
-                #[pyo3(signature = ())]
-                fn new() -> (Self, PySignal) {
-                    (Self, PySignal::new(Signal::<$ty>::new(), PyInterp::Linear))
+                #[pyo3(signature = (*, interpolation_method = "linear"))]
+                fn new(interpolation_method: &str) -> PyResult<(Self, PySignal)> {
+                    let interp = match interpolation_method {
+                        "linear" => PyInterp::Linear,
+                        "constant" => PyInterp::Constant,
+                        _ => return Err(PyValueError::new_err(format!("unsupported interpolation method `{}`", interpolation_method))),
+                    };
+                    Ok((Self, PySignal::new(Signal::<$ty>::new(), interp)))
                 }
 
                 /// Create a new signal with constant value
                 #[classmethod]
-                fn constant(_: &PyType, py: Python<'_>, value: $ty) -> PyResult<Py<Self>> {
+                #[pyo3(signature = (value, *, interpolation_method = "linear"))]
+                fn constant(_: &PyType, py: Python<'_>, value: $ty, interpolation_method: &str) -> PyResult<Py<Self>> {
+                    let interp = match interpolation_method {
+                        "linear" => PyInterp::Linear,
+                        "constant" => PyInterp::Constant,
+                        _ => return Err(PyValueError::new_err(format!("unsupported interpolation method `{}`", interpolation_method))),
+                    };
                     Py::new(
                         py,
-                        (Self, PySignal::new(Signal::constant(value), PyInterp::Linear))
+                        (Self, PySignal::new(Signal::constant(value), interp))
                     )
                 }
 
                 /// Create a new signal from some finite number of samples
                 #[classmethod]
-                fn from_samples(_: &PyType, samples: Vec<(f64, $ty)>) -> PyResult<Py<Self>> {
+                #[pyo3(signature = (samples, *, interpolation_method = "linear"))]
+                fn from_samples(_: &PyType, samples: Vec<(f64, $ty)>, interpolation_method: &str) -> PyResult<Py<Self>> {
                     let ret: Signal::<$ty> = Signal::<$ty>::try_from_iter(samples
                         .into_iter()
                         .map(|(t, v)| (core::time::Duration::try_from_secs_f64(t).unwrap_or_else(|err| panic!("Value = {}, {}", t, err)), v))
                     ).map_err(PyArgusError::from)?;
+
+                    let interp = match interpolation_method {
+                        "linear" => PyInterp::Linear,
+                        "constant" => PyInterp::Constant,
+                        _ => return Err(PyValueError::new_err(format!("unsupported interpolation method `{}`", interpolation_method))),
+                    };
                     Python::with_gil(|py| {
                         Py::new(
                             py,
-                            (Self, PySignal::new(ret, PyInterp::Linear))
+                            (Self, PySignal::new(ret, interp))
                         )
                     })
                 }
@@ -171,6 +191,7 @@ macro_rules! impl_signals {
                     let time = core::time::Duration::from_secs_f64(time);
                     match super_.interpolation {
                         PyInterp::Linear => signal.interpolate_at::<Linear>(time),
+                        PyInterp::Constant => signal.interpolate_at::<Constant>(time),
                     }
                 }