Skip to main content

impulse_query_engine.model.series.sample_series

SampleSeries class implementation

SampleSeries​

class SampleSeries()

__init__​

def __init__(tstarts: Sized, tends: Sized, values: Sized)

Initialize the SampleSeries object.

Arguments:

  • tstarts (Sized): Array-like of interval start times.
  • tends (Sized): Array-like of interval end times.
  • values (Sized): Array-like of sample values.

dtype​

def dtype()

Returns the Spark data type for SampleSeries.

Returns:

pyspark.sql.types.BinaryType: Spark BinaryType for serialized SampleSeries.

get_data​

def get_data() -> list

Returns the series as a list of [tstart, tend, value] lists.

Returns:

list: List of [tstart, tend, value] triples.

sparse​

def sparse() -> SampleSeries

Returns a sparse version of this SampleSeries, merging consecutive samples with the same value.

Returns:

SampleSeries: New sparse SampleSeries.

__len__​

def __len__() -> int

Returns the number of samples in the series.

Returns:

int: Number of samples.

__add__​

def __add__(other: SampleSeries | float) -> SampleSeries

Add another SampleSeries or scalar to this series.

Arguments:

  • other (SampleSeries or float): Operand to add.

Returns:

SampleSeries: Resulting SampleSeries.

__radd__​

def __radd__(other: SampleSeries | float) -> SampleSeries

Add this series to another SampleSeries or scalar (reversed operands).

Arguments:

  • other (SampleSeries or float): Operand to add.

Returns:

SampleSeries: Resulting SampleSeries.

__sub__​

def __sub__(other: SampleSeries | float) -> SampleSeries

Subtract another SampleSeries or scalar from this series.

Arguments:

  • other (SampleSeries or float): Operand to subtract.

Returns:

SampleSeries: Resulting SampleSeries.

__rsub__​

def __rsub__(other: SampleSeries | float) -> SampleSeries

Subtract this series from another SampleSeries or scalar (reversed operands).

Arguments:

  • other (SampleSeries or float): Operand to subtract.

Returns:

SampleSeries: Resulting SampleSeries.

__mul__​

def __mul__(other: SampleSeries | float) -> SampleSeries

Multiply this series by another SampleSeries or scalar.

Arguments:

  • other (SampleSeries or float): Operand to multiply.

Returns:

SampleSeries: Resulting SampleSeries.

__rmul__​

def __rmul__(other: SampleSeries | float) -> SampleSeries

Multiply another SampleSeries or scalar by this series (reversed operands).

Arguments:

  • other (SampleSeries or float): Operand to multiply.

Returns:

SampleSeries: Resulting SampleSeries.

__truediv__​

def __truediv__(other: SampleSeries | float) -> SampleSeries

Divide this series by another SampleSeries or scalar.

Arguments:

  • other (SampleSeries or float): Operand to divide.

Returns:

SampleSeries: Resulting SampleSeries.

__rtruediv__​

def __rtruediv__(other: SampleSeries | float) -> SampleSeries

Divide another SampleSeries or scalar by this series (reversed operands).

Arguments:

  • other (SampleSeries or float): Operand to divide.

Returns:

SampleSeries: Resulting SampleSeries.

__mod__​

def __mod__(other: int | float | SampleSeries) -> Intervals

Return the modulus of this series and another SampleSeries or scalar.

Arguments:

  • other (int, float, or SampleSeries): Operand for modulus.

Returns:

SampleSeries: Resulting SampleSeries.

__rmod__​

def __rmod__(other: int | float | SampleSeries) -> Intervals

Return the modulus of another SampleSeries or scalar and this series.

Arguments:

  • other (int, float, or SampleSeries): Operand for modulus (reversed operands).

Returns:

SampleSeries: Resulting SampleSeries.

__gt__​

def __gt__(other: int | float | SampleSeries) -> Intervals

Return intervals where this series is greater than another.

Arguments:

  • other (int, float, or SampleSeries): Operand for comparison.

Returns:

Intervals: Intervals where condition holds.

__ge__​

def __ge__(other: int | float | SampleSeries) -> Intervals

Return intervals where this series is greater than or equal to another.

Arguments:

  • other (int, float, or SampleSeries): Operand for comparison.

Returns:

Intervals: Intervals where condition holds.

__lt__​

def __lt__(other: int | float | SampleSeries) -> Intervals

Return intervals where this series is less than another.

Arguments:

  • other (int, float, or SampleSeries): Operand for comparison.

Returns:

Intervals: Intervals where condition holds.

__le__​

def __le__(other: int | float | SampleSeries) -> Intervals

Return intervals where this series is less than or equal to another.

Arguments:

  • other (int, float, or SampleSeries): Operand for comparison.

Returns:

Intervals: Intervals where condition holds.

__eq__​

def __eq__(other: int | float | SampleSeries) -> Intervals

Return intervals where this series is equal to another.

Arguments:

  • other (int, float, or SampleSeries): Operand for comparison.

Returns:

Intervals: Intervals where condition holds.

__ne__​

def __ne__(other: int | float | SampleSeries) -> Intervals

Return intervals where this series is not equal to another.

Arguments:

  • other (int, float, or SampleSeries): Operand for comparison.

Returns:

Intervals: Intervals where condition holds.

sample_count​

def sample_count() -> int

Returns the number of samples in this SampleSeries.

Returns:

int: Number of samples.

unique_times​

def unique_times() -> npt.NDArray

Returns a sorted array of all unique start and end times.

Returns:

numpy.ndarray: Array of unique times.

start_time​

def start_time() -> FloatOrNaN

Returns the start time of the first sample.

Returns:

float: Start time or NaN if empty.

end_time​

def end_time() -> FloatOrNaN

Returns the end time of the last sample.

Returns:

float: End time or NaN if empty.

duration_ms​

def duration_ms() -> FloatOrNaN

Returns the total duration in milliseconds.

Returns:

float: Duration in milliseconds.

nan_ratio​

def nan_ratio() -> FloatOrNaN

Returns the ratio of NaN samples to all samples, weighted by duration.

Returns:

float: Ratio of NaN durations to total duration.

durations​

def durations() -> npt.NDArray

Returns an array of durations for all samples.

Returns:

numpy.ndarray: Array of durations (in seconds).

sample_rate​

def sample_rate() -> FloatOrNaN

Returns the average sample rate (mean duration).

Returns:

float: Average sample rate or NaN if empty.

sum​

def sum() -> FloatOrNaN

Returns the sum of all values, weighted by duration.

Returns:

float: Weighted sum of values.

min​

def min() -> FloatOrNaN

Returns the minimum value in the series.

Returns:

float: Minimum value or NaN if empty.

max​

def max() -> FloatOrNaN

Returns the maximum value in the series.

Returns:

float: Maximum value or NaN if empty.

mean​

def mean() -> FloatOrNaN

Returns the mean value, weighted by durations.

Returns:

float: Weighted mean value or NaN if empty.

rising_edges​

def rising_edges() -> PointsInTime

Returns points in time where the value rises compared to the previous sample.

Returns:

PointsInTime: Points where rising edges occur.

falling_edges​

def falling_edges() -> PointsInTime

Returns points in time where the value falls compared to the previous sample.

Returns:

PointsInTime: Points where falling edges occur.

rising_edge​

def rising_edge() -> PointsInTime

Alias for rising_edges().

Returns:

PointsInTime: Points where rising edges occur.

falling_edge​

def falling_edge() -> PointsInTime

Alias for falling_edges().

Returns:

PointsInTime: Points where falling edges occur.

intervals_between_falling_edges​

def intervals_between_falling_edges() -> Intervals

Build intervals [tstart, tend] from falling edges of the series.

Processes each continuous interval of the series separately. Within each continuous block, each interval starts at a falling edge and ends at the timestamp before the next falling edge. The first interval in a block starts at the block's first timestamp; the last interval in a block ends at the block's last timestamp. Blocks with no falling edges contribute no intervals.

Returns:

Intervals: Intervals between consecutive falling edges.

diff​

def diff() -> "SampleSeries"

Calculate the difference between consecutive values.

Returns:

SampleSeries: New SampleSeries with difference values, preserving original timestamps. The first value of each continuous segment is 0 (no previous value to diff from).

histogram​

def histogram(bins: npt.ArrayLike = None,
              weights: SampleSeries = None,
              weight_type: str = None) -> tuple[npt.NDArray, npt.NDArray]

Compute a histogram of the sample values using the specified bins.

Arguments:

  • bins (array_like): Bin edges for the histogram. If None, uses [-np.inf, np.inf].
  • weights (SampleSeries): Custom weights series. If None, uses sample durations as weights.
  • weight_type (str): Type of weighting to use. Options:
  • None (default): Use weights values directly (or durations if weights is None)
  • 'time': Multiply weights values by their durations

Returns:

ndarray: The values of the histogram.

histogram2d​

def histogram2d(
        y_series: SampleSeries,
        x_bins: npt.ArrayLike,
        y_bins: npt.ArrayLike,
        weights: SampleSeries = None,
        weight_type: str = None
) -> tuple[npt.NDArray, npt.NDArray, npt.NDArray]

Compute a bi-dimensional histogram of the sample values using the specified x and y bins.

Arguments:

  • y_series (SampleSeries): The second sample series for the y-axis.
  • x_bins (array_like): Bin edges for the x-axis.
  • y_bins (array_like): Bin edges for the y-axis.
  • weights (SampleSeries): Custom weights series. If None, uses sample durations as weights.
  • weight_type (str): Type of weighting to use. Options:
  • None (default): Use weights values directly (or durations if weights is None)
  • 'time': Multiply weights values by their durations

Returns:

ndarray: The 2D histogram array.

synchronized​

def synchronized(other: SampleSeries)

Synchronize this series with another, aligning intervals.

Arguments:

  • other (SampleSeries): Series to synchronize with.

Returns:

tuple of SampleSeries: Synchronized SampleSeries objects.

synchronized_all​

def synchronized_all(others: list[SampleSeries])

Synchronize this series with multiple other SampleSeries.

Arguments:

  • others (list of SampleSeries): List of series to synchronize.

Returns:

tuple of SampleSeries: Synchronized SampleSeries objects.

where​

def where(other: Intervals) -> SampleSeries

Returns a SampleSeries where the given intervals are defined.

Arguments:

  • other (Intervals): Intervals to filter by.

Returns:

SampleSeries: Filtered SampleSeries.

resample​

def resample(sample_rate=1.0)

Resample the series at a given sample rate.

Arguments:

  • sample_rate (float): Desired sample rate (default 1.0).

Returns:

SampleSeries: Resampled SampleSeries.

rolling_average​

def rolling_average(window_size=1.0, evenly_spaced=False) -> SampleSeries

Compute rolling average over a window.

Arguments:

  • window_size (float): Size of the rolling window (default 1.0).
  • evenly_spaced (bool): If True, use evenly spaced weights (default False).

Returns:

SampleSeries: Rolling average SampleSeries.

rolling_stats​

def rolling_stats(
        window_size=1.0,
        evenly_spaced=False
) -> tuple[SampleSeries, SampleSeries, SampleSeries]

Compute rolling min, max, and average over a window.

Arguments:

  • window_size (float): Size of the rolling window (default 1.0).
  • evenly_spaced (bool): If True, use evenly spaced weights (default False).

Returns:

SampleSeries: Rolling minimum values.

trapz​

def trapz() -> float

Perform discrete integration using the composite trapezoidal rule.

Returns:

float: Integrated value.

cumtrapz​

def cumtrapz() -> SampleSeries

Perform cumulative discrete integration using the trapezoidal rule.

Returns:

SampleSeries: Cumulative integrated SampleSeries.

__str__​

def __str__() -> str

Returns a string representation of the SampleSeries.

Returns:

str: String representation.

__repr__​

def __repr__() -> str

Returns a string representation for debugging.

Returns:

str: String representation.

serialize​

def serialize()

Serialize and compress the SampleSeries.

Returns:

bytes: Compressed serialized data.

deserialize​

def deserialize(d)

Deserialize a compressed SampleSeries.

Arguments:

  • d (bytes): Compressed serialized data.

Returns:

SampleSeries: Deserialized SampleSeries object.

to_pickle​

def to_pickle(uri: str)

Write the SampleSeries to disk at the given URI.

Arguments:

  • uri (str): File path to write to.

from_pickle​

def from_pickle(uri: str)

Read a pickled SampleSeries from disk.

Arguments:

  • uri (str): File path to read from.

Returns:

SampleSeries: Loaded SampleSeries object.

from_timestamps​

def from_timestamps(times, values)

Create a SampleSeries from timestamps and values.

Arguments:

  • times (array-like): Array of timestamps.
  • values (array-like): Array of values.

Returns:

SampleSeries: Constructed SampleSeries.

empty​

def empty() -> SampleSeries

Returns an empty SampleSeries.

Returns:

SampleSeries: Empty SampleSeries object.