From bcedf794a52074e2dc3f120f38285d97bbad666d Mon Sep 17 00:00:00 2001 From: x01dc Date: Fri, 10 Jul 2026 15:46:47 +0200 Subject: [PATCH] fixes/version2 sim devices s Please enter the commit message for your changes. Lines starting --- .../simulated_omny/simulated_flomni.yaml | 2 +- .../epics/delay_generator_csaxs/ddg_1.py | 3 +- csaxs_bec/devices/sim/sim_cameras.py | 54 ++- csaxs_bec/devices/sim/sim_galil.py | 150 +++++-- csaxs_bec/devices/sim/sim_rt_flomni.py | 84 +++- csaxs_bec/devices/sim/sim_smaract.py | 56 ++- tests/sim_flomni_harness.py | 378 ++++++++++++++++++ 7 files changed, 655 insertions(+), 72 deletions(-) create mode 100644 tests/sim_flomni_harness.py diff --git a/csaxs_bec/device_configs/simulated_omny/simulated_flomni.yaml b/csaxs_bec/device_configs/simulated_omny/simulated_flomni.yaml index c57eadc..2890aed 100644 --- a/csaxs_bec/device_configs/simulated_omny/simulated_flomni.yaml +++ b/csaxs_bec/device_configs/simulated_omny/simulated_flomni.yaml @@ -295,7 +295,7 @@ ftransy: sim_velocity: 5 sim_initial_position: 0 sim_analog_inputs: - 1: -1.6 + "1": -1.6 enabled: true onFailure: buffer readOnly: false diff --git a/csaxs_bec/devices/epics/delay_generator_csaxs/ddg_1.py b/csaxs_bec/devices/epics/delay_generator_csaxs/ddg_1.py index 05d1054..ad05299 100644 --- a/csaxs_bec/devices/epics/delay_generator_csaxs/ddg_1.py +++ b/csaxs_bec/devices/epics/delay_generator_csaxs/ddg_1.py @@ -37,7 +37,7 @@ import traceback from typing import TYPE_CHECKING from bec_lib.logger import bec_logger -from bec_server.scan_server.scans.scan_base import ScanInfo as ScanServerScanInfo +#from bec_server.scan_server.scans.scan_base import ScanInfo as ScanServerScanInfo from ophyd import Component as Cpt from ophyd import EpicsSignalRO, Kind from ophyd_devices import CompareStatus, DeviceStatus, StatusBase, TransitionStatus @@ -60,6 +60,7 @@ from csaxs_bec.devices.utils.utils import fetch_scan_info if TYPE_CHECKING: # pragma: no cover from bec_lib.devicemanager import DeviceManagerBase, ScanInfo + from bec_server.scan_server.scans.scan_base import ScanInfo as ScanServerScanInfo from csaxs_bec.devices.epics.mcs_card.mcs_card_csaxs import MCSCardCSAXS diff --git a/csaxs_bec/devices/sim/sim_cameras.py b/csaxs_bec/devices/sim/sim_cameras.py index 6219915..31cdfb4 100644 --- a/csaxs_bec/devices/sim/sim_cameras.py +++ b/csaxs_bec/devices/sim/sim_cameras.py @@ -40,11 +40,25 @@ def make_test_pattern(height: int, width: int, rgb: bool) -> np.ndarray: class SimWebcamViewer(WebcamViewer): - """WebcamViewer serving a static synthetic frame instead of an MJPEG stream.""" + """WebcamViewer serving a static synthetic frame instead of an MJPEG stream. - def __init__(self, url: str, name: str, sim_shape=(480, 640), **kwargs) -> None: + The signature lists all deviceConfig keys explicitly; the BEC device server only + passes config keys present in inspect.signature(dev_cls). + """ + + def __init__( + self, + url: str = "http://simulated", + name: str = "", + num_rotation_90=0, + transpose=False, + sim_shape=(480, 640), + **kwargs, + ) -> None: self._sim_frame = make_test_pattern(int(sim_shape[0]), int(sim_shape[1]), rgb=True) - super().__init__(url=url, name=name, **kwargs) + super().__init__( + url=url, name=name, num_rotation_90=num_rotation_90, transpose=transpose, **kwargs + ) def _update_loop(self) -> None: while not self._shutdown_event.is_set(): @@ -93,8 +107,34 @@ class _SimIDSBackend: class SimIDSCamera(IDSCamera): """IDSCamera with the pyueye backend replaced by a synthetic frame source.""" - def __init__(self, *, name: str, camera_id: int = 0, sim_shape=(1024, 1280), **kwargs): - super().__init__(name=name, camera_id=camera_id, **kwargs) - rgb = not kwargs.get("force_monochrome", False) - self.cam = _SimIDSBackend(int(sim_shape[1]), int(sim_shape[0]), rgb=rgb) + def __init__( + self, + *, + name: str, + camera_id: int = 0, + prefix: str = "", + scan_info=None, + m_n_colormode=1, + bits_per_pixel=24, + live_mode: bool = False, + num_rotation_90: int = 0, + transpose: bool = False, + force_monochrome: bool = False, + sim_shape=(1024, 1280), + **kwargs, + ): + super().__init__( + name=name, + camera_id=camera_id, + prefix=prefix, + scan_info=scan_info, + m_n_colormode=m_n_colormode, + bits_per_pixel=bits_per_pixel, + live_mode=live_mode, + num_rotation_90=num_rotation_90, + transpose=transpose, + force_monochrome=force_monochrome, + **kwargs, + ) + self.cam = _SimIDSBackend(int(sim_shape[1]), int(sim_shape[0]), rgb=not force_monochrome) self.cam.force_monochrome = self._force_monochrome diff --git a/csaxs_bec/devices/sim/sim_galil.py b/csaxs_bec/devices/sim/sim_galil.py index 456786b..1474287 100644 --- a/csaxs_bec/devices/sim/sim_galil.py +++ b/csaxs_bec/devices/sim/sim_galil.py @@ -470,44 +470,136 @@ class SimGalilSocket(SimSocketBase): return ":" -def _pop_sim_kwargs(kwargs: dict) -> dict: +def _sim_state_kwargs( + sim_initial_position, + sim_velocity, + sim_stppermm, + sim_referenced, + sim_analog_inputs, + sim_digital_inputs, + sim_limit_low_active, + sim_limit_high_active, + sign, +) -> dict: + initial_position = None + if sim_initial_position is not None: + initial_position = sim_initial_position * sign return { - "initial_position": kwargs.pop("sim_initial_position", None), - "velocity": kwargs.pop("sim_velocity", None), - "stppermm": kwargs.pop("sim_stppermm", None), - "referenced": kwargs.pop("sim_referenced", True), - "analog_inputs": kwargs.pop("sim_analog_inputs", None), - "digital_inputs": kwargs.pop("sim_digital_inputs", None), - "limit_low_active": kwargs.pop("sim_limit_low_active", None), - "limit_high_active": kwargs.pop("sim_limit_high_active", None), + "initial_position": initial_position, + "velocity": sim_velocity, + "stppermm": sim_stppermm, + "referenced": sim_referenced, + "analog_inputs": sim_analog_inputs, + "digital_inputs": sim_digital_inputs, + "limit_low_active": sim_limit_low_active, + "limit_high_active": sim_limit_high_active, } class SimFlomniGalilMotor(FlomniGalilMotor): - """FlomniGalilMotor connected to the simulated Galil controller.""" + """FlomniGalilMotor connected to the simulated Galil controller. - def __init__(self, axis_Id, *args, **kwargs): - sim_kwargs = _pop_sim_kwargs(kwargs) - kwargs["socket_cls"] = SimGalilSocket - super().__init__(axis_Id, *args, **kwargs) - state = SimStateRegistry.get(SimGalilState, kwargs.get("host"), kwargs.get("port")) - if sim_kwargs["initial_position"] is not None: - sim_kwargs["initial_position"] = sim_kwargs["initial_position"] * self.sign - state.configure_axis(self.axis_Id_numeric, **sim_kwargs) + Note: the signature must list all deviceConfig keys explicitly. The BEC device + server only passes config keys that appear in inspect.signature(dev_cls) and only + injects device_manager if it is an explicit parameter. + """ + + def __init__( + self, + axis_Id, + prefix="", + *, + name, + host="mpc2844.psi.ch", + port=8081, + limits=None, + sign=1, + device_manager=None, + sim_initial_position=None, + sim_velocity=None, + sim_stppermm=None, + sim_referenced=True, + sim_analog_inputs=None, + sim_digital_inputs=None, + sim_limit_low_active=None, + sim_limit_high_active=None, + **kwargs, + ): + super().__init__( + axis_Id, + prefix, + name=name, + host=host, + port=port, + limits=limits, + sign=sign, + socket_cls=SimGalilSocket, + device_manager=device_manager, + **kwargs, + ) + state = SimStateRegistry.get(SimGalilState, host, port) + state.configure_axis( + self.axis_Id_numeric, + **_sim_state_kwargs( + sim_initial_position, + sim_velocity, + sim_stppermm, + sim_referenced, + sim_analog_inputs, + sim_digital_inputs, + sim_limit_low_active, + sim_limit_high_active, + self.sign, + ), + ) class SimFuprGalilMotor(FuprGalilMotor): """FuprGalilMotor (rotation stage) connected to the simulated Galil controller.""" - def __init__(self, axis_Id, *args, **kwargs): - sim_kwargs = _pop_sim_kwargs(kwargs) - kwargs["socket_cls"] = SimGalilSocket - super().__init__(axis_Id, *args, **kwargs) - state = SimStateRegistry.get(SimGalilState, kwargs.get("host"), kwargs.get("port")) - sim_kwargs["stppermm"] = self.MOTOR_RESOLUTION + def __init__( + self, + axis_Id, + prefix="", + *, + name, + host="mpc2844.psi.ch", + port=8084, + limits=None, + sign=1, + device_manager=None, + sim_initial_position=None, + sim_velocity=None, + sim_referenced=True, + **kwargs, + ): + super().__init__( + axis_Id, + prefix, + name=name, + host=host, + port=port, + limits=limits, + sign=sign, + socket_cls=SimGalilSocket, + device_manager=device_manager, + **kwargs, + ) + state = SimStateRegistry.get(SimGalilState, host, port) state.axis(self.axis_Id_numeric).auto_motor_off = False - if sim_kwargs["velocity"] is None: - sim_kwargs["velocity"] = 25.0 # deg/s, SPA=resol*25 after referencing - if sim_kwargs["initial_position"] is not None: - sim_kwargs["initial_position"] = sim_kwargs["initial_position"] * self.sign - state.configure_axis(self.axis_Id_numeric, **sim_kwargs) + if sim_velocity is None: + sim_velocity = 25.0 # deg/s, SPA=resol*25 after referencing + state.configure_axis( + self.axis_Id_numeric, + **_sim_state_kwargs( + sim_initial_position, + sim_velocity, + self.MOTOR_RESOLUTION, + sim_referenced, + None, + None, + None, + None, + self.sign, + ), + ) diff --git a/csaxs_bec/devices/sim/sim_rt_flomni.py b/csaxs_bec/devices/sim/sim_rt_flomni.py index 292bfba..873d88a 100644 --- a/csaxs_bec/devices/sim/sim_rt_flomni.py +++ b/csaxs_bec/devices/sim/sim_rt_flomni.py @@ -240,33 +240,75 @@ class SimRtFlomniSocket(SimSocketBase): ) -def _pop_sim_kwargs(kwargs: dict) -> dict: - return { - "point_dwell_s": kwargs.pop("sim_point_dwell_s", None), - "pid_x_voltage": kwargs.pop("sim_pid_x_voltage", None), - "ssi_signal": kwargs.pop("sim_ssi_signal", None), - } - - class SimRtFlomniMotor(RtFlomniMotor): - """RtFlomniMotor connected to the simulated Orchestra server.""" + """RtFlomniMotor connected to the simulated Orchestra server. - def __init__(self, axis_Id, *args, **kwargs): - sim_kwargs = _pop_sim_kwargs(kwargs) - kwargs["socket_cls"] = SimRtFlomniSocket - super().__init__(axis_Id, *args, **kwargs) - SimStateRegistry.get(SimRtFlomniState, kwargs.get("host"), kwargs.get("port")).configure( - **sim_kwargs + The signature lists all deviceConfig keys explicitly; the BEC device server only + passes config keys present in inspect.signature(dev_cls). + """ + + def __init__( + self, + axis_Id, + prefix="", + *, + name, + host="mpc2844.psi.ch", + port=2222, + limits=None, + sign=1, + device_manager=None, + sim_point_dwell_s=None, + sim_pid_x_voltage=None, + sim_ssi_signal=None, + **kwargs, + ): + super().__init__( + axis_Id, + prefix, + name=name, + host=host, + port=port, + limits=limits, + sign=sign, + socket_cls=SimRtFlomniSocket, + device_manager=device_manager, + **kwargs, + ) + SimStateRegistry.get(SimRtFlomniState, host, port).configure( + point_dwell_s=sim_point_dwell_s, + pid_x_voltage=sim_pid_x_voltage, + ssi_signal=sim_ssi_signal, ) class SimRtFlomniFlyer(RtFlomniFlyer): """RtFlomniFlyer connected to the simulated Orchestra server.""" - def __init__(self, *args, **kwargs): - sim_kwargs = _pop_sim_kwargs(kwargs) - kwargs["socket_cls"] = SimRtFlomniSocket - super().__init__(*args, **kwargs) - SimStateRegistry.get(SimRtFlomniState, kwargs.get("host"), kwargs.get("port")).configure( - **sim_kwargs + def __init__( + self, + prefix="", + *, + name, + host="mpc2844.psi.ch", + port=2222, + device_manager=None, + sim_point_dwell_s=None, + sim_pid_x_voltage=None, + sim_ssi_signal=None, + **kwargs, + ): + super().__init__( + prefix, + name=name, + host=host, + port=port, + socket_cls=SimRtFlomniSocket, + device_manager=device_manager, + **kwargs, + ) + SimStateRegistry.get(SimRtFlomniState, host, port).configure( + point_dwell_s=sim_point_dwell_s, + pid_x_voltage=sim_pid_x_voltage, + ssi_signal=sim_ssi_signal, ) diff --git a/csaxs_bec/devices/sim/sim_smaract.py b/csaxs_bec/devices/sim/sim_smaract.py index 192c4f8..2f13ed6 100644 --- a/csaxs_bec/devices/sim/sim_smaract.py +++ b/csaxs_bec/devices/sim/sim_smaract.py @@ -192,17 +192,47 @@ class SimSmaractSocket(SimSocketBase): class SimSmaractMotor(SmaractMotor): - """SmaractMotor connected to the simulated MCS controller.""" + """SmaractMotor connected to the simulated MCS controller. - def __init__(self, axis_Id, *args, **kwargs): - sim_kwargs = { - "initial_position": kwargs.pop("sim_initial_position", None), - "velocity": kwargs.pop("sim_velocity", None), - "referenced": kwargs.pop("sim_referenced", True), - } - kwargs["socket_cls"] = SimSmaractSocket - super().__init__(axis_Id, *args, **kwargs) - state = SimStateRegistry.get(SimSmaractState, kwargs.get("host"), kwargs.get("port")) - if sim_kwargs["initial_position"] is not None: - sim_kwargs["initial_position"] = sim_kwargs["initial_position"] * self.sign - state.configure_axis(self.axis_Id_numeric, **sim_kwargs) + The signature lists all deviceConfig keys explicitly; the BEC device server only + passes config keys present in inspect.signature(dev_cls). + """ + + def __init__( + self, + axis_Id, + prefix="", + *, + name, + host="mpc2844.psi.ch", + port=3334, + limits=None, + sign=1, + device_manager=None, + sim_initial_position=None, + sim_velocity=None, + sim_referenced=True, + **kwargs, + ): + super().__init__( + axis_Id, + prefix, + name=name, + host=host, + port=port, + limits=limits, + sign=sign, + socket_cls=SimSmaractSocket, + device_manager=device_manager, + **kwargs, + ) + state = SimStateRegistry.get(SimSmaractState, host, port) + initial_position = None + if sim_initial_position is not None: + initial_position = sim_initial_position * self.sign + state.configure_axis( + self.axis_Id_numeric, + initial_position=initial_position, + velocity=sim_velocity, + referenced=sim_referenced, + ) diff --git a/tests/sim_flomni_harness.py b/tests/sim_flomni_harness.py new file mode 100644 index 0000000..72bd599 --- /dev/null +++ b/tests/sim_flomni_harness.py @@ -0,0 +1,378 @@ +""" +Offline harness for the simulated flOMNI hardware stack. + +Instantiates the real device classes wired to the protocol simulations +(csaxs_bec.devices.sim) without a running BEC deployment and exercises the code +paths used by the flomni IPython-client scripts: + + 1. Galil coarse-stage moves and readback (fgalil protocol, both controllers) + 2. FUPR rotation incl. rotation-angle propagation to the RT controller + 3. drive_axis_to_limit / referencing (#FES) and limit-switch checks + 4. Smaract OSA moves + 5. RT feedback_enable_with_reset + laser tracker (cross-controller choreography) + 6. Fermat-like position scan through RtFlomniFlyer kickoff/complete readout + 7. Gripper sample transfer routine (#MNTMODE, #GRGET incl. confirm handshake) + 8. Simulated cameras (frame generation) + +Run with the repo root on PYTHONPATH: python tests/sim_flomni_harness.py +""" + +from __future__ import annotations + +import threading +import time + +import numpy as np + + +class FakeConfiguredDevice: + """Minimal stand-in for a bec_lib configured device (dm.devices.).""" + + def __init__(self, obj, user_parameter=None): + self.obj = obj + self.enabled = True + self.read_only = False + self.user_parameter = dict(user_parameter or {}) + + def update_user_parameter(self, update: dict): + self.user_parameter.update(update) + + +class FakeDeviceContainer(dict): + def __getattr__(self, name): + try: + return self[name] + except KeyError as exc: + raise AttributeError(name) from exc + + +class FakeDeviceManager: + def __init__(self): + self.devices = FakeDeviceContainer() + + def add(self, name, obj, user_parameter=None): + self.devices[name] = FakeConfiguredDevice(obj, user_parameter) + return obj + + +def check(condition, message): + status = "OK " if condition else "FAIL" + print(f"[{status}] {message}") + if not condition: + raise AssertionError(message) + + +def main(): # noqa: C901 + from csaxs_bec.devices.sim.sim_cameras import SimIDSCamera, SimWebcamViewer + from csaxs_bec.devices.sim.sim_galil import SimFlomniGalilMotor, SimFuprGalilMotor + from csaxs_bec.devices.sim.sim_rt_flomni import SimRtFlomniFlyer, SimRtFlomniMotor + from csaxs_bec.devices.sim.sim_smaract import SimSmaractMotor + + dm = FakeDeviceManager() + host = "mpc2844.psi.ch" + + # --- instantiate the simulated endstation ----------------------------------- + fsamx = dm.add( + "fsamx", + SimFlomniGalilMotor( + "E", + name="fsamx", + host=host, + port=8081, + sign=1, + limits=[-162, 0], + device_manager=dm, + sim_stppermm=2560, + sim_velocity=50, + sim_initial_position=-1.14, + ), + user_parameter={"in": -1.14}, + ) + fsamy = dm.add( + "fsamy", + SimFlomniGalilMotor( + "F", + name="fsamy", + host=host, + port=8081, + sign=1, + limits=[2, 3.8], + device_manager=dm, + sim_stppermm=1280000, + sim_velocity=0.1, + sim_initial_position=3, + ), + user_parameter={"in": 3}, + ) + ftransy = dm.add( + "ftransy", + SimFlomniGalilMotor( + "A", + name="ftransy", + host=host, + port=8081, + sign=1, + limits=[-100, 0], + device_manager=dm, + sim_stppermm=25600, + sim_velocity=5, + sim_initial_position=0, + sim_analog_inputs={1: -1.6}, + ), + user_parameter={"sensor_voltage": -1.6}, + ) + ftray = dm.add( + "ftray", + SimFlomniGalilMotor( + "D", + name="ftray", + host=host, + port=8081, + sign=-1, + limits=[-200, 0], + device_manager=dm, + sim_stppermm=25600, + sim_velocity=8, + sim_initial_position=0, + sim_limit_low_active=True, + ), + ) + foptx = dm.add( + "foptx", + SimFlomniGalilMotor( + "B", + name="foptx", + host=host, + port=8082, + sign=1, + limits=[-17, -12], + device_manager=dm, + sim_stppermm=6400, + sim_velocity=1, + sim_initial_position=-13.831, + ), + ) + fopty = dm.add( + "fopty", + SimFlomniGalilMotor( + "F", + name="fopty", + host=host, + port=8082, + sign=1, + limits=[0, 4], + device_manager=dm, + sim_stppermm=128000, + sim_velocity=0.05, + sim_initial_position=0.42, + ), + ) + foptz = dm.add( + "foptz", + SimFlomniGalilMotor( + "A", + name="foptz", + host=host, + port=8082, + sign=1, + limits=[0, 27], + device_manager=dm, + sim_stppermm=6400, + sim_velocity=1, + sim_initial_position=23, + ), + ) + ftrackz = dm.add( + "ftrackz", + SimFlomniGalilMotor( + "G", + name="ftrackz", + host=host, + port=8082, + sign=1, + limits=[4.5, 5.5], + device_manager=dm, + sim_stppermm=115200, + sim_velocity=0.02, + sim_initial_position=5, + ), + ) + fsamroy = dm.add( + "fsamroy", + SimFuprGalilMotor( + "A", + name="fsamroy", + host=host, + port=8084, + sign=-1, + limits=[-5, 365], + device_manager=dm, + sim_initial_position=0, + ), + ) + fosax = dm.add( + "fosax", + SimSmaractMotor( + "A", + name="fosax", + host=host, + port=3334, + sign=-1, + limits=[10.2, 10.6], + device_manager=dm, + sim_initial_position=10.4, + ), + ) + rtx = dm.add( + "rtx", + SimRtFlomniMotor( + "A", name="rtx", host=host, port=2222, sign=1, limits=[-200, 200], device_manager=dm + ), + user_parameter={"low_signal": 8500, "min_signal": 8000, "rt_pid_voltage": -0.06219}, + ) + rty = dm.add( + "rty", + SimRtFlomniMotor( + "B", name="rty", host=host, port=2222, sign=1, limits=[-100, 100], device_manager=dm + ), + user_parameter={"tomo_additional_offsety": 0}, + ) + rtz = dm.add( + "rtz", + SimRtFlomniMotor( + "C", name="rtz", host=host, port=2222, sign=1, limits=[-100, 100], device_manager=dm + ), + ) + flyer = dm.add( + "rt_positions", + SimRtFlomniFlyer( + name="rt_positions", host=host, port=2222, device_manager=dm, sim_point_dwell_s=0.005 + ), + ) + + for device in (fsamx, fsamy, ftransy, ftray, foptx, fopty, foptz, ftrackz, fsamroy, fosax): + device.wait_for_connection() + for device in (rtx, rty, rtz): + device.wait_for_connection() + flyer.controller.on() + print("--- all simulated devices connected ---") + + # --- 1. galil moves ---------------------------------------------------------- + check(np.isclose(fsamx.readback.get(), -1.14, atol=1e-3), "fsamx boots at -1.14 (in)") + t0 = time.time() + fsamx.move(-2.14, wait=True) + check(np.isclose(fsamx.readback.get(), -2.14, atol=1e-3), "fsamx moved to -2.14") + fsamx.move(-1.14, wait=True) + t0 = time.time() + fsamy.move(3.05, wait=True) + duration = time.time() - t0 + check(np.isclose(fsamy.readback.get(), 3.05, atol=1e-3), "fsamy moved to 3.05") + check(duration > 0.3, f"fsamy move at 0.1 mm/s took {duration:.2f} s (constant velocity)") + sensor = float(ftransy.controller.socket_put_and_receive("MG@AN[1]").strip()) + check(np.isclose(sensor, -1.6), "ftransy gripper sensor voltage reads -1.6") + + # --- 2. FUPR rotation + RT angle propagation --------------------------------- + fsamroy.move(90, wait=True) + check(np.isclose(fsamroy.readback.get(), 90, atol=0.1), "fsamroy rotated to 90 deg") + rt_state = rtx.controller.sock.state + check( + np.isclose(rt_state.angle_rad, 90 / 180 * np.pi, atol=1e-3), + "RT angle updated via FUPR readback", + ) + fsamroy.move(0, wait=True) + + # --- 3. referencing / limit switches ------------------------------------------ + check(ftray.controller.get_motor_limit_switch("D")[0], "ftray boots with 'IN' limit active") + foptz.controller.drive_axis_to_limit(0, "reverse") + check(np.isclose(foptz.readback.get(), 0.0, atol=1e-3), "foptz referenced to 0 at limit") + check(foptz.controller.get_motor_limit_switch("A")[0], "foptz reverse limit switch active") + curtain = foptz.controller.fosaz_light_curtain_is_triggered() + check(curtain, "fosaz light curtain reads triggered") + + # --- 4. smaract --------------------------------------------------------------- + fosax.move(10.5, wait=True) + check(np.isclose(fosax.readback.get(), 10.5, atol=1e-4), "fosax moved to 10.5") + check(fosax.controller.axis_is_referenced(0), "fosax referenced") + + # --- 5. RT feedback + tracker -------------------------------------------------- + rtx.controller.feedback_enable_with_reset() + check(rtx.controller.feedback_is_running(), "interferometer feedback running") + check(rtx.controller.laser_tracker_check_enabled(), "laser tracker enabled") + check(dm.devices["fsamx"].read_only, "fsamx set to read-only with feedback on") + signal_status = rtx.controller.laser_tracker_check_signalstrength(verbose=False) + check(signal_status == "ok", f"tracker signal strength '{signal_status}'") + + rtx.move(5.0, wait=True) + check(np.isclose(rtx.readback.get(), 5.0, atol=1e-3), "rtx moved to 5.0 with feedback") + rtx.controller.move_samx_to_scan_region(5.0) + + # --- 6. fermat-like scan through the flyer ------------------------------------- + num_points = 400 + positions = [[5.0 + 3 * np.sin(i * 0.3), 2 * np.cos(i * 0.3), 0.0] for i in range(num_points)] + rtx.controller.clear_trajectory_generator() + rtx.controller.add_pos_to_scan(positions) + flyer.stage() + status = flyer.kickoff() + status.wait(timeout=10) + complete_status = flyer.complete() + deadline = time.time() + 30 + while not complete_status.done and time.time() < deadline: + time.sleep(0.1) + check(complete_status.done, "flyer completed the scan readout") + data = flyer.data.get() + check(rtx.controller.get_scan_status()[0] == 0, "scan mode back to 0 after scan") + print(f" last data row: {data}") + + # --- 7. gripper transfer routine ----------------------------------------------- + con = ftransy.controller + con.socket_put_confirmed("XQ#MNTMODE") + time.sleep(0.1) + check(bool(float(con.socket_put_and_receive("MG mntmod").strip())), "controller in mount mode") + con.socket_put_confirmed("getaprch=-17.5") + con.socket_put_confirmed("XQ#GRGET,3") + + def confirmer(): + for _ in range(200): + confirm = int(float(con.socket_put_and_receive("MG confirm").strip())) + if confirm == -1: + con.socket_put_confirmed("confirm=1") + in_progress = bool(float(con.socket_put_and_receive("MG mntprgs").strip())) + if not in_progress: + return + time.sleep(0.1) + + confirm_thread = threading.Thread(target=confirmer) + confirm_thread.start() + deadline = time.time() + 30 + while time.time() < deadline: + if not bool(float(con.socket_put_and_receive("MG mntprgs").strip())): + break + time.sleep(0.2) + confirm_thread.join() + check( + not bool(float(con.socket_put_and_receive("MG mntprgs").strip())), + "gripper #GRGET routine finished (confirm handshakes served)", + ) + con.socket_put_confirmed("XQ#POSMODE") + time.sleep(0.1) + check( + not bool(float(con.socket_put_and_receive("MG mntmod").strip())), + "controller back in positioning mode", + ) + + # --- 8. cameras ----------------------------------------------------------------- + webcam = SimWebcamViewer(url="http://sim/video", name="cam_flomni_gripper") + ids_cam = SimIDSCamera(name="cam_xeye", camera_id=11, force_monochrome=True) + frame = ids_cam.cam.get_image_data() + check(frame.ndim == 2 and frame.max() > 200, f"IDS sim frame {frame.shape}, max {frame.max()}") + check(webcam._sim_frame.shape == (480, 640, 3), "webcam sim frame 480x640 RGB") + + # --- feedback disable puts stages back to writable ------------------------------- + rtx.controller.feedback_disable() + check(not dm.devices["fsamx"].read_only, "fsamx writable again after feedback_disable") + + print("\nAll harness checks passed.") + + +if __name__ == "__main__": + main()