perda.core_data_structures.data_instance#

pydantic model perda.core_data_structures.data_instance.DataInstance[source]#

Bases: BaseModel

A single time-series variable, pairing a 1D timestamp array with a 1D value array.

Notes

Timestamps are raw integers in the log’s native unit, SingleRunData.timestamp_unit (microseconds for current logs), not seconds.

Arithmetic (+ - * / **) accepts only another DataInstance or a scalar, never a raw NumPy array. Between two DataInstances, the right one is interpolated onto the left one’s timestamps, so the result has the left one’s length. To combine with an array, do the math on .value_np and wrap the result in a new DataInstance.

The constructor takes keyword arguments only.

Examples

>>> from perda.core_data_structures import DataInstance
>>> di = aly.data["pcm.wheelSpeeds.frontRight"]
>>> average = (di + aly.data["pcm.wheelSpeeds.frontLeft"]) / 2.0
>>> squared = DataInstance(
...     timestamp_np=di.timestamp_np, value_np=di.value_np**2, label="Speed squared"
... )
Config:

  • arbitrary_types_allowed: bool = True

Fields:
Validators:
field timestamp_np: NDArray [Required]#

Timestamps as a 1D NumPy array

Validated by:
field value_np: NDArray [Required]#

Values as a 1D NumPy array

Validated by:
field label: str | None = None#

Human-readable label for this variable

field var_id: int | None = None#

Unique variable ID

field cpp_name: str | None = None#

C++ variable name

validator validate_timestamp  »  timestamp_np[source]#

Validate that timestamp array is 1-dimensional, positive, and strictly increasing.

Return type:

TypeAliasType

Parameters:

v (Any)

validator validate_value  »  value_np[source]#

Validate that value array is 1-dimensional

Return type:

TypeAliasType

Parameters:

v (Any)

model_post_init(_DataInstance__context)[source]#

Post-initialization validation that timestamp and value arrays have the same length.

Return type:

None

Parameters:

_DataInstance__context (Any)

trim(ts_start=None, ts_end=None)[source]#

Return a new DataInstance containing only points within the given timestamp range.

Parameters:

  • ts_start (float | None, optional) – Lower bound in raw timestamp units (inclusive). Default is None (no lower bound).

  • ts_end (float | None, optional) – Upper bound in raw timestamp units (inclusive). Default is None (no upper bound).

Returns:

New DataInstance with only in-range data points.

Return type:

DataInstance

Examples

>>> clipped = di.trim(ts_start=10_000, ts_end=30_000)
resample_to_freq(freq_hz, source_time_unit, method=ResampleMethod.LINEAR)[source]#

Return a new DataInstance resampled onto a uniform frequency grid.

Parameters:

  • freq_hz (float) – Target sampling frequency in Hz

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

  • method (ResampleMethod, optional) – Interpolation method. Default is LINEAR.

Returns:

New DataInstance with values resampled onto a uniform timestamp grid

Return type:

DataInstance

Examples

>>> uniform = di.resample_to_freq(freq_hz=100.0, source_time_unit=aly.data.timestamp_unit)
perda.core_data_structures.data_instance.apply_ufunc_inner_join(left, right, ufunc, *, tolerance)[source]#

Apply a binary operation to two DataInstances using inner join.

Parameters:

  • left (DataInstance) – Left DataInstance

  • right (DataInstance) – Right DataInstance

  • ufunc (Callable) – NumPy universal function to apply (e.g., np.add, np.subtract)

  • tolerance (float) – Maximum allowed distance between left and right timestamps for a match, in raw timestamp units.

Returns:

New DataInstance with combined values

Return type:

DataInstance

Examples

>>> import numpy as np
>>> from perda.core_data_structures import apply_ufunc_inner_join
>>> larger = apply_ufunc_inner_join(a, b, np.maximum, tolerance=1_000)
perda.core_data_structures.data_instance.apply_ufunc_left_join(left, right, ufunc)[source]#

Apply a binary operation to two DataInstances using left join.

Parameters:

  • left (DataInstance) – Left DataInstance (all timestamps are kept)

  • right (DataInstance) – Right DataInstance (values interpolated to left)

  • ufunc (Callable) – NumPy universal function to apply (e.g., np.add, np.subtract)

Returns:

New DataInstance with combined values

Return type:

DataInstance

Examples

>>> import numpy as np
>>> from perda.core_data_structures import apply_ufunc_left_join
>>> larger = apply_ufunc_left_join(a, b, np.maximum)
perda.core_data_structures.data_instance.apply_ufunc_outer_join(left, right, ufunc, *, drop_nan=True, fill=0.0)[source]#

Apply a binary operation to two DataInstances using outer join.

Parameters:

  • left (DataInstance) – Left DataInstance

  • right (DataInstance) – Right DataInstance

  • ufunc (Callable) – NumPy universal function to apply (e.g., np.add, np.subtract)

  • drop_nan (bool, optional) – If True, drop rows where either series has NaN after interpolation. Default is True.

  • fill (float, optional) – Fill value for NaNs when drop_nan is False. Default is 0.0.

Returns:

New DataInstance with combined values

Return type:

DataInstance

Examples

>>> import numpy as np
>>> from perda.core_data_structures import apply_ufunc_outer_join
>>> larger = apply_ufunc_outer_join(a, b, np.maximum)
perda.core_data_structures.data_instance.inner_join_data_instances(left, right, *, tolerance, method=ResampleMethod.LINEAR)[source]#

Inner join two DataInstances: keep only left timestamps with matching right timestamps.

Parameters:

  • left (DataInstance) – Left DataInstance

  • right (DataInstance) – Right DataInstance

  • tolerance (float) – Maximum allowed distance between left and right timestamps for a match. Timestamps with distance > tolerance are dropped.

  • method (ResampleMethod, optional) – Interpolation method for right values. Default is LINEAR.

Return type:

tuple[DataInstance, DataInstance]

Returns:

  • left_result (DataInstance) – Left DataInstance with only matched timestamps

  • right_result (DataInstance) – Right DataInstance with only matched timestamps

Notes

tolerance is in raw timestamp units, aly.data.timestamp_unit.

Examples

>>> from perda.core_data_structures import inner_join_data_instances
>>> a_matched, b_matched = inner_join_data_instances(a, b, tolerance=1_000)
perda.core_data_structures.data_instance.left_join_data_instances(left, right, *, method=ResampleMethod.LINEAR)[source]#

Left join one or more DataInstances onto the left timestamp grid.

All right series are interpolated onto the left series timestamps. To resample onto a uniform frequency grid first, call resample_to_freq on the left instance before passing it here.

Parameters:

  • left (DataInstance) – Left DataInstance (defines the target timestamp grid)

  • right (DataInstance or list of DataInstance) – One or more right DataInstances to align to the left grid

  • method (ResampleMethod, optional) – Interpolation method. Default is LINEAR.

Returns:

First element is the left DataInstance (unchanged), followed by one aligned DataInstance per right input, in the same order.

Return type:

tuple of DataInstance

Notes

Signals are logged at different rates, so two DataInstances usually have different lengths and timestamps. Join them first whenever you combine their arrays: a boolean mask built from one signal’s value_np cannot index another’s until both share a timestamp grid.

Examples

>>> from perda.core_data_structures import ResampleMethod, left_join_data_instances
>>> speed = aly.data["pcm.wheelSpeeds.frontRight"]
>>> torque = aly.data["pcm.moc.motor.requestedTorque"]
>>> speed, torque = left_join_data_instances(speed, torque)
>>> torque_while_moving = torque.value_np[speed.value_np > 1.0]
>>> left_a, right_b, right_c = left_join_data_instances(a, [b, c])
>>> left_a, right_b = left_join_data_instances(a, b, method=ResampleMethod.ZOH)
perda.core_data_structures.data_instance.outer_join_data_instances(left, right, *, drop_nan=True, fill=0.0)[source]#

Outer join two DataInstances: union of timestamps with interpolation.

Parameters:

  • left (DataInstance) – Left DataInstance

  • right (DataInstance) – Right DataInstance

  • drop_nan (bool, optional) – If True, drop rows where either series has NaN after interpolation. Default is True.

  • fill (float, optional) – Fill value for NaNs when drop_nan is False. Default is 0.0.

Return type:

tuple[DataInstance, DataInstance]

Returns:

  • left_result (DataInstance) – Left DataInstance with values interpolated to union timestamps

  • right_result (DataInstance) – Right DataInstance with values interpolated to union timestamps

Examples

>>> from perda.core_data_structures import outer_join_data_instances
>>> a_union, b_union = outer_join_data_instances(a, b)