Replay Tabs:

- Add stability detection and threshold classification features with corresponding tests

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

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,
+    )

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
+        

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 ]