perda.utils.integrate#

perda.utils.integrate.average_over_time_range(data_instance, source_time_unit, start_time=0, end_time=-1, target_time_unit=Timescale.S)[source]#

Get average value over time using integral divided by time range.

Parameters:

  • data_instance (DataInstance) – DataInstance to average

  • source_time_unit (Timescale) – Time unit of data_instance.timestamp_np. Pass aly.data.timestamp_unit.

  • start_time (int, optional) – Start of the window, as an absolute timestamp in target_time_unit. Default is 0

  • end_time (int, optional) – End of the window, as an absolute timestamp in target_time_unit. -1 means end of data. Default is -1

  • target_time_unit (Timescale, optional) – Time unit of start_time and end_time. Default is Timescale.S

Returns:

Time-weighted average value over the time range

Return type:

float

Notes

Uses numpy.trapezoid (trapezoidal rule) for numerical integration of discrete data.

Examples

>>> from perda.utils import average_over_time_range
>>> voltage = aly.data["bms.pack.voltage"]
>>> mean_voltage = average_over_time_range(
...     voltage, source_time_unit=aly.data.timestamp_unit
... )
perda.utils.integrate.get_data_slice_by_timestamp(original_instance, start_time=0, end_time=-1)[source]#

Get a new DataInstance with data in [start_time, end_time).

Parameters:

  • original_instance (DataInstance) – Original DataInstance to slice

  • start_time (int, optional) – Start time (inclusive). Default is 0

  • end_time (int, optional) – End time (exclusive). -1 means till end. Default is -1

Returns:

New DataInstance containing only data within the specified time range

Return type:

DataInstance

Notes

start_time and end_time are raw timestamps, in aly.data.timestamp_unit.

Examples

>>> from perda.utils import get_data_slice_by_timestamp
>>> di = aly.data["pcm.wheelSpeeds.frontRight"]
>>> first_half = get_data_slice_by_timestamp(di, end_time=int(di.timestamp_np[len(di) // 2]))
perda.utils.integrate.integrate_over_time_range(data_instance, source_time_unit, start_time=0, end_time=-1, target_time_unit=Timescale.S)[source]#

Get integral of the value over time using the trapezoidal rule.

Parameters:

  • data_instance (DataInstance) – DataInstance to integrate

  • source_time_unit (Timescale) – Time unit of data_instance.timestamp_np. Pass aly.data.timestamp_unit.

  • start_time (int, optional) – Start of the window, as an absolute timestamp in target_time_unit. Default is 0

  • end_time (int, optional) – End of the window, as an absolute timestamp in target_time_unit. -1 means end of data. Default is -1

  • target_time_unit (Timescale, optional) – Time unit to integrate over, e.g. seconds gives value * seconds. Default is Timescale.S

Returns:

Integral of the value over the time range

Return type:

float

Notes

Uses numpy.trapezoid (trapezoidal rule) for numerical integration of discrete data.

Examples

>>> from perda.utils import integrate_over_time_range
>>> current = aly.data["bms.pack.current"]
>>> charge_amp_seconds = integrate_over_time_range(
...     current, source_time_unit=aly.data.timestamp_unit
... )
perda.utils.integrate.smoothed_filtered_integration(data, source_time_unit, target_time_unit=Timescale.S, filter_window_size=10, n_sigmas=3, smoothing_window_len=11, smoothing_poly_order=2)[source]#

Integrate a time-series signal after outlier removal and smoothing.

Cleans spikes via rolling MAD-based outlier detection, applies Savitzky-Golay smoothing, then computes the cumulative trapezoidal integral over time.

Parameters:

  • data (DataInstance) – Time-series signal to integrate.

  • source_time_unit (Timescale) – Time unit of data.timestamp_np. Pass aly.data.timestamp_unit.

  • target_time_unit (Timescale) – Target time unit for integration result. Default is Timescale.S.

  • filter_window_size (int) – Rolling window size used for median and MAD outlier detection.

  • n_sigmas (float) – Number of scaled MAD units beyond which a point is considered an outlier.

  • smoothing_window_len (int) – Window length for Savitzky-Golay smoothing filter.

  • smoothing_poly_order (int) – Polynomial order for Savitzky-Golay smoothing filter.

Returns:

A tuple of (timestamps, smoothed values, cumulative integral), all of the same length as the input signal.

Return type:

tuple[NDArray[Float64], NDArray[Float64], NDArray[Float64]]

Examples

>>> from perda.utils import smoothed_filtered_integration
>>> power = aly.data["bms.pack.voltage"] * aly.data["bms.pack.current"]
>>> ts, smoothed_power, energy_joules = smoothed_filtered_integration(
...     power, source_time_unit=aly.data.timestamp_unit
... )