2cbc143c20
Implement DeviceController and SessionController to bridge QML requests with backend state, supporting playback replay tracking, cluster averaging, and slot error boundaries.
36 lines
1.1 KiB
Python
36 lines
1.1 KiB
Python
"""Dynamic Time Warping comparision between two temperature runs."""
|
|
|
|
import numpy as np
|
|
from dtw import dtw
|
|
|
|
|
|
def compare_runs(series_a: list[float], series_b: list[float]) -> dict:
|
|
"""Compute DTW alignment between 2 average-temperature series.
|
|
|
|
Returns:
|
|
dict with keys:
|
|
- distance: float - warping distance
|
|
- index_a: list[int] - warping path indices into series_a
|
|
- index_b: list[int] - warping path indices into series_b
|
|
- warping_path: list[tuple[int,int]] - paired indices
|
|
"""
|
|
x = np.array(series_a, dtype=float)
|
|
y = np.array(series_b, dtype=float)
|
|
|
|
# Filter out NaN/Inf values so dtw doesn't choke on them.
|
|
mask_x = np.isfinite(x)
|
|
mask_y = np.isfinite(y)
|
|
if not mask_x.any() or not mask_y.any():
|
|
return {"distance": float('inf'), "index_a": [], "index_b": [], "warping_path": []}
|
|
|
|
x = x[mask_x]
|
|
y = y[mask_y]
|
|
|
|
alignment = dtw(x, y, keep_internals=True)
|
|
return {
|
|
"distance": float(alignment.distance),
|
|
"index_a": alignment.index1.tolist(),
|
|
"index_b": alignment.index2.tolist(),
|
|
"warping_path": list(zip(alignment.index1.tolist(), alignment.index2.tolist()))
|
|
}
|