feat(flomni): add continuous free-running raw-sample mode to PandaBoxOMNY

Adds an opt-in raw_stream_mode to omny_panda alongside the existing
per-gate averaged burst acquisition, so a burst window can instead be
captured as continuous raw samples plus a detector-gate bit and a
monotonic frame counter for completeness checking. Client-side
coalescing bounds the message rate to the BEC message bus regardless of
PandA's own network-frame batching, and completion uses a settle-based
check since an exact expected sample count isn't predictable for
continuous capture. Defaults to off, preserving today's behavior
exactly; hardware wiring/layout and the raw data schema are documented
under docs/developer/ pending PandA hardware access to configure and
validate against.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

csaxs_bec/device_configs/ptycho_flomni.yaml

@@ -646,33 +646,43 @@ calculated_signal:
 ############################################################
 #################### OMNY Pandabox #########################
 ############################################################
-# omny_panda:
-#   readoutPriority: async
-#   deviceClass: csaxs_bec.devices.panda_box.panda_box_omny.PandaBoxOMNY
-#   deviceConfig:
-#     host: omny-panda.psi.ch
-#     signal_alias:
-#       FMC_IN.VAL1.Min: cap_voltage_fzp_y_min
-#       FMC_IN.VAL1.Max: cap_voltage_fzp_y_max
-#       FMC_IN.VAL1.Mean: cap_voltage_fzp_y_mean
-#       FMC_IN.VAL2.Min: cap_voltage_fzp_x_min
-#       FMC_IN.VAL2.Max: cap_voltage_fzp_x_max
-#       FMC_IN.VAL2.Mean: cap_voltage_fzp_x_mean
-#       INENC1.VAL.Max: interf_st_fzp_y_max
-#       INENC1.VAL.Mean: interf_st_fzp_y_mean
-#       INENC1.VAL.Min: interf_st_fzp_y_min
-#       INENC2.VAL.Max: interf_st_fzp_x_max
-#       INENC2.VAL.Mean: interf_st_fzp_x_mean
-#       INENC2.VAL.Min: interf_st_fzp_x_min
-#       INENC3.VAL.Max: interf_st_rotz_max
-#       INENC3.VAL.Mean: interf_st_rotz_mean
-#       INENC3.VAL.Min: interf_st_rotz_min
-#       INENC4.VAL.Max: interf_st_rotx_max
-#       INENC4.VAL.Mean: interf_st_rotx_mean
-#       INENC4.VAL.Min: interf_st_rotx_min
-#       PCAP.GATE_DURATION.Value: pcap_gate_duration_value
-#   enabled: true
-#   readOnly: false
-#   softwareTrigger: false
-#   deviceTags:
-#     - ptycho_flomni
+omny_panda:
+  readoutPriority: async
+  deviceClass: csaxs_bec.devices.panda_box.panda_box_omny.PandaBoxOMNY
+  deviceConfig:
+    host: omny-panda.psi.ch
+    # raw_stream_mode selects the free-running raw-sample acquisition mode described in
+    # docs/developer/panda_box_free_running_setup.md. Defaults to false (legacy per-gate
+    # aggregate capture, as configured below) - do not enable until the PandaBox layout has
+    # been reconfigured for continuous capture per that document.
+    raw_stream_mode: false
+    signal_alias:
+      FMC_IN.VAL1.Min: cap_voltage_fzp_y_min
+      FMC_IN.VAL1.Max: cap_voltage_fzp_y_max
+      FMC_IN.VAL1.Mean: cap_voltage_fzp_y_mean
+      FMC_IN.VAL2.Min: cap_voltage_fzp_x_min
+      FMC_IN.VAL2.Max: cap_voltage_fzp_x_max
+      FMC_IN.VAL2.Mean: cap_voltage_fzp_x_mean
+      INENC1.VAL.Max: interf_st_fzp_y_max
+      INENC1.VAL.Mean: interf_st_fzp_y_mean
+      INENC1.VAL.Min: interf_st_fzp_y_min
+      INENC2.VAL.Max: interf_st_fzp_x_max
+      INENC2.VAL.Mean: interf_st_fzp_x_mean
+      INENC2.VAL.Min: interf_st_fzp_x_min
+      INENC3.VAL.Max: interf_st_rotz_max
+      INENC3.VAL.Mean: interf_st_rotz_mean
+      INENC3.VAL.Min: interf_st_rotz_min
+      INENC4.VAL.Max: interf_st_rotx_max
+      INENC4.VAL.Mean: interf_st_rotx_mean
+      INENC4.VAL.Min: interf_st_rotx_min
+      PCAP.GATE_DURATION.Value: pcap_gate_duration_value
+      # Free-running mode (raw_stream_mode: true) signals. Block numbers below are
+      # placeholders (TODO) pending the actual PandaBox wiring/layout - see
+      # docs/developer/panda_box_free_running_setup.md.
+      PCAP.BITS0.Value: gate_detector_active
+      COUNTER1.OUT.Value: frame_counter
+  enabled: true
+  readOnly: false
+  softwareTrigger: false
+  deviceTags:
+    - ptycho_flomni

csaxs_bec/devices/panda_box/panda_box_omny.py

@@ -1,11 +1,14 @@
 """Module to integrate the PandaBox for cSAXS measurements."""
 
+import threading
 import time
+from typing import Any
 
 from bec_lib.logger import bec_logger
 from bec_server.scan_server.scans.scan_base import ScanInfo as ScanServerScanInfo
 from ophyd_devices import StatusBase
 from ophyd_devices.devices.panda_box.panda_box import PandaBox, PandaState
+from pandablocks.responses import FrameData
 
 from csaxs_bec.devices.utils.utils import fetch_scan_info
 
@@ -13,21 +16,81 @@ logger = bec_logger.logger
 
 
 class PandaBoxOMNY(PandaBox):
-    """PandaBox integration for OMNY. This class implements OMNY specific logic for the PandaBox integration."""
+    """PandaBox integration for OMNY. This class implements OMNY specific logic for the PandaBox integration.
+
+    Two operation modes are supported, selected via ``raw_stream_mode``:
+
+    - Legacy mode (``raw_stream_mode=False``, default): unchanged behavior. Each PCAP gate
+      (one per scan point / burst sub-frame) produces one aggregate statistic
+      (Mean/Min/Max/Sum/Diff) that is published immediately, and completion is checked
+      against the exact expected number of gates (``num_points * frames_per_trigger``).
+    - Free-running mode (``raw_stream_mode=True``): the PCAP module is expected to be
+      configured on the PandaBox itself to capture raw ``Value`` samples continuously at a
+      much higher rate for the duration of a burst window, alongside a digital gate-input bit
+      and a monotonic frame counter (see docs/developer/panda_box_free_running_setup.md).
+      Incoming samples are coalesced client-side (by row count or time, whichever comes
+      first) before being published, to keep the message rate to the BEC message bus
+      bounded regardless of how the PandaBox itself batches network frames. Completion is
+      checked by waiting for the hardware-reported captured-frame count to stop increasing,
+      since an exact expected sample count can't be predicted client-side for continuous
+      capture.
+    """
+
+    def __init__(
+        self,
+        *,
+        name: str,
+        host: str,
+        signal_alias: dict[str, str] | None = None,
+        scan_info: ScanServerScanInfo | None = None,
+        device_manager: Any | None = None,
+        raw_stream_mode: bool = False,
+        raw_stream_flush_interval: float = 0.05,
+        raw_stream_flush_row_count: int = 500,
+        **kwargs,
+    ) -> None:
+        # Free-running raw-stream configuration. Defaults preserve today's behavior:
+        # raw_stream_mode=False means _receive_frame_data/on_stage/on_complete are
+        # byte-for-byte identical to before this mode was added.
+        self.raw_stream_mode = raw_stream_mode
+        self.raw_stream_flush_interval = raw_stream_flush_interval
+        self.raw_stream_flush_row_count = raw_stream_flush_row_count
+        self._raw_stream_buffer: dict[str, dict[str, Any]] = {}
+        self._raw_stream_buffer_row_count = 0
+        self._raw_stream_buffer_lock = threading.Lock()
+        self._raw_stream_last_flush = time.monotonic()
+
+        super().__init__(
+            name=name,
+            host=host,
+            signal_alias=signal_alias,
+            scan_info=scan_info,
+            device_manager=device_manager,
+            **kwargs,
+        )
 
     def on_init(self):
         super().on_init()
         self._acquisition_group = "burst"
         self._timeout_on_completed = 10
+        self._settle_timeout_in_s = 0.2
         self.scan_parameters: ScanServerScanInfo | None = None
 
     def on_stage(self):
         start_time = time.time()
         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()
+
         # TODO, adjust as seen fit.
         # Adjust the acquisition group based on scan parameters if needed
-        if self.scan_parameters.scan_type == "hardware_triggered":
+        if self.raw_stream_mode:
+            self._acquisition_group = "free_running"
+        elif self.scan_parameters.scan_type == "hardware_triggered":
             self._acquisition_group = "fly"
         elif self.scan_parameters.scan_type == "software_triggered":
             if self.scan_parameters.frames_per_trigger == 1:
@@ -37,8 +100,19 @@ class PandaBoxOMNY(PandaBox):
 
         logger.info(f"PandaBox {self.name} on_stage completed in {time.time() - start_time:.3f}s.")
 
+    def on_unstage(self):
+        """Make sure no buffered raw-stream data is left behind before resetting the device."""
+        self._flush_raw_stream_buffer()
+        return super().on_unstage()
+
     def on_complete(self):
         """On complete is called after the scan is complete. We need to wait for the capture to complete before we can disarm the PandaBox."""
+        if self.raw_stream_mode:
+            return self._on_complete_free_running()
+        return self._on_complete_burst()
+
+    def _on_complete_burst(self):
+        """Legacy completion check: wait for the exact expected number of gated captures."""
 
         def _check_capture_complete():
             captured = 0
@@ -66,6 +140,93 @@ class PandaBoxOMNY(PandaBox):
         self.cancel_on_stop(status_captured)
         return status_captured
 
+    def _on_complete_free_running(self):
+        """Completion check for continuous free-running capture: wait until the
+        hardware-reported captured-frame count stops increasing (settles), since the exact
+        expected sample count can't be predicted client-side for continuous capture at an
+        arbitrary rate. The frame_counter signal remains the authoritative record of total
+        frames captured for later completeness verification.
+        """
+
+        def _check_capture_settled():
+            captured = -1
+            last_change = time.monotonic()
+            start_time = time.monotonic()
+            try:
+                while True:
+                    ret = self.send_raw("*PCAP.CAPTURED?")
+                    new_captured = int(ret[0].split("=")[-1])
+                    now = time.monotonic()
+                    if new_captured != captured:
+                        captured = new_captured
+                        last_change = now
+                    elif now - last_change > self._settle_timeout_in_s:
+                        break
+                    if now - start_time > self._timeout_on_completed:
+                        raise TimeoutError(
+                            f"Pandabox {self.name} capture did not settle after {self._timeout_on_completed}s, "
+                            f"last captured count: {captured}."
+                        )
+                    time.sleep(0.01)
+            finally:
+                self._flush_raw_stream_buffer()
+                self._disarm()
+
+        status_captured = self.task_handler.submit_task(_check_capture_settled, run=True)
+        self.cancel_on_stop(status_captured)
+        return status_captured
+
+    def _receive_frame_data(self, data: FrameData) -> None:
+        """Callback to receive frame data from the PandaBox.
+
+        In legacy mode this is identical to the base class: publish immediately, one
+        FrameData in, one `.data.put()` out. In free-running mode, converted rows are
+        accumulated in a buffer and only published once a row-count or time threshold is
+        reached, to bound the message rate to the BEC message bus.
+        """
+        if not self.raw_stream_mode:
+            super()._receive_frame_data(data)
+            return
+
+        converted = self.convert_frame_data(frame_data=data)
+        with self._raw_stream_buffer_lock:
+            self._merge_into_raw_stream_buffer(converted)
+            elapsed = time.monotonic() - self._raw_stream_last_flush
+            if (
+                self._raw_stream_buffer_row_count >= self.raw_stream_flush_row_count
+                or elapsed >= self.raw_stream_flush_interval
+            ):
+                self._flush_raw_stream_buffer_locked()
+
+    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`."""
+        row_count = 0
+        for key, entry in converted.items():
+            buffered = self._raw_stream_buffer.setdefault(
+                key, {"value": [], "timestamp": entry["timestamp"]}
+            )
+            buffered["value"].extend(entry["value"])
+            buffered["timestamp"] = entry["timestamp"]
+            row_count = max(row_count, len(entry["value"]))
+        self._raw_stream_buffer_row_count += row_count
+
+    def _flush_raw_stream_buffer(self) -> None:
+        """Flush any pending buffered raw-stream data, taking the lock. Safe to call from
+        any thread, e.g. on_complete/on_unstage, even if nothing is buffered."""
+        with self._raw_stream_buffer_lock:
+            self._flush_raw_stream_buffer_locked()
+
+    def _flush_raw_stream_buffer_locked(self) -> None:
+        """Flush any pending buffered raw-stream data. Must be called while holding
+        `_raw_stream_buffer_lock`."""
+        self._raw_stream_last_flush = time.monotonic()
+        if not self._raw_stream_buffer:
+            return
+        out, self._raw_stream_buffer = self._raw_stream_buffer, {}
+        self._raw_stream_buffer_row_count = 0
+        self.data.put(out, acquisition_group=self._acquisition_group)
+
 
 if __name__ == "__main__":
     import time

docs/developer/panda_box_free_running_data_schema.md

@@ -0,0 +1,47 @@
+# PandABox free-running (`raw_stream_mode`) raw data schema
+
+> **Status: schema for the acquisition side only.** This document defines
+> what raw data `omny_panda` writes to its HDF5 async dataset when
+> `raw_stream_mode: true` (see `docs/developer/panda_box_free_running_setup.md`
+> for the hardware/mode setup). Reconstructing the previous
+> averaged/stdev-per-position view from this raw data (binning by the
+> `gate_detector_active` transitions or timestamps) is a separate,
+> not-yet-planned analysis task and is intentionally out of scope here.
+
+## Where this data lands
+
+`omny_panda` is configured with `readoutPriority: async`
+(`csaxs_bec/device_configs/ptycho_flomni.yaml`). Its `data` signal is a
+`DynamicSignal` (`ophyd_devices/devices/panda_box/panda_box.py`) that BEC's
+async file writer appends to an HDF5 dataset per scan, under this device's
+async data group. In `raw_stream_mode`, each write corresponds to one
+coalesced flush (see the setup doc's mode-comparison table), not one raw
+PandA network frame — the row-level data below is still preserved exactly,
+just batched differently for publishing efficiency.
+
+## Per-row fields
+
+Each row corresponds to one PCAP capture tick (one raw sample at the
+free-running capture rate, e.g. ~20 kHz — see the setup doc for the actual
+configured rate):
+
+| Field (BEC signal name) | Source PandA block | Meaning |
+|---|---|---|
+| *(raw analog/position value(s), name TBD per setup doc)* | e.g. `FMC_IN.VAL1.Value` / `INENC1.VAL.Value` | The instantaneous raw sample at this capture tick, replacing the Mode A aggregate (Mean/Min/Max) for whichever channel(s) are switched to raw capture. |
+| `gate_detector_active` | `PCAP.BITS0.Value` (block TBD) | The other detectors' gate signal, sampled at this same tick: `1` while the other detectors are actively gated/acquiring, `0` otherwise. This is what a future reconstruction step would use to delimit which raw samples belong to which detector-gate window. |
+| `frame_counter` | `COUNTER1.OUT.Value` (block TBD) | Monotonically incrementing hardware counter, one increment per capture tick. Used to verify data completeness after acquisition: a gap-free, strictly increasing sequence means no samples were dropped between the PandA and the HDF5 file; a gap indicates lost data (e.g. from the Redis stream's `MAXLEN` trimming if the file writer fell behind — see the acquisition plan's data-path notes). |
+| `PCAP.TS_TRIG.Value` (existing) | `PCAP.TS_TRIG` | Absolute/relative timestamp already captured today; usable for time-alignment against other detectors' data during future reconstruction. |
+| `PCAP.GATE_DURATION.Value` (existing, aliased as `pcap_gate_duration_value`) | `PCAP.GATE_DURATION` | Existing field, retained; less meaningful once the gate is held open continuously for the whole burst window rather than pulsed per point (see setup doc, Mode B gate/arm row) — re-evaluate whether this still needs to be captured once real hardware is available. |
+
+All rows across a burst window share the same set of fields; there is no
+per-point boundary marker in this raw stream other than what can be derived
+from `gate_detector_active` transitions and/or absolute timestamps — no
+explicit "scan point index" field is captured by the PandA itself.
+
+## What this does *not* include
+
+- No on-the-fly or post-scan computation of Mean/Min/Max/stdev from the raw
+  samples — the raw stream is the complete record; deriving the previous
+  per-point statistics from it is a separate future task.
+- No changes to how Mode A (legacy) data is written — this schema only
+  applies when `raw_stream_mode: true`.

docs/developer/panda_box_free_running_setup.md

@@ -0,0 +1,122 @@
+# PandABox (`omny_panda`) burst acquisition: two operation modes
+
+> **Status: design + software implemented, hardware not yet configured or
+> tested.** The PandA layout changes and target sample rate below are
+> placeholders (marked TODO) until real PandA hardware is available to
+> configure and measure against. This document should be updated with actual
+> values once that happens, per the flomni burst-acquisition redesign plan.
+
+## Context
+
+flomni burst scans acquire `frames_per_trigger` sub-frames per scan point.
+Historically, the PandABox (`omny_panda`, device class `PandaBoxOMNY` in
+`csaxs_bec/devices/panda_box/panda_box_omny.py`) reports one PCAP-aggregated
+statistic (Mean/Min/Max/Sum/Diff) per gate, discarding the time structure
+within a gate. This document describes the two supported operation modes —
+the existing per-gate aggregate mode, and a new continuous free-running raw
+mode — and how the PandA hardware and `ptycho_flomni.yaml` device config must
+be set up for each.
+
+Mode selection is a single software flag (`raw_stream_mode` in the device
+config); it does **not** by itself reconfigure the PandA hardware. The PandA
+layout must independently match the intended mode (see "Open question" at
+the end) before that mode can be used.
+
+## Mode comparison
+
+| | Mode A: legacy (default) | Mode B: free-running (new) |
+|---|---|---|
+| Selected by | `raw_stream_mode` unset / `false` | `raw_stream_mode: true` |
+| PCAP capture fields | `Mean`/`Min`/`Max`/`Sum`/`Diff` per gate | `Value` (raw instantaneous sample) + gate bit + frame counter |
+| Gate/arm timing | Pulsed per point / sub-frame (rate = scan rate × `frames_per_trigger`) | PCAP held open continuously for the burst window, free-running at a high internal sample rate (target: **TODO**, e.g. ~20 kHz) |
+| `_acquisition_group` (BEC async metadata) | `fly` / `monitored` / `burst`, chosen from `scan_type`/`frames_per_trigger` | `free_running` |
+| Message publishing | One `.data.put()` per PandA network frame, unchanged | Coalesced client-side: flushed every `raw_stream_flush_row_count` rows or `raw_stream_flush_interval` seconds, whichever comes first (mitigates Redis/HDF5 message-rate risk — see the acquisition plan's data-path notes) |
+| Completion check (`on_complete`) | Exact count: waits for `*PCAP.CAPTURED?` to reach `num_points * frames_per_trigger` | Settle-based: waits for `*PCAP.CAPTURED?` to stop changing for `_settle_timeout_in_s`, since an exact expected sample count can't be predicted client-side for continuous capture |
+| Data completeness verification | Implicit in the exact-count completion check | The `frame_counter` signal (Mode B only) — a monotonically incrementing hardware counter — is the record to check for gaps/drops after the fact |
+
+Both modes are implemented in the same `PandaBoxOMNY` class; `raw_stream_mode`
+defaults to `false` so any device config that doesn't set it keeps today's
+exact behavior unchanged.
+
+## Physical wiring (Mode B only)
+
+**TODO**: which PandA digital input connector/pin the other detectors' gate
+TTL signal is patched into. Fill in once the cabling is done.
+
+## Block configuration
+
+### Mode A (legacy) — reference baseline
+
+The blocks already configured and aliased today, for reference / so this
+layout can be restored unambiguously (see `ptycho_flomni.yaml`):
+
+- `FMC_IN.VAL1`/`FMC_IN.VAL2` (capacitive sensor analog inputs) — `Min`,
+  `Max`, `Mean` capture fields.
+- `INENC1`-`INENC4.VAL` (position-capture encoder inputs) — `Min`, `Max`,
+  `Mean` capture fields.
+- `PCAP.GATE_DURATION` — `Value`.
+- PCAP gate is pulsed once per point / sub-frame by the existing external
+  trigger electronics (unchanged by this work).
+
+### Mode B (new) — free-running raw capture
+
+- **Gate-bit input**: the other detectors' gate signal (wired above) →
+  **TODO: which `PCAP.BITSx` block and bit position** — set
+  `CAPTURE=Value` so every captured row records the instantaneous gate
+  state (0/1).
+- **Frame counter**: dedicate **TODO: which `COUNTERx.OUT` block** (any of
+  `COUNTER1`-`COUNTER8` not already in use) to free-run, incrementing once
+  per PCAP capture tick, with `CAPTURE=Value`. This is the
+  data-completeness signal (`frame_counter` in `signal_alias`, see below).
+- **Raw analog/position capture**: switch **TODO: which of
+  `FMC_IN.VAL1/VAL2`/`INENC1`-`4.VAL`** from aggregate capture to
+  `CAPTURE=Value`, so the raw time series is recorded instead of (or in
+  addition to) the per-gate statistics.
+- **Continuous gate/arm**: instead of pulsing PCAP per point, keep the gate
+  open for the whole burst window (armed once per point-group / scan) so
+  PCAP free-runs at the target sample rate — **TODO: confirm target rate**
+  (design target: ~20 kHz) and how the continuous gate is generated (e.g. a
+  PandA-internal clock/pulse-generator block feeding PCAP's gate input,
+  rather than the external per-point trigger used in Mode A).
+
+## Saving / restoring layouts
+
+Use `ophyd_devices/devices/panda_box/utility_scripts.py` to save the current
+PandA configuration to a `.ini` layout file and reload it later, so a
+configuration survives a PandA power cycle. **TODO**: once Mode B's block
+configuration is finalized against real hardware, save it as a named layout
+(e.g. `flomni_panda_free_running.ini`) alongside the existing Mode A layout,
+and record the exact save/load commands used here.
+
+## Signal alias mapping
+
+The physical PandA block feeding each BEC signal name, kept here and in
+`csaxs_bec/device_configs/ptycho_flomni.yaml` (`omny_panda.deviceConfig.signal_alias`)
+in sync as a single source of truth:
+
+| PandA block | Mode | BEC signal name |
+|---|---|---|
+| `FMC_IN.VAL1.Min`/`.Max`/`.Mean` | A | `cap_voltage_fzp_y_min`/`_max`/`_mean` |
+| `FMC_IN.VAL2.Min`/`.Max`/`.Mean` | A | `cap_voltage_fzp_x_min`/`_max`/`_mean` |
+| `INENC1.VAL.Min`/`.Max`/`.Mean` | A | `interf_st_fzp_y_min`/`_max`/`_mean` |
+| `INENC2.VAL.Min`/`.Max`/`.Mean` | A | `interf_st_fzp_x_min`/`_max`/`_mean` |
+| `INENC3.VAL.Min`/`.Max`/`.Mean` | A | `interf_st_rotz_min`/`_max`/`_mean` |
+| `INENC4.VAL.Min`/`.Max`/`.Mean` | A | `interf_st_rotx_min`/`_max`/`_mean` |
+| `PCAP.GATE_DURATION.Value` | A | `pcap_gate_duration_value` |
+| `PCAP.BITS0.Value` (**TODO: confirm bit block**) | B | `gate_detector_active` |
+| `COUNTER1.OUT.Value` (**TODO: confirm counter block**) | B | `frame_counter` |
+
+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`.
+
+## Open question to resolve once hardware is available
+
+Can both modes coexist under a single static PandA layout — i.e. is mode
+switching purely the `raw_stream_mode` software flag, with the PandA always
+capturing both the aggregate and raw fields simultaneously — or do Mode A
+and Mode B require two distinct saved layouts that must be loaded before a
+scan depending on the desired mode? If two layouts are required,
+`raw_stream_mode` alone is necessary but not sufficient to switch modes, and
+the operator must also load the matching PandA layout beforehand — update
+this document with the answer and, if applicable, the load procedure.

docs/user/ptychography/flomni.md

@@ -233,6 +233,14 @@ The basic scan function can be called by `scans.flomni_fermat_scan()` and offers
 Example:  
 `scans.flomni_fermat_scan(fovx=20, fovy=25, cenx=0.02, ceny=0, zshift=0, angle=0, step=0.5, exp_time=0.01, frames_per_trigger=1)`
 
+`frames_per_trigger` controls burst acquisition for detectors generally.
+Separately, the `omny_panda` PandABox device has its own `raw_stream_mode`
+device-config flag (`csaxs_bec/device_configs/ptycho_flomni.yaml`) that
+switches it between the default per-gate averaged acquisition and a
+continuous, high-rate raw-sample acquisition mode — see
+`docs/developer/panda_box_free_running_setup.md` for what each mode captures
+and how to enable it.
+
 #### Overview of the alignment steps
 
 There are several corrections applied to maintain the sample in the FOV:

docs/user/ptychography/omny.md

@@ -338,6 +338,14 @@ The basic scan function can be called by `scans.omny_fermat_scan()` and offers a
 Example:  
 `scans.omny_fermat_scan(fovx=20, fovy=25, cenx=0.02, ceny=0, zshift=0, angle=0, step=0.5, exp_time=0.01, frames_per_trigger=1, readout_time=0)`
 
+`frames_per_trigger` controls burst acquisition for detectors generally.
+Separately, the `omny_panda` PandABox device has its own `raw_stream_mode`
+device-config flag (`csaxs_bec/device_configs/ptycho_flomni.yaml`) that
+switches it between the default per-gate averaged acquisition and a
+continuous, high-rate raw-sample acquisition mode — see
+`docs/developer/panda_box_free_running_setup.md` for what each mode captures
+and how to enable it.
+
 #### Overview of the alignment steps
 
 There are several corrections applied to maintain the sample in the FOV:

tests/tests_devices/test_panda.py

@@ -5,14 +5,26 @@ from __future__ import annotations
 
 from unittest import mock
 
+import numpy as np
 import pytest
 from ophyd import Staged
+from pandablocks.responses import FrameData
 
 from csaxs_bec.devices.panda_box.panda_box import PandaBoxCSAXS
 from csaxs_bec.devices.panda_box.panda_box_omny import PandaBoxOMNY
 from csaxs_bec.devices.utils.utils import fetch_scan_info
 
 
+def _make_frame_data(n_rows: int, start: int = 0) -> FrameData:
+    """Build a synthetic FrameData batch with `n_rows` rows for the frame_counter /
+    gate_detector_active signals used by the raw-stream tests below."""
+    data = np.array(
+        [(np.float64(i), np.float64(i % 2)) for i in range(start, start + n_rows)],
+        dtype=[("COUNTER1.OUT.Value", "<f8"), ("PCAP.BITS0.Value", "<f8")],
+    )
+    return FrameData(data)
+
+
 @pytest.fixture
 def panda_omny():
     dev_name = "panda_omny"
@@ -31,6 +43,25 @@ def panda_omny():
     yield dev
 
 
+@pytest.fixture
+def panda_omny_raw_stream():
+    dev_name = "panda_omny_raw"
+    dev = PandaBoxOMNY(
+        name=dev_name,
+        host="omny-panda-box.psi.ch",
+        signal_alias={
+            "PCAP.BITS0.Value": "gate_detector_active",
+            "COUNTER1.OUT.Value": "frame_counter",
+        },
+        raw_stream_mode=True,
+        raw_stream_flush_row_count=5,
+        # Effectively disable the time-based flush so tests are deterministic and only
+        # exercise the row-count threshold.
+        raw_stream_flush_interval=1000,
+    )
+    yield dev
+
+
 @pytest.fixture
 def panda_csaxs():
     dev_name = "panda_csaxs"
@@ -124,6 +155,138 @@ def test_panda_omny_complete(panda_omny):
         assert status.success is True
 
 
+def test_panda_omny_default_raw_stream_mode_is_off(panda_omny):
+    """New devices must default to legacy behavior unless raw_stream_mode is explicitly set."""
+    assert panda_omny.raw_stream_mode is False
+
+
+def test_panda_omny_legacy_receive_frame_data_is_unbuffered(panda_omny):
+    """Backward-compatibility regression: without raw_stream_mode, each FrameData must still
+    be published immediately, exactly as before this feature was added."""
+    with mock.patch.object(panda_omny.data, "put") as mock_put:
+        for i in range(3):
+            panda_omny._receive_frame_data(_make_frame_data(2, start=i * 2))
+        assert mock_put.call_count == 3
+
+
+def test_panda_omny_raw_stream_acquisition_group(panda_omny_raw_stream):
+    """raw_stream_mode must select the free_running acquisition group regardless of
+    scan_type/frames_per_trigger."""
+    panda_omny_raw_stream.scan_info.msg.info["scan_type"] = "software_triggered"
+    panda_omny_raw_stream.scan_info.msg.info["frames_per_trigger"] = 10
+    panda_omny_raw_stream.stage()
+
+    assert panda_omny_raw_stream._acquisition_group == "free_running"
+    assert panda_omny_raw_stream.staged == Staged.yes
+
+
+def test_panda_omny_raw_stream_coalesces_by_row_count(panda_omny_raw_stream):
+    """Multiple small FrameData batches must be coalesced into a single .data.put() call
+    once the row-count threshold is reached, not published one-by-one."""
+    dev = panda_omny_raw_stream
+    with mock.patch.object(dev.data, "put") as mock_put:
+        dev._receive_frame_data(_make_frame_data(2, start=0))
+        dev._receive_frame_data(_make_frame_data(2, start=2))
+        mock_put.assert_not_called()  # 4 rows buffered so far, threshold is 5
+
+        dev._receive_frame_data(_make_frame_data(2, start=4))
+        mock_put.assert_called_once()  # 6 rows >= 5, flush triggered
+
+        out = mock_put.call_args.args[0]
+        assert [v for v in out["frame_counter"]["value"]] == [0.0, 1.0, 2.0, 3.0, 4.0, 5.0]
+        assert [v for v in out["gate_detector_active"]["value"]] == [0.0, 1.0, 0.0, 1.0, 0.0, 1.0]
+
+
+def test_panda_omny_raw_stream_flush_on_demand(panda_omny_raw_stream):
+    """Buffered data below the threshold must still be flushed on demand (as on_complete/
+    on_unstage do), and flushing an empty buffer must not publish again."""
+    dev = panda_omny_raw_stream
+    with mock.patch.object(dev.data, "put") as mock_put:
+        dev._receive_frame_data(_make_frame_data(2, start=0))
+        mock_put.assert_not_called()
+
+        dev._flush_raw_stream_buffer()
+        mock_put.assert_called_once()
+        out = mock_put.call_args.args[0]
+        assert len(out["frame_counter"]["value"]) == 2
+
+        dev._flush_raw_stream_buffer()
+        mock_put.assert_called_once()  # still 1: nothing left to flush
+
+
+def test_panda_omny_unstage_flushes_pending_raw_stream_data(panda_omny_raw_stream):
+    dev = panda_omny_raw_stream
+    dev.stage()
+    with (
+        mock.patch.object(dev.data, "put") as mock_put,
+        mock.patch.object(dev, "_disarm", return_value=None),
+    ):
+        dev._receive_frame_data(_make_frame_data(2, start=0))
+        mock_put.assert_not_called()
+
+        dev.unstage()
+        mock_put.assert_called_once()
+
+
+def test_panda_omny_complete_free_running_settles(panda_omny_raw_stream):
+    """Completion in free-running mode waits for the captured count to stop increasing,
+    rather than an exact expected count."""
+    dev = panda_omny_raw_stream
+    dev.scan_info.msg.info["num_points"] = 1
+    dev.scan_info.msg.info["frames_per_trigger"] = 1
+    dev.scan_parameters = fetch_scan_info(dev.scan_info)
+    dev._timeout_on_completed = 5
+    dev._settle_timeout_in_s = 0.1
+
+    call_count = {"n": 0}
+
+    def _send_raw(*args, **kwargs):
+        # Ramp up for the first 3 calls, then plateau indefinitely at "=2" so the
+        # settle-timeout branch (rather than running out of a fixed side_effect list) is
+        # what ends the loop.
+        call_count["n"] += 1
+        return [f"={min(call_count['n'] - 1, 2)}"]
+
+    with (
+        mock.patch.object(dev, "send_raw", side_effect=_send_raw),
+        mock.patch.object(dev, "_disarm", return_value=None) as mock_disarm,
+        mock.patch.object(dev, "_flush_raw_stream_buffer") as mock_flush,
+    ):
+        status = dev.on_complete()
+        status.wait(timeout=4)
+        assert status.done is True
+        assert status.success is True
+        mock_disarm.assert_called_once()
+        mock_flush.assert_called_once()
+
+
+def test_panda_omny_complete_free_running_timeout(panda_omny_raw_stream):
+    """If the captured count never settles, the overall timeout must still fire."""
+    dev = panda_omny_raw_stream
+    dev.scan_info.msg.info["num_points"] = 1
+    dev.scan_info.msg.info["frames_per_trigger"] = 1
+    dev.scan_parameters = fetch_scan_info(dev.scan_info)
+    dev._timeout_on_completed = 0.3
+    dev._settle_timeout_in_s = 10  # never settles within the overall timeout
+
+    with (
+        mock.patch.object(dev, "send_raw", side_effect=lambda *a, **k: ["=0"]),
+        mock.patch.object(dev, "_disarm", return_value=None) as mock_disarm,
+    ):
+        status = dev.on_complete()
+        with pytest.raises(TimeoutError):
+            status.wait(timeout=4)
+        mock_disarm.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
+    assert "frame_counter" in all_signal_names
+    assert "PCAP.BITS0.Value" not in all_signal_names
+    assert "COUNTER1.OUT.Value" not in all_signal_names
+
+
 def test_panda_csaxs(panda_csaxs):
     assert panda_csaxs.name == "panda_csaxs"
     assert panda_csaxs.host == "csaxs-panda-box.psi.ch"