feat: add general signal types

argus-core/Cargo.toml

@@ -5,6 +5,7 @@ edition = "2021"
 
 [dependencies]
 derive_more = "0.99.17"
+num-traits = "0.2.15"
 thiserror = "1.0.39"
 
 [dev-dependencies]

argus-core/src/lib.rs

@@ -1,4 +1,7 @@
 pub mod expr;
+pub mod signals;
+
+use std::time::Duration;
 
 use thiserror::Error;
 
@@ -8,8 +11,13 @@ pub enum Error {
     IdentifierRedeclaration,
     #[error("insufficient number of arguments")]
     IncompleteArgs,
+
+    #[error(
+        "trying to create a non-monotonically signal, signal end time ({end_time:?}) > sample time point \
+         ({current_sample:?})"
+    )]
+    NonMonotonicSignal { end_time: Duration, current_sample: Duration },
 }
 
 pub type ArgusError = Error;
 pub type ArgusResult<T> = Result<T, Error>;
-

argus-core/src/signals.rs

@@ -0,0 +1,251 @@
+//! Concrete signal types
+//!
+//! In Argus, there are essentially 2 kinds of signals:
+//!
+//! 1. [`Signal<T>`] is a variable length signal with finitely many sampled points. This
+//!    implies that the signal has a fixed start and end point (both inclusive) and can
+//!    be iterated over.
+//! 2. [`ConstantSignal<T>`] is a signal that maintains a constant value throughtout
+//!    its domain, and thus, do not require interpolation and extrapolation. Moreover,
+//!    since they are defined over the entire time domain, they cannot be iterated over.
+pub mod iter;
+pub mod traits;
+
+use std::{
+    ops::{RangeFull, RangeInclusive},
+    time::Duration,
+};
+
+use crate::{ArgusResult, Error};
+
+use self::traits::{BaseSignal, LinearInterpolatable};
+
+#[derive(Debug, Clone, Copy)]
+pub enum InterpolationMethod {
+    Linear,
+    Nearest,
+}
+
+impl InterpolationMethod {
+    pub(crate) fn at<T>(self, time: Duration, a: &Option<Sample<T>>, b: &Option<Sample<T>>) -> Option<T>
+    where
+        T: Copy + LinearInterpolatable,
+    {
+        use InterpolationMethod::*;
+        match (self, a, b) {
+            (Nearest, Some(ref a), Some(ref b)) => {
+                assert!(a.time < time && time < b.time);
+                if (b.time - time) > (time - a.time) {
+                    // a is closer to the required time than b
+                    Some(a.value)
+                } else {
+                    // b is closer
+                    Some(b.value)
+                }
+            }
+            (Nearest, Some(nearest), None) | (Nearest, None, Some(nearest)) => Some(nearest.value),
+            (Linear, Some(a), Some(b)) => Some(T::interpolate_at(a, b, time)),
+            _ => None,
+        }
+    }
+}
+
+#[derive(Copy, Clone, Debug)]
+pub struct Sample<T> {
+    pub time: Duration,
+    pub value: T,
+}
+
+/// A signal is a sequence of time points ([`Duration`](core::time::Duration)) and
+/// corresponding value samples.
+#[derive(Default, Debug, Clone)]
+pub struct Signal<T> {
+    pub(crate) values: Vec<T>,
+    pub(crate) time_points: Vec<Duration>,
+}
+
+impl<T> Signal<T> {
+    /// Create a new empty signal
+    pub fn new() -> Self {
+        Self {
+            values: Default::default(),
+            time_points: Default::default(),
+        }
+    }
+
+    /// Create a new empty signal with the specified capacity
+    pub fn new_with_capacity(size: usize) -> Self {
+        Self {
+            values: Vec::with_capacity(size),
+            time_points: Vec::with_capacity(size),
+        }
+    }
+
+    /// Create an iterator over the pairs of time points and values of the signal.
+    pub fn iter(&self) -> impl Iterator<Item = (&Duration, &T)> {
+        self.into_iter()
+    }
+
+    /// Try to create a signal from the input iterator
+    ///
+    /// Returns an `Err` if the input samples are not in strictly monotonically
+    /// increasing order.
+    pub fn try_from_iter<I>(iter: I) -> ArgusResult<Self>
+    where
+        I: IntoIterator<Item = (Duration, T)>,
+    {
+        let iter = iter.into_iter();
+        let mut signal = Signal::new_with_capacity(iter.size_hint().0);
+        for (time, value) in iter.into_iter() {
+            signal.push(time, value)?;
+        }
+        Ok(signal)
+    }
+}
+
+impl<T> BaseSignal for Signal<T> {
+    type Value = T;
+    type Bounds = RangeInclusive<Duration>;
+
+    fn at(&self, time: Duration) -> Option<&Self::Value> {
+        assert_eq!(
+            self.time_points.len(),
+            self.values.len(),
+            "invariant: number of time points must equal number of samples"
+        );
+        // if there are no sample points, then there is no sample point (nor neighboring
+        // sample points) to return
+        if self.time_points.is_empty() {
+            return None;
+        }
+
+        // We will use binary search to find the appropriate index
+        match self.time_points.binary_search(&time) {
+            Ok(idx) => self.values.get(idx),
+            Err(_) => None,
+        }
+    }
+
+    fn interpolate_at(&self, time: Duration, interp: InterpolationMethod) -> Option<Self::Value>
+    where
+        Self::Value: Copy + LinearInterpolatable,
+    {
+        assert_eq!(
+            self.time_points.len(),
+            self.values.len(),
+            "invariant: number of time points must equal number of samples"
+        );
+        // if there are no sample points, then there is no sample point (nor neighboring
+        // sample points) to return
+        if self.time_points.is_empty() {
+            return None;
+        }
+
+        // We will use binary search to find the appropriate index
+        let hint_idx = match self.time_points.binary_search(&time) {
+            Ok(idx) => return self.values.get(idx).copied(),
+            Err(idx) => idx,
+        };
+
+        // We have an hint as to where the sample _should have been_.
+        // So, lets check if there is a preceding and/or following sample.
+        let (first, second) = if hint_idx == 0 {
+            // Sample appears before the start of the signal
+            // So, let's return just the following sample, which is the first sample
+            // (since we know that the signal is non-empty).
+            let preceding = None;
+            let following = Some(Sample {
+                time: self.time_points[hint_idx],
+                value: self.values[hint_idx],
+            });
+            (preceding, following)
+        } else if hint_idx == self.time_points.len() {
+            // Sample appears past the end of the signal
+            // So, let's return just the preceding sample, which is the last sample
+            // (since we know the signal is non-empty)
+            let preceding = Some(Sample {
+                time: self.time_points[hint_idx - 1],
+                value: self.values[hint_idx - 1],
+            });
+            let following = None;
+            (preceding, following)
+        } else {
+            // The sample should exist within the signal.
+            assert!(self.time_points.len() >= 2, "There should be at least 2 elements");
+            let preceding = Some(Sample {
+                time: self.time_points[hint_idx - 1],
+                value: self.values[hint_idx - 1],
+            });
+            let following = Some(Sample {
+                time: self.time_points[hint_idx],
+                value: self.values[hint_idx],
+            });
+            (preceding, following)
+        };
+
+        interp.at(time, &first, &second)
+    }
+
+    fn bounds(&self) -> Self::Bounds {
+        let first = self.time_points.first();
+        let last = self.time_points.last();
+        match (first, last) {
+            (None, None) => Duration::from_secs(1)..=Duration::from_secs(0),
+            (Some(first), Some(last)) => *first..=*last,
+            (..) => unreachable!("there is either 0 time points or some time points"),
+        }
+    }
+
+    fn push(&mut self, time: Duration, value: Self::Value) -> ArgusResult<bool> {
+        assert_eq!(self.time_points.len(), self.values.len());
+
+        let last_time = self.time_points.last();
+        match last_time {
+            Some(last_t) if last_t > &time => Err(Error::NonMonotonicSignal {
+                end_time: *last_t,
+                current_sample: time,
+            }),
+            _ => {
+                self.time_points.push(time);
+                self.values.push(value);
+                Ok(true)
+            }
+        }
+    }
+}
+
+#[derive(Debug, Clone)]
+pub struct ConstantSignal<T> {
+    pub value: T,
+}
+
+impl<T> ConstantSignal<T> {
+    pub fn new(value: T) -> Self {
+        Self { value }
+    }
+}
+
+impl<T> BaseSignal for ConstantSignal<T> {
+    type Value = T;
+
+    type Bounds = RangeFull;
+
+    fn at(&self, _time: Duration) -> Option<&Self::Value> {
+        Some(&self.value)
+    }
+
+    fn bounds(&self) -> Self::Bounds {
+        ..
+    }
+
+    fn interpolate_at(&self, _time: Duration, _interp: InterpolationMethod) -> Option<Self::Value>
+    where
+        Self::Value: Copy + LinearInterpolatable,
+    {
+        Some(self.value)
+    }
+
+    fn push(&mut self, _time: Duration, _value: Self::Value) -> ArgusResult<bool> {
+        Ok(false)
+    }
+}

argus-core/src/signals/iter.rs

@@ -0,0 +1,42 @@
+use std::iter::Zip;
+use std::time::Duration;
+
+use super::Signal;
+
+pub struct Iter<'a, T> {
+    iter: Zip<core::slice::Iter<'a, Duration>, core::slice::Iter<'a, T>>,
+}
+
+impl<'a, T> Iterator for Iter<'a, T> {
+    type Item = (&'a Duration, &'a T);
+
+    fn next(&mut self) -> Option<Self::Item> {
+        self.iter.next()
+    }
+}
+
+impl<'a, T> IntoIterator for &'a Signal<T> {
+    type IntoIter = Iter<'a, T>;
+    type Item = <Self::IntoIter as Iterator>::Item;
+
+    fn into_iter(self) -> Self::IntoIter {
+        Iter {
+            iter: self.time_points.iter().zip(self.values.iter()),
+        }
+    }
+}
+
+impl<T> FromIterator<(Duration, T)> for Signal<T>
+where
+    T: Copy,
+{
+    /// Takes a sequence of sample points and creates a signal.
+    ///
+    /// # Panics
+    ///
+    /// If the input data does not contain strictly monotonically increasing time
+    /// stamps. If this isn't desired, sort and deduplicate the input data.
+    fn from_iter<I: IntoIterator<Item = (Duration, T)>>(iter: I) -> Self {
+        Self::try_from_iter(iter).unwrap()
+    }
+}

argus-core/src/signals/traits.rs

@@ -0,0 +1,158 @@
+use num_traits::Num;
+use std::{ops::RangeBounds, time::Duration};
+
+use crate::ArgusResult;
+
+use super::{InterpolationMethod, Sample};
+
+/// A general Signal trait
+pub trait BaseSignal {
+    /// Type of the values contained in the signal.
+    ///
+    /// For example, a signal that implements `BaseSignal<Value = f64, ...>` contains a
+    /// sequence of timestamped `f64` values.
+    type Value;
+
+    /// A type that implements [`RangeBounds`] to determine the duration bounds of the
+    /// signal.
+    ///
+    /// In practice, this should only be either [`RangeFull`](core::ops::RangeFull)
+    /// (returned by constant signals) or [`Range`](core::ops::Range) (returned by
+    /// sampled signals).
+    type Bounds: RangeBounds<Duration>;
+
+    /// Get the value of the signal at the given time point
+    ///
+    /// If there exists a sample at the given time point then `Some(value)` is returned.
+    /// Otherwise, `None` is returned. If the goal is to interpolate the value at the
+    /// a given time, see [`interpolate_at`](Self::interpolate_at).
+    fn at(&self, time: Duration) -> Option<&Self::Value>;
+
+    /// Interpolate the value of the signal at the given time point
+    ///
+    /// If there exists a sample at the given time point then `Some(value)` is returned
+    /// with the value of the signal at the point. Otherwise, a the
+    /// [`InterpolationMethod`] is used to compute the value. If the given interpolation
+    /// method cannot be used at the given time (for example, if we use
+    /// [`InterpolationMethod::Linear`] and the `time` point is outside the signal
+    /// domain), then a `None` is returned.
+    fn interpolate_at(&self, time: Duration, interp: InterpolationMethod) -> Option<Self::Value>
+    where
+        Self::Value: Copy + LinearInterpolatable;
+
+    /// Get the bounds for the signal
+    fn bounds(&self) -> Self::Bounds;
+
+    /// Push a new sample to the signal at the given time point
+    ///
+    /// The method should enforce the invariant that the time points of the signal must
+    /// have strictly monotonic increasing values, otherwise it returns an error without
+    /// adding the sample point.
+    ///
+    /// The result contains `true` if the sample was successfully added. For example,
+    /// pusing a value to a [constant signal](crate::signals::constant) will be a no-op
+    /// and return `false`.
+    fn push(&mut self, time: Duration, value: Self::Value) -> ArgusResult<bool>;
+
+    /// Check if the signal is empty
+    fn is_empty(&self) -> bool {
+        use core::ops::Bound::*;
+        let bounds = self.bounds();
+        match (bounds.start_bound(), bounds.end_bound()) {
+            (Included(start), Included(end)) => start > end,
+            (Included(start), Excluded(end)) | (Excluded(start), Included(end)) | (Excluded(start), Excluded(end)) => {
+                start >= end
+            }
+
+            (Unbounded, Unbounded) => false,
+            bound => unreachable!("Argus doesn't support signals with bound {:?}", bound),
+        }
+    }
+
+    /// Get the time at which the given signal starts.
+    fn start_time(&self) -> core::ops::Bound<Duration> {
+        self.bounds().start_bound().cloned()
+    }
+
+    /// Get the time at which the given signal ends.
+    fn end_time(&self) -> core::ops::Bound<Duration> {
+        self.bounds().end_bound().cloned()
+    }
+}
+
+/// A Boolean signal
+pub trait BaseBooleanSignal: BaseSignal {}
+
+/// A numeric signal
+pub trait BaseNumericSignal: BaseSignal {
+    type Value: Num;
+}
+
+/// Trait for values that are linear interpolatable
+pub trait LinearInterpolatable {
+    fn interpolate_at(a: &Sample<Self>, b: &Sample<Self>, time: Duration) -> Self
+    where
+        Self: Sized;
+}
+
+impl LinearInterpolatable for bool {
+    fn interpolate_at(a: &Sample<Self>, b: &Sample<Self>, time: Duration) -> Self
+    where
+        Self: Sized,
+    {
+        use InterpolationMethod::Nearest;
+        assert!(a.time < time && time < b.time);
+        // We can't linear interpolate a boolean, so we return the nearest.
+        Nearest.at(time, &Some(*a), &Some(*b)).unwrap()
+    }
+}
+
+macro_rules! interpolate_for_num {
+    ($ty:ty) => {
+        impl LinearInterpolatable for $ty {
+            fn interpolate_at(first: &Sample<Self>, second: &Sample<Self>, time: Duration) -> Self
+            where
+                Self: Sized,
+            {
+                use num_traits::cast;
+                // We will need to cast the samples to f64 values (along with the time
+                // window) to be able to interpolate correctly.
+                // TODO(anand): Verify this works.
+                let t1 = first.time.as_secs_f64();
+                let t2 = second.time.as_secs_f64();
+                let at = time.as_secs_f64();
+                assert!((t1..=t2).contains(&at));
+
+                // We need to do stable linear interpolation
+                // https://www.open-std.org/jtc1/sc22/wg21/docs/papers/2019/p0811r3.html
+                let a: f64 = cast(first.value).unwrap();
+                let b: f64 = cast(second.value).unwrap();
+
+                // Set t to a value in [0, 1]
+                let t = (at - t1) / (t2 - t1);
+                assert!((0.0..=1.0).contains(&t));
+
+                let val = if (a <= 0.0 && b >= 0.0) || (a >= 0.0 && b <= 0.0) {
+                    t * b + (1.0 - t) * a
+                } else if t == 1.0 {
+                    b
+                } else {
+                    a + t * (b - a)
+                };
+
+                cast(val).unwrap()
+            }
+        }
+    };
+}
+
+interpolate_for_num!(i8);
+interpolate_for_num!(i16);
+interpolate_for_num!(i32);
+interpolate_for_num!(i64);
+interpolate_for_num!(u8);
+interpolate_for_num!(u16);
+interpolate_for_num!(u32);
+interpolate_for_num!(u64);
+interpolate_for_num!(f32);
+interpolate_for_num!(f64);