From 9cd3170e8a56bdc37aceb7ecba1d3b251f6882d4 Mon Sep 17 00:00:00 2001 From: jack Date: Thu, 4 Jun 2026 13:25:11 -0700 Subject: [PATCH] Replay Tabs: - Add stability detection and threshold classification features with corresponding tests --- src/pygui/backend/frame.py | 6 ++-- src/pygui/backend/frame_stats.py | 37 ++++++++++++++++++++ src/pygui/backend/stability_detector.py | 34 ++++++++++++++++++ src/pygui/backend/threshold_classifier.py | 42 +++++++++++++++++++++++ tests/test_data_parser.py | 10 +++--- tests/test_frame_stats.py | 22 ++++++++++++ tests/test_stability__detector.py | 35 +++++++++++++++++++ tests/test_threshold_classifier.py | 29 ++++++++++++++++ 8 files changed, 207 insertions(+), 8 deletions(-) create mode 100644 src/pygui/backend/frame_stats.py create mode 100644 src/pygui/backend/stability_detector.py create mode 100644 src/pygui/backend/threshold_classifier.py create mode 100644 tests/test_frame_stats.py create mode 100644 tests/test_stability__detector.py create mode 100644 tests/test_threshold_classifier.py diff --git a/src/pygui/backend/frame.py b/src/pygui/backend/frame.py index eb44475..824a480 100644 --- a/src/pygui/backend/frame.py +++ b/src/pygui/backend/frame.py @@ -6,6 +6,6 @@ from dataclasses import dataclass class Frame: """One sample across all sensors ata a point in time""" - seq: int - t: float - values: list[float] + seq: int # monotonically increasing + t: float # seconds (relative or epoch) + values: list[float] # one per sensor, in sensor-layout order diff --git a/src/pygui/backend/frame_stats.py b/src/pygui/backend/frame_stats.py new file mode 100644 index 0000000..0cfef4b --- /dev/null +++ b/src/pygui/backend/frame_stats.py @@ -0,0 +1,37 @@ +"""Per-frame descriptive statistics""" + +from __future__ import annotations +import math +from dataclasses import dataclass + + +@dataclass(frozen=True) +class Stats: + min: float; min_index: int + max: float; max_index: int + diff: float; avg: float + sigma: float; three_sigma: float + + +def compute_stats(values: list[float]) -> Stats: + clean = [(i, v) for i, v in enumerate(values) if not math.isnan(v)] + if not clean: + return Stats(0.0, -1, 0.0, -1, 0.0, 0.0, 0.0, 0.0) + + min_index, min_v = min(clean, key=lambda iv: iv[1]) + max_index, max_v = max(clean, key=lambda iv: iv[1]) + nums = [v for _, v in clean] + avg = sum(nums) / len(nums) + variance = sum((v - avg) ** 2 for v in nums) / len(nums) + sigma = math.sqrt(variance) + + return Stats( + min=min_v, + min_index=min_index, + max=max_v, + max_index=max_index, + diff=max_v - min_v, + avg=avg, + sigma=sigma, + three_sigma=3 * sigma, + ) diff --git a/src/pygui/backend/stability_detector.py b/src/pygui/backend/stability_detector.py new file mode 100644 index 0000000..b233699 --- /dev/null +++ b/src/pygui/backend/stability_detector.py @@ -0,0 +1,34 @@ +"""Detect process state (Idle, Ramp, Set) from the running average temp""" +from __future__ import annotations +from typing import Optional + +STATE_IDLE = "idle" +STATE_RAMP = "ramp" +STATE_SET = "set" + + +class StabilityDetector: + def __init__(self, idle_below: float = 50.0, tolerance: float = 1.0, + settle_seconds: float = 10.0) -> None: + self._idle_below = idle_below + self._tolerance = tolerance + self._settle_seconds = settle_seconds + self._near_since: Optional[float] = None # When avg entered the +- tolerance band + + + def reset(self) -> None: + self._near_since = None + + def update(self, avg: float, time: float, set_point: float) ->str: + if avg < self._idle_below: + self._near_since = None + return STATE_IDLE + if abs(avg - set_point) <= self._tolerance: + if self._near_since is None: + self._near_since = time + if time - self._near_since >= self._settle_seconds: + return STATE_SET + return STATE_RAMP + self._near_since = None + return STATE_RAMP + \ No newline at end of file diff --git a/src/pygui/backend/threshold_classifier.py b/src/pygui/backend/threshold_classifier.py new file mode 100644 index 0000000..d2d6560 --- /dev/null +++ b/src/pygui/backend/threshold_classifier.py @@ -0,0 +1,42 @@ +"""Classify sensor values into three bands around (target, margin) + +Auto mode derives target=mean, margin=1 +""" + +from __future__ import annotations +import math +from dataclasses import dataclass + +BAND_IN = "in_range" +BAND_HIGH = "high" +BAND_LOW = "low" + + +@dataclass(frozen=True) +class ThresholdConfig: + set_point: float = 149.0 # process target: used as band TARGET when auto=False + margin: float = 1.0 # used as band MARGIN when auto=False + auto: bool = True # auto=True: target=frame mean, margin=frame 1σ + +def resolve_bounds(values: list[float], cfg: ThresholdConfig) -> tuple[float, float]: + if not cfg.auto: + return cfg.set_point, cfg.margin + clean = [v for v in values if not math.isnan(v)] + if not clean: + return cfg.set_point, cfg.margin + mean = sum(clean) / len(clean) + variance = sum((v - mean) ** 2 for v in clean) / len(clean) + return mean, math.sqrt(variance) + +def classify(value: float, target: float, margin: float) -> str: + if math.isnan(value): + return BAND_IN + if value > target + margin: + return BAND_HIGH + if value < target - margin: + return BAND_LOW + return BAND_IN + +def classify_all(values: list[float], cfg: ThresholdConfig) -> list[str]: + target, margin = resolve_bounds(values, cfg) + return [classify(v, target, margin)for v in values ] \ No newline at end of file diff --git a/tests/test_data_parser.py b/tests/test_data_parser.py index 5dc64cc..9e14617 100644 --- a/tests/test_data_parser.py +++ b/tests/test_data_parser.py @@ -1,7 +1,7 @@ """Tests for serialcomm/data_parser.py binary parsing pipeline.""" import pytest -from serialcomm.data_parser import ( +from pygui.serialcomm.data_parser import ( csv_column_count, parse_binary_data, convert_to_temperatures, @@ -12,7 +12,6 @@ from serialcomm.data_parser import ( MAXDUT_X, ) - # ── csv_column_count ────────────────────────────────────────────────────────── @@ -203,6 +202,7 @@ class TestConvertToTemperatures: def test_p_family_single_block(self): data = _make_p_block(1, value=0x0100) hex_data = parse_binary_data(data, "P") + assert hex_data is not None result = convert_to_temperatures(hex_data, "P") assert len(result) == 1 assert all(isinstance(v, str) for v in result[0]) @@ -270,9 +270,9 @@ class TestSaveToCsv: result = save_to_csv(data, family, f"{family}00001", str(tmp_path)) assert result is not None, f"save_to_csv returned None for {family}" headers = open(result).readline().strip().split(",") - assert len(headers) == expected_cols, ( - f"{family}: expected {expected_cols} headers, got {len(headers)}" - ) + assert ( + len(headers) == expected_cols + ), f"{family}: expected {expected_cols} headers, got {len(headers)}" assert headers[0] == "Sensor1" assert headers[-1] == f"Sensor{expected_cols}" diff --git a/tests/test_frame_stats.py b/tests/test_frame_stats.py new file mode 100644 index 0000000..d36c687 --- /dev/null +++ b/tests/test_frame_stats.py @@ -0,0 +1,22 @@ +import math +import pytest +from pygui.backend.frame_stats import compute_stats, Stats + + +def test_basic_stat(): + s = compute_stats([148.0, 150.0, 149.0]) + assert s.min == 148.0 and s.min_index == 0 + assert s.max == 150.0 and s.max_index == 1 + assert s.diff == pytest.approx(2.0) + assert s.avg == pytest.approx(149.0) + assert s.sigma == pytest.approx(math.sqrt(2 / 3)) + assert s.three_sigma == pytest.approx(3 * math.sqrt(2 / 3)) + + +def test_empty_values_returns_zeros(): + s = compute_stats([]) + assert s == Stats(0.0, -1, 0.0, -1, 0.0, 0.0, 0.0, 0.0) + + +def test_ignores_nan(): + s = compute_stats([149.0, float("nan"), 151.0]) diff --git a/tests/test_stability__detector.py b/tests/test_stability__detector.py new file mode 100644 index 0000000..c74aa0a --- /dev/null +++ b/tests/test_stability__detector.py @@ -0,0 +1,35 @@ +from pygui.backend.stability_detector import ( + StabilityDetector, STATE_IDLE, STATE_RAMP, STATE_SET, +) + + +SET_POINT = 149.0 + +def make(): + return StabilityDetector(idle_below = 50.0, tolerance=1.0, settle_seconds=2.0) + +def test_idle_when_cold(): + d = make() + assert d.update(avg=25.0, time=0.0, set_point=SET_POINT) == STATE_IDLE + +def test_ramp_while_far_from_setpoint(): + d = make() + d.update(avg=100.0, time=0.0, set_point=SET_POINT) + +def test_ramp_until_settle_time_elapses(): + d = make() + assert d.update(avg=149.2, time=0.0, set_point=SET_POINT) == STATE_RAMP + assert d.update(avg=148.9, time=0.0, set_point=SET_POINT) == STATE_RAMP + +def test_set_after_holding_near_setpoint(): + d = make() + d.update(avg=149.2, time=0.0, set_point=SET_POINT) + d.update(avg=148.9, time=1.0, set_point=SET_POINT) + assert d.update(avg=149.0, time=2.5, set_point=SET_POINT) == STATE_SET + + +def test_back_to_ramp_on_disturbance(): + d = make() + for t in (0.0, 1.0, 2.5): + d.update(149.0, t, set_point=SET_POINT) + assert d.update(avg=160.0, time=3.0, set_point=SET_POINT) == STATE_RAMP \ No newline at end of file diff --git a/tests/test_threshold_classifier.py b/tests/test_threshold_classifier.py new file mode 100644 index 0000000..16ab171 --- /dev/null +++ b/tests/test_threshold_classifier.py @@ -0,0 +1,29 @@ +import math + +from pygui.backend.threshold_classifier import ( + ThresholdConfig, classify, classify_all, resolve_bounds, + BAND_IN, BAND_HIGH, BAND_LOW +) + + +def test_classify_about_target_margin(): + assert classify(149.0, target=149.0, margin=1.0) ==BAND_IN # exactly + assert classify(149.9, target=149.0, margin=1.0) ==BAND_IN # within + assert classify(150.5, target=149.0, margin=1.0) ==BAND_HIGH # 1.5 above + assert classify(147.5, target=149.0, margin=1.0) ==BAND_LOW # 1.5 below + + +def test_manual_bounds_use_set_point_and_margin(): + cfg = ThresholdConfig(set_point=149.0, margin=1.0, auto=False) + assert resolve_bounds([200.0, 0.0], cfg) == (149.0, 1.0) + +def test_auto_bounds_use_mean_and_sigma(): + cfg = ThresholdConfig(auto=True) + target, margin = resolve_bounds([148.0, 150.0, 149.0], cfg) + assert target == 149.0 + assert margin == math.sqrt(2 / 3) + +def test_classify_all_manual(): + cfg = ThresholdConfig(set_point=149.0, margin=1.0, auto=False) + assert classify_all([149.0, 151.0, 147.0], cfg) == [BAND_IN, BAND_HIGH, BAND_LOW] + \ No newline at end of file