fix(panda): drop stale frame_counter carryover in Mode B raw stream

Diagnosed against a real scan (S06313): grouping raw interf_st_fzp_x/_y
samples by frame_counter + gate_detector_active reproduces rt_positions'
trusted average_x_st_fzp/_y to R² > 0.9999, confirming that method -- but one
point (167) was off by ~116 nm because PandA's COUNTER1.OUT is not cleared by
Arm(). The first raw samples of an acquisition can briefly carry a stale
frame_counter value left over from the previous acquisition's tail (observed:
36 samples reading a stale 167 before dropping to 0), which silently
contaminates whichever real point later in the same scan happens to share
that counter value.

PandaBoxOMNY now drops raw rows captured before it observes frame_counter ==
0 for the first time in each acquisition -- on every arm, including
on_pre_scan retries, since each Arm() reopens the same stale-value window --
and drops leftover buffered rows from an abandoned attempt at the same point.
Verified against S06313: point 167's residual drops from ~116 nm to ~0.1 nm
after the fix, in line with the rest of the scan.

Write-up, including the diagnostic method and a possible PandA-layout-level
alternative fix, in docs/developer/panda_box_free_running_setup.md.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01NPZ9gzN2rt3wgiMnTKyrug

csaxs_bec/devices/panda_box/panda_box_omny.py

@@ -56,6 +56,20 @@ class PandaBoxOMNY(PandaBox):
     # connecting, so a missing READY means the readout thread/box is in a stale state.
     _PRE_SCAN_ARM_ATTEMPTS = 2
 
+    # PandA's COUNTER1.OUT (the block behind frame_counter) is not cleared by Arm() --
+    # COUNTER1.ENABLE=PCAP.ACTIVE only gates counting, not the persisted output value, and
+    # COUNTER1.SET is wired to a constant (never pulsed) in the saved layout (see
+    # layouts/omny_freerun.ini). So the first raw samples of a new acquisition can briefly
+    # carry a stale frame_counter value left over from the previous acquisition's tail --
+    # confirmed 2026-09-21 against a real scan (S06313): 36 samples reading a stale value
+    # before dropping to 0 and counting correctly from there. If that stale value collides
+    # with a real point index later in the same scan, grouping raw samples by frame_counter
+    # silently mixes those stale rows into that point's average (see
+    # docs/developer/panda_box_free_running_setup.md, "Known data-quality caveat"). Rows are
+    # dropped client-side until the first observed frame_counter == 0 per acquisition to
+    # guard against this; see _reset_raw_stream_state/_drop_stale_frame_counter_rows below.
+    _FRAME_COUNTER_SIGNAL_NAME = "frame_counter"
+
     def __init__(
         self,
         *,
@@ -80,6 +94,10 @@ class PandaBoxOMNY(PandaBox):
         self._raw_stream_buffer_row_count = 0
         self._raw_stream_buffer_lock = threading.Lock()
         self._raw_stream_last_flush = time.monotonic()
+        # Whether frame_counter == 0 has been observed yet for the current acquisition; see
+        # _FRAME_COUNTER_SIGNAL_NAME above. Reset for every fresh arm, not just once per stage,
+        # since on_pre_scan retries re-arm the box (and thus reset COUNTER1's counting) too.
+        self._raw_stream_seen_reset = False
 
         # Seconds to wait for the PandA READY event per arming attempt in on_pre_scan.
         self.pre_scan_timeout = pre_scan_timeout
@@ -108,10 +126,7 @@ class PandaBoxOMNY(PandaBox):
         super().on_stage()
         self.scan_parameters = fetch_scan_info(self.scan_info)
 
-        with self._raw_stream_buffer_lock:
-            self._raw_stream_buffer = {}
-            self._raw_stream_buffer_row_count = 0
-            self._raw_stream_last_flush = time.monotonic()
+        self._reset_raw_stream_state()
 
         # TODO, adjust as seen fit.
         # Adjust the acquisition group based on scan parameters if needed
@@ -200,6 +215,10 @@ class PandaBoxOMNY(PandaBox):
             self.remove_status_callback(cb_id)
             logger.warning(f"PandaBox {self.name} arming attempt failed: {exc!r}")
             return False
+        # Reset raw-stream state right before arming, not just once in on_stage: a retry here
+        # re-arms the box (and thus resets COUNTER1's counting) too, so the stale-value window
+        # applies again on every attempt, not just the first.
+        self._reset_raw_stream_state()
         # Sent from here rather than from a status callback (as the base class does), so the
         # scan is only released after Arm() actually went out.
         self._arm()
@@ -320,6 +339,11 @@ class PandaBoxOMNY(PandaBox):
             return
 
         converted = self.convert_frame_data(frame_data=data)
+        converted = self._drop_stale_frame_counter_rows(converted)
+        if converted is None:
+            # Every row in this chunk still carried a stale frame_counter from the previous
+            # acquisition; nothing here belongs to the current one.
+            return
         with self._raw_stream_buffer_lock:
             self._merge_into_raw_stream_buffer(converted)
             elapsed = time.monotonic() - self._raw_stream_last_flush
@@ -329,6 +353,53 @@ class PandaBoxOMNY(PandaBox):
             ):
                 self._flush_raw_stream_buffer_locked()
 
+    def _reset_raw_stream_state(self) -> None:
+        """Drop any leftover buffered rows and mark that frame_counter == 0 has not yet been
+        observed, for a fresh acquisition. Called from on_stage (once per scan) and from
+        _try_arm right before every Arm() (including on_pre_scan retries), since each Arm()
+        resets COUNTER1's counting and thus reopens the stale-value window described at
+        _FRAME_COUNTER_SIGNAL_NAME above. Also drops whatever was buffered from a previous,
+        now-abandoned attempt, since that data belongs to an acquisition that never completed.
+        """
+        with self._raw_stream_buffer_lock:
+            self._raw_stream_buffer = {}
+            self._raw_stream_buffer_row_count = 0
+            self._raw_stream_last_flush = time.monotonic()
+        self._raw_stream_seen_reset = False
+
+    def _drop_stale_frame_counter_rows(
+        self, converted: dict[str, dict[str, Any]]
+    ) -> dict[str, dict[str, Any]] | None:
+        """Drop rows captured before frame_counter first reads 0 for this acquisition.
+
+        See the class-level note on `_FRAME_COUNTER_SIGNAL_NAME`: PandA's COUNTER1.OUT is not
+        cleared by Arm(), so the first raw samples of a new acquisition can briefly carry a
+        stale value left over from the previous acquisition's tail. Truncating every signal's
+        row list at the same index keeps all channels aligned.
+
+        Returns the (possibly truncated) dict, or None if every row in this chunk is still
+        stale and should be dropped entirely.
+        """
+        if self._raw_stream_seen_reset:
+            return converted
+        counter_entry = converted.get(self._FRAME_COUNTER_SIGNAL_NAME)
+        if counter_entry is None:
+            # This layout/config doesn't capture frame_counter -- nothing to filter against.
+            self._raw_stream_seen_reset = True
+            return converted
+        reset_index = next(
+            (i for i, v in enumerate(counter_entry["value"]) if v == 0), None
+        )
+        if reset_index is None:
+            return None
+        self._raw_stream_seen_reset = True
+        if reset_index == 0:
+            return converted
+        return {
+            key: {**entry, "value": entry["value"][reset_index:]}
+            for key, entry in converted.items()
+        }
+
     def _merge_into_raw_stream_buffer(self, converted: dict[str, dict[str, Any]]) -> None:
         """Append newly converted rows into the pending raw-stream buffer. Must be called
         while holding `_raw_stream_buffer_lock`."""

docs/developer/panda_box_free_running_setup.md

@@ -220,6 +220,70 @@ The BEC-side data schema produced by Mode B (what ends up in the HDF5 async
 dataset) is documented separately in
 `docs/developer/panda_box_free_running_data_schema.md`.
 
+## Known data-quality caveat: stale `frame_counter` carryover (found + fixed 2026-09-21)
+
+Diagnosed against a real scan (`S06313`): grouping raw `interf_st_fzp_x`/`_y`
+samples by `frame_counter` (restricted to `gate_detector_active == 1`) and
+averaging per group reproduces `rt_positions`'s trusted
+`average_x_st_fzp`/`average_y_st_fzp` to R² > 0.9999 once a single constant
+per-axis offset is removed -- an expected reference-zero difference between
+the PandA/`INENC` counter and RT's own internal tracking, irrelevant to
+reconstruction since it's the same constant for every point in a scan. This
+also confirms the grouping/filtering method itself: naively treating every
+raw row as its own position (without grouping by `frame_counter` and
+filtering to `gate_detector_active == 1`) does not reproduce the true
+positions and cannot be fixed by any scale/offset correction, since most raw
+rows are samples taken between real exposures, not at them (see "Continuous
+gate/arm" above).
+
+One point out of 472 was a clear outlier under that method: its
+`frame_counter` group had 38 extra samples (1036 vs. the usual ~998) and its
+recomputed position was off by ~116 nm in Y, against ~5 nm typical.
+
+Root cause: PandA's `COUNTER1.OUT` (the block behind `frame_counter`) is not
+cleared by `Arm()`. `COUNTER1.ENABLE=PCAP.ACTIVE` only gates *counting*, not
+the persisted output value, and `COUNTER1.SET` -- the block's reset-trigger
+input -- is wired to the constant `ZERO` in the saved layout (see
+`layouts/omny_freerun.ini`), i.e. never pulsed on arm, unlike `ENABLE` which
+*is* wired to a live source (`PCAP.ACTIVE`). So the first raw samples of a new
+acquisition can briefly carry a stale `frame_counter` value left over from the
+previous acquisition's tail -- observed directly in `S06313`'s raw data: 36
+samples reading a stale `167` before dropping to `0` and counting correctly
+from there. Harmless on its own, except that stale value collided with a real
+point index later in the *same* scan (point 167), so grouping by
+`frame_counter` silently mixed ~38 stale rows into that one point's average.
+
+Fix, implemented in `panda_box_omny.py` (software-only, no PandA layout
+change needed): every time PCAP is armed -- including `on_pre_scan` retries,
+since each `Arm()` reopens the same stale-value window -- `PandaBoxOMNY` now
+drops any raw rows captured before it observes `frame_counter == 0` for the
+first time in that acquisition (`_drop_stale_frame_counter_rows`, called from
+`_receive_frame_data`), and drops any leftover buffered rows from an
+abandoned previous attempt at the same point (`_reset_raw_stream_state`,
+called from `on_stage` and from `_try_arm` right before every `Arm()`). Every
+signal's row list is truncated at the same index to keep channels aligned.
+Verified against `S06313`: dropping the (37, in that scan) leading stale rows
+before grouping brings point 167's residual from ~116 nm down to ~0.1 nm, in
+line with the rest of the scan.
+
+A more thorough fix would rewire `COUNTER1.SET` on the PandA layout itself
+(from the constant `ZERO` to `PCAP.ACTIVE`'s rising edge, matching how
+`ENABLE` is already wired) so `OUT` resets at the source on every arm, rather
+than relying on the client to detect and discard the stale prefix. Not
+implemented here since it requires a PandA layout change to be tested against
+live hardware; the software-side drop is a correct and sufficient fix in the
+meantime, and remains a reasonable defense-in-depth even if the layout is
+fixed later. If this is revisited: re-verify against a range of scans
+(especially ones with `frame_counter` staying below a few hundred, where a
+stale-value collision like point 167's is most likely) before removing the
+software-side check.
+
+This also means the "gap in an otherwise-monotonic sequence" completeness
+check described in the mode-comparison table's "Data completeness
+verification" row needs one caveat: a single backward step at the very start
+of the raw sequence, from the previous acquisition's terminal value down to
+`0`, is this expected (now-filtered) artifact, not a sign of dropped rows.
+
 ## Switching between modes
 
 `omny_panda` (Mode A) and `omny_panda_continuous` (Mode B) need different PandA hardware

tests/tests_devices/test_panda.py

@@ -384,6 +384,106 @@ def test_panda_omny_pre_scan_stop_is_not_retried(panda_omny, capsys):
     assert "Forcing abort/reset" not in capsys.readouterr().out
 
 
+def test_panda_omny_raw_stream_drops_stale_leading_frame_counter_rows(panda_omny_raw_stream):
+    """PandA's COUNTER1.OUT is not cleared by Arm(), so the first raw samples of a new
+    acquisition can briefly carry a stale frame_counter value left over from the previous
+    acquisition's tail (confirmed 2026-09-21 against a real scan, S06313). Rows before the
+    first observed frame_counter == 0 must be dropped, and every channel must be truncated
+    at the same row so they stay aligned."""
+    dev = panda_omny_raw_stream
+    # Two stale rows carrying the previous acquisition's terminal counter value (167), then
+    # the real acquisition starting at frame_counter 0.
+    data = np.array(
+        [(167.0, 1.0), (167.0, 0.0), (0.0, 1.0), (0.0, 0.0), (1.0, 1.0)],
+        dtype=[("COUNTER1.OUT.Value", "<f8"), ("PCAP.BITS0.Value", "<f8")],
+    )
+    with mock.patch.object(dev.data, "put") as mock_put:
+        dev._receive_frame_data(FrameData(data))
+        dev._flush_raw_stream_buffer()
+
+        out = mock_put.call_args.args[0]
+        assert [v for v in out["frame_counter"]["value"]] == [0.0, 0.0, 1.0]
+        assert [v for v in out["gate_detector_active"]["value"]] == [1, 0, 1]
+
+
+def test_panda_omny_raw_stream_drops_whole_chunk_if_still_stale(panda_omny_raw_stream):
+    """If a single FrameData batch is entirely stale rows (no frame_counter == 0 observed
+    yet), it must be dropped in full rather than partially buffered, and the next batch
+    must still be checked for the reset."""
+    dev = panda_omny_raw_stream
+    all_stale = np.array(
+        [(167.0, 0.0), (167.0, 1.0)], dtype=[("COUNTER1.OUT.Value", "<f8"), ("PCAP.BITS0.Value", "<f8")]
+    )
+    real_start = np.array(
+        [(0.0, 1.0), (1.0, 0.0)], dtype=[("COUNTER1.OUT.Value", "<f8"), ("PCAP.BITS0.Value", "<f8")]
+    )
+    with mock.patch.object(dev.data, "put") as mock_put:
+        dev._receive_frame_data(FrameData(all_stale))
+        assert dev._raw_stream_buffer == {}
+        assert dev._raw_stream_seen_reset is False
+
+        dev._receive_frame_data(FrameData(real_start))
+        dev._flush_raw_stream_buffer()
+
+        out = mock_put.call_args.args[0]
+        assert [v for v in out["frame_counter"]["value"]] == [0.0, 1.0]
+
+
+def test_panda_omny_raw_stream_passes_through_once_reset_seen(panda_omny_raw_stream):
+    """After the first frame_counter == 0 has been observed, later rows must never be
+    filtered again, even if a later point's real frame_counter happens to collide with the
+    earlier stale value (e.g. point 167 in a scan with more than 167 points)."""
+    dev = panda_omny_raw_stream
+    with mock.patch.object(dev.data, "put"):
+        dev._receive_frame_data(_make_frame_data(3, start=0))  # observes the reset at row 0
+        assert dev._raw_stream_seen_reset is True
+        dev._flush_raw_stream_buffer()
+
+    # A later chunk whose frame_counter happens to read 167 again must pass through
+    # untouched -- filtering only ever applies before the first reset.
+    collide = np.array(
+        [(167.0, 1.0), (168.0, 0.0)], dtype=[("COUNTER1.OUT.Value", "<f8"), ("PCAP.BITS0.Value", "<f8")]
+    )
+    with mock.patch.object(dev.data, "put") as mock_put:
+        dev._receive_frame_data(FrameData(collide))
+        dev._flush_raw_stream_buffer()
+        out = mock_put.call_args.args[0]
+        assert [v for v in out["frame_counter"]["value"]] == [167.0, 168.0]
+
+
+def test_panda_omny_raw_stream_state_reset_on_stage(panda_omny_raw_stream):
+    """on_stage must reopen the stale-value window (new acquisition) even if a previous
+    acquisition already observed its reset."""
+    dev = panda_omny_raw_stream
+    dev._raw_stream_seen_reset = True
+    dev._raw_stream_buffer = {"leftover": {"value": [1, 2, 3], "timestamp": 0.0}}
+
+    dev.stage()
+
+    assert dev._raw_stream_seen_reset is False
+    assert dev._raw_stream_buffer == {}
+
+
+def test_panda_omny_pre_scan_arm_resets_raw_stream_state(panda_omny):
+    """Each Arm() resets COUNTER1's counting on the PandA side, so every successful arm in
+    on_pre_scan -- not just the one implied by on_stage -- must reopen the stale-frame_counter
+    window right before Arm() is sent. This matters most for a retried arm after a forced
+    abort/reset, since on_stage's reset has already happened by then."""
+    dev = panda_omny
+    dev.pre_scan_timeout = 2
+    with (
+        mock.patch.object(dev, "_arm") as mock_arm,
+        mock.patch.object(dev, "_reset_panda"),
+        mock.patch.object(dev, "_reset_raw_stream_state") as mock_reset_raw,
+    ):
+        _deliver_ready_after(dev, 0.05)
+        status = dev.on_pre_scan()
+        status.wait(timeout=4)
+        assert status.success is True
+        mock_arm.assert_called_once()
+        mock_reset_raw.assert_called_once()
+
+
 def test_panda_omny_raw_stream_signal_alias(panda_omny_raw_stream):
     all_signal_names = [name for name, _ in panda_omny_raw_stream.data.signals]
     assert "gate_detector_active" in all_signal_names