bec/csaxs_bec
0
0
Fork 2
Code Issues 17 Pull Requests 2 Actions Packages Projects Releases Wiki Activity

fixes/version2 sim devices

Browse Source 
s Please enter the commit message for your changes. Lines starting
This commit is contained in:
x01dc
2026-07-12 07:52:37 +02:00
committed by holler
co-authored by holler
parent 704826c94a
commit bcedf794a5
7 changed files with 655 additions and 72 deletions
Download Patch File Download Diff File Expand all files Collapse all files
csaxs_bec/device_configs/simulated_omny/simulated_flomni.yaml
+1 -1
View File
@@ -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
csaxs_bec/devices/epics/delay_generator_csaxs/ddg_1.py
+2 -1
View File
@@ -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
csaxs_bec/devices/sim/sim_cameras.py
+47 -7
View File
@@ -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
csaxs_bec/devices/sim/sim_galil.py
+121 -29
View File
@@ -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,
),
)
csaxs_bec/devices/sim/sim_rt_flomni.py
+63 -21
View File
@@ -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,
)
csaxs_bec/devices/sim/sim_smaract.py
+43 -13
View File
@@ -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,
)
tests/sim_flomni_harness.py
+378
View File
@@ -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.<name>)."""
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()