diff --git a/csaxs_bec/bec_ipython_client/plugins/LamNI/lamni.py b/csaxs_bec/bec_ipython_client/plugins/LamNI/lamni.py index babd886e..df8db921 100644 --- a/csaxs_bec/bec_ipython_client/plugins/LamNI/lamni.py +++ b/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: diff --git a/csaxs_bec/device_configs/keithley_2400.yaml b/csaxs_bec/device_configs/keithley_2400.yaml new file mode 100644 index 00000000..73c1b02d --- /dev/null +++ b/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 diff --git a/csaxs_bec/device_configs/simulated_omny/simulated_lamni_keithley.yaml b/csaxs_bec/device_configs/simulated_omny/simulated_lamni_keithley.yaml new file mode 100644 index 00000000..2f1cba07 --- /dev/null +++ b/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 diff --git a/csaxs_bec/devices/keithley/__init__.py b/csaxs_bec/devices/keithley/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/csaxs_bec/devices/keithley/keithley_2400.py b/csaxs_bec/devices/keithley/keithley_2400.py new file mode 100644 index 00000000..9464295b --- /dev/null +++ b/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" diff --git a/docs/user/ptychography/lamni.md b/docs/user/ptychography/lamni.md index a481ce5e..31248dd8 100644 --- a/docs/user/ptychography/lamni.md +++ b/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] diff --git a/tests/tests_bec_ipython_client/test_lamni_tomo_queue.py b/tests/tests_bec_ipython_client/test_lamni_tomo_queue.py index 71321b33..854fe751 100644 --- a/tests/tests_bec_ipython_client/test_lamni_tomo_queue.py +++ b/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)" + ) diff --git a/tests/tests_bec_widgets/test_tomo_params_keithley_queue.py b/tests/tests_bec_widgets/test_tomo_params_keithley_queue.py new file mode 100644 index 00000000..c5468fdb --- /dev/null +++ b/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) diff --git a/tests/tests_devices/test_keithley_2400.py b/tests/tests_devices/test_keithley_2400.py new file mode 100644 index 00000000..51bff74b --- /dev/null +++ b/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?")