ykcheungab
76 lines
2.7 KiB
Python
76 lines
2.7 KiB
Python
"""Tests for time-windowed MetricHistory (force plot buffer)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import pytest
|
|
import numpy as np
|
|
import warnings
|
|
from orisys_app_shared.force.history import MetricHistory
|
|
|
|
|
|
def test_unavailable_channels_keep_gaps_without_empty_slice_warnings():
|
|
history = MetricHistory()
|
|
history.push(1, np.nan, np.nan, np.nan)
|
|
with warnings.catch_warnings():
|
|
warnings.simplefilter("error", RuntimeWarning)
|
|
state = history.state()
|
|
assert np.isnan(state.values).all()
|
|
assert all(np.isfinite(bounds).all() and bounds[0] < bounds[1]
|
|
for bounds in state.y_ranges)
|
|
|
|
|
|
def _push_at_rate(hist: MetricHistory, *, fps: float, duration_sec: float, t0: float = 1000.0) -> None:
|
|
dt = 1.0 / fps
|
|
n = int(round(duration_sec * fps)) + 1
|
|
for i in range(n):
|
|
hist.push(t0 + i * dt, 1.0, 0.0, -1.0)
|
|
|
|
|
|
def test_fill_phase_x_range_is_fixed_window():
|
|
hist = MetricHistory(window_sec=4.0, sample_rate_hz=30.0, lowpass_cutoff_hz=0.0, max_fps=90.0)
|
|
_push_at_rate(hist, fps=30.0, duration_sec=2.0)
|
|
state = hist.state()
|
|
assert state is not None
|
|
assert state.x_range == pytest.approx((0.0, 4.0))
|
|
assert float(state.times[-1]) < 4.0
|
|
|
|
|
|
def test_high_fps_after_scroll_covers_x_range():
|
|
window = 4.0
|
|
hist = MetricHistory(window_sec=window, sample_rate_hz=30.0, lowpass_cutoff_hz=0.0, max_fps=90.0)
|
|
_push_at_rate(hist, fps=60.0, duration_sec=window + 1.0)
|
|
state = hist.state()
|
|
assert state is not None
|
|
x0, x1 = state.x_range
|
|
assert x1 - x0 == pytest.approx(window)
|
|
assert float(state.times[0]) == pytest.approx(x0)
|
|
assert float(state.times[-1]) == pytest.approx(x1)
|
|
|
|
|
|
def test_nominal_30hz_after_scroll_covers_x_range():
|
|
window = 4.0
|
|
hist = MetricHistory(window_sec=window, sample_rate_hz=30.0, lowpass_cutoff_hz=0.0, max_fps=90.0)
|
|
_push_at_rate(hist, fps=30.0, duration_sec=window + 2.0)
|
|
state = hist.state()
|
|
assert state is not None
|
|
x0, x1 = state.x_range
|
|
assert x1 - x0 == pytest.approx(window)
|
|
assert float(state.times[0]) == pytest.approx(x0)
|
|
|
|
|
|
def test_set_window_sec_prunes_older_samples():
|
|
hist = MetricHistory(window_sec=4.0, sample_rate_hz=30.0, lowpass_cutoff_hz=0.0, max_fps=90.0)
|
|
_push_at_rate(hist, fps=30.0, duration_sec=5.0)
|
|
before = hist.state()
|
|
assert before is not None
|
|
assert float(before.times[-1]) >= 4.0
|
|
|
|
hist.set_window_sec(1.0)
|
|
after = hist.state()
|
|
assert after is not None
|
|
assert after.x_range[1] - after.x_range[0] == pytest.approx(1.0)
|
|
assert float(after.times[0]) == pytest.approx(after.x_range[0])
|
|
# All retained samples must lie within the new window ending at t_end.
|
|
t_end = float(after.times[-1])
|
|
assert float(after.times[0]) >= t_end - 1.0 - 1e-9
|