Accept stable residual pressure when a zero target cannot be reached exactly

CHANGELOG.md

@@ -8,6 +8,8 @@ All notable changes are documented here. The project follows semantic versioning
 
 ## [Unreleased]
 
+- Accept a stable near-zero residual for zero targets after 10 seconds within +/-0.5 bar and a 0.01 bar stability band; display ZERO STABLE while retaining the measured pressure.
+
 - Add independent Stop buttons for long term measurements, keeping the last result visible until a new measurement starts.
 - Display available short term averaging history out of five minutes and long term history from 0 / 120 min.
 

cross_platform/docs/PACE_Controller_Manual.md

@@ -253,6 +253,12 @@ Rows execute from top to bottom. Dwell begins only after the PACE reports in-lim
 
 At completion the software requests MEASURE unless **Keep CONTROL after final step** was explicitly selected. Timeouts and user interruption request MEASURE.
 
+### Stable residual at target zero
+
+When the requested target is exactly 0 bar, the application also accepts a near-zero plateau even if the instrument never reports IN LIMIT. The residual must remain within +/-0.5 bar and the pressure range over 10 seconds must not exceed 0.01 bar, with continuous valid readings. A pressure still decreasing or a plateau outside this tolerance does not qualify.
+
+The target card shows **ZERO STABLE** (Italian: **ZERO STABILE**). The measured residual remains displayed and logged at its actual value; it is not replaced by zero and the instrument's IN LIMIT flag is not altered. Manual, indenting and routine zero steps can complete, or start their requested dwell, using this condition. Starting another step, leaving CONTROL, missing telemetry or disconnecting clears the stability history. Nonzero targets retain the instrument's normal IN LIMIT criterion.
+
 ## 12. Software protection rules
 
 ### 12.1 Large target change

cross_platform/src/pace_controller/i18n.py

@@ -39,6 +39,7 @@
         "valve_effort": "Valve effort",
         "state": "State",
         "target_state": "Target",
+        "zero_target_stable": "ZERO STABLE",
         "source_margin": "Source margin",
         "manual": "1  MANUAL",
         "indenting": "2  INDENTING",
@@ -182,6 +183,7 @@
         "valve_effort": "Sforzo valvole",
         "state": "Stato",
         "target_state": "Target",
+        "zero_target_stable": "ZERO STABILE",
         "source_margin": "Margine sorgente",
         "manual": "1  MANUALE",
         "indenting": "2  INDENTING",

cross_platform/src/pace_controller/models.py

@@ -70,6 +70,7 @@ class Telemetry:
     valve_effort_percent: float = nan
     control: bool = False
     in_limits: bool = False
+    zero_target_stable: bool = False
     source_margin_bar: float = nan
 
 

cross_platform/src/pace_controller/service.py

@@ -25,6 +25,7 @@
     configure_dedicated_adapter,
     restore_dedicated_adapter,
 )
+from .stability import ZeroTargetStability
 from .storage import DataLogger
 from .transports import ScpiTransport, TransportError, create_transport
 
@@ -96,6 +97,8 @@ def __init__(
         self._poll_failures = 0
         self._supply_interlock_latched = False
         self._automation: dict[str, Any] | None = None
+        self._zero_target_monitor = ZeroTargetStability(maximum_gap_seconds=max(3.0, poll_interval * 3.0))
+        self._zero_target_reported = False
 
     @property
     def telemetry(self) -> Telemetry:
@@ -253,6 +256,9 @@ def _connect(self, config: ConnectionConfig, module: int) -> None:
             self.busy_changed.emit(False)
 
     def _close_device(self, request_measure: bool, emit: bool = True) -> None:
+        self._zero_target_monitor.reset()
+        self._telemetry.zero_target_stable = False
+        self._zero_target_reported = False
         self._cancel_automation()
         if request_measure and self._connected:
             try:
@@ -386,6 +392,9 @@ def _start_current_step(self) -> None:
     def _apply_pressure_step(
         self, step: PressureStep, parameters: ControlParameters
     ) -> None:
+        self._zero_target_monitor.reset()
+        self._telemetry.zero_target_stable = False
+        self._zero_target_reported = False
         module = self._module
         self._write("*CLS")
         self._write(f":UNIT{module}:PRES BAR")
@@ -417,7 +426,7 @@ def _process_automation(self, now: float) -> None:
             index = int(self._automation["index"])
             step = steps[index]
             if state == "waiting":
-                if self._telemetry.in_limits:
+                if self._telemetry.in_limits or (step.target_bar == 0.0 and self._telemetry.zero_target_stable):
                     if step.dwell_seconds > 0:
                         self._automation["state"] = "dwelling"
                         self._automation["dwell_end"] = now + step.dwell_seconds
@@ -477,6 +486,9 @@ def _start_vent(self, parameters: ControlParameters) -> None:
         self._write_log(f"VENT started at {parameters.vent_rate_bar_s} bar/s.")
 
     def _set_measure(self) -> None:
+        self._zero_target_monitor.reset()
+        self._telemetry.zero_target_stable = False
+        self._zero_target_reported = False
         self._require_connected()
         self._write(f":OUTP{self._module}:STAT OFF")
         self._assert_no_error()
@@ -511,11 +523,23 @@ def _poll(self) -> None:
                 in_limits=in_limits,
                 source_margin_bar=margin,
             )
+            self._telemetry.zero_target_stable = self._zero_target_monitor.add(
+                time.monotonic(), current, target, control
+            )
+            if self._telemetry.zero_target_stable and not in_limits:
+                if not self._zero_target_reported:
+                    self._write_log(f"Zero target accepted at stable residual pressure {current:.6f} bar (within +/-0.5 bar, stable for 10 s).")
+                self._zero_target_reported = True
+            else:
+                self._zero_target_reported = False
             self._poll_failures = 0
             self.telemetry_received.emit(self._telemetry)
             self.logger.telemetry(self._telemetry)
             self._check_source_interlock()
         except Exception as exc:
+            self._zero_target_monitor.reset()
+            self._zero_target_reported = False
+            self._telemetry.zero_target_stable = False
             self._poll_failures += 1
             was_control = self._telemetry.control
             self._write_log(f"Telemetry failure {self._poll_failures}: {exc}")

cross_platform/src/pace_controller/stability.py

@@ -0,0 +1,41 @@
+"""Recognition of a stable residual when a zero target is not attainable."""
+
+from collections import deque
+from math import isfinite
+
+
+class ZeroTargetStability:
+    """Accept a near-zero plateau, never a stalled pressure outside the tolerance."""
+
+    def __init__(self, maximum_gap_seconds: float = 3.0) -> None:
+        self.tolerance_bar = 0.5
+        self.duration_seconds = 10.0
+        self.band_bar = 0.01
+        self.maximum_gap_seconds = maximum_gap_seconds
+        self.samples: deque[tuple[float, float]] = deque()
+
+    def reset(self) -> None:
+        self.samples.clear()
+
+    def add(self, timestamp: float, pressure_bar: float, target_bar: float, control: bool) -> bool:
+        if (
+            not control
+            or target_bar != 0.0
+            or not isfinite(timestamp)
+            or not isfinite(pressure_bar)
+            or abs(pressure_bar) > self.tolerance_bar + 1e-9
+        ):
+            self.reset()
+            return False
+        if self.samples:
+            gap = timestamp - self.samples[-1][0]
+            if gap <= 0 or gap > self.maximum_gap_seconds:
+                self.reset()
+        self.samples.append((timestamp, pressure_bar))
+        cutoff = timestamp - self.duration_seconds
+        # Retain the point spanning the beginning of the observation interval.
+        while len(self.samples) > 1 and self.samples[1][0] <= cutoff:
+            self.samples.popleft()
+        elapsed = timestamp - self.samples[0][0]
+        pressures = [value for _, value in self.samples]
+        return elapsed >= self.duration_seconds and max(pressures) - min(pressures) <= self.band_bar + 1e-9

cross_platform/src/pace_controller/ui.py

@@ -1113,9 +1113,11 @@ def on_telemetry(self, telemetry: object) -> None:
         self.metrics["state"].value.setStyleSheet(
             "color: #b42318;" if telemetry.control else "color: #2459e0;"
         )
-        self.metrics["target_state"].value.setText("IN LIMIT" if telemetry.in_limits else "MOVING")
+        self.metrics["target_state"].value.setText(
+            "IN LIMIT" if telemetry.in_limits else self.t("zero_target_stable") if telemetry.zero_target_stable else "MOVING"
+        )
         self.metrics["target_state"].value.setStyleSheet(
-            "color: #078419;" if telemetry.in_limits else "color: #d56a00;"
+            "color: #078419;" if telemetry.in_limits or telemetry.zero_target_stable else "color: #d56a00;"
         )
         self.metrics["source_margin"].value.setText(self.format_metric(telemetry.source_margin_bar, "bar"))
         self.metrics["source_margin"].value.setStyleSheet(

cross_platform/tests/test_zero_target.py

@@ -0,0 +1,81 @@
+import pytest
+
+from pace_controller.models import ControlParameters, PressureStep, Telemetry
+from pace_controller.service import PaceService
+from pace_controller.stability import ZeroTargetStability
+
+
+def test_zero_accepts_stable_residual_only_after_ten_seconds():
+    monitor = ZeroTargetStability()
+    for timestamp in range(10):
+        assert not monitor.add(timestamp, 0.5, 0.0, True)
+    assert monitor.add(10.0, 0.5, 0.0, True)
+    assert not monitor.add(11.0, 0.48, 0.0, True)
+
+
+@pytest.mark.parametrize("pressure,target,control", [
+    (0.51, 0.0, True), (5.0, 0.0, True),
+    (0.5, 0.1, True), (0.5, 0.0, False), (float("nan"), 0.0, True),
+])
+def test_zero_rejects_outside_tolerance_nonzero_target_and_invalid_readings(pressure, target, control):
+    monitor = ZeroTargetStability()
+    for timestamp in range(15):
+        assert not monitor.add(timestamp, pressure, target, control)
+
+
+def test_zero_does_not_accept_a_slow_ramp_or_missing_telemetry():
+    monitor = ZeroTargetStability()
+    for timestamp in range(20):
+        assert not monitor.add(timestamp, 0.49 - timestamp * 0.002, 0.0, True)
+    monitor.reset()
+    for timestamp in range(11):
+        stable = monitor.add(timestamp, 0.49 + (0.002 if timestamp % 2 else 0.0), 0.0, True)
+    assert stable
+    assert not monitor.add(20.0, 0.49, 0.0, True)
+    assert not monitor.add(21.0, 0.49, 0.0, True)
+
+
+@pytest.mark.parametrize("target,stable,dwell,expected", [
+    (0.0, True, 0.0, "complete"),
+    (0.0, True, 5.0, "dwelling"),
+    (0.0, False, 0.0, "waiting"),
+    (1.0, True, 0.0, "waiting"),
+])
+def test_automation_accepts_zero_plateau_without_hardware_in_limit(monkeypatch, tmp_path, target, stable, dwell, expected):
+    monkeypatch.setenv("PACE_CONTROLLER_DATA_DIR", str(tmp_path))
+    service = PaceService()
+    service._telemetry = Telemetry(current_pressure_bar=0.5, target_pressure_bar=target, control=True, in_limits=False, zero_target_stable=stable)
+    service._automation = {"steps": [PressureStep(target, 0.1, dwell)], "index": 0, "state": "waiting", "deadline": 100.0}
+    completed = []
+    service._complete_step = lambda: completed.append(True)
+    service._process_automation(20.0)
+    if expected == "complete":
+        assert completed
+    else:
+        assert not completed
+        assert service._automation["state"] == expected
+
+
+def test_poll_recognizes_zero_plateau_and_preserves_true_pressure(monkeypatch, tmp_path):
+    monkeypatch.setenv("PACE_CONTROLLER_DATA_DIR", str(tmp_path))
+    service = PaceService()
+    service._connected = True
+    clock = [0.0]
+    monkeypatch.setattr("pace_controller.service.time.monotonic", lambda: clock[0])
+    values = {
+        ":SENS1:PRES:CONT?": "0.5", ":SOUR1:PRES?": "0",
+        ":OUTP1:STAT?": "1", ":SOUR1:PRES:COMP1?": "50",
+        ":SENS1:PRES:INL?": "0",
+    }
+    service._query = lambda command: values.get(command, "0")
+    for timestamp in range(11):
+        clock[0] = float(timestamp)
+        service._poll()
+    assert service.telemetry.zero_target_stable
+    assert not service.telemetry.in_limits
+    assert service.telemetry.current_pressure_bar == 0.5
+    service._write = lambda command: None
+    service._assert_no_error = lambda: None
+    service._apply_pressure_step(PressureStep(0.0, 0.1), ControlParameters())
+    assert not service.telemetry.zero_target_stable
+    assert not service._zero_target_monitor.samples