from __future__ import annotations
from typing import Any, Callable
import numpy as np
from numpy.typing import NDArray
from pydantic import BaseModel, ConfigDict, Field, field_validator
from ..units import Timescale, from_seconds
from .joins import inner_join, left_join, outer_join
from .resampling import ResampleMethod, _interpolate
[docs]
class DataInstance(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"
... )
"""
model_config = ConfigDict(arbitrary_types_allowed=True)
timestamp_np: NDArray = Field(description="Timestamps as a 1D NumPy array")
value_np: NDArray = Field(description="Values as a 1D NumPy array")
label: str | None = Field(
default=None, description="Human-readable label for this variable"
)
var_id: int | None = Field(default=None, description="Unique variable ID")
cpp_name: str | None = Field(default=None, description="C++ variable name")
[docs]
@field_validator("timestamp_np")
@classmethod
def validate_timestamp(cls, v: Any) -> NDArray:
"""
Validate that timestamp array is 1-dimensional, positive, and strictly increasing.
"""
v = v.astype(np.int64)
if v.ndim != 1:
raise ValueError("timestamp_np must be 1-dimensional array.")
if not np.all(np.diff(v) >= 0):
raise ValueError("timestamp_np cannot be decreasing.")
if len(v) and v[0] < 0:
raise ValueError("timestamp_np must be non-negative.")
return v
[docs]
@field_validator("value_np")
@classmethod
def validate_value(cls, v: Any) -> NDArray:
"""
Validate that value array is 1-dimensional
"""
v = v.astype(np.float64)
if v.ndim != 1:
raise ValueError("value_np must be 1-dimensional array.")
return v
[docs]
def model_post_init(self, __context: Any) -> None:
"""
Post-initialization validation that timestamp and value arrays have the same length.
"""
if self.timestamp_np.shape[0] != self.value_np.shape[0]:
raise ValueError("timestamp_np and value_np must have the same length.")
def __str__(self) -> str:
"""
Human-readable one-line summary of this DataInstance.
Returns
-------
str
``Label | ID: <var_id> | C++ Name: <cpp_name>``
"""
return f"{self.label} | ID: {self.var_id} | C++ Name: {self.cpp_name}"
def __len__(self) -> int:
"""
Get number of data points in this DataInstance.
"""
return self.timestamp_np.shape[0]
def __add__(self, other: DataInstance | int | float) -> DataInstance:
"""
Add two DataInstances or add a scalar to a DataInstance.
"""
if isinstance(other, DataInstance):
return apply_ufunc_left_join(self, other, np.add)
if np.isscalar(other):
return DataInstance(
timestamp_np=self.timestamp_np,
value_np=np.add(self.value_np, other),
label=self.label,
var_id=self.var_id,
)
raise TypeError("add expects (DataInstance, DataInstance|scalar) in any order")
def __sub__(self, other: DataInstance | int | float) -> DataInstance:
"""
Subtract a DataInstance or scalar from this DataInstance.
"""
if isinstance(other, DataInstance):
return apply_ufunc_left_join(self, other, np.subtract)
if np.isscalar(other):
return DataInstance(
timestamp_np=self.timestamp_np,
value_np=np.subtract(self.value_np, other),
label=self.label,
var_id=self.var_id,
)
raise TypeError("sub expects (DataInstance, DataInstance|scalar)")
def __mul__(self, other: DataInstance | int | float) -> DataInstance:
"""
Multiply two DataInstances or multiply a DataInstance by a scalar.
"""
if isinstance(other, DataInstance):
return apply_ufunc_left_join(self, other, np.multiply)
if np.isscalar(other):
return DataInstance(
timestamp_np=self.timestamp_np,
value_np=np.multiply(self.value_np, other),
label=self.label,
var_id=self.var_id,
)
raise TypeError("mul expects (DataInstance, DataInstance|scalar)")
def __truediv__(self, other: DataInstance | int | float) -> DataInstance:
"""
Divide this DataInstance by another DataInstance or scalar.
"""
if isinstance(other, DataInstance):
return apply_ufunc_left_join(self, other, np.true_divide)
if np.isscalar(other):
return DataInstance(
timestamp_np=self.timestamp_np,
value_np=np.true_divide(self.value_np, other),
label=self.label,
var_id=self.var_id,
)
raise TypeError("div expects (DataInstance, DataInstance|scalar)")
def __pow__(self, other: DataInstance | int | float) -> DataInstance:
"""
Raise this DataInstance to the power of another DataInstance or scalar.
"""
if isinstance(other, DataInstance):
return apply_ufunc_left_join(self, other, np.power)
if np.isscalar(other):
return DataInstance(
timestamp_np=self.timestamp_np,
value_np=np.power(self.value_np, other),
label=self.label,
var_id=self.var_id,
)
raise TypeError("pow_ expects (DataInstance, DataInstance|scalar)")
def __neg__(self) -> DataInstance:
"""
Negate all values in this DataInstance.
"""
return DataInstance(
timestamp_np=self.timestamp_np,
value_np=np.negative(self.value_np),
label=self.label,
var_id=self.var_id,
)
[docs]
def trim(
self,
ts_start: float | None = None,
ts_end: float | None = None,
) -> DataInstance:
"""
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
-------
DataInstance
New DataInstance with only in-range data points.
Examples
--------
>>> clipped = di.trim(ts_start=10_000, ts_end=30_000)
"""
if ts_start is not None and ts_end is not None:
mask = (self.timestamp_np >= ts_start) & (self.timestamp_np <= ts_end)
elif ts_start is not None:
mask = self.timestamp_np >= ts_start
else:
mask = self.timestamp_np <= ts_end # type: ignore[operator]
return DataInstance(
timestamp_np=self.timestamp_np[mask],
value_np=self.value_np[mask],
label=self.label,
var_id=self.var_id,
cpp_name=self.cpp_name,
)
[docs]
def resample_to_freq(
self,
freq_hz: float,
source_time_unit: Timescale,
method: ResampleMethod = ResampleMethod.LINEAR,
) -> DataInstance:
"""
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
-------
DataInstance
New DataInstance with values resampled onto a uniform timestamp grid
Examples
--------
>>> uniform = di.resample_to_freq(freq_hz=100.0, source_time_unit=aly.data.timestamp_unit)
"""
dt = from_seconds(1.0 / freq_hz, source_time_unit)
target_ts = np.arange(
self.timestamp_np[0], self.timestamp_np[-1], dt, dtype=np.float64
).astype(np.int64)
resampled_val = _interpolate(
target_ts.astype(np.float64),
self.timestamp_np.astype(np.float64),
self.value_np,
method,
)
return DataInstance(
timestamp_np=target_ts,
value_np=resampled_val,
label=self.label,
var_id=self.var_id,
cpp_name=self.cpp_name,
)
[docs]
def left_join_data_instances(
left: DataInstance,
right: DataInstance | list[DataInstance],
*,
method: ResampleMethod = ResampleMethod.LINEAR,
) -> tuple[DataInstance, ...]:
"""
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
-------
tuple of DataInstance
First element is the left DataInstance (unchanged), followed by one
aligned DataInstance per right input, in the same order.
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)
"""
rights = [right] if isinstance(right, DataInstance) else list(right)
ts, left_val, first_right_val = left_join(
left.timestamp_np,
left.value_np,
rights[0].timestamp_np,
rights[0].value_np,
method=method,
)
results: list[DataInstance] = [
DataInstance(
timestamp_np=ts, value_np=left_val, label=left.label, var_id=left.var_id
),
DataInstance(
timestamp_np=ts,
value_np=first_right_val,
label=rights[0].label,
var_id=rights[0].var_id,
),
]
# Interpolate remaining right series onto the already-computed target grid
target_f = ts.astype(np.float64)
for di in rights[1:]:
vals = _interpolate(
target_f, di.timestamp_np.astype(np.float64), di.value_np, method
)
results.append(
DataInstance(
timestamp_np=ts, value_np=vals, label=di.label, var_id=di.var_id
)
)
return tuple(results)
[docs]
def outer_join_data_instances(
left: DataInstance,
right: DataInstance,
*,
drop_nan: bool = True,
fill: float = 0.0,
) -> tuple[DataInstance, DataInstance]:
"""
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.
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)
"""
ts, left_val, right_val = outer_join(
left.timestamp_np,
left.value_np,
right.timestamp_np,
right.value_np,
drop_nan=drop_nan,
fill=fill,
)
left_result = DataInstance(
timestamp_np=ts, value_np=left_val, label=left.label, var_id=left.var_id
)
right_result = DataInstance(
timestamp_np=ts, value_np=right_val, label=right.label, var_id=right.var_id
)
return left_result, right_result
[docs]
def inner_join_data_instances(
left: DataInstance,
right: DataInstance,
*,
tolerance: float,
method: ResampleMethod = ResampleMethod.LINEAR,
) -> tuple[DataInstance, DataInstance]:
"""
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.
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)
"""
ts, left_val, right_val = inner_join(
left.timestamp_np,
left.value_np,
right.timestamp_np,
right.value_np,
tolerance=tolerance,
method=method,
)
left_result = DataInstance(
timestamp_np=ts, value_np=left_val, label=left.label, var_id=left.var_id
)
right_result = DataInstance(
timestamp_np=ts, value_np=right_val, label=right.label, var_id=right.var_id
)
return left_result, right_result
[docs]
def apply_ufunc_left_join(
left: DataInstance,
right: DataInstance,
ufunc: Callable,
) -> DataInstance:
"""
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
-------
DataInstance
New DataInstance with combined values
Examples
--------
>>> import numpy as np
>>> from perda.core_data_structures import apply_ufunc_left_join
>>> larger = apply_ufunc_left_join(a, b, np.maximum)
"""
left_aligned, right_aligned = left_join_data_instances(left, right)
result_val = ufunc(left_aligned.value_np, right_aligned.value_np)
return DataInstance(
timestamp_np=left_aligned.timestamp_np,
value_np=result_val,
label=left.label,
var_id=left.var_id,
)
[docs]
def apply_ufunc_outer_join(
left: DataInstance,
right: DataInstance,
ufunc: Callable,
*,
drop_nan: bool = True,
fill: float = 0.0,
) -> DataInstance:
"""
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
-------
DataInstance
New DataInstance with combined values
Examples
--------
>>> import numpy as np
>>> from perda.core_data_structures import apply_ufunc_outer_join
>>> larger = apply_ufunc_outer_join(a, b, np.maximum)
"""
left_aligned, right_aligned = outer_join_data_instances(
left, right, drop_nan=drop_nan, fill=fill
)
result_val = ufunc(left_aligned.value_np, right_aligned.value_np)
return DataInstance(
timestamp_np=left_aligned.timestamp_np,
value_np=result_val,
label=left.label,
var_id=left.var_id,
)
[docs]
def apply_ufunc_inner_join(
left: DataInstance,
right: DataInstance,
ufunc: Callable,
*,
tolerance: float,
) -> DataInstance:
"""
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
-------
DataInstance
New DataInstance with combined values
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)
"""
left_aligned, right_aligned = inner_join_data_instances(
left, right, tolerance=tolerance
)
result_val = ufunc(left_aligned.value_np, right_aligned.value_np)
return DataInstance(
timestamp_np=left_aligned.timestamp_np,
value_np=result_val,
label=left.label,
var_id=left.var_id,
)