added docs for burst acquisition

csaxs_bec/bec_ipython_client/plugins/LamNI/lamni_optics_mixin.py

@@ -9,9 +9,11 @@ from csaxs_bec.bec_ipython_client.plugins.cSAXS import epics_put, fshclose
 
 # import builtins to avoid linter errors
 dev = builtins.__dict__.get("dev")
-umv = builtins.__dict__.get("umv")
 bec = builtins.__dict__.get("bec")
+scans = builtins.__dict__.get("scans")
 
+def umv(*args):
+    return scans.umv(*args, relative=False)
 
 class LamNIInitError(Exception):
     pass

csaxs_bec/bec_ipython_client/plugins/cSAXS/cSAXSDLPCA200.py

@@ -0,0 +1,443 @@
+"""
+csaxs_dlpca200.py
+=================
+BEC control script for FEMTO DLPCA-200 Variable Gain Low Noise Current Amplifiers
+connected to Galil RIO digital outputs.
+
+DLPCA-200 Remote Control (datasheet page 4)
+-------------------------------------------
+Sub-D pin → function:
+  Pin 10  →  gain LSB   (digital out channel, index 0 in bit-tuple)
+  Pin 11  →  gain MID   (digital out channel, index 1 in bit-tuple)
+  Pin 12  →  gain MSB   (digital out channel, index 2 in bit-tuple)
+  Pin 13  →  coupling   LOW = AC,  HIGH = DC
+  Pin 14  →  speed mode HIGH = low noise (Pin14=1), LOW = high speed (Pin14=0)
+
+Gain truth table (MSB, MID, LSB):
+  0,0,0  →  low-noise: 1e3   high-speed: 1e5
+  0,0,1  →  low-noise: 1e4   high-speed: 1e6
+  0,1,0  →  low-noise: 1e5   high-speed: 1e7
+  0,1,1  →  low-noise: 1e6   high-speed: 1e8
+  1,0,0  →  low-noise: 1e7   high-speed: 1e9
+  1,0,1  →  low-noise: 1e8   high-speed: 1e10
+  1,1,0  →  low-noise: 1e9   high-speed: 1e11
+
+Strategy: prefer low-noise mode (1e3–1e9). For 1e10 and 1e11,
+automatically fall back to high-speed mode.
+
+Device wiring example (galilrioesxbox):
+  bpm4:  Pin10→ch0, Pin11→ch1, Pin12→ch2, Pin13→ch3, Pin14→ch4
+  bim:   Pin10→ch6, Pin11→ch7, Pin12→ch8, Pin13→ch9, Pin14→ch10
+
+Usage examples
+--------------
+  csaxs_amp = cSAXSDLPCA200(client)
+
+  csaxs_amp.set_gain("bpm4", 1e7)          # low-noise if possible
+  csaxs_amp.set_gain("bim",  1e10)         # auto high-speed
+  csaxs_amp.set_coupling("bpm4", "DC")
+  csaxs_amp.set_coupling("bim",  "AC")
+  csaxs_amp.info("bpm4")                   # print current settings
+  csaxs_amp.info_all()                     # print all configured amplifiers
+"""
+
+import builtins
+
+from bec_lib import bec_logger
+
+logger = bec_logger.logger
+
+bec = builtins.__dict__.get("bec")
+dev = builtins.__dict__.get("dev")
+
+
+# ---------------------------------------------------------------------------
+# Amplifier registry
+# ---------------------------------------------------------------------------
+# Each entry describes one DLPCA-200 amplifier connected to a Galil RIO.
+#
+# Keys inside "channels":
+#   gain_lsb   → digital output channel number wired to DLPCA-200 Pin 10
+#   gain_mid   → digital output channel number wired to DLPCA-200 Pin 11
+#   gain_msb   → digital output channel number wired to DLPCA-200 Pin 12
+#   coupling   → digital output channel number wired to DLPCA-200 Pin 13
+#   speed_mode → digital output channel number wired to DLPCA-200 Pin 14
+#
+# To add a new amplifier, simply extend this dict.
+# ---------------------------------------------------------------------------
+DLPCA200_AMPLIFIER_CONFIG: dict[str, dict] = {
+    "bpm4": {
+        "rio_device": "galilrioesxbox",
+        "description": "Beam Position Monitor 4 current amplifier",
+        "channels": {
+            "gain_lsb":   0,   # Pin 10 → Galil ch0
+            "gain_mid":   1,   # Pin 11 → Galil ch1
+            "gain_msb":   2,   # Pin 12 → Galil ch2
+            "coupling":   3,   # Pin 13 → Galil ch3
+            "speed_mode": 4,   # Pin 14 → Galil ch4
+        },
+    },
+    "bim": {
+        "rio_device": "galilrioesxbox",
+        "description": "Beam Intensity Monitor current amplifier",
+        "channels": {
+            "gain_lsb":   6,   # Pin 10 → Galil ch6
+            "gain_mid":   7,   # Pin 11 → Galil ch7
+            "gain_msb":   8,   # Pin 12 → Galil ch8
+            "coupling":   9,   # Pin 13 → Galil ch9
+            "speed_mode": 10,  # Pin 14 → Galil ch10
+        },
+    },
+}
+
+# ---------------------------------------------------------------------------
+# DLPCA-200 gain encoding tables
+# ---------------------------------------------------------------------------
+# (msb, mid, lsb) → gain in V/A
+_GAIN_BITS_LOW_NOISE: dict[tuple, int] = {
+    (0, 0, 0): int(1e3),
+    (0, 0, 1): int(1e4),
+    (0, 1, 0): int(1e5),
+    (0, 1, 1): int(1e6),
+    (1, 0, 0): int(1e7),
+    (1, 0, 1): int(1e8),
+    (1, 1, 0): int(1e9),
+}
+
+_GAIN_BITS_HIGH_SPEED: dict[tuple, int] = {
+    (0, 0, 0): int(1e5),
+    (0, 0, 1): int(1e6),
+    (0, 1, 0): int(1e7),
+    (0, 1, 1): int(1e8),
+    (1, 0, 0): int(1e9),
+    (1, 0, 1): int(1e10),
+    (1, 1, 0): int(1e11),
+}
+
+# Inverse maps: gain → (msb, mid, lsb, low_noise_flag)
+# low_noise_flag: True = Pin14 HIGH, False = Pin14 LOW
+_GAIN_TO_BITS: dict[int, tuple] = {}
+for _bits, _gain in _GAIN_BITS_LOW_NOISE.items():
+    _GAIN_TO_BITS[_gain] = (*_bits, True)
+for _bits, _gain in _GAIN_BITS_HIGH_SPEED.items():
+    if _gain not in _GAIN_TO_BITS:          # low-noise takes priority
+        _GAIN_TO_BITS[_gain] = (*_bits, False)
+
+VALID_GAINS = sorted(_GAIN_TO_BITS.keys())
+
+
+class cSAXSDLPCA200Error(Exception):
+    pass
+
+
+class cSAXSDLPCA200:
+    """
+    Control class for FEMTO DLPCA-200 current amplifiers connected via Galil RIO
+    digital outputs in a BEC environment.
+
+    Supports:
+    - Forward control: set_gain(), set_coupling()
+    - Readback reporting: info(), info_all(), read_settings()
+    - Robust error handling and logging following cSAXS conventions.
+    """
+
+    TAG = "[DLPCA200]"
+
+    def __init__(self, client, config: dict | None = None) -> None:
+        """
+        Parameters
+        ----------
+        client : BEC client object (passed through for future use)
+        config : optional override for DLPCA200_AMPLIFIER_CONFIG.
+                 Falls back to the module-level dict if not provided.
+        """
+        self.client = client
+        self._config: dict[str, dict] = config if config is not None else DLPCA200_AMPLIFIER_CONFIG
+
+    # ------------------------------------------------------------------
+    # Internal helpers
+    # ------------------------------------------------------------------
+
+    def _require_dev(self) -> None:
+        if dev is None:
+            raise cSAXSDLPCA200Error(
+                f"{self.TAG} BEC 'dev' namespace is not available in this session."
+            )
+
+    def _get_cfg(self, amp_name: str) -> dict:
+        """Return config dict for a named amplifier, raising on unknown names."""
+        if amp_name not in self._config:
+            known = ", ".join(sorted(self._config.keys()))
+            raise cSAXSDLPCA200Error(
+                f"{self.TAG} Unknown amplifier '{amp_name}'. Known: [{known}]"
+            )
+        return self._config[amp_name]
+
+    def _get_rio(self, amp_name: str):
+        """Return the live RIO device object for a given amplifier."""
+        self._require_dev()
+        cfg = self._get_cfg(amp_name)
+        rio_name = cfg["rio_device"]
+        try:
+            rio = getattr(dev, rio_name)
+        except AttributeError:
+            raise cSAXSDLPCA200Error(
+                f"{self.TAG} RIO device '{rio_name}' not found in BEC 'dev'."
+            )
+        return rio
+
+    def _dout_get(self, rio, ch: int) -> int:
+        """Read one digital output channel (returns 0 or 1)."""
+        attr = getattr(rio.digital_out, f"ch{ch}")
+        val = attr.get()
+        return int(val)
+
+    def _dout_set(self, rio, ch: int, value: bool) -> None:
+        """Write one digital output channel (True=HIGH=1, False=LOW=0)."""
+        attr = getattr(rio.digital_out, f"ch{ch}")
+        attr.set(value)
+
+    def _read_gain_bits(self, amp_name: str) -> tuple[int, int, int, int]:
+        """
+        Read current gain bit-state from hardware.
+
+        Returns
+        -------
+        (msb, mid, lsb, speed_mode)
+            speed_mode: 1 = low-noise (Pin14=HIGH), 0 = high-speed (Pin14=LOW)
+        """
+        rio = self._get_rio(amp_name)
+        ch = self._get_cfg(amp_name)["channels"]
+        msb        = self._dout_get(rio, ch["gain_msb"])
+        mid        = self._dout_get(rio, ch["gain_mid"])
+        lsb        = self._dout_get(rio, ch["gain_lsb"])
+        speed_mode = self._dout_get(rio, ch["speed_mode"])
+        return msb, mid, lsb, speed_mode
+
+    def _decode_gain(self, msb: int, mid: int, lsb: int, speed_mode: int) -> int | None:
+        """
+        Decode hardware bit-state into gain value (V/A).
+
+        speed_mode=1 → low-noise table, speed_mode=0 → high-speed table.
+        Returns None if the bit combination is not in the table.
+        """
+        bits = (msb, mid, lsb)
+        if speed_mode:
+            return _GAIN_BITS_LOW_NOISE.get(bits)
+        else:
+            return _GAIN_BITS_HIGH_SPEED.get(bits)
+
+    # ------------------------------------------------------------------
+    # Public API – control
+    # ------------------------------------------------------------------
+
+    def set_gain(self, amp_name: str, gain: float, force_high_speed: bool = False) -> None:
+        """
+        Set the transimpedance gain of a DLPCA-200 amplifier.
+
+        The method automatically selects low-noise mode (Pin14=HIGH) whenever
+        the requested gain is achievable in low-noise mode (1e3 – 1e9 V/A).
+        For gains of 1e10 and 1e11 V/A, high-speed mode is used automatically.
+
+        Parameters
+        ----------
+        amp_name : str
+            Amplifier name as defined in DLPCA200_AMPLIFIER_CONFIG (e.g. "bpm4").
+        gain : float or int
+            Target gain in V/A. Must be one of:
+            1e3, 1e4, 1e5, 1e6, 1e7, 1e8, 1e9, 1e10, 1e11.
+        force_high_speed : bool, optional
+            If True, force high-speed (low-noise=False) mode even for gains
+            below 1e10. Default: False (prefer low-noise).
+
+        Examples
+        --------
+        csaxs_amp.set_gain("bpm4", 1e7)          # low-noise mode (automatic)
+        csaxs_amp.set_gain("bim",  1e10)         # high-speed mode (automatic)
+        csaxs_amp.set_gain("bpm4", 1e7, force_high_speed=True)  # override to high-speed
+        """
+        gain_int = int(gain)
+        if gain_int not in _GAIN_TO_BITS:
+            valid_str = ", ".join(f"1e{int(round(__import__('math').log10(g)))}" for g in VALID_GAINS)
+            raise cSAXSDLPCA200Error(
+                f"{self.TAG} Invalid gain {gain:.2e} V/A for '{amp_name}'. "
+                f"Valid values: {valid_str}"
+            )
+
+        msb, mid, lsb, low_noise_preferred = _GAIN_TO_BITS[gain_int]
+
+        # Apply force_high_speed override
+        if force_high_speed and low_noise_preferred:
+            # Check if this gain is achievable in high-speed mode
+            hs_entry = next(
+                (bits for bits, g in _GAIN_BITS_HIGH_SPEED.items() if g == gain_int), None
+            )
+            if hs_entry is None:
+                raise cSAXSDLPCA200Error(
+                    f"{self.TAG} Gain {gain:.2e} V/A is not achievable in high-speed mode "
+                    f"for '{amp_name}'."
+                )
+            msb, mid, lsb = hs_entry
+            low_noise_preferred = False
+
+        use_low_noise = low_noise_preferred and not force_high_speed
+
+        try:
+            rio = self._get_rio(amp_name)
+            ch  = self._get_cfg(amp_name)["channels"]
+
+            self._dout_set(rio, ch["gain_msb"],   bool(msb))
+            self._dout_set(rio, ch["gain_mid"],   bool(mid))
+            self._dout_set(rio, ch["gain_lsb"],   bool(lsb))
+            self._dout_set(rio, ch["speed_mode"], use_low_noise)  # True=low-noise
+
+            mode_str = "low-noise" if use_low_noise else "high-speed"
+            logger.info(
+                f"{self.TAG} [{amp_name}] gain set to {gain_int:.2e} V/A "
+                f"({mode_str} mode, bits MSB={msb} MID={mid} LSB={lsb})"
+            )
+            print(
+                f"{amp_name}: gain → {gain_int:.2e} V/A  [{mode_str}]  "
+                f"(bits: MSB={msb} MID={mid} LSB={lsb})"
+            )
+
+        except cSAXSDLPCA200Error:
+            raise
+        except Exception as exc:
+            raise cSAXSDLPCA200Error(
+                f"{self.TAG} Failed to set gain on '{amp_name}': {exc}"
+            ) from exc
+
+    def set_coupling(self, amp_name: str, coupling: str) -> None:
+        """
+        Set AC or DC coupling on a DLPCA-200 amplifier.
+
+        Parameters
+        ----------
+        amp_name : str
+            Amplifier name (e.g. "bpm4", "bim").
+        coupling : str
+            "AC" or "DC" (case-insensitive).
+            DC → Pin13 HIGH,  AC → Pin13 LOW.
+
+        Examples
+        --------
+        csaxs_amp.set_coupling("bpm4", "DC")
+        csaxs_amp.set_coupling("bim",  "AC")
+        """
+        coupling_upper = coupling.strip().upper()
+        if coupling_upper not in ("AC", "DC"):
+            raise cSAXSDLPCA200Error(
+                f"{self.TAG} Invalid coupling '{coupling}' for '{amp_name}'. "
+                f"Use 'AC' or 'DC'."
+            )
+
+        pin13_high = coupling_upper == "DC"
+
+        try:
+            rio = self._get_rio(amp_name)
+            ch  = self._get_cfg(amp_name)["channels"]
+            self._dout_set(rio, ch["coupling"], pin13_high)
+
+            logger.info(f"{self.TAG} [{amp_name}] coupling set to {coupling_upper}")
+            print(f"{amp_name}: coupling → {coupling_upper}")
+
+        except cSAXSDLPCA200Error:
+            raise
+        except Exception as exc:
+            raise cSAXSDLPCA200Error(
+                f"{self.TAG} Failed to set coupling on '{amp_name}': {exc}"
+            ) from exc
+
+    # ------------------------------------------------------------------
+    # Public API – readback / reporting
+    # ------------------------------------------------------------------
+
+    def read_settings(self, amp_name: str) -> dict:
+        """
+        Read back the current settings from hardware digital outputs.
+
+        Returns
+        -------
+        dict with keys:
+            "amp_name"   : str
+            "gain"       : int or None    – gain in V/A (None if unknown bit pattern)
+            "mode"       : str            – "low-noise" or "high-speed"
+            "coupling"   : str            – "AC" or "DC"
+            "bits"       : dict           – raw bit values {msb, mid, lsb, speed_mode, coupling}
+        """
+        rio = self._get_rio(amp_name)
+        ch  = self._get_cfg(amp_name)["channels"]
+
+        msb        = self._dout_get(rio, ch["gain_msb"])
+        mid        = self._dout_get(rio, ch["gain_mid"])
+        lsb        = self._dout_get(rio, ch["gain_lsb"])
+        speed_mode = self._dout_get(rio, ch["speed_mode"])
+        coupling_bit = self._dout_get(rio, ch["coupling"])
+
+        gain = self._decode_gain(msb, mid, lsb, speed_mode)
+        mode = "low-noise" if speed_mode else "high-speed"
+        coupling = "DC" if coupling_bit else "AC"
+
+        return {
+            "amp_name": amp_name,
+            "gain":     gain,
+            "mode":     mode,
+            "coupling": coupling,
+            "bits": {
+                "msb":        msb,
+                "mid":        mid,
+                "lsb":        lsb,
+                "speed_mode": speed_mode,
+                "coupling":   coupling_bit,
+            },
+        }
+
+    def info(self, amp_name: str) -> None:
+        """
+        Print a plain summary of the current settings for one amplifier.
+
+        Example output
+        --------------
+        Amplifier   : bpm4
+        Description : Beam Position Monitor 4 current amplifier
+        RIO device  : galilrioesxbox
+        Gain        : 1.00e+07 V/A
+        Mode        : low-noise
+        Coupling    : DC
+        Raw bits    : MSB=1 MID=0 LSB=0  speed=1 coup=1
+        """
+        cfg = self._get_cfg(amp_name)
+
+        try:
+            s = self.read_settings(amp_name)
+        except Exception as exc:
+            print(f"{self.TAG} [{amp_name}] Could not read settings: {exc}")
+            return
+
+        gain_str = (
+            f"{s['gain']:.2e} V/A" if s["gain"] is not None else "UNKNOWN (invalid bit pattern)"
+        )
+        bits = s["bits"]
+
+        print(f"  {'Amplifier':<12}: {amp_name}")
+        print(f"  {'Description':<12}: {cfg.get('description', '')}")
+        print(f"  {'RIO device':<12}: {cfg['rio_device']}")
+        print(f"  {'Gain':<12}: {gain_str}")
+        print(f"  {'Mode':<12}: {s['mode']}")
+        print(f"  {'Coupling':<12}: {s['coupling']}")
+        print(f"  {'Raw bits':<12}: MSB={bits['msb']} MID={bits['mid']} LSB={bits['lsb']}  speed={bits['speed_mode']} coup={bits['coupling']}")
+
+    def info_all(self) -> None:
+        """
+        Print a plain summary for ALL configured amplifiers.
+        """
+        print("\nDLPCA-200 Amplifier Status Report")
+        print("-" * 40)
+        for amp_name in sorted(self._config.keys()):
+            self.info(amp_name)
+            print()
+
+    def list_amplifiers(self) -> list[str]:
+        """Return sorted list of configured amplifier names."""
+        return sorted(self._config.keys())

csaxs_bec/bec_ipython_client/plugins/cSAXS/filter_transmission.py

@@ -41,8 +41,10 @@ import builtins
 if builtins.__dict__.get("bec") is not None:
     bec = builtins.__dict__.get("bec")
     dev = builtins.__dict__.get("dev")
-    umv = builtins.__dict__.get("umv")
-    umvr = builtins.__dict__.get("umvr")
+    scans = builtins.__dict__.get("scans")
+
+def umv(*args):
+    return scans.umv(*args, relative=False)
 
 class cSAXSFilterTransmission:
     """

csaxs_bec/bec_ipython_client/plugins/cSAXS/smaract.py

@@ -8,11 +8,14 @@ from bec_lib import bec_logger
 logger = bec_logger.logger
 
 # Pull BEC globals if present
-bec = builtins.__dict__.get("bec")
-dev = builtins.__dict__.get("dev")
-umv = builtins.__dict__.get("umv")
-umvr = builtins.__dict__.get("umvr")
+if builtins.__dict__.get("bec") is not None:
+    bec = builtins.__dict__.get("bec")
+    dev = builtins.__dict__.get("dev")
+    scans = builtins.__dict__.get("scans")
 
+def umv(*args):
+    return scans.umv(*args, relative=False)
+    
 
 class cSAXSInitSmaractStagesError(Exception):
     pass
@@ -383,7 +386,6 @@ class cSAXSInitSmaractStages:
         if not self._yesno("Proceed with the motions listed above?", "y"):
             logger.info("[cSAXS] Motion to initial position aborted by user.")
             return
-
         # --- Execution phase (SIMULTANEOUS MOTION) ---
         if umv is None:
             logger.error("[cSAXS] 'umv' is not available in this session.")

csaxs_bec/bec_ipython_client/plugins/flomni/flomni.py

@@ -22,8 +22,10 @@ logger = bec_logger.logger
 if builtins.__dict__.get("bec") is not None:
     bec = builtins.__dict__.get("bec")
     dev = builtins.__dict__.get("dev")
-    umv = builtins.__dict__.get("umv")
-    umvr = builtins.__dict__.get("umvr")
+    scans = builtins.__dict__.get("scans")
+
+def umv(*args):
+    return scans.umv(*args, relative=False)
 
 
 class FlomniToolsError(Exception):

csaxs_bec/bec_ipython_client/plugins/flomni/gui_tools.py

@@ -7,8 +7,10 @@ from bec_widgets.cli.client import BECDockArea
 if builtins.__dict__.get("bec") is not None:
     bec = builtins.__dict__.get("bec")
     dev = builtins.__dict__.get("dev")
-    umv = builtins.__dict__.get("umv")
-    umvr = builtins.__dict__.get("umvr")
+    scans = builtins.__dict__.get("scans")
+
+def umv(*args):
+    return scans.umv(*args, relative=False)
 
 
 class flomniGuiToolsError(Exception):

csaxs_bec/bec_ipython_client/plugins/flomni/x_ray_eye_align.py

@@ -13,8 +13,10 @@ logger = bec_logger.logger
 # import builtins to avoid linter errors
 bec = builtins.__dict__.get("bec")
 dev = builtins.__dict__.get("dev")
-umv = builtins.__dict__.get("umv")
-umvr = builtins.__dict__.get("umvr")
+scans = builtins.__dict__.get("scans")
+
+def umv(*args):
+    return scans.umv(*args, relative=False)
 
 if TYPE_CHECKING:
     from bec_ipython_client.plugins.flomni import Flomni

csaxs_bec/bec_ipython_client/plugins/omny/gui_tools.py

@@ -7,8 +7,10 @@ from bec_widgets.cli.client import BECDockArea
 if builtins.__dict__.get("bec") is not None:
     bec = builtins.__dict__.get("bec")
     dev = builtins.__dict__.get("dev")
-    umv = builtins.__dict__.get("umv")
-    umvr = builtins.__dict__.get("umvr")
+    scans = builtins.__dict__.get("scans")
+
+def umv(*args):
+    return scans.umv(*args, relative=False)
 
 
 class OMNYGuiToolsError(Exception):

csaxs_bec/bec_ipython_client/plugins/omny/omny.py

@@ -27,9 +27,10 @@ logger = bec_logger.logger
 if builtins.__dict__.get("bec") is not None:
     bec = builtins.__dict__.get("bec")
     dev = builtins.__dict__.get("dev")
-    umv = builtins.__dict__.get("umv")
-    umvr = builtins.__dict__.get("umvr")
+    scans = builtins.__dict__.get("scans")
 
+def umv(*args):
+    return scans.umv(*args, relative=False)
 
 class OMNYInitError(Exception):
     pass

csaxs_bec/bec_ipython_client/plugins/omny/omny_general_tools.py

@@ -16,8 +16,10 @@ from rich.table import Table
 if builtins.__dict__.get("bec") is not None:
     bec = builtins.__dict__.get("bec")
     dev = builtins.__dict__.get("dev")
-    umv = builtins.__dict__.get("umv")
-    umvr = builtins.__dict__.get("umvr")
+    scans = builtins.__dict__.get("scans")
+
+def umv(*args):
+    return scans.umv(*args, relative=False)
 
 
 class OMNYToolsError(Exception):

csaxs_bec/bec_ipython_client/plugins/omny/omny_sample_transfer_mixin.py

@@ -16,8 +16,10 @@ from csaxs_bec.bec_ipython_client.plugins.cSAXS import epics_get, epics_put, fsh
 if builtins.__dict__.get("bec") is not None:
     bec = builtins.__dict__.get("bec")
     dev = builtins.__dict__.get("dev")
-    umv = builtins.__dict__.get("umv")
-    umvr = builtins.__dict__.get("umvr")
+    scans = builtins.__dict__.get("scans")
+
+def umv(*args):
+    return scans.umv(*args, relative=False)
 
 
 class OMNYTransferError(Exception):

csaxs_bec/bec_ipython_client/plugins/omny/x_ray_eye_align.py

@@ -13,8 +13,10 @@ logger = bec_logger.logger
 # import builtins to avoid linter errors
 bec = builtins.__dict__.get("bec")
 dev = builtins.__dict__.get("dev")
-umv = builtins.__dict__.get("umv")
-umvr = builtins.__dict__.get("umvr")
+scans = builtins.__dict__.get("scans")
+
+def umv(*args):
+    return scans.umv(*args, relative=False)
 
 if TYPE_CHECKING:
     from bec_ipython_client.plugins.omny import OMNY

csaxs_bec/bec_ipython_client/startup/post_startup.py

@@ -30,29 +30,74 @@ logger = bec_logger.logger
 
 logger.info("Using the cSAXS startup script.")
 
-# pylint: disable=import-error
-_args = _main_dict["args"]
-
-_session_name = "cSAXS"
-if _args.session.lower() == "lamni":
-    from csaxs_bec.bec_ipython_client.plugins.cSAXS import *
-    from csaxs_bec.bec_ipython_client.plugins.LamNI import *
-
-    _session_name = "LamNI"
-    lamni = LamNI(bec)
-    logger.success("LamNI session loaded.")
-
-elif _args.session.lower() == "csaxs":
-    print("Loading cSAXS session")
-    from csaxs_bec.bec_ipython_client.plugins.cSAXS import *
-
-    logger.success("cSAXS session loaded.")
 
 from csaxs_bec.bec_ipython_client.plugins.tool_box.debug_tools import DebugTools
 
 debug = DebugTools()
 logger.success("Debug tools loaded. Use 'debug' to access them.")
 
+# pylint: disable=import-error
+_args = _main_dict["args"]
+
+_session_name = "cSAXS"
+
+print("Loading cSAXS session")
+from csaxs_bec.bec_ipython_client.plugins.cSAXS.cSAXS import cSAXS
+csaxs = cSAXS(bec)
+logger.success("cSAXS session loaded.")
+
+
+if _args.session.lower() == "lamni":
+    from csaxs_bec.bec_ipython_client.plugins.LamNI import LamNI
+
+    _session_name = "LamNI"
+    lamni = LamNI(bec)
+    logger.success("LamNI session loaded.")
+    print(r"""
+    ██████╗ ███████╗ ██████╗     ██╗      █████╗ ███╗   ███╗███╗   ██╗██╗
+    ██╔══██╗██╔════╝██╔════╝     ██║     ██╔══██╗████╗ ████║████╗  ██║██║
+    ██████╔╝█████╗  ██║          ██║     ███████║██╔████╔██║██╔██╗ ██║██║
+    ██╔══██╗██╔══╝  ██║          ██║     ██╔══██║██║╚██╔╝██║██║╚██╗██║██║
+    ██████╔╝███████╗╚██████╗     ███████╗██║  ██║██║ ╚═╝ ██║██║ ╚████║██║
+    ╚═════╝ ╚══════╝ ╚═════╝     ╚══════╝╚═╝  ╚═╝╚═╝     ╚═╝╚═╝  ╚═══╝╚═╝
+                                                                            
+                                B E C   L a m N I
+    """)
+
+elif _args.session.lower() == "omny":
+    from csaxs_bec.bec_ipython_client.plugins.flomni import OMNY
+
+    _session_name = "OMNY"
+    omny = OMNY(bec)
+    logger.success("OMNY session loaded.")
+    print(r"""
+    ██████╗ ███████╗ ██████╗      ██████╗ ███╗   ███╗███╗   ██╗██╗   ██╗
+    ██╔══██╗██╔════╝██╔════╝     ██╔═══██╗████╗ ████║████╗  ██║╚██╗ ██╔╝
+    ██████╔╝█████╗  ██║          ██║   ██║██╔████╔██║██╔██╗ ██║ ╚████╔╝ 
+    ██╔══██╗██╔══╝  ██║          ██║   ██║██║╚██╔╝██║██║╚██╗██║  ╚██╔╝  
+    ██████╔╝███████╗╚██████╗     ╚██████╔╝██║ ╚═╝ ██║██║ ╚████║   ██║   
+    ╚═════╝ ╚══════╝ ╚═════╝      ╚═════╝ ╚═╝     ╚═╝╚═╝  ╚═══╝   ╚═╝   
+                                                                        
+                                B E C   O M N Y
+    """)
+
+elif _args.session.lower() == "flomni":
+    from csaxs_bec.bec_ipython_client.plugins.flomni import Flomni
+
+    _session_name = "flomni"
+    flomni = Flomni(bec)
+    logger.success("flomni session loaded.")
+    print(r"""
+    ██████╗ ███████╗ ██████╗     ███████╗██╗      ██████╗ ███╗   ███╗███╗   ██╗██╗
+    ██╔══██╗██╔════╝██╔════╝     ██╔════╝██║     ██╔═══██╗████╗ ████║████╗  ██║██║
+    ██████╔╝█████╗  ██║          █████╗  ██║     ██║   ██║██╔████╔██║██╔██╗ ██║██║
+    ██╔══██╗██╔══╝  ██║          ██╔══╝  ██║     ██║   ██║██║╚██╔╝██║██║╚██╗██║██║
+    ██████╔╝███████╗╚██████╗     ██║     ███████╗╚██████╔╝██║ ╚═╝ ██║██║ ╚████║██║
+    ╚═════╝ ╚══════╝ ╚═════╝     ╚═╝     ╚══════╝ ╚═════╝ ╚═╝     ╚═╝╚═╝  ╚═══╝╚═╝
+                                                                                    
+                                B E C   f l O M N I
+    """)
+
 
 # SETUP BEAMLINE INFO
 from bec_ipython_client.plugins.SLS.sls_info import OperatorInfo, SLSInfo

csaxs_bec/device_configs/bl_detectors.yaml

@@ -9,27 +9,27 @@ eiger_1_5:
   readoutPriority: async
   softwareTrigger: False
 
-eiger_9:
-  description: Eiger 9M detector 
-  deviceClass: csaxs_bec.devices.jungfraujoch.eiger_9m.Eiger9M
-  deviceConfig:
-    detector_distance: 100
-    beam_center: [0, 0]
-  onFailure: raise
-  enabled: true
-  readoutPriority: async
-  softwareTrigger: False
+# eiger_9:
+#   description: Eiger 9M detector 
+#   deviceClass: csaxs_bec.devices.jungfraujoch.eiger_9m.Eiger9M
+#   deviceConfig:
+#     detector_distance: 100
+#     beam_center: [0, 0]
+#   onFailure: raise
+#   enabled: true
+#   readoutPriority: async
+#   softwareTrigger: False
 
-ids_cam:
-  description: IDS camera for live image acquisition
-  deviceClass: csaxs_bec.devices.ids_cameras.IDSCamera
-  deviceConfig:
-    camera_id: 201
-    bits_per_pixel: 24
-    m_n_colormode: 1
-    live_mode: True
-  onFailure: raise
-  enabled: true
-  readoutPriority: async
-  softwareTrigger: True
+# ids_cam:
+#   description: IDS camera for live image acquisition
+#   deviceClass: csaxs_bec.devices.ids_cameras.IDSCamera
+#   deviceConfig:
+#     camera_id: 201
+#     bits_per_pixel: 24
+#     m_n_colormode: 1
+#     live_mode: True
+#   onFailure: raise
+#   enabled: true
+#   readoutPriority: async
+#   softwareTrigger: True
 

csaxs_bec/device_configs/bl_general.yaml

@@ -0,0 +1,25 @@
+############################################################
+##################### EPS ##################################
+############################################################
+x12saEPS:
+  description: X12SA EPS info and control
+  deviceClass: csaxs_bec.devices.epics.eps.EPS
+  deviceConfig: {}
+  enabled: true
+  onFailure: buffer
+  readOnly: false
+  readoutPriority: baseline
+############################################################
+##################### GalilRIO #############################
+############################################################
+
+galilrioesxbox:
+  description: Galil RIO for remote gain switching and slow reading ES XBox
+  deviceClass: csaxs_bec.devices.omny.galil.galil_rio.GalilRIO
+  deviceConfig:
+    host: galilrioesxbox.psi.ch
+  enabled: true
+  onFailure: raise
+  readOnly: false
+  readoutPriority: baseline
+  connectionTimeout: 20

csaxs_bec/device_configs/main.yaml

@@ -1,11 +1,11 @@
 # This is the main configuration file that is
 # commented or uncommented according to the type of experiment
 
-optics:
-  - !include ./bl_optics_hutch.yaml
+# optics:
+#   - !include ./bl_optics_hutch.yaml
 
-frontend:
-  - !include ./bl_frontend.yaml
+# frontend:
+#   - !include ./bl_frontend.yaml
 
 endstation:
   - !include ./bl_endstation.yaml
@@ -16,8 +16,8 @@ detectors:
 #sastt:
 #  - !include ./sastt.yaml
 
-#flomni:
-#  - !include ./ptycho_flomni.yaml
+flomni:
+ - !include ./ptycho_flomni.yaml
 
 #omny:
 #  - !include ./ptycho_omny.yaml

csaxs_bec/device_configs/ptycho_flomni.yaml

@@ -471,7 +471,7 @@ omnyfsh:
 #################### GUI Signals ###########################
 ############################################################
 omny_xray_gui:
-  description: Gui Epics signals
+  description: Gui signals
   deviceClass: csaxs_bec.devices.omny.xray_epics_gui.OMNYXRayEpicsGUI
   deviceConfig: {}
   enabled: true
@@ -486,4 +486,25 @@ calculated_signal:
     compute_method: "def just_rand():\n return 42"
   enabled: true
   readOnly: false
-  readoutPriority: baseline
+  readoutPriority: baseline
+
+############################################################
+#################### 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
+  deviceTags:
+    - detector
+  enabled: true
+  readOnly: false
+  softwareTrigger: false

csaxs_bec/devices/omny/galil/galil_ophyd.py

@@ -4,6 +4,7 @@ This module contains the base class for Galil controllers as well as the signals
 
 import functools
 import time
+from typing import Any
 
 from bec_lib import bec_logger
 from ophyd.utils import ReadOnlyError
@@ -347,7 +348,7 @@ class GalilSignalBase(SocketSignal):
     def __init__(self, signal_name, **kwargs):
         self.signal_name = signal_name
         super().__init__(**kwargs)
-        self.controller = self.parent.controller
+        self.controller = self.root.controller if hasattr(self.root, "controller") else None
 
 
 class GalilSignalRO(GalilSignalBase):

csaxs_bec/devices/omny/galil/galil_rio.py

@@ -6,24 +6,26 @@ Link to the Galil RIO vendor page:
 https://www.galil.com/plcs/remote-io/rio-471xx
 
 This module provides the GalilRIOController for communication with the RIO controller
-over TCP/IP. It also provides a device integration that interfaces to these
-8 analog channels.
+over TCP/IP. It also provides a device integration that interfaces to its
+8 analog channels, and 16 digital output channels. Some PLCs may have 24 digital output channels,
+which can be easily supported by changing the _NUM_DIGITAL_OUTPUT_CHANNELS variable.
 """
 
 from __future__ import annotations
 
-import time
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, Literal
 
 from bec_lib.logger import bec_logger
-from ophyd import Component as Cpt
+from ophyd import DynamicDeviceComponent as DDC
+from ophyd import Kind
+from ophyd.utils import ReadOnlyError
 from ophyd_devices import PSIDeviceBase
 from ophyd_devices.utils.controller import Controller, threadlocked
 from ophyd_devices.utils.socket import SocketIO
 
 from csaxs_bec.devices.omny.galil.galil_ophyd import (
     GalilCommunicationError,
-    GalilSignalRO,
+    GalilSignalBase,
     retry_once,
 )
 
@@ -35,15 +37,11 @@ logger = bec_logger.logger
 
 
 class GalilRIOController(Controller):
-    """
-    Controller Class for Galil RIO controller communication.
-
-    Multiple controllers are in use at the cSAXS beamline:
-    - 129.129.98.64 (port 23)
-    """
+    """Controller Class for Galil RIO controller communication."""
 
     @threadlocked
     def socket_put(self, val: str) -> None:
+        """Socker put method."""
         self.sock.put(f"{val}\r".encode())
 
     @retry_once
@@ -64,21 +62,95 @@ class GalilRIOController(Controller):
             )
 
 
-class GalilRIOSignalRO(GalilSignalRO):
+class GalilRIOAnalogSignalRO(GalilSignalBase):
     """
-    Read-only Signal for reading a single analog input channel from the Galil RIO controller.
-    It always read all 8 analog channels at once, and updates the reabacks of all channels.
-    New readbacks are only fetched from the controller if the last readback is older than
-    _READ_TIMEOUT seconds, otherwise the last cached readback is returned to reduce network traffic.
+    Signal for reading analog input channels of the Galil RIO controller. This signal is read-only, so
+    the set method raises a ReadOnlyError. The get method retrieves the values of all analog
+    channels in a single socket command. The readback values of all channels are updated based
+    on the response, and subscriptions are run for all channels. Readings are cached as implemented
+    in the SocketSignal class, so that multiple reads of the same channel within an update cycle do
+    not result in multiple socket calls.
 
     Args:
-        signal_name (str): Name of the signal.
-        channel (int): Analog channel number (0-7).
-        parent (GalilRIO): Parent GalilRIO device.
+        signal_name (str): The name of the signal, e.g. "ch0", "ch1", ..., "ch7"
+        channel (int): The channel number corresponding to the signal, e.g. 0 for "ch0", 1 for "ch1", ...
+        parent (GalilRIO): The parent device instance that this signal belongs to.
     """
 
     _NUM_ANALOG_CHANNELS = 8
-    _READ_TIMEOUT = 0.1  # seconds
+
+    def __init__(self, signal_name: str, channel: int, parent: GalilRIO, **kwargs):
+        super().__init__(signal_name=signal_name, parent=parent, **kwargs)
+        self._channel = channel
+        self._metadata["connected"] = False
+        self._metadata["write_access"] = False
+
+    def _socket_set(self, val):
+        """Read-only signal, so set method raises an error."""
+        raise ReadOnlyError(f"Signal {self.name} is read-only.")
+
+    def _socket_get(self) -> float:
+        """Get command for the readback signal"""
+        cmd = "MG@" + ", @".join([f"AN[{ii}]" for ii in range(self._NUM_ANALOG_CHANNELS)])
+        ret = self.controller.socket_put_and_receive(cmd)
+        values = [float(val) for val in ret.strip().split(" ")]
+        # Run updates for all channels. This also updates the _readback and metadata timestamp
+        # value of this channel.
+        self._update_all_channels(values)
+        return self._readback
+
+    # pylint: disable=protected-access
+    def _update_all_channels(self, values: list[float]) -> None:
+        """
+        Method to receive a list of readback values for channels 0 to 7. Updates for each channel idx
+        are applied to the corresponding GalilRIOAnalogSignalRO signal with matching attr_name "ch{idx}".
+
+        We also update the _last_readback attribute of each of the signals, to avoid multiple socket calls,
+        but rather use the cached value of the combined reading for all channels.
+
+        Args:
+            values (list[float]): List of new readback values for all channels, where the
+                index corresponds to the channel number (0-7).
+        """
+        updates: dict[str, tuple[float, float]] = {}  # attr_name -> (new_val, old_val)
+        # Update all readbacks first
+        for walk in self.parent.walk_signals():
+            if isinstance(walk.item, GalilRIOAnalogSignalRO):
+                idx = int(walk.item.attr_name[-1])
+                if 0 <= idx < len(values):
+                    old_val = walk.item._readback
+                    new_val = values[idx]
+                    walk.item._metadata["timestamp"] = self._last_readback
+                    walk.item._last_readback = self._last_readback
+                    walk.item._readback = new_val
+                    if (
+                        idx != self._channel
+                    ):  # Only run subscriptions on other channels, not on itself
+                        # as this is handled by the SocketSignal and we want to avoid running multiple
+                        # subscriptions for the same channel update
+                        updates[walk.item.attr_name] = (new_val, old_val)
+                else:
+                    logger.warning(
+                        f"Received {len(values)} values but found channel index {idx} in signal {walk.item.name}. Skipping update for this signal."
+                    )
+
+        # Run subscriptions after all readbacks have been updated
+        # on all channels except the one that triggered the update
+        for walk in self.parent.walk_signals():
+            if walk.item.attr_name in updates:
+                new_val, old_val = updates[walk.item.attr_name]
+                walk.item._run_subs(
+                    sub_type=walk.item.SUB_VALUE,
+                    old_value=old_val,
+                    value=new_val,
+                    timestamp=self._last_readback,
+                )
+
+
+class GalilRIODigitalOutSignal(GalilSignalBase):
+    """Signal for controlling digital outputs of the Galil RIO controller."""
+
+    _NUM_DIGITAL_OUTPUT_CHANNELS = 16
 
     def __init__(self, signal_name: str, channel: int, parent: GalilRIO, **kwargs):
         super().__init__(signal_name, parent=parent, **kwargs)
@@ -87,81 +159,83 @@ class GalilRIOSignalRO(GalilSignalRO):
 
     def _socket_get(self) -> float:
         """Get command for the readback signal"""
-        cmd = "MG@" + ",@".join([f"AN[{ii}]" for ii in range(self._NUM_ANALOG_CHANNELS)])
+        cmd = f"MG@OUT[{self._channel}]"
         ret = self.controller.socket_put_and_receive(cmd)
-        values = [float(val) for val in ret.strip().split(" ")]
-        # This updates all channels' readbacks, including self._readback
-        self._update_all_channels(values)
+        self._readback = float(ret.strip())
         return self._readback
 
-    def get(self):
-        """Get current analog channel values from the Galil RIO controller."""
-        # If the last readback has happend more than _READ_TIMEOUT seconds ago, read all channels again
-        if time.monotonic() - self.parent.last_readback > self._READ_TIMEOUT:
-            self._readback = self._socket_get()
-        return self._readback
+    def _socket_set(self, val: Literal[0, 1]) -> None:
+        """Set command for the digital output signal. Value should be 0 or 1."""
 
-    # pylint: disable=protected-access
-    def _update_all_channels(self, values: list[float]) -> None:
-        """
-        Update all analog channel readbacks based on the provided list of values.
-        List of values must be in order from an_ch0 to an_ch7.
+        if val not in (0, 1):
+            raise ValueError("Digital output value must be 0 or 1.")
+        cmd = f"SB{self._channel}" if val == 1 else f"CB{self._channel}"
+        self.controller.socket_put_confirmed(cmd)
 
-        We first have to update the _last_readback timestamp of the GalilRIO parent device.
-        Then we update all readbacks of all an_ch channels, before we run any subscriptions.
-        This ensures that all readbacks are updated before any subscriptions are run, which
-        may themselves read other channels.
 
-        Args:
-            values (list[float]): List of 8 float values corresponding to the analog channels.
-                                  They must be in order from an_ch0 to an_ch7.
-        """
-        timestamp = time.time()
-        # Update parent's last readback before running subscriptions!!
-        self.parent._last_readback = time.monotonic()
-        updates: dict[str, tuple[float, float]] = {}  # attr_name -> (new_val, old_val)
-        # Update all readbacks first
-        for walk in self.parent.walk_signals():
-            if walk.item.attr_name.startswith("an_ch"):
-                idx = int(walk.item.attr_name[-1])
-                if 0 <= idx < len(values):
-                    old_val = walk.item._readback
-                    new_val = values[idx]
-                    walk.item._metadata["timestamp"] = timestamp
-                    walk.item._readback = new_val
-                    updates[walk.item.attr_name] = (new_val, old_val)
+def _create_analog_channels(num_channels: int) -> dict[str, tuple]:
+    """
+    Helper method to create a dictionary of analog channel definitions for the DynamicDeviceComponent.
 
-        # Run subscriptions after all readbacks have been updated
-        for walk in self.parent.walk_signals():
-            if walk.item.attr_name in updates:
-                new_val, old_val = updates[walk.item.attr_name]
-                walk.item._run_subs(
-                    sub_type=walk.item.SUB_VALUE,
-                    old_value=old_val,
-                    value=new_val,
-                    timestamp=timestamp,
-                )
+    Args:
+        num_channels (int): The number of analog channels to create.
+    """
+    an_channels = {}
+    for i in range(0, num_channels):
+        an_channels[f"ch{i}"] = (
+            GalilRIOAnalogSignalRO,
+            f"ch{i}",
+            {"kind": Kind.normal, "notify_bec": True, "channel": i, "doc": f"Analog channel {i}."},
+        )
+    return an_channels
+
+
+def _create_digital_output_channels(num_channels: int) -> dict[str, tuple]:
+    """
+    Helper method to create a dictionary of digital output channel definitions for the DynamicDeviceComponent.
+
+    Args:
+        num_channels (int): The number of digital output channels to create.
+    """
+    di_out_channels = {}
+    for i in range(0, num_channels):
+        di_out_channels[f"ch{i}"] = (
+            GalilRIODigitalOutSignal,
+            f"ch{i}",
+            {
+                "kind": Kind.config,
+                "notify_bec": True,
+                "channel": i,
+                "doc": f"Digital output channel {i}.",
+            },
+        )
+    return di_out_channels
 
 
 class GalilRIO(PSIDeviceBase):
     """
-    Galil RIO controller integration with 8 analog input channels. To implement the device,
-    please provide the appropriate host and port (default port is 23).
+    Galil RIO controller integration with 16 digital output channels and 8 analog input channels.
+    The default port for the controller is 23.
 
     Args:
         host (str): Hostname or IP address of the Galil RIO controller.
+        port (int, optional): Port number for the TCP/IP connection. Defaults to 23.
+        socket_cls (type[SocketIO], optional): Socket class to use for communication. Defaults to SocketIO.
+        scan_info (ScanInfo, optional): ScanInfo object for the device.
+        device_manager (DeviceManagerDS): The device manager instance that manages this device.
+        **kwargs: Additional keyword arguments passed to the PSIDeviceBase constructor.
     """
 
     SUB_CONNECTION_CHANGE = "connection_change"
 
-    an_ch0 = Cpt(GalilRIOSignalRO, signal_name="an_ch0", channel=0, doc="Analog input channel 0")
-    an_ch1 = Cpt(GalilRIOSignalRO, signal_name="an_ch1", channel=1, doc="Analog input channel 1")
-    an_ch2 = Cpt(GalilRIOSignalRO, signal_name="an_ch2", channel=2, doc="Analog input channel 2")
-    an_ch3 = Cpt(GalilRIOSignalRO, signal_name="an_ch3", channel=3, doc="Analog input channel 3")
-    an_ch4 = Cpt(GalilRIOSignalRO, signal_name="an_ch4", channel=4, doc="Analog input channel 4")
-    an_ch5 = Cpt(GalilRIOSignalRO, signal_name="an_ch5", channel=5, doc="Analog input channel 5")
-    an_ch6 = Cpt(GalilRIOSignalRO, signal_name="an_ch6", channel=6, doc="Analog input channel 6")
-    an_ch7 = Cpt(GalilRIOSignalRO, signal_name="an_ch7", channel=7, doc="Analog input channel 7")
+    #############################
+    ### Analog input channels ###
+    #############################
+
+    analog_in = DDC(_create_analog_channels(GalilRIOAnalogSignalRO._NUM_ANALOG_CHANNELS))
+    digital_out = DDC(
+        _create_digital_output_channels(GalilRIODigitalOutSignal._NUM_DIGITAL_OUTPUT_CHANNELS)
+    )
 
     def __init__(
         self,
@@ -177,19 +251,18 @@ class GalilRIO(PSIDeviceBase):
         if port is None:
             port = 23  # Default port for Galil RIO controller
         self.controller = GalilRIOController(
-            socket_cls=socket_cls, socket_host=host, socket_port=port, device_manager=device_manager
+            name=f"GalilRIOController_{name}",
+            socket_cls=socket_cls,
+            socket_host=host,
+            socket_port=port,
+            device_manager=device_manager,
         )
-        self._last_readback: float = time.monotonic()
+        self._readback_metadata: dict[str, float] = {"last_readback": 0.0}
         super().__init__(name=name, device_manager=device_manager, scan_info=scan_info, **kwargs)
         self.controller.subscribe(
             self._update_connection_state, event_type=self.SUB_CONNECTION_CHANGE
         )
 
-    @property
-    def last_readback(self) -> float:
-        """Return the time of the last readback from the controller."""
-        return self._last_readback
-
     # pylint: disable=arguments-differ
     def wait_for_connection(self, timeout: float = 30.0) -> None:
         """Wait for the RIO controller to be connected within timeout period."""
@@ -207,7 +280,7 @@ class GalilRIO(PSIDeviceBase):
 
 
 if __name__ == "__main__":
-    HOST_NAME = "129.129.98.64"
+    HOST_NAME = "129.129.122.14"
     from bec_server.device_server.tests.utils import DMMock
 
     dm = DMMock()

csaxs_bec/devices/panda_box/panda_box.py

@@ -0,0 +1,101 @@
+"""Module to integrate the PandaBox for cSAXS measurements."""
+
+import time
+
+from bec_lib.logger import bec_logger
+from ophyd_devices import AsyncMultiSignal, StatusBase
+from ophyd_devices.devices.panda_box.panda_box import PandaBox, PandaState
+from pandablocks.responses import FrameData
+
+logger = bec_logger.logger
+
+
+class PandaBoxCSAXS(PandaBox):
+
+    def on_init(self):
+        super().on_init()
+        self._acquisition_group = "burst"
+        self._timeout_on_completed = 10
+
+    def on_stage(self):
+        # TODO, adjust as seen fit.
+        # Adjust the acquisition group based on scan parameters if needed
+        if self.scan_info.msg.scan_type == "fly":
+            self._acquisition_group = "fly"
+        elif self.scan_info.msg.scan_type == "step":
+            if self.scan_info.msg.scan_parameters["frames_per_trigger"] == 1:
+                self._acquisition_group = "monitored"
+            else:
+                self._acquisition_group = "burst"
+
+    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."""
+
+        def _check_capture_complete():
+            captured = 0
+            start_time = time.monotonic()
+            try:
+                expected_points = int(self.scan_info.msg.num_points * self.scan_info.msg.scan_parameters.get("frames_per_trigger",1))
+                while captured < expected_points:
+                    logger.info(
+                        f"Run with captured {captured} and expected points : {expected_points}."
+                    )
+                    ret = self.send_raw("*PCAP.CAPTURED?")
+                    captured = int(ret[0].split("=")[-1])
+                    time.sleep(0.01)
+                    if (time.monotonic() - start_time) > self._timeout_on_completed:
+                        raise TimeoutError(f"Pandabox {self.name} did not complete after {self._timeout_on_completed} with points captured {captured}/{expected_points}")
+            finally:
+                self._disarm()
+
+        _check_capture_complete()
+
+        # NOTE: This utility class allows to submit a blocking function to a thread and return a status object
+        # that can be awaited for. This allows for asynchronous waiting for the capture to complete without blocking
+        # the main duty cycle of the device server. The device server knows how to handle the status object (future)
+        # and will wait for it to complete.
+        # status = self.task_handler.submit_task(_check_capture_complete, run=True)
+
+        # status_panda_state = StatusBase(obj=self)
+        # self.add_status_callback(
+        #     status, success=[PandaState.END, PandaState.DISARMED], failure=[PandaState.READY]
+        # )
+        # ret_status = status & status_panda_state
+        # self.cancel_on_stop(ret_status)
+        # return ret_status
+
+
+if __name__ == "__main__":
+    import time
+
+    panda = PandaBoxCSAXS(
+        name="omny_panda",
+        host="omny-panda.psi.ch",
+        signal_alias={
+            "FMC_IN.VAL2.Value": "alias",
+            "FMC_IN.VAL1.Min": "alias2",
+            "FMC_IN.VAL1.Max": "alias3",
+            "FMC_IN.VAL1.Mean": "alias4",
+        },
+    )
+    panda.on_connected()
+    status = StatusBase(obj=panda)
+    panda.add_status_callback(
+        status=status, success=[PandaState.DISARMED], failure=[PandaState.READY]
+    )
+    panda.stop()
+    status.wait(timeout=2)
+    panda.unstage()
+    logger.info(f"Panda connected")
+    ret = panda.stage()
+    logger.info(f"Panda staged")
+    ret = panda.pre_scan()
+    ret.wait(timeout=5)
+    logger.info(f"Panda pre scan done")
+    time.sleep(5)
+    panda.stop()
+    st = panda.complete()
+    st.wait(timeout=5)
+    logger.info(f"Measurement completed")
+    panda.unstage()
+    logger.info(f"Panda Unstaged")

csaxs_bec/devices/panda_box/panda_box_omny.py

@@ -0,0 +1,102 @@
+"""Module to integrate the PandaBox for cSAXS measurements."""
+
+import time
+
+from bec_lib.logger import bec_logger
+from ophyd_devices import AsyncMultiSignal, StatusBase
+from ophyd_devices.devices.panda_box.panda_box import PandaBox, PandaState
+from pandablocks.responses import FrameData
+
+logger = bec_logger.logger
+
+
+class PandaBoxOMNY(PandaBox):
+
+    def on_init(self):
+        super().on_init()
+        self._acquisition_group = "burst"
+        self._timeout_on_completed = 10
+
+    def on_stage(self):
+        
+        # TODO, adjust as seen fit.
+        # Adjust the acquisition group based on scan parameters if needed
+        if self.scan_info.msg.scan_type == "fly":
+            self._acquisition_group = "fly"
+        elif self.scan_info.msg.scan_type == "step":
+            if self.scan_info.msg.scan_parameters["frames_per_trigger"] == 1:
+                self._acquisition_group = "monitored"
+            else:
+                self._acquisition_group = "burst"
+
+    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."""
+
+        def _check_capture_complete():
+            captured = 0
+            start_time = time.monotonic()
+            try:
+                expected_points = int(self.scan_info.msg.num_points * self.scan_info.msg.scan_parameters.get("frames_per_trigger",1))
+                while captured < expected_points:
+                    logger.info(
+                        f"Run with captured {captured} and expected points : {expected_points}."
+                    )
+                    ret = self.send_raw("*PCAP.CAPTURED?")
+                    captured = int(ret[0].split("=")[-1])
+                    time.sleep(0.01)
+                    if (time.monotonic() - start_time) > self._timeout_on_completed:
+                        raise TimeoutError(f"Pandabox {self.name} did not complete after {self._timeout_on_completed} with points captured {captured}/{expected_points}")
+            finally:
+                self._disarm()
+
+        _check_capture_complete()
+
+        # NOTE: This utility class allows to submit a blocking function to a thread and return a status object
+        # that can be awaited for. This allows for asynchronous waiting for the capture to complete without blocking
+        # the main duty cycle of the device server. The device server knows how to handle the status object (future)
+        # and will wait for it to complete.
+        # status = self.task_handler.submit_task(_check_capture_complete, run=True)
+
+        # status_panda_state = StatusBase(obj=self)
+        # self.add_status_callback(
+        #     status, success=[PandaState.END, PandaState.DISARMED], failure=[PandaState.READY]
+        # )
+        # ret_status = status & status_panda_state
+        # self.cancel_on_stop(ret_status)
+        # return ret_status
+
+
+if __name__ == "__main__":
+    import time
+
+    panda = PandaBoxCSAXS(
+        name="omny_panda",
+        host="omny-panda.psi.ch",
+        signal_alias={
+            "FMC_IN.VAL2.Value": "alias",
+            "FMC_IN.VAL1.Min": "alias2",
+            "FMC_IN.VAL1.Max": "alias3",
+            "FMC_IN.VAL1.Mean": "alias4",
+        },
+    )
+    panda.on_connected()
+    status = StatusBase(obj=panda)
+    panda.add_status_callback(
+        status=status, success=[PandaState.DISARMED], failure=[PandaState.READY]
+    )
+    panda.stop()
+    status.wait(timeout=2)
+    panda.unstage()
+    logger.info(f"Panda connected")
+    ret = panda.stage()
+    logger.info(f"Panda staged")
+    ret = panda.pre_scan()
+    ret.wait(timeout=5)
+    logger.info(f"Panda pre scan done")
+    time.sleep(5)
+    panda.stop()
+    st = panda.complete()
+    st.wait(timeout=5)
+    logger.info(f"Measurement completed")
+    panda.unstage()
+    logger.info(f"Panda Unstaged")

csaxs_bec/scans/LamNIFermatScan.py

@@ -210,7 +210,7 @@ class LamNIFermatScan(ScanBase, LamNIMixin):
     arg_input = {}
     arg_bundle_size = {"bundle": len(arg_input), "min": None, "max": None}
 
-    def __init__(self, *args, parameter: dict = None, **kwargs):
+    def __init__(self, *args, parameter: dict = None, frames_per_trigger:int=1, exp_time:float=0,**kwargs):
         """
         A LamNI scan following Fermat's spiral.
 
@@ -230,10 +230,10 @@ class LamNIFermatScan(ScanBase, LamNIMixin):
 
         Examples:
             >>> scans.lamni_fermat_scan(fov_size=[20], step=0.5, exp_time=0.1)
-            >>> scans.lamni_fermat_scan(fov_size=[20, 25], center_x=0.02, center_y=0, shift_x=0, shift_y=0, angle=0, step=0.5, fov_circular=0, exp_time=0.1)
+            >>> scans.lamni_fermat_scan(fov_size=[20, 25], center_x=0.02, center_y=0, shift_x=0, shift_y=0, angle=0, step=0.5, fov_circular=0, exp_time=0.1, frames_per_trigger=1)
         """
 
-        super().__init__(parameter=parameter, **kwargs)
+        super().__init__(parameter=parameter, frames_per_trigger=frames_per_trigger, exp_time=exp_time,**kwargs)
         self.axis = []
         scan_kwargs = parameter.get("kwargs", {})
         self.fov_size = scan_kwargs.get("fov_size")
@@ -482,6 +482,7 @@ class LamNIFermatScan(ScanBase, LamNIMixin):
         yield from self.open_scan()
         yield from self.stage()
         yield from self.run_baseline_reading()
+        yield from self.pre_scan()
         yield from self.scan_core()
         yield from self.finalize()
         yield from self.unstage()

csaxs_bec/scans/flomni_fermat_scan.py

@@ -52,6 +52,7 @@ class FlomniFermatScan(SyncFlyScanBase):
         angle: float = None,
         corridor_size: float = 3,
         parameter: dict = None,
+        frames_per_trigger:int=1,
         **kwargs,
     ):
         """
@@ -62,7 +63,8 @@ class FlomniFermatScan(SyncFlyScanBase):
             fovy(float) [um]: Fov in the piezo plane (i.e. piezo range). Max 100 um
             cenx(float) [um]: center position in x.
             ceny(float) [um]: center position in y.
-            exp_time(float) [s]: exposure time
+            exp_time(float) [s]: exposure time per burst frame
+            frames_per_trigger(int) : Number of burst frames per point
             step(float) [um]: stepsize
             zshift(float) [um]: shift in z
             angle(float) [deg]: rotation angle (will rotate first)
@@ -71,10 +73,10 @@ class FlomniFermatScan(SyncFlyScanBase):
         Returns:
 
         Examples:
-            >>> scans.flomni_fermat_scan(fovx=20, fovy=25, cenx=0.02, ceny=0, zshift=0, angle=0, step=0.5, exp_time=0.01)
+            >>> 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)
         """
 
-        super().__init__(parameter=parameter, exp_time=exp_time, **kwargs)
+        super().__init__(parameter=parameter, exp_time=exp_time, frames_per_trigger=frames_per_trigger, **kwargs)
         self.show_live_table = False
         self.axis = []
         self.fovx = fovx
@@ -323,6 +325,7 @@ class FlomniFermatScan(SyncFlyScanBase):
         yield from self.stage()
         yield from self.run_baseline_reading()
         yield from self._prepare_setup_part2()
+        yield from self.pre_scan()
         yield from self.scan_core()
         yield from self.finalize()
         yield from self.unstage()

csaxs_bec/scans/omny_fermat_scan.py

@@ -51,6 +51,7 @@ class OMNYFermatScan(SyncFlyScanBase):
         angle: float = None,
         corridor_size: float = 3,
         parameter: dict = None,
+        frames_per_trigger:int=1,
         **kwargs,
     ):
         """
@@ -62,6 +63,7 @@ class OMNYFermatScan(SyncFlyScanBase):
             cenx(float) [um]: center position in x.
             ceny(float) [um]: center position in y.
             exp_time(float) [s]: exposure time
+            frames_per_trigger:int: Number of burst frames per trigger, defaults to 1.
             step(float) [um]: stepsize
             zshift(float) [um]: shift in z
             angle(float) [deg]: rotation angle (will rotate first)
@@ -73,7 +75,7 @@ class OMNYFermatScan(SyncFlyScanBase):
             >>> scans.omny_fermat_scan(fovx=20, fovy=25, cenx=10, ceny=0, zshift=0, angle=0, step=2, exp_time=0.01)
         """
 
-        super().__init__(parameter=parameter, **kwargs)
+        super().__init__(parameter=parameter, exp_time=exp_time, frames_per_trigger=frames_per_trigger, **kwargs)
         self.axis = []
         self.fovx = fovx
         self.fovy = fovy
@@ -299,6 +301,7 @@ class OMNYFermatScan(SyncFlyScanBase):
         yield from self.stage()
         yield from self.run_baseline_reading()
         yield from self._prepare_setup_part2()
+        yield from self.pre_scan()
         yield from self.scan_core()
         yield from self.finalize()
         yield from self.unstage()

docs/user/ptychography/flomni.md

@@ -193,14 +193,15 @@ The basic scan function can be called by `scans.flomni_fermat_scan()` and offers
 | fovy (float) | Fov in the piezo plane (i.e. piezo range). Max 100 um |
 | cenx (float) | center position in x |
 | ceny (float) | center position in y |
-| exp_time (float) | exposure time |
+| exp_time (float) | exposure time per frame |
+| frames_per_trigger(int) | Number of burst frames per position |
 | step (float) | stepsize |
 | zshift (float) | shift in z |
 | angle (float) | rotation angle (will rotate first) |
 | corridor_size (float) | corridor size for the corridor optimization. Default 3 um |
 
 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)`
+`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)`
 
 #### Overview of the alignment steps
 

docs/user/ptychography/omny.md

@@ -327,14 +327,15 @@ The basic scan function can be called by `scans.omny_fermat_scan()` and offers a
 | fovy (float) | Fov in the piezo plane (i.e. piezo range). Max 100 um |
 | cenx (float) | center position in x |
 | ceny (float) | center position in y |
-| exp_time (float) | exposure time |
+| exp_time (float) | exposure time per frame |
+| frames_per_trigger(int) | Number of burst frames per position |
 | step (float) | stepsize |
 | zshift (float) | shift in z |
 | angle (float) | rotation angle (will rotate first) |
 | corridor_size (float) | corridor size for the corridor optimization. Default 3 um |
 
 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)`
+`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)`
 
 #### Overview of the alignment steps
 

tests/conftest.py

@@ -1,14 +0,0 @@
-"""
-Conftest runs for all tests in this directory and subdirectories. Thereby, we know for
-certain that the SocketSignal.READBACK_TIMEOUT is set to 0 for all tests, which prevents
-hanging tests when a readback is attempted on a non-connected socket.
-"""
-
-# conftest.py
-import pytest
-from ophyd_devices.utils.socket import SocketSignal
-
-
-@pytest.fixture(autouse=True)
-def patch_socket_timeout(monkeypatch):
-    monkeypatch.setattr(SocketSignal, "READBACK_TIMEOUT", 0.0)

tests/tests_devices/test_fupr_ophyd.py

@@ -2,6 +2,7 @@ from unittest import mock
 
 import pytest
 from ophyd_devices.tests.utils import SocketMock
+from ophyd_devices.utils.socket import SocketSignal
 
 from csaxs_bec.devices.omny.galil.fupr_ophyd import FuprGalilController, FuprGalilMotor
 
@@ -17,6 +18,11 @@ def fsamroy(dm_with_devices):
         socket_cls=SocketMock,
         device_manager=dm_with_devices,
     )
+    for walk in fsamroy_motor.walk_signals():
+        if isinstance(walk.item, SocketSignal):
+            walk.item._readback_timeout = (
+                0.0  # Set the readback timeout to 0 to avoid waiting during tests
+            )
     fsamroy_motor.controller.on()
     assert isinstance(fsamroy_motor.controller, FuprGalilController)
     yield fsamroy_motor

tests/tests_devices/test_galil.py

@@ -9,7 +9,11 @@ from ophyd_devices.tests.utils import SocketMock
 from csaxs_bec.devices.npoint.npoint import NPointAxis, NPointController
 from csaxs_bec.devices.omny.galil.fgalil_ophyd import FlomniGalilController, FlomniGalilMotor
 from csaxs_bec.devices.omny.galil.fupr_ophyd import FuprGalilController, FuprGalilMotor
-from csaxs_bec.devices.omny.galil.galil_rio import GalilRIO, GalilRIOController, GalilRIOSignalRO
+from csaxs_bec.devices.omny.galil.galil_rio import (
+    GalilRIO,
+    GalilRIOAnalogSignalRO,
+    GalilRIOController,
+)
 from csaxs_bec.devices.omny.galil.lgalil_ophyd import LamniGalilController, LamniGalilMotor
 from csaxs_bec.devices.omny.galil.ogalil_ophyd import OMNYGalilController, OMNYGalilMotor
 from csaxs_bec.devices.omny.galil.sgalil_ophyd import GalilController, SGalilMotor
@@ -272,26 +276,27 @@ def test_galil_rio_signal_read(galil_rio):
     ## Test read of all channels
     ###########
 
-    assert galil_rio.an_ch0._READ_TIMEOUT == 0.1  # Default read timeout of 100ms
+    assert galil_rio.analog_in.ch0._readback_timeout == 0.1  # Default read timeout of 100ms
     # Mock the socket to return specific values
-    galil_rio.controller.sock.buffer_recv = [b" 1.234 2.345 3.456 4.567 5.678 6.789 7.890 8.901"]
-    galil_rio._last_readback = 0  # Force read from controller
-
+    analog_bufffer = b" 1.234 2.345 3.456 4.567 5.678 6.789 7.890 8.901\r\n"
+    galil_rio.controller.sock.buffer_recv = []  # Clear any existing buffer
+    galil_rio.controller.sock.buffer_recv.append(analog_bufffer)
     read_values = galil_rio.read()
     assert len(read_values) == 8  # 8 channels
+
     expected_values = {
-        galil_rio.an_ch0.name: {"value": 1.234},
-        galil_rio.an_ch1.name: {"value": 2.345},
-        galil_rio.an_ch2.name: {"value": 3.456},
-        galil_rio.an_ch3.name: {"value": 4.567},
-        galil_rio.an_ch4.name: {"value": 5.678},
-        galil_rio.an_ch5.name: {"value": 6.789},
-        galil_rio.an_ch6.name: {"value": 7.890},
-        galil_rio.an_ch7.name: {"value": 8.901},
+        galil_rio.analog_in.ch0.name: {"value": 1.234},
+        galil_rio.analog_in.ch1.name: {"value": 2.345},
+        galil_rio.analog_in.ch2.name: {"value": 3.456},
+        galil_rio.analog_in.ch3.name: {"value": 4.567},
+        galil_rio.analog_in.ch4.name: {"value": 5.678},
+        galil_rio.analog_in.ch5.name: {"value": 6.789},
+        galil_rio.analog_in.ch6.name: {"value": 7.890},
+        galil_rio.analog_in.ch7.name: {"value": 8.901},
     }
     # All timestamps should be the same
     assert all(
-        ret["timestamp"] == read_values[galil_rio.an_ch0.name]["timestamp"]
+        ret["timestamp"] == read_values[galil_rio.analog_in.ch0.name]["timestamp"]
         for signal_name, ret in read_values.items()
     )
     # Check values
@@ -301,7 +306,7 @@ def test_galil_rio_signal_read(galil_rio):
 
     # Check communication command to socker
     assert galil_rio.controller.sock.buffer_put == [
-        b"MG@AN[0],@AN[1],@AN[2],@AN[3],@AN[4],@AN[5],@AN[6],@AN[7]\r"
+        b"MG@AN[0], @AN[1], @AN[2], @AN[3], @AN[4], @AN[5], @AN[6], @AN[7]\r"
     ]
 
     ###########
@@ -313,11 +318,11 @@ def test_galil_rio_signal_read(galil_rio):
 
     def value_callback(value, old_value, **kwargs):
         obj = kwargs.get("obj")
-        galil = obj.parent
+        galil = obj.parent.parent
         readback = galil.read()
         value_callback_buffer.append(readback)
 
-    galil_rio.an_ch0.subscribe(value_callback, run=False)
+    galil_rio.analog_in.ch0.subscribe(value_callback, run=False)
     galil_rio.controller.sock.buffer_recv = [b" 2.5 2.6 2.7 2.8 2.9 3.0 3.1 3.2"]
     expected_values = [2.5, 2.6, 2.7, 2.8, 2.9, 3.0, 3.1, 3.2]
 
@@ -327,13 +332,15 @@ def test_galil_rio_signal_read(galil_rio):
 
     # Should have used the cached value
     for walk in galil_rio.walk_signals():
-        walk.item._READ_TIMEOUT = 10  # Make sure cached read is used
-    ret = galil_rio.an_ch0.read()
+        walk.item._readback_timeout = 10  # Make sure cached read is used
+    ret = galil_rio.analog_in.ch0.read()
 
     # Should not trigger callback since value did not change
-    assert np.isclose(ret[galil_rio.an_ch0.name]["value"], 1.234)
+    assert np.isclose(ret[galil_rio.analog_in.ch0.name]["value"], 1.234)
     # Same timestamp as for another channel as this is cached read
-    assert np.isclose(ret[galil_rio.an_ch0.name]["timestamp"], galil_rio.an_ch7.timestamp)
+    assert np.isclose(
+        ret[galil_rio.analog_in.ch0.name]["timestamp"], galil_rio.analog_in.ch7.timestamp
+    )
     assert len(value_callback_buffer) == 0
 
     ##################
@@ -341,10 +348,10 @@ def test_galil_rio_signal_read(galil_rio):
     ##################
 
     # Now force a read from the controller
-    galil_rio._last_readback = 0  # Force read from controller
-    ret = galil_rio.an_ch0.read()
+    galil_rio.analog_in.ch0._last_readback = 0  # Force read from controller
+    ret = galil_rio.analog_in.ch0.read()
 
-    assert np.isclose(ret[galil_rio.an_ch0.name]["value"], 2.5)
+    assert np.isclose(ret[galil_rio.analog_in.ch0.name]["value"], 2.5)
 
     # Check callback invocation, but only 1 callback even with galil_rio.read() call in callback
     assert len(value_callback_buffer) == 1
@@ -352,7 +359,45 @@ def test_galil_rio_signal_read(galil_rio):
     assert np.isclose(values, expected_values).all()
     assert all(
         [
-            value["timestamp"] == value_callback_buffer[0][galil_rio.an_ch0.name]["timestamp"]
+            value["timestamp"]
+            == value_callback_buffer[0][galil_rio.analog_in.ch0.name]["timestamp"]
             for value in value_callback_buffer[0].values()
         ]
     )
+
+
+def test_galil_rio_digital_out_signal(galil_rio):
+    """
+    Test that the Galil RIO digital output signal can be set correctly.
+    """
+    ## Test Read from digital output channels
+    buffer_receive = []
+    excepted_put_buffer = []
+    for ii in range(galil_rio.digital_out.ch0._NUM_DIGITAL_OUTPUT_CHANNELS):
+        cmd = f"MG@OUT[{ii}]\r".encode()
+        excepted_put_buffer.append(cmd)
+        recv = " 1.000".encode()
+        buffer_receive.append(recv)
+
+    galil_rio.controller.sock.buffer_recv = buffer_receive  # Mock response for readback
+
+    digital_read = galil_rio.read_configuration()  # Read to populate readback values
+
+    for walk in galil_rio.digital_out.walk_signals():
+        assert np.isclose(digital_read[walk.item.name]["value"], 1.0)
+
+    assert galil_rio.controller.sock.buffer_put == excepted_put_buffer
+
+    # Test writing to digital output channels
+    galil_rio.controller.sock.buffer_put = []  # Clear buffer put
+    galil_rio.controller.sock.buffer_recv = [b":"]  # Mock response for readback
+
+    # Set digital output channel 0 to high
+    galil_rio.digital_out.ch0.put(1)
+    assert galil_rio.controller.sock.buffer_put == [b"SB0\r"]
+
+    # Set digital output channel 0 to low
+    galil_rio.controller.sock.buffer_put = []  # Clear buffer put
+    galil_rio.controller.sock.buffer_recv = [b":"]  # Mock response for readback
+    galil_rio.digital_out.ch0.put(0)
+    assert galil_rio.controller.sock.buffer_put == [b"CB0\r"]