feat(lamni): Keithley 2400 current source and queue actions

Add a Keithley 2400 device, a positioner in mA over the TCP-serial bridge,
with voltage readback and output_on/output_off/status user commands. The
output and level are left alone on connect, stop and restart.

Add the LamNI tomo queue actions set_keithley_current and sleep, available
in the command job dialog, and a config with only the Keithley plus one
combining it with the simulated LamNI.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01M8zoDzTXoPAqgdFjHUFFF8

csaxs_bec/bec_ipython_client/plugins/LamNI/lamni.py

@@ -207,6 +207,7 @@ class LamNI(
     # overwrite). loptz is deliberately NOT in _TOMO_QUEUE_MOVE_DEVICES, so
     # this is the only way to queue a loptz move and the new shift can't be
     # forgotten. See TomoQueueMixin._queue_action_move() for "move".
+    # "set_keithley_current" is not a "move" because it carries its own sleep.
     _TOMO_QUEUE_ACTIONS = {
         "move": {
             "func": "_queue_action_move",
@@ -255,8 +256,76 @@ class LamNI(
                 },
             },
         },
+        "set_keithley_current": {
+            "func": "_queue_action_set_keithley_current",
+            "idempotent": True,  # absolute level, re-runs harmlessly
+            "help": (
+                "Set the Keithley source current, switching its output on if it is "
+                "off, then sleep."
+            ),
+            "params": {
+                "current_mA": {
+                    "type": "float",
+                    "default": 0.0,
+                    "min": 0.0,
+                    "max": 3.0,
+                    "unit": "mA",
+                    "label": "Current",
+                    "device_limits_from": "keithley",
+                },
+                "sleep_s": {
+                    "type": "float",
+                    "default": 0.0,
+                    "min": 0.0,
+                    "max": 86400.0,
+                    "unit": "s",
+                    "label": "Sleep afterwards",
+                },
+            },
+        },
+        "sleep": {
+            "func": "_queue_action_sleep",
+            "idempotent": True,  # re-running only waits again
+            "help": "Wait for a number of seconds.",
+            "params": {
+                "seconds": {
+                    "type": "float",
+                    "default": 60.0,
+                    "min": 0.0,
+                    "max": 86400.0,
+                    "unit": "s",
+                    "label": "Sleep",
+                },
+            },
+        },
     }
 
+    def _queue_action_sleep(self, seconds: float) -> None:
+        """Wait, the queue stays on this job."""
+        if seconds < 0:
+            raise ValueError(f"Queue sleep: seconds={seconds} is negative.")
+        print(f"Queue: sleeping {seconds} s.")
+        time.sleep(seconds)
+
+    def _queue_action_set_keithley_current(self, current_mA: float, sleep_s: float = 0.0) -> None:
+        """Set the Keithley current and sleep.
+
+        The move fails if the current isn't reached (e.g. voltage compliance),
+        which pauses the queue before the next job. Re-checks the live limits
+        at execution time.
+        """
+        if "keithley" not in dev:
+            raise ValueError("Queue set_keithley_current: device 'keithley' is not loaded.")
+        lo, hi = dev.keithley.limits
+        if not (lo <= current_mA <= hi):
+            raise ValueError(
+                f"Queue set_keithley_current: current_mA={current_mA} is outside "
+                f"keithley's current limits ({lo}, {hi})."
+            )
+        umv(dev.keithley, current_mA)
+        if sleep_s > 0:
+            self._queue_action_sleep(sleep_s)
+
     def _queue_action_move_loptz_with_shift(
         self, loptz_target: float, loptz_shift_x: float, loptz_shift_y: float
     ) -> None:

csaxs_bec/device_configs/keithley_2400.yaml

@@ -0,0 +1,25 @@
+# Keithley 2400 as a current source, RS-232 through the TCP-serial bridge on mpc2844
+# (service flomni-serial-keithley, temporary).
+# compliance_voltage: the output never drives more than this, also with the load disconnected.
+# remote_sense: true needs the sense leads connected at the load.
+keithley:
+  description: Keithley 2400 current source
+  deviceClass: csaxs_bec.devices.keithley.keithley_2400.Keithley2400
+  deviceConfig:
+    host: mpc2844.psi.ch
+    port: 3336
+    limits:
+    - 0
+    - 3
+    tolerance: 0.01
+    compliance_voltage: 5
+    source_range_ma: 10
+    remote_sense: false
+  enabled: true
+  onFailure: buffer
+  readOnly: false
+  readoutPriority: baseline
+  softwareTrigger: false
+  connectionTimeout: 20
+  deviceTags:
+  - keithley

csaxs_bec/device_configs/simulated_omny/simulated_lamni_keithley.yaml

@@ -0,0 +1,6 @@
+# Simulated LamNI endstation with the real Keithley 2400.
+lamni:
+  - !include ./simulated_lamni.yaml
+
+keithley:
+  - !include ../keithley_2400.yaml

csaxs_bec/devices/keithley/keithley_2400.py

@@ -0,0 +1,423 @@
+"""
+Keithley 2400 SourceMeter used as a current source, RS-232 reached through a TCP-serial bridge.
+
+The device is a positioner in mA, so ``umv(dev.keithley, 1.5)`` sources 1.5 mA. Voltage and
+current are measured concurrently while the output is on.
+
+Output handling, deliberate because a load hangs on this instrument:
+- ``:MEAS?`` switches the output on by itself and is therefore never sent. Readbacks use ``:READ?``
+  and only while the output is on (it fails with the output off), reporting 0 otherwise.
+- ``:SOUR:CLE:AUTO`` is kept off, otherwise ``:READ?`` would toggle the output.
+- Connecting, disconnecting, stopping or a BEC restart leave the output and the level untouched. If
+  the output is on at connect, the source settings are not changed at all.
+- A move programs the level first and only then switches the output on, so a stale level is
+  never sourced. The output is only switched off by ``output_off()``.
+
+Commands end in CR. Replies end in CR LF, set by the RS-232 terminator on the front panel.
+"""
+
+from __future__ import annotations
+
+import threading
+import time
+
+import numpy as np
+from bec_lib.logger import bec_logger
+from ophyd import Component as Cpt
+from ophyd import Device, PositionerBase, Signal, SignalRO
+from ophyd.status import MoveStatus
+from ophyd.utils import LimitError, ReadOnlyError
+from ophyd.utils.errors import InvalidState
+from ophyd_devices.utils.controller import Controller, ControllerCommunicationError, threadlocked
+from ophyd_devices.utils.socket import SocketIO, SocketSignal, raise_if_disconnected
+from prettytable import PrettyTable
+
+logger = bec_logger.logger
+
+
+class Keithley2400Error(Exception):
+    """Instrument in an unexpected state, or the requested current could not be reached."""
+
+
+class Keithley2400Controller(Controller):
+    """SCPI communication with the Keithley 2400."""
+
+    _axes_per_controller = 1
+    READBACK_MAX_AGE = 0.1  # s, the voltage and current signals share one :READ?
+
+    def _initialize(self):
+        super()._initialize()
+        self._reading = (0.0, 0.0)
+        self._reading_time = 0.0
+
+    @threadlocked
+    def socket_put(self, val: str) -> None:
+        self.command_history.append(f"[PUT]: {val}")
+        self.sock.put(f"{val}\r".encode())
+
+    @threadlocked
+    def socket_get(self) -> str:
+        """Read one reply. At 9600 baud a reply can arrive in several TCP segments."""
+        buffer = b""
+        while not buffer.endswith(b"\n"):
+            chunk = self.sock.receive()
+            if not chunk:
+                raise ConnectionError("Connection closed while waiting for a reply.")
+            buffer += chunk
+        response = buffer.decode().strip()
+        self.command_history.append(f"[GET]: {response}")
+        return response
+
+    @threadlocked
+    def socket_put_and_receive(self, val: str, remove_trailing_chars: bool = True) -> str:
+        """Query, not retried: a late reply to a repeated query would shift all following replies."""
+        try:
+            self.socket_put(val)
+            return self.socket_get()
+        except Exception as exc:
+            logger.error(f"Keithley query {val} failed: {exc}. {list(self.command_history)}")
+            raise ControllerCommunicationError(f"Keithley query {val} failed: {exc}") from exc
+
+    @threadlocked
+    def configure(
+        self, compliance_voltage: float, source_range_ma: float, remote_sense: bool
+    ) -> None:
+        """Set up the current source. With the output on only harmless settings are sent."""
+        idn = self.socket_put_and_receive("*IDN?")
+        if "MODEL 2400" not in idn:
+            raise Keithley2400Error(f"Unexpected instrument identification: {idn}")
+        self.socket_put(":SOUR:CLE:AUTO OFF")
+        self.socket_put(":FORM:ELEM VOLT,CURR")
+
+        if self.output_enabled():
+            source_function = self.socket_put_and_receive(":SOUR:FUNC?")
+            if source_function != "CURR":
+                raise Keithley2400Error(
+                    f"Output is on with source function {source_function}, not CURR. "
+                    "Check the instrument and switch the output off."
+                )
+            actual = self.get_compliance_voltage()
+            if not np.isclose(actual, compliance_voltage):
+                logger.warning(
+                    f"Keithley output is on, compliance stays at {actual} V "
+                    f"(configured: {compliance_voltage} V)."
+                )
+        else:
+            for command in (
+                ":SOUR:FUNC CURR",
+                ":SOUR:CURR:MODE FIX",
+                ":SOUR:CURR:RANG:AUTO OFF",
+                f":SOUR:CURR:RANG {source_range_ma * 1e-3:.6E}",
+                ":SENS:FUNC:CONC ON",
+                ':SENS:FUNC:ON "VOLT","CURR"',
+                f":SYST:RSEN {'ON' if remote_sense else 'OFF'}",
+                ":ARM:COUN 1",
+                ":TRIG:COUN 1",
+            ):
+                self.socket_put(command)
+            self.set_compliance_voltage(compliance_voltage)
+            actual = self.get_compliance_voltage()
+            if not np.isclose(actual, compliance_voltage):
+                raise Keithley2400Error(
+                    f"Compliance is {actual} V, expected {compliance_voltage} V."
+                )
+
+        error = self.socket_put_and_receive(":SYST:ERR?")
+        if not error.startswith("0,"):
+            raise Keithley2400Error(f"Instrument error after setup: {error}")
+
+    def output_enabled(self) -> bool:
+        return self.socket_put_and_receive(":OUTP?") == "1"
+
+    def set_output(self, enabled: bool) -> None:
+        self.socket_put(f":OUTP {'ON' if enabled else 'OFF'}")
+        self._reading_time = 0.0
+
+    def get_level(self) -> float:
+        """Programmed source level in A."""
+        return float(self.socket_put_and_receive(":SOUR:CURR?"))
+
+    def set_level(self, amps: float) -> None:
+        self.socket_put(f":SOUR:CURR {amps:.6E}")
+        self._reading_time = 0.0
+
+    def get_compliance_voltage(self) -> float:
+        return float(self.socket_put_and_receive(":SENS:VOLT:PROT?"))
+
+    def set_compliance_voltage(self, volts: float) -> None:
+        self.socket_put(f":SENS:VOLT:PROT {volts:.6E}")
+
+    def in_compliance(self) -> bool:
+        return self.socket_put_and_receive(":SENS:VOLT:PROT:TRIP?") == "1"
+
+    @threadlocked
+    def read_output(self, max_age: float | None = None) -> tuple[float, float]:
+        """Measured (volts, amps), (0, 0) with the output off."""
+        max_age = self.READBACK_MAX_AGE if max_age is None else max_age
+        if time.monotonic() - self._reading_time > max_age:
+            if self.output_enabled():
+                volts, amps = self.socket_put_and_receive(":READ?").split(",")[:2]
+                self._reading = (float(volts), float(amps))
+            else:
+                self._reading = (0.0, 0.0)
+            self._reading_time = time.monotonic()
+        return self._reading
+
+
+class Keithley2400SignalBase(SocketSignal):
+    def __init__(self, signal_name, **kwargs):
+        self.signal_name = signal_name
+        super().__init__(**kwargs)
+        self.controller: Keithley2400Controller = self.parent.controller
+
+
+class Keithley2400SignalRO(Keithley2400SignalBase):
+    def __init__(self, signal_name, **kwargs):
+        super().__init__(signal_name, **kwargs)
+        self._metadata["write_access"] = False
+
+    def _socket_set(self, val):
+        raise ReadOnlyError("Read-only signals cannot be set")
+
+
+class Keithley2400ReadbackSignal(Keithley2400SignalRO):
+    """Measured current in mA."""
+
+    def _socket_get(self):
+        return self.controller.read_output()[1] * 1e3
+
+
+class Keithley2400VoltageSignal(Keithley2400SignalRO):
+    """Measured voltage in V."""
+
+    def _socket_get(self):
+        return self.controller.read_output()[0]
+
+
+class Keithley2400OutputSignal(Keithley2400SignalRO):
+    def _socket_get(self):
+        return int(self.controller.output_enabled())
+
+
+class Keithley2400ComplianceTrippedSignal(Keithley2400SignalRO):
+    def _socket_get(self):
+        return int(self.controller.in_compliance())
+
+
+class Keithley2400SetpointSignal(Keithley2400SignalBase):
+    """Programmed source level in mA."""
+
+    def _socket_get(self):
+        return self.controller.get_level() * 1e3
+
+    def _socket_set(self, val):
+        self.parent.check_value(val)
+        self.controller.set_level(val * 1e-3)
+
+
+class Keithley2400ComplianceVoltageSignal(Keithley2400SignalBase):
+    """Voltage limit in V."""
+
+    def _socket_get(self):
+        return self.controller.get_compliance_voltage()
+
+    def _socket_set(self, val):
+        self.controller.set_compliance_voltage(val)
+
+
+class Keithley2400IsMoving(SignalRO):
+    def set_is_moving(self, value: int) -> None:
+        self._readback = value
+
+
+class Keithley2400(Device, PositionerBase):
+    """
+    Keithley 2400 as a current source, positioner in mA.
+
+    Args:
+        limits: Allowed source current in mA.
+        tolerance: Allowed difference between measured and requested current in mA.
+        compliance_voltage: Voltage limit in V.
+        source_range_ma: Fixed source range in mA, fixed so level changes cause no range glitch.
+        remote_sense: True for 4-wire sensing, needs the sense leads connected at the load.
+    """
+
+    USER_ACCESS = ["output_on", "output_off", "status"]
+
+    readback = Cpt(Keithley2400ReadbackSignal, signal_name="readback", kind="hinted")
+    user_setpoint = Cpt(Keithley2400SetpointSignal, signal_name="setpoint")
+    voltage = Cpt(Keithley2400VoltageSignal, signal_name="voltage", kind="normal")
+    output_enabled = Cpt(Keithley2400OutputSignal, signal_name="output_enabled", kind="normal")
+    compliance_tripped = Cpt(
+        Keithley2400ComplianceTrippedSignal, signal_name="compliance_tripped", kind="normal"
+    )
+    compliance_voltage = Cpt(
+        Keithley2400ComplianceVoltageSignal, signal_name="compliance_voltage", kind="config"
+    )
+    motor_is_moving = Cpt(Keithley2400IsMoving, value=0, kind="normal")
+    high_limit_travel = Cpt(Signal, value=0, kind="omitted")
+    low_limit_travel = Cpt(Signal, value=0, kind="omitted")
+
+    SUB_READBACK = "readback"
+    SUB_CONNECTION_CHANGE = "connection_change"
+    _default_sub = SUB_READBACK
+
+    SETTLE_TIMEOUT = 2.0  # s
+
+    def __init__(
+        self,
+        prefix="",
+        *,
+        name,
+        kind=None,
+        read_attrs=None,
+        configuration_attrs=None,
+        parent=None,
+        host="mpc2844.psi.ch",
+        port=3336,
+        limits=(0.0, 3.0),
+        tolerance: float = 0.01,
+        compliance_voltage: float = 5.0,
+        source_range_ma: float = 10.0,
+        remote_sense: bool = False,
+        socket_cls=SocketIO,
+        device_manager=None,
+        **kwargs,
+    ):
+        self.controller = Keithley2400Controller(
+            socket_cls=socket_cls, socket_host=host, socket_port=port, device_manager=device_manager
+        )
+        self.controller.set_axis(axis=self, axis_nr=0)
+        self.tolerance = tolerance
+        self._compliance_voltage = compliance_voltage
+        self._source_range_ma = source_range_ma
+        self._remote_sense = remote_sense
+
+        super().__init__(
+            prefix,
+            name=name,
+            kind=kind,
+            read_attrs=read_attrs,
+            configuration_attrs=configuration_attrs,
+            parent=parent,
+            **kwargs,
+        )
+        self.readback.name = self.name
+        self.controller.subscribe(
+            self._update_connection_state, event_type=self.SUB_CONNECTION_CHANGE
+        )
+        self._update_connection_state()
+        self.low_limit_travel.put(limits[0])
+        self.high_limit_travel.put(limits[1])
+
+    def wait_for_connection(self, timeout: float = 30.0) -> bool:
+        self.controller.on(timeout=timeout)
+        self.controller.configure(
+            self._compliance_voltage, self._source_range_ma, self._remote_sense
+        )
+
+    def destroy(self):
+        """Close the socket. The output stays as it is."""
+        self.controller.off(update_config=False)
+        return super().destroy()
+
+    @property
+    def limits(self):
+        return (self.low_limit_travel.get(), self.high_limit_travel.get())
+
+    @property
+    def low_limit(self):
+        return self.limits[0]
+
+    @property
+    def high_limit(self):
+        return self.limits[1]
+
+    def check_value(self, pos):
+        low_limit, high_limit = self.limits
+        if not low_limit <= pos <= high_limit:
+            raise LimitError(f"current={pos} mA not within limits {self.limits}")
+
+    def _update_connection_state(self, **kwargs):
+        for walk in self.walk_signals():
+            walk.item._metadata["connected"] = self.controller.connected
+
+    @raise_if_disconnected
+    def move(self, position, wait=True, **kwargs) -> MoveStatus:
+        """Source ``position`` mA, switching the output on if it is off."""
+        timeout = kwargs.pop("timeout", 10)
+        status = super().move(position, timeout=timeout, **kwargs)
+        self.user_setpoint.put(position, wait=False)  # level first, then the output
+
+        def apply_and_finish():
+            self.motor_is_moving.set_is_moving(1)
+            error = None
+            try:
+                self.controller.set_output(True)
+                self._wait_for_current(position)
+            except Exception as exc:  # pylint: disable=broad-except
+                error = exc
+            self.motor_is_moving.set_is_moving(0)
+            val = self.readback.read()
+            self._run_subs(sub_type=self.SUB_READBACK, value=val, timestamp=time.time())
+            if error is None:
+                self._done_moving(success=True)
+                return
+            # BEC forwards this exception to the client, _done_moving would lose the reason.
+            logger.error(f"{self.name}: {error}")
+            try:
+                status.set_exception(error)
+            except InvalidState:
+                pass  # stopped or timed out in the meantime
+            self._reset_sub(self._SUB_REQ_DONE)
+
+        threading.Thread(target=apply_and_finish, daemon=True).start()
+        if wait:
+            status.wait()
+        return status
+
+    def _wait_for_current(self, position: float) -> None:
+        deadline = time.monotonic() + self.SETTLE_TIMEOUT
+        while True:
+            volts, amps = self.controller.read_output(max_age=0)
+            if np.isclose(amps * 1e3, position, atol=self.tolerance, rtol=0):
+                return
+            if time.monotonic() > deadline:
+                break
+            time.sleep(0.05)
+        hint = (
+            f" Voltage compliance reached at {volts} V: check the wiring, or the load needs "
+            "more than the compliance voltage."
+            if self.controller.in_compliance()
+            else ""
+        )
+        raise Keithley2400Error(f"Requested {position} mA, measured {amps * 1e3} mA.{hint}")
+
+    def output_on(self) -> None:
+        """Switch the output on at the programmed level. Use ``umv`` to set a new level."""
+        self.controller.set_output(True)
+
+    def output_off(self) -> None:
+        """Switch the output off. The programmed level is kept."""
+        self.controller.set_output(False)
+
+    def status(self) -> None:
+        """Print the state of the instrument."""
+        volts, amps = self.controller.read_output(max_age=0)
+        table = PrettyTable()
+        table.header = False
+        table.add_rows(
+            [
+                ["Output", "on" if self.controller.output_enabled() else "off"],
+                ["Programmed current", f"{self.controller.get_level() * 1e3:.4f} mA"],
+                ["Measured current", f"{amps * 1e3:.4f} mA"],
+                ["Measured voltage", f"{volts:.4f} V"],
+                ["Compliance voltage", f"{self.controller.get_compliance_voltage()} V"],
+                ["In compliance", "YES" if self.controller.in_compliance() else "no"],
+            ]
+        )
+        print(table)
+
+    @property
+    def egu(self):
+        return "mA"

docs/user/ptychography/lamni.md

@@ -333,6 +333,8 @@ Only actions from a fixed, reviewed registry can be queued — not arbitrary cod
 | --- | --- | --- |
 | `move` | Move device(s) to absolute position(s). Only devices on the allow-list below can be targeted. | `positions`: `{device: target}`, e.g. `{"mokev": 6.2}` |
 | `optimize_idgap` | Scan the undulator gap over a range and move to the peak. **Not yet implemented (no-op stub).** | `search_range`: mm, default 0.5, range 0–2 |
+| `set_keithley_current` | Set the Keithley 2400 current, switching its output on if it is off, then sleep. The queue pauses if the current is not reached, e.g. in voltage compliance. | `current_mA`: mA, range from the device limits; `sleep_s`: s, default 0 |
+| `sleep` | Wait. | `seconds`: s, default 60 |
 
 Devices allowed for `move`: `mokev` (energy, keV), `idgap` (undulator gap, mm) — the
 same beamline-wide devices as flOMNI's queue. A `move` naming any other device is
@@ -347,6 +349,15 @@ lamni.tomo_queue_add_command(
 )
 ```
 
+Example — change the sample current and wait for it to act, then wait further:
+```python
+lamni.tomo_queue_add_command(
+    {"action": "set_keithley_current", "kwargs": {"current_mA": 1.5, "sleep_s": 120}},
+    label="1.5 mA",
+)
+lamni.tomo_queue_add_command({"action": "sleep", "kwargs": {"seconds": 600}}, label="wait 10 min")
+```
+
 `tomo_queue_show()` lists command jobs alongside tomogram jobs, e.g.:
 ```text
 [2]    pending  reconfigure to 6.2 keV  CMD move{'positions': {'mokev': 6.2}} > optimize_idgap{'search_range': 0.5}  [idem]

tests/tests_bec_ipython_client/test_lamni_tomo_queue.py

@@ -9,6 +9,8 @@ tomo-queue backend actually needs: self.client + _init_tomo_queue().
 
 import types
 
+import pytest
+
 import csaxs_bec.bec_ipython_client.plugins.LamNI.lamni as lamni_module
 import csaxs_bec.bec_ipython_client.plugins.OMNY_shared.tomo_queue_mixin as tomo_queue_mixin
 from csaxs_bec.bec_ipython_client.plugins.LamNI.lamni import LamNI, LamNIError, _ProgressProxy
@@ -438,3 +440,75 @@ def test_format_command_step_shows_units():
     move_step = {"action": "move", "kwargs": {"positions": {"ccm_energy": 6.2}}}
     assert tomo_queue_mixin.format_command_step(move_step, registry) == "move(ccm_energy=6.2 keV)"
     assert tomo_queue_mixin.format_command_step({"action": "noop", "kwargs": {}}) == "noop"
+
+
+def _setup_keithley_action(monkeypatch):
+    fake_dev = _FakeDev()
+    fake_dev.keithley = types.SimpleNamespace(limits=(0.0, 3.0))
+    lamni = make_lamni()
+    lamni.device_manager = types.SimpleNamespace(devices={"keithley": fake_dev.keithley})
+    calls = []
+    monkeypatch.setattr(lamni_module, "dev", fake_dev, raising=False)
+    monkeypatch.setattr(lamni_module, "umv", lambda *a: calls.append(("umv", a)), raising=False)
+    monkeypatch.setattr(lamni_module.time, "sleep", lambda s: calls.append(("sleep", s)))
+    return lamni, fake_dev, calls
+
+
+def test_tomo_queue_set_keithley_current_sets_then_sleeps(monkeypatch):
+    lamni, fake_dev, calls = _setup_keithley_action(monkeypatch)
+    lamni.tomo_queue_add_command(
+        {"action": "set_keithley_current", "kwargs": {"current_mA": 1.5, "sleep_s": 30}}
+    )
+    lamni.tomo_queue_execute()
+    assert calls == [("umv", (fake_dev.keithley, 1.5)), ("sleep", 30)]
+    assert lamni._tomo_queue_proxy[0]["status"] == "done"
+    assert lamni._tomo_queue_proxy[0]["idempotent"] is True
+
+
+def test_tomo_queue_set_keithley_current_without_sleep(monkeypatch):
+    lamni, fake_dev, calls = _setup_keithley_action(monkeypatch)
+    lamni._queue_action_set_keithley_current(0.0)
+    assert calls == [("umv", (fake_dev.keithley, 0.0))]
+
+
+def test_tomo_queue_set_keithley_current_rejects_out_of_limit(monkeypatch):
+    lamni, _fake_dev, calls = _setup_keithley_action(monkeypatch)
+    with pytest.raises(ValueError, match="above the maximum"):
+        lamni.tomo_queue_add_command(
+            {"action": "set_keithley_current", "kwargs": {"current_mA": 3.5}}
+        )
+    with pytest.raises(ValueError, match="current limits"):
+        lamni._queue_action_set_keithley_current(3.5)
+    assert calls == []
+
+
+def test_tomo_queue_set_keithley_current_without_device_pauses_the_queue(monkeypatch):
+    lamni, fake_dev, calls = _setup_keithley_action(monkeypatch)
+    del fake_dev.keithley
+    lamni.device_manager.devices.clear()
+    lamni.tomo_queue_add_command(
+        {"action": "set_keithley_current", "kwargs": {"current_mA": 1.0}}
+    )
+    with pytest.raises(ValueError, match="not loaded"):
+        lamni.tomo_queue_execute()
+    assert lamni._tomo_queue_proxy[0]["status"] == "incomplete"
+    assert calls == []
+
+
+def test_tomo_queue_sleep_action(monkeypatch):
+    lamni, _fake_dev, calls = _setup_keithley_action(monkeypatch)
+    lamni.tomo_queue_add_command({"action": "sleep", "kwargs": {"seconds": 5}})
+    lamni.tomo_queue_execute()
+    assert calls == [("sleep", 5)]
+    with pytest.raises(ValueError, match="below the minimum"):
+        lamni.tomo_queue_add_command({"action": "sleep", "kwargs": {"seconds": -1}})
+
+
+def test_format_command_step_keithley_units():
+    lamni = make_lamni()
+    registry = lamni._describe_tomo_queue_actions()
+    step = {"action": "set_keithley_current", "kwargs": {"current_mA": 1.5, "sleep_s": 30}}
+    assert (
+        tomo_queue_mixin.format_command_step(step, registry)
+        == "set_keithley_current(current_mA=1.5 mA, sleep_s=30 s)"
+    )

tests/tests_bec_widgets/test_tomo_params_keithley_queue.py

@@ -0,0 +1,62 @@
+"""The LamNI keithley and sleep queue actions must be buildable in the command job dialog, which
+is generated from the published action registry."""
+
+import os
+from unittest import mock
+
+import pytest
+
+os.environ.setdefault("QT_QPA_PLATFORM", "offscreen")
+
+from qtpy.QtWidgets import QApplication, QDoubleSpinBox  # noqa: E402
+
+from csaxs_bec.bec_ipython_client.plugins.LamNI.lamni import LamNI  # noqa: E402
+from csaxs_bec.bec_widgets.widgets.tomo_params.tomo_params import (  # noqa: E402
+    CommandJobBuilderDialog,
+)
+
+
+@pytest.fixture(scope="module")
+def qapp():
+    return QApplication.instance() or QApplication([])
+
+
+@pytest.fixture
+def dialog(qapp):
+    registry = {
+        "actions": {
+            name: {key: spec[key] for key in ("idempotent", "help", "params")}
+            for name, spec in LamNI._TOMO_QUEUE_ACTIONS.items()
+        },
+        "move_devices": {},
+    }
+    client = mock.MagicMock()
+    client.get_global_var.return_value = registry
+    client.device_manager.devices = {"keithley": mock.MagicMock(limits=(0.0, 3.0))}
+    return CommandJobBuilderDialog(client)
+
+
+def _add_step(dialog, action, **values):
+    dialog._action_combo.setCurrentIndex(dialog._action_combo.findData(action))
+    for name, value in values.items():
+        widget, _field = dialog._step_field_widgets[name]
+        assert isinstance(widget, QDoubleSpinBox)
+        widget.setValue(value)
+    dialog._on_add_step()
+
+
+def test_keithley_and_sleep_steps_build_a_queue_job(dialog):
+    _add_step(dialog, "set_keithley_current", current_mA=1.5, sleep_s=30)
+    _add_step(dialog, "sleep", seconds=60)
+    dialog._on_accept()
+    assert dialog.job["steps"] == [
+        {"action": "set_keithley_current", "kwargs": {"current_mA": 1.5, "sleep_s": 30}},
+        {"action": "sleep", "kwargs": {"seconds": 60}},
+    ]
+    assert dialog.job["idempotent"] is True
+
+
+def test_current_field_is_bounded_by_the_device_limits(dialog):
+    dialog._action_combo.setCurrentIndex(dialog._action_combo.findData("set_keithley_current"))
+    widget, _field = dialog._step_field_widgets["current_mA"]
+    assert (widget.minimum(), widget.maximum()) == (0.0, 3.0)

tests/tests_devices/test_keithley_2400.py

@@ -0,0 +1,191 @@
+from unittest import mock
+
+import pytest
+from ophyd.utils import LimitError
+from ophyd_devices.utils.controller import ControllerCommunicationError
+
+from csaxs_bec.devices.keithley.keithley_2400 import (
+    Keithley2400,
+    Keithley2400Controller,
+    Keithley2400Error,
+)
+
+
+class FakeKeithleySocket:
+    """Minimal 2400: replies CR LF terminated and split in two segments, as seen through the bridge."""
+
+    instances = []
+    output = False
+    level = 0.0
+    source_function = "VOLT"
+    compliance = 21.0
+    max_volts = 100.0  # load limit, above it the instrument sits in compliance
+    resistance = 1000.0
+
+    def __init__(self, host, port, socket_timeout=2):
+        self.is_open = False
+        self.commands = []
+        self.replies = []
+        type(self).instances.append(self)
+
+    def open(self, timeout=10):
+        self.is_open = True
+
+    def close(self):
+        self.is_open = False
+
+    def receive(self, buffer_length=1024):
+        return self.replies.pop(0) if self.replies else b""
+
+    def _measured(self):
+        volts = min(self.level * self.resistance, self.compliance, self.max_volts)
+        return volts, volts / self.resistance
+
+    def put(self, msg):
+        command = msg.decode().strip()
+        self.commands.append(command)
+        reply = {
+            "*IDN?": "KEITHLEY INSTRUMENTS INC.,MODEL 2400,4009094,C32",
+            ":OUTP?": "1" if self.output else "0",
+            ":SOUR:FUNC?": self.source_function,
+            ":SOUR:CURR?": f"{self.level:E}",
+            ":SENS:VOLT:PROT?": f"{self.compliance:E}",
+            ":SYST:ERR?": '0,"No error"',
+        }.get(command)
+        if command == ":READ?":
+            assert self.output, "the 2400 refuses :READ? with the output off"
+            reply = "{:E},{:E}".format(*self._measured())
+        elif command == ":SENS:VOLT:PROT:TRIP?":
+            reply = "1" if self._measured()[0] < self.level * self.resistance else "0"
+        elif command in (":OUTP ON", ":OUTP OFF"):
+            type(self).output = command == ":OUTP ON"
+        elif command == ":SOUR:FUNC CURR":
+            type(self).source_function = "CURR"
+        elif command.startswith(":SOUR:CURR ") and "RANG" not in command:
+            type(self).level = float(command.split()[1])
+        elif command.startswith(":SENS:VOLT:PROT "):
+            type(self).compliance = float(command.split()[1])
+        if reply is not None:
+            data = f"{reply}\r\n".encode()
+            self.replies += [part for part in (data[:3], data[3:]) if part]
+
+
+@pytest.fixture
+def socket_cls():
+    """A fresh socket class per test, the instrument state lives on the class."""
+    Keithley2400Controller._reset_controller()
+    return type("Socket", (FakeKeithleySocket,), {"instances": []})
+
+
+@pytest.fixture
+def keithley(socket_cls):
+    dev = Keithley2400(
+        name="keithley", socket_cls=socket_cls, host="fake", port=1, device_manager=mock.MagicMock()
+    )
+    yield dev
+    dev.controller.off(update_config=False)
+
+
+def commands(socket_cls):
+    return socket_cls.instances[-1].commands
+
+
+def test_connect_with_output_off_configures_and_keeps_it_off(keithley, socket_cls):
+    keithley.wait_for_connection()
+    assert socket_cls.source_function == "CURR"
+    assert socket_cls.compliance == 5.0
+    assert socket_cls.output is False
+    assert not any("OUTP ON" in cmd or "MEAS" in cmd or cmd == "*RST" for cmd in commands(socket_cls))
+
+
+def test_connect_with_output_on_leaves_the_source_alone(keithley, socket_cls):
+    socket_cls.output, socket_cls.level, socket_cls.source_function = True, 2e-3, "CURR"
+    keithley.wait_for_connection()
+    assert (socket_cls.output, socket_cls.level, socket_cls.compliance) == (True, 2e-3, 21.0)
+    sent = commands(socket_cls)
+    assert not any(cmd.startswith((":SOUR:FUNC ", ":SOUR:CURR", ":OUTP ", "*RST")) for cmd in sent)
+    assert keithley.readback.get() == pytest.approx(2.0)
+    assert keithley.user_setpoint.get() == pytest.approx(2.0)
+
+
+def test_connect_refuses_voltage_source_with_output_on(keithley, socket_cls):
+    socket_cls.output = True
+    with pytest.raises(Keithley2400Error, match="source function VOLT"):
+        keithley.wait_for_connection()
+    assert socket_cls.output is True
+
+
+def test_connect_refuses_other_instrument(keithley, socket_cls, monkeypatch):
+    monkeypatch.setattr(keithley.controller, "socket_put_and_receive", lambda val: "SOMETHING ELSE")
+    keithley.controller.on()
+    with pytest.raises(Keithley2400Error, match="identification"):
+        keithley.controller.configure(5.0, 10.0, False)
+
+
+def test_readbacks_with_output_off_do_not_trigger(keithley, socket_cls):
+    keithley.wait_for_connection()
+    assert keithley.readback.get() == 0.0
+    assert keithley.voltage.get() == 0.0
+    assert ":READ?" not in commands(socket_cls)
+
+
+def test_move_programs_the_level_before_the_output(keithley, socket_cls):
+    keithley.wait_for_connection()
+    keithley.move(1.5).wait(timeout=5)
+    sent = commands(socket_cls)
+    assert sent.index(":SOUR:CURR 1.500000E-03") < sent.index(":OUTP ON")
+    assert keithley.readback.get() == pytest.approx(1.5)
+    assert keithley.voltage.get() == pytest.approx(1.5)
+    assert keithley.output_enabled.get() == 1
+
+
+def test_move_outside_limits_changes_nothing(keithley, socket_cls):
+    keithley.wait_for_connection()
+    with pytest.raises(LimitError):
+        keithley.move(3.1)
+    with pytest.raises(LimitError):
+        keithley.user_setpoint.put(3.1)
+    assert socket_cls.level == 0.0
+    assert socket_cls.output is False
+
+
+def test_move_in_compliance_fails_with_the_reason(keithley, socket_cls, monkeypatch):
+    monkeypatch.setattr(keithley, "SETTLE_TIMEOUT", 0.1)
+    keithley.wait_for_connection()
+    with pytest.raises(Keithley2400Error, match="compliance"):
+        socket_cls.max_volts = 1.0
+        keithley.move(2.0).wait(timeout=5)
+
+
+def test_output_off_keeps_the_level(keithley, socket_cls):
+    keithley.wait_for_connection()
+    keithley.move(1.0).wait(timeout=5)
+    keithley.output_off()
+    assert socket_cls.output is False
+    assert socket_cls.level == pytest.approx(1e-3)
+    assert keithley.readback.get() == 0.0
+
+
+def test_status_prints_the_state(keithley, socket_cls, capsys):
+    keithley.wait_for_connection()
+    keithley.status()
+    assert "Compliance voltage" in capsys.readouterr().out
+
+
+def test_stop_and_destroy_leave_the_output_on(socket_cls):
+    dev = Keithley2400(
+        name="keithley", socket_cls=socket_cls, host="fake", port=1, device_manager=mock.MagicMock()
+    )
+    dev.wait_for_connection()
+    dev.move(1.0).wait(timeout=5)
+    dev.stop()
+    dev.destroy()
+    assert socket_cls.output is True
+    assert socket_cls.level == pytest.approx(1e-3)
+
+
+def test_closed_connection_raises(keithley, socket_cls):
+    keithley.controller.on()
+    socket_cls.instances[-1].put = lambda msg: None
+    with pytest.raises(ControllerCommunicationError):
+        keithley.controller.socket_put_and_receive(":OUTP?")