"""Utilities for handling diagnostic time-step sequences with overlaps."""
from __future__ import annotations
import numpy as np
[docs]
def build_last_occurrence_step_mask(step_values) -> np.ndarray:
"""
Return indices that sort by step value and keep the last duplicate occurrence.
This is useful for diagnostics that may contain overlapping sections in time.
Parameters
----------
step_values : array-like
1D or broadcastable array of step indices / timestamps.
Returns
-------
np.ndarray
Integer indices into the original sequence, ordered by increasing step
value, with duplicate step values reduced to their last occurrence.
"""
steps = np.asarray(step_values).reshape(-1)
if steps.size <= 1:
return np.arange(steps.size, dtype=int)
# Stable sort by step value; later duplicates overwrite earlier ones.
order = np.argsort(steps, kind="stable")
last_for_value = {}
for idx in order:
key = steps[idx].item() if hasattr(steps[idx], "item") else steps[idx]
last_for_value[key] = int(idx)
# Return mask ordered by increasing step value.
return np.array([last_for_value[k] for k in sorted(last_for_value.keys())], dtype=int)
[docs]
def select_interval_sample_indices(time_values, *, time_window=None, selected_frame_index: int | None = None) -> np.ndarray:
"""
Return deduplicated sample indices needed for one interval selection.
Parameters
----------
time_values : array-like
Monotonic or mostly monotonic per-sample physical times after any
duplicate-step reduction. The values correspond to sample endpoints;
intervals are formed between adjacent entries.
time_window : tuple[float, float] or None, optional
Optional physical-time window. Intervals overlapping the window are
selected. If omitted, ``selected_frame_index`` chooses one interval.
selected_frame_index : int or None, optional
GUI-style frame index into the deduplicated sample sequence. Frame 0 is
promoted to the first valid interval because interval rates require a
previous sample.
Returns
-------
np.ndarray
Integer indices into ``time_values`` containing the minimal contiguous
sample subset needed for the selected interval range.
"""
time = np.asarray(time_values, dtype=float).reshape(-1)
if time.size < 2:
raise ValueError("Need at least two time samples to select diagnostic intervals.")
dt = np.diff(time)
if not np.any(dt > 0.0):
raise ValueError("Time samples do not contain positive intervals.")
interval_mask = dt > 0.0
if time_window is not None:
t0, t1 = [float(value) for value in time_window]
interval_mask &= (time[1:] > t0) & (time[:-1] < t1)
else:
end_idx = int(selected_frame_index) if selected_frame_index is not None else (time.size - 1)
end_idx = max(1, min(end_idx, time.size - 1))
interval_mask = np.zeros_like(dt, dtype=bool)
interval_mask[end_idx - 1] = dt[end_idx - 1] > 0.0
if not np.any(interval_mask):
mids = 0.5 * (time[:-1] + time[1:])
if time_window is None:
if selected_frame_index is None:
k = int(np.argmax(dt))
else:
end_idx = max(1, min(int(selected_frame_index), time.size - 1))
k = end_idx - 1
else:
k = int(np.argmin(np.abs(mids - 0.5 * (float(time_window[0]) + float(time_window[1])))))
interval_mask = np.zeros_like(dt, dtype=bool)
interval_mask[k] = dt[k] > 0.0
chosen = np.flatnonzero(interval_mask)
start = int(chosen[0])
stop = int(chosen[-1]) + 1
return np.arange(start, stop + 1, dtype=int)