ykcheungab
395 lines
16 KiB
Python
395 lines
16 KiB
Python
"""Display-only force baseline / deadband (value cards and curves)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import numpy as np
|
|
from orisys_app_shared.force.presentation import ForcePresentation, ZeroEstimate
|
|
|
|
#: The idle triplet from the customer's screenshot at run start: the numbers the
|
|
#: cards were showing because the estimate was not applied yet.
|
|
REPORTED_RAW_BIAS = (1118.0, 34.16, -186.1)
|
|
REPORTED_NEWTON_BIAS = (0.4, -0.25, 0.1)
|
|
|
|
|
|
def test_fit_zero_rejects_insufficient_samples() -> None:
|
|
presentation = ForcePresentation(enabled=True, min_samples=30)
|
|
estimate = presentation.fit_zero([(1.0, 2.0, 3.0), (1.1, 2.1, 3.1)])
|
|
assert estimate.quality == "insufficient"
|
|
|
|
|
|
def test_fit_zero_and_accept_deadband() -> None:
|
|
rng = np.random.default_rng(0)
|
|
bias = np.array([1262.0, -11.0, -353.0])
|
|
noise = rng.normal(0.0, 20.0, size=(80, 3))
|
|
samples = (bias + noise).tolist()
|
|
presentation = ForcePresentation(
|
|
enabled=True,
|
|
min_samples=30,
|
|
baseline_seconds=0.0,
|
|
min_deadband_raw=50.0,
|
|
deadband_percentile=99.0,
|
|
)
|
|
estimate = presentation.fit_zero(samples)
|
|
assert estimate.quality == "ok"
|
|
presentation.apply_estimate(estimate)
|
|
idle = presentation.accept(tuple(bias), unit="raw")
|
|
assert idle.zero_zone
|
|
assert idle.values == (0.0, 0.0, 0.0)
|
|
# Curves / CSV keep the measurement: accept never mutates the input tuple.
|
|
measured = (1262.0, -11.0, -353.0)
|
|
presentation.accept(measured, unit="raw")
|
|
assert measured == (1262.0, -11.0, -353.0)
|
|
|
|
|
|
def test_observe_skips_contact_contamination() -> None:
|
|
presentation = ForcePresentation(
|
|
enabled=True,
|
|
min_samples=5,
|
|
baseline_seconds=0.1,
|
|
require_no_contact=True,
|
|
)
|
|
for i in range(10):
|
|
out = presentation.observe(
|
|
(1000.0, 0.0, 0.0),
|
|
t_rel_s=i * 0.05,
|
|
contact_found=True,
|
|
flow_p95=0.1,
|
|
)
|
|
assert out is None
|
|
assert not presentation.ready
|
|
|
|
|
|
def test_observe_freezes_after_quiet_window() -> None:
|
|
presentation = ForcePresentation(
|
|
enabled=True,
|
|
min_samples=10,
|
|
baseline_seconds=0.5,
|
|
min_deadband_raw=10.0,
|
|
)
|
|
estimate: ZeroEstimate | None = None
|
|
for i in range(40):
|
|
estimate = presentation.observe(
|
|
(100.0 + (i % 3), -5.0, 2.0),
|
|
t_rel_s=i * 0.02,
|
|
flow_p95=0.1,
|
|
contact_found=False,
|
|
)
|
|
if estimate is not None:
|
|
break
|
|
assert estimate is not None
|
|
assert estimate.quality == "ok"
|
|
assert presentation.ready
|
|
# Further observes do not chase a new zero once frozen.
|
|
later = presentation.observe((5000.0, 0.0, 0.0), t_rel_s=10.0, flow_p95=0.0)
|
|
assert later is None
|
|
|
|
|
|
def test_the_estimate_is_kept_per_display_unit() -> None:
|
|
"""A newton baseline and a raw baseline are different scales, not one number."""
|
|
presentation = ForcePresentation(enabled=True, min_samples=5, baseline_seconds=1.0)
|
|
for i in range(12):
|
|
presentation.observe(
|
|
(1000.0 + i * 0.1, 0.0, 0.0), t_rel_s=i * 0.02, unit="raw"
|
|
)
|
|
assert presentation.unit == "raw"
|
|
assert presentation.ready
|
|
assert presentation.accept((1000.55, 0.0, 0.0), unit="raw").values[0] == 0.0
|
|
|
|
# The other unit has no estimate of its own and must show what was measured.
|
|
assert presentation.accept((1000.55, 0.0, 0.0), unit="N").values == (
|
|
1000.55,
|
|
0.0,
|
|
0.0,
|
|
)
|
|
presentation.unit = "N"
|
|
assert not presentation.ready, "the raw estimate made newtons ready"
|
|
presentation.unit = "raw"
|
|
|
|
# ...and it collects its own window, in its own scale.
|
|
for i in range(12):
|
|
presentation.observe((0.5 + i * 0.01, 0.0, 0.0), t_rel_s=i * 0.02, unit="N")
|
|
presentation.unit = "N"
|
|
assert presentation.ready
|
|
assert presentation.accept((0.555, 0.0, 0.0), unit="N").values[0] == 0.0
|
|
|
|
|
|
def test_the_newton_floor_is_not_the_raw_floor() -> None:
|
|
"""50 raw counts is a noise floor; 50 N would be a contact."""
|
|
presentation = ForcePresentation(enabled=True)
|
|
assert presentation.min_deadband_for("raw") == 50.0
|
|
assert presentation.min_deadband_for("N") == 0.0
|
|
|
|
rng = np.random.default_rng(3)
|
|
quiet = (0.5 + rng.normal(0.0, 0.002, size=(60, 3))).tolist()
|
|
estimate = presentation.fit_zero(quiet, unit="N")
|
|
assert estimate.quality == "ok"
|
|
assert max(estimate.deadbands) < 0.01, estimate.deadbands
|
|
|
|
|
|
def test_present_holds_a_resting_block_at_zero_and_keeps_an_excursion() -> None:
|
|
rng = np.random.default_rng(7)
|
|
bias = np.array([9555.0, -13368.0, -353.0])
|
|
quiet = (bias + rng.normal(0.0, 20.0, size=(80, 3))).tolist()
|
|
presentation = ForcePresentation(
|
|
enabled=True, min_samples=30, baseline_seconds=0.0, deadband_percentile=99.0
|
|
)
|
|
estimate = presentation.fit_zero(quiet)
|
|
assert estimate.quality == "ok"
|
|
presentation.apply_estimate(estimate)
|
|
|
|
block = np.asarray(
|
|
[bias, bias + 5.0, bias + 4000.0, [np.nan, np.nan, np.nan]], dtype=float
|
|
)
|
|
displayed, suppressed = presentation.present(block, unit="raw")
|
|
assert displayed[0].tolist() == [0.0, 0.0, 0.0], "a resting sample drifted"
|
|
assert suppressed[0].all()
|
|
assert displayed[1].tolist() == [0.0, 0.0, 0.0], "noise outside the deadband?"
|
|
# A real excursion survives, centred on the estimated offset.
|
|
assert 3900.0 < displayed[2][0] < 4100.0, displayed[2]
|
|
assert not suppressed[2].any()
|
|
# A gap is a gap: an unavailable axis must never become a zero.
|
|
assert np.isnan(displayed[3]).all()
|
|
assert not suppressed[3].any()
|
|
|
|
|
|
def test_present_is_a_display_copy_and_never_touches_the_input() -> None:
|
|
presentation = ForcePresentation(enabled=True, min_samples=5, baseline_seconds=1.0)
|
|
for i in range(12):
|
|
presentation.observe((100.0 + i * 0.01, 0.0, 0.0), t_rel_s=i * 0.02)
|
|
source = np.array([[100.055, 0.0, 0.0]])
|
|
displayed, _suppressed = presentation.present(source, unit="raw")
|
|
assert source.tolist() == [[100.055, 0.0, 0.0]], "present mutated the measurement"
|
|
assert displayed[0][0] == 0.0
|
|
|
|
|
|
def test_the_operator_floor_moves_the_frozen_deadband() -> None:
|
|
presentation = ForcePresentation(enabled=True, min_samples=5, baseline_seconds=1.0)
|
|
for i in range(12):
|
|
presentation.observe((100.0 + i * 0.01, 0.0, 0.0), t_rel_s=i * 0.02, unit="raw")
|
|
assert presentation.ready
|
|
|
|
presentation.set_min_deadband("raw", 500.0)
|
|
assert presentation.accept((300.0, 0.0, 0.0), unit="raw").values[0] == 0.0, (
|
|
"200 raw counts above the device's own band but inside the operator's floor"
|
|
)
|
|
# Back to auto: the floor is a floor, so the estimated band returns.
|
|
presentation.set_min_deadband("raw", 0.0)
|
|
assert presentation.min_deadband_for("raw") == 0.0
|
|
assert presentation.accept((300.0, 0.0, 0.0), unit="raw").values[0] != 0.0
|
|
|
|
|
|
def test_format_force_stable_digits() -> None:
|
|
presentation = ForcePresentation(enabled=True)
|
|
assert presentation.format_force(0.0, unit="raw") == "0"
|
|
assert presentation.format_force(1262.4, unit="raw") == "1262"
|
|
assert presentation.format_force(1.2345, unit="N") == "1.234"
|
|
|
|
|
|
# ------------------------------------------------------- the reported defect
|
|
def _quiet_until_zeroed(
|
|
presentation: ForcePresentation,
|
|
bias: tuple[float, float, float],
|
|
*,
|
|
unit: str,
|
|
limit: int = 29,
|
|
) -> int:
|
|
"""Feed quiet samples until ``accept`` reads zero; return the sample count.
|
|
|
|
Returns ``limit + 1`` (i.e. "never within the limit") if it did not zero.
|
|
"""
|
|
for index in range(limit):
|
|
presentation.observe(bias, t_rel_s=index / 30.0, flow_p95=0.1, unit=unit)
|
|
if presentation.accept(bias, unit=unit).values == (0.0, 0.0, 0.0):
|
|
return index + 1
|
|
return limit + 1
|
|
|
|
|
|
def test_a_resting_raw_channel_reads_zero_from_the_first_samples() -> None:
|
|
"""FN 1118 / FX 34.16 / FY -186.1 must be zero long before 30 samples / 3 s.
|
|
|
|
This is the reported defect: the cards kept the device's raw bias for the
|
|
whole startup window, so a session that started at rest looked like it was
|
|
under load.
|
|
"""
|
|
presentation = ForcePresentation(enabled=True, min_samples=30, baseline_seconds=3.0)
|
|
zeroed_at = _quiet_until_zeroed(presentation, REPORTED_RAW_BIAS, unit="raw")
|
|
|
|
assert zeroed_at <= 3, f"the raw bias was still on the cards after {zeroed_at} samples"
|
|
assert not presentation.ready, "zeroing should not have needed the whole quiet window"
|
|
assert presentation.display_ready, "the display policy was not applied while collecting"
|
|
assert presentation.display_quality == "provisional"
|
|
# ...and no invented zero: the offset is the one the quiet samples measured.
|
|
offsets = presentation.last_estimate.offsets
|
|
assert offsets[0] == REPORTED_RAW_BIAS[0]
|
|
assert offsets[1] == REPORTED_RAW_BIAS[1]
|
|
assert offsets[2] == REPORTED_RAW_BIAS[2]
|
|
|
|
|
|
def test_a_resting_newton_channel_reads_zero_from_the_first_samples() -> None:
|
|
"""The same aid in newtons — the estimate is per display unit."""
|
|
presentation = ForcePresentation(
|
|
enabled=True, unit="N", min_samples=30, baseline_seconds=3.0
|
|
)
|
|
zeroed_at = _quiet_until_zeroed(presentation, REPORTED_NEWTON_BIAS, unit="N")
|
|
|
|
assert zeroed_at <= 3, f"the newton bias survived {zeroed_at} samples"
|
|
assert not presentation.ready
|
|
assert presentation.display_quality == "provisional"
|
|
|
|
|
|
def test_a_noisy_channel_is_zeroed_from_the_first_samples_too() -> None:
|
|
"""The provisional band covers the samples it was fitted from.
|
|
|
|
With a handful of samples the p99 residual is essentially the largest one, so
|
|
the newest sample of the window is inside the band: the first tenths of a
|
|
second are zeroed, and the *first* sample is not claimed to be zero because
|
|
nothing has been estimated from it yet.
|
|
"""
|
|
rng = np.random.default_rng(11)
|
|
bias = np.array(REPORTED_RAW_BIAS)
|
|
presentation = ForcePresentation(enabled=True, min_samples=30, baseline_seconds=3.0)
|
|
for index in range(8):
|
|
sample = bias + rng.normal(0.0, 30.0, size=3)
|
|
presentation.observe(tuple(sample), t_rel_s=index / 30.0, flow_p95=0.1)
|
|
if index < 2:
|
|
# Nothing has been estimated from two samples: those frames honestly
|
|
# show the measurement (and are asserted to do so elsewhere).
|
|
continue
|
|
displayed = presentation.accept(tuple(sample), unit="raw")
|
|
assert displayed.values == (0.0, 0.0, 0.0), (
|
|
f"sample {index} of a resting noisy channel was drawn"
|
|
)
|
|
|
|
|
|
def test_a_contact_present_at_session_start_is_never_zeroed() -> None:
|
|
"""A press at run start must stay visible: no estimate may be invented for it.
|
|
|
|
The contact guard decides what feeds the estimate, so a window that has only
|
|
ever seen a contact has nothing to subtract and displays the measurement.
|
|
"""
|
|
presentation = ForcePresentation(enabled=True, min_samples=30, baseline_seconds=3.0)
|
|
pressed = (6000.0, 40.0, -180.0)
|
|
for index in range(20):
|
|
assert (
|
|
presentation.observe(
|
|
pressed, t_rel_s=index / 30.0, contact_found=True, flow_p95=0.1
|
|
)
|
|
is None
|
|
)
|
|
displayed = presentation.accept(pressed, unit="raw")
|
|
assert displayed.values[0] == pressed[0], "a contact at startup was zeroed away"
|
|
assert not displayed.suppressed[0]
|
|
assert not presentation.display_ready
|
|
assert presentation.display_quality == "none"
|
|
|
|
|
|
def test_a_contaminated_window_never_falls_back_to_the_raw_bias() -> None:
|
|
"""A window that closes badly keeps the estimate it already worked out.
|
|
|
|
Falling back to the raw offset is the reported defect. Both ways a startup
|
|
window can be spoiled — a contact, or too much flow — must leave the display
|
|
zeroed and say so through ``display_quality``.
|
|
"""
|
|
bias = REPORTED_RAW_BIAS
|
|
pressed = (6000.0, 40.0, -180.0)
|
|
for spoiled_by in ("contact", "flow"):
|
|
presentation = ForcePresentation(
|
|
enabled=True, min_samples=30, baseline_seconds=3.0
|
|
)
|
|
for index in range(3):
|
|
presentation.observe(bias, t_rel_s=index / 30.0, flow_p95=0.1)
|
|
assert presentation.display_ready and not presentation.ready
|
|
|
|
for index in range(90): # the window is spoiled before it can close
|
|
presentation.observe(
|
|
pressed,
|
|
t_rel_s=0.5 + index / 30.0,
|
|
contact_found=(spoiled_by == "contact"),
|
|
flow_p95=0.9 if spoiled_by == "flow" else 0.1,
|
|
)
|
|
estimate = presentation.observe(bias, t_rel_s=4.0, flow_p95=0.1)
|
|
assert estimate is not None, spoiled_by
|
|
assert estimate.quality == "contaminated", (spoiled_by, estimate)
|
|
assert not presentation.ready, spoiled_by
|
|
assert presentation.display_quality == "contaminated", spoiled_by
|
|
assert presentation.accept(bias, unit="raw").values == (0.0, 0.0, 0.0), spoiled_by
|
|
|
|
|
|
def test_a_window_that_closed_short_still_applies_what_it_learned() -> None:
|
|
"""A session shorter than the quiet window is not a reason to show the bias."""
|
|
presentation = ForcePresentation(enabled=True, min_samples=30, baseline_seconds=0.5)
|
|
for index in range(3):
|
|
presentation.observe(REPORTED_RAW_BIAS, t_rel_s=index / 30.0, flow_p95=0.1)
|
|
|
|
estimate = presentation.observe(REPORTED_RAW_BIAS, t_rel_s=0.6, flow_p95=0.1)
|
|
assert estimate is not None and estimate.quality == "insufficient"
|
|
assert presentation.display_quality == "insufficient"
|
|
assert presentation.accept(REPORTED_RAW_BIAS, unit="raw").values == (0.0, 0.0, 0.0)
|
|
|
|
|
|
def test_a_frozen_window_stops_chasing_the_data() -> None:
|
|
"""The provisional phase is a start, not a licence to re-zero a press."""
|
|
presentation = ForcePresentation(
|
|
enabled=True, min_samples=5, baseline_seconds=3.0, min_deadband_raw=50.0
|
|
)
|
|
for index in range(10):
|
|
presentation.observe(REPORTED_RAW_BIAS, t_rel_s=index / 30.0, flow_p95=0.1)
|
|
assert presentation.ready
|
|
assert presentation.display_quality == "ok"
|
|
# A held press after the window is a reading, not a new zero.
|
|
assert presentation.accept((5000.0, 34.16, -186.1), unit="raw").values[0] > 3000.0
|
|
|
|
|
|
def test_the_cards_and_the_curves_are_the_same_presentation() -> None:
|
|
"""What ``accept`` puts on a card is the row ``present`` draws on its curve.
|
|
|
|
This is what the extra "the estimate is ready" gate in the value cards broke:
|
|
the card kept the raw measurement while the curve was already drawn from the
|
|
applied estimate, so the two disagreed about the same sample.
|
|
"""
|
|
bias = REPORTED_RAW_BIAS
|
|
#: The samples the acquisition would flag as a contact: they must not feed
|
|
#: the idle estimate, or the display would zero the press away.
|
|
presses = {
|
|
3: (bias[0] + 4000.0, bias[1] + 10.0, bias[2] - 10.0),
|
|
5: (bias[0] - 4000.0, bias[1], bias[2]),
|
|
}
|
|
stream = [
|
|
bias,
|
|
bias,
|
|
bias,
|
|
presses[3],
|
|
bias,
|
|
presses[5],
|
|
bias,
|
|
]
|
|
presentation = ForcePresentation(enabled=True, min_samples=30, baseline_seconds=3.0)
|
|
for index, sample in enumerate(stream):
|
|
presentation.observe(
|
|
sample,
|
|
t_rel_s=index / 30.0,
|
|
flow_p95=0.1,
|
|
contact_found=index in presses,
|
|
)
|
|
card = presentation.accept(sample, unit="raw")
|
|
curve, suppressed = presentation.present([sample], unit="raw")
|
|
assert card.values == tuple(float(value) for value in curve[0]), index
|
|
assert card.suppressed == tuple(bool(flag) for flag in suppressed[0]), index
|
|
# The assertion is not vacuous: the excursion really is on both.
|
|
assert presentation.accept(stream[3], unit="raw").values[0] > 3000.0
|
|
|
|
|
|
def test_present_uses_the_provisional_estimate_before_the_window_closes() -> None:
|
|
"""The curves get the same early zero as the cards, not only after 3 s."""
|
|
presentation = ForcePresentation(enabled=True, min_samples=30, baseline_seconds=3.0)
|
|
block = np.asarray([REPORTED_RAW_BIAS] * 4, dtype=float)
|
|
raw, _suppressed = presentation.present(block, unit="raw")
|
|
assert raw[0].tolist() == list(REPORTED_RAW_BIAS), "nothing was learned yet"
|
|
|
|
for index in range(3):
|
|
presentation.observe(REPORTED_RAW_BIAS, t_rel_s=index / 30.0, flow_p95=0.1)
|
|
displayed, suppressed = presentation.present(block, unit="raw")
|
|
assert displayed.tolist() == [[0.0, 0.0, 0.0]] * 4
|
|
assert suppressed.all()
|
|
assert not presentation.ready, "the whole quiet window was not needed"
|