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

feat(sim): add simulated flOMNI endstation (galil, smaract, rt, cameras)

Browse Source 
Protocol-level simulation of the flOMNI hardware injected via socket_cls:
real device/controller classes run unchanged against state machines
implementing fgalil.dmc, galil_micos_upr.dmc, the Smaract MCS protocol
and the Orchestra CommunicationServer. Cameras reuse the real device
classes with a synthetic frame source. Includes simulated_flomni device
config (boots referenced at 'in' positions) and an offline harness
covering moves, referencing, rt feedback/tracker, flyer scan readout
and the gripper transfer routine.
This commit is contained in:
x01dc
2026-07-12 07:52:37 +02:00
committed by holler
parent 6722d1573a
commit 704826c94a
6 changed files with 1777 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
csaxs_bec/device_configs/simulated_omny/simulated_flomni.yaml
+574
View File
@@ -0,0 +1,574 @@
############################################################
# Simulated flOMNI endstation
#
# Mirrors ptycho_flomni.yaml with the hardware replaced by protocol-level
# simulations (csaxs_bec.devices.sim.*). The real ophyd device and controller
# classes are used; only the socket transport / frame source is simulated.
#
# sim_* keys in deviceConfig:
# sim_initial_position boot position in user units (axes start referenced)
# sim_velocity axis speed in user units per second (from fgalil.dmc)
# sim_stppermm microsteps per mm (from fgalil.dmc #INIT tables)
# sim_referenced boot referenced state (default true)
# sim_point_dwell_s RT scan time per position
#
# Not included in iteration 1: flomni_samples (EPICS), flomni_temphum (EPICS),
# omny_panda, omny_xray_gui.
############################################################
############################################################
#################### flOMNI Galil motors ###################
############################################################
feyex:
description: Xray eye X (simulated)
deviceClass: csaxs_bec.devices.sim.sim_galil.SimFlomniGalilMotor
deviceConfig:
axis_Id: D
host: mpc2844.psi.ch
limits:
- -30
- -1
port: 8082
sign: 1
sim_stppermm: 25600
sim_velocity: 5
sim_initial_position: -1
enabled: true
onFailure: buffer
readOnly: false
readoutPriority: baseline
connectionTimeout: 20
userParameter:
in: -16.453
out: -1
fttrx_in: 2.3
fttrx_out: -24
feyey:
description: Xray eye Y (simulated)
deviceClass: csaxs_bec.devices.sim.sim_galil.SimFlomniGalilMotor
deviceConfig:
axis_Id: E
host: mpc2844.psi.ch
limits:
- -1
- -10
port: 8082
sign: 1
sim_stppermm: 25600
sim_velocity: 5
sim_initial_position: -1
enabled: true
onFailure: buffer
readOnly: false
readoutPriority: baseline
connectionTimeout: 20
userParameter:
in: -10.09
fheater:
description: Heater Y (simulated)
deviceClass: csaxs_bec.devices.sim.sim_galil.SimFlomniGalilMotor
deviceConfig:
axis_Id: C
host: mpc2844.psi.ch
limits:
- -15
- 0
port: 8082
sign: -1
sim_stppermm: 10240
sim_velocity: 0.1
sim_initial_position: 0
enabled: true
onFailure: buffer
readOnly: false
readoutPriority: baseline
connectionTimeout: 20
foptx:
description: Optics X (simulated)
deviceClass: csaxs_bec.devices.sim.sim_galil.SimFlomniGalilMotor
deviceConfig:
axis_Id: B
host: mpc2844.psi.ch
limits:
- -17
- -12
port: 8082
sign: 1
sim_stppermm: 6400
sim_velocity: 1
sim_initial_position: -13.831
enabled: true
onFailure: buffer
readOnly: true
readoutPriority: baseline
connectionTimeout: 20
userParameter:
#170 micron, 60 nm
in: -13.831
out: -13.831
fopty:
description: Optics Y (simulated)
deviceClass: csaxs_bec.devices.sim.sim_galil.SimFlomniGalilMotor
deviceConfig:
axis_Id: F
host: mpc2844.psi.ch
limits:
- 0
- 4
port: 8082
sign: 1
sim_stppermm: 128000
sim_velocity: 0.05
sim_initial_position: 0.42
enabled: true
onFailure: buffer
readOnly: true
readoutPriority: baseline
connectionTimeout: 20
userParameter:
#170 micron, 60 nm
in: 0.42
out: 0.57
foptz:
description: Optics Z (simulated)
deviceClass: csaxs_bec.devices.sim.sim_galil.SimFlomniGalilMotor
deviceConfig:
axis_Id: A
host: mpc2844.psi.ch
limits:
- 0
- 27
port: 8082
sign: 1
sim_stppermm: 6400
sim_velocity: 1
sim_initial_position: 23
enabled: true
onFailure: buffer
readOnly: false
readoutPriority: baseline
connectionTimeout: 20
userParameter:
in: 23
fsamroy:
description: Sample rotation (simulated)
deviceClass: csaxs_bec.devices.sim.sim_galil.SimFuprGalilMotor
deviceConfig:
axis_Id: A
host: mpc2844.psi.ch
limits:
- -5
- 365
port: 8084
sign: -1
sim_velocity: 25
sim_initial_position: 0
enabled: true
onFailure: buffer
readOnly: false
readoutPriority: baseline
connectionTimeout: 20
fsamx:
description: Sample coarse X (simulated)
deviceClass: csaxs_bec.devices.sim.sim_galil.SimFlomniGalilMotor
deviceConfig:
axis_Id: E
host: mpc2844.psi.ch
limits:
- -162
- 0
port: 8081
sign: 1
sim_stppermm: 2560
sim_velocity: 50
sim_initial_position: -1.14
enabled: true
onFailure: buffer
readOnly: true
readoutPriority: baseline
connectionTimeout: 20
userParameter:
in: -1.14
fsamy:
description: Sample coarse Y (simulated)
deviceClass: csaxs_bec.devices.sim.sim_galil.SimFlomniGalilMotor
deviceConfig:
axis_Id: F
host: mpc2844.psi.ch
limits:
- 2
- 3.8
port: 8081
sign: 1
sim_stppermm: 1280000
sim_velocity: 0.1
sim_initial_position: 3
enabled: true
onFailure: buffer
readOnly: true
readoutPriority: baseline
connectionTimeout: 20
userParameter:
in: 3
ftracky:
description: Laser Tracker coarse Y (simulated)
deviceClass: csaxs_bec.devices.sim.sim_galil.SimFlomniGalilMotor
deviceConfig:
axis_Id: H
host: mpc2844.psi.ch
limits:
- 2.2
- 2.8
port: 8082
sign: 1
sim_stppermm: 115200
sim_velocity: 0.02
sim_initial_position: 2.5
enabled: true
onFailure: buffer
readOnly: false
readoutPriority: baseline
connectionTimeout: 20
ftrackz:
description: Laser Tracker coarse Z (simulated)
deviceClass: csaxs_bec.devices.sim.sim_galil.SimFlomniGalilMotor
deviceConfig:
axis_Id: G
host: mpc2844.psi.ch
limits:
- 4.5
- 5.5
port: 8082
sign: 1
sim_stppermm: 115200
sim_velocity: 0.02
sim_initial_position: 5
enabled: true
onFailure: buffer
readOnly: false
readoutPriority: baseline
connectionTimeout: 20
ftransx:
description: Sample transer X (simulated)
deviceClass: csaxs_bec.devices.sim.sim_galil.SimFlomniGalilMotor
deviceConfig:
axis_Id: C
host: mpc2844.psi.ch
limits:
- 0
- 50
port: 8081
sign: 1
sim_stppermm: 25600
sim_velocity: 4
sim_initial_position: 0
enabled: true
onFailure: buffer
readOnly: false
readoutPriority: baseline
connectionTimeout: 20
ftransy:
description: Sample transer Y (simulated)
deviceClass: csaxs_bec.devices.sim.sim_galil.SimFlomniGalilMotor
deviceConfig:
axis_Id: A
host: mpc2844.psi.ch
limits:
- -100
- 0
port: 8081
sign: 1
sim_stppermm: 25600
sim_velocity: 5
sim_initial_position: 0
sim_analog_inputs:
1: -1.6
enabled: true
onFailure: buffer
readOnly: false
readoutPriority: baseline
connectionTimeout: 20
userParameter:
sensor_voltage: -1.6
ftransz:
description: Sample transer Z (simulated)
deviceClass: csaxs_bec.devices.sim.sim_galil.SimFlomniGalilMotor
deviceConfig:
axis_Id: B
host: mpc2844.psi.ch
limits:
- 0
- 145
port: 8081
sign: 1
sim_stppermm: 25600
sim_velocity: 8
sim_initial_position: 0
enabled: true
onFailure: buffer
readOnly: false
readoutPriority: baseline
connectionTimeout: 20
ftray:
description: Sample transfer tray (simulated)
deviceClass: csaxs_bec.devices.sim.sim_galil.SimFlomniGalilMotor
deviceConfig:
axis_Id: D
host: mpc2844.psi.ch
limits:
- -200
- 0
port: 8081
sign: -1
sim_stppermm: 25600
sim_velocity: 8
sim_initial_position: 0
# tray boots at the 'IN' position (low limit switch active, see check_tray_in)
sim_limit_low_active: true
enabled: true
onFailure: buffer
readOnly: false
readoutPriority: baseline
connectionTimeout: 20
############################################################
#################### flOMNI Smaract motors #################
############################################################
fosax:
description: OSA X (simulated)
deviceClass: csaxs_bec.devices.sim.sim_smaract.SimSmaractMotor
deviceConfig:
axis_Id: A
host: mpc2844.psi.ch
limits:
- 10.2
- 10.6
port: 3334
sign: -1
sim_velocity: 1
sim_initial_position: 10.4
enabled: true
onFailure: buffer
readOnly: false
readoutPriority: baseline
connectionTimeout: 20
userParameter:
#170 micron, 60 nm, 7.9 kev
in: 8.727079
out: 5.1
fosay:
description: OSA Y (simulated)
deviceClass: csaxs_bec.devices.sim.sim_smaract.SimSmaractMotor
deviceConfig:
axis_Id: B
host: mpc2844.psi.ch
limits:
- -3.1
- -2.9
port: 3334
sign: -1
sim_velocity: 1
sim_initial_position: -3
enabled: true
onFailure: buffer
readOnly: false
readoutPriority: baseline
connectionTimeout: 20
userParameter:
#170 micron, 60 nm, 7.9 kev
in: -0.04603
fosaz:
description: OSA Z (simulated)
deviceClass: csaxs_bec.devices.sim.sim_smaract.SimSmaractMotor
deviceConfig:
axis_Id: C
host: mpc2844.psi.ch
limits:
- -6
- -4
port: 3334
sign: 1
sim_velocity: 1
sim_initial_position: -5
enabled: true
onFailure: buffer
readOnly: false
readoutPriority: baseline
connectionTimeout: 20
userParameter:
#170 micron, 60 nm, 7.9 kev, foptz 16.9, probe size 7.5 mu
in: 13.1
out: 6
############################################################
#################### flOMNI RT motors ######################
############################################################
rtx:
description: flomni rt (simulated)
deviceClass: csaxs_bec.devices.sim.sim_rt_flomni.SimRtFlomniMotor
deviceConfig:
limits:
- -200
- 200
axis_Id: A
host: mpc2844.psi.ch
port: 2222
sign: 1
enabled: true
onFailure: buffer
readOnly: false
readoutPriority: on_request
connectionTimeout: 20
userParameter:
low_signal: 8500
min_signal: 8000
rt_pid_voltage: -0.06219
rty:
description: flomni rt (simulated)
deviceClass: csaxs_bec.devices.sim.sim_rt_flomni.SimRtFlomniMotor
deviceConfig:
limits:
- -100
- 100
axis_Id: B
host: mpc2844.psi.ch
port: 2222
sign: 1
enabled: true
onFailure: buffer
readOnly: false
readoutPriority: on_request
connectionTimeout: 20
userParameter:
tomo_additional_offsety: 0
rtz:
description: flomni rt (simulated)
deviceClass: csaxs_bec.devices.sim.sim_rt_flomni.SimRtFlomniMotor
deviceConfig:
limits:
- -100
- 100
axis_Id: C
host: mpc2844.psi.ch
port: 2222
sign: 1
enabled: true
onFailure: buffer
readOnly: false
readoutPriority: on_request
connectionTimeout: 20
rt_positions:
deviceClass: csaxs_bec.devices.sim.sim_rt_flomni.SimRtFlomniFlyer
deviceConfig:
host: mpc2844.psi.ch
port: 2222
# time per scan position; determines simulated scan duration
sim_point_dwell_s: 0.02
readoutPriority: async
connectionTimeout: 20
enabled: true
readOnly: False
############################################################
####################### Cameras ############################
############################################################
cam_flomni_gripper:
description: Camera sample changer (simulated)
deviceClass: csaxs_bec.devices.sim.sim_cameras.SimWebcamViewer
deviceConfig:
url: http://flomnicamserver:5000/video_high
num_rotation_90: 3
transpose: false
enabled: true
onFailure: buffer
readOnly: false
readoutPriority: on_request
cam_flomni_overview:
description: Camera flomni overview (simulated)
deviceClass: csaxs_bec.devices.sim.sim_cameras.SimWebcamViewer
deviceConfig:
url: http://flomnicamserver:5001/video_high
num_rotation_90: 3
transpose: false
enabled: true
onFailure: buffer
readOnly: false
readoutPriority: on_request
cam_xeye:
description: Camera flOMNI Xray eye ID1 (simulated)
deviceClass: csaxs_bec.devices.sim.sim_cameras.SimIDSCamera
deviceConfig:
camera_id: 11
bits_per_pixel: 24
num_rotation_90: 0
transpose: true
force_monochrome: true
m_n_colormode: 1
enabled: true
onFailure: buffer
readOnly: false
readoutPriority: async
userParameter:
pixel_calibration: 0.00044247787610619477 # mm/pixel (= 0.05 / 113)
############################################################
#################### Computed signal #######################
############################################################
calculated_signal:
description: Calculated signal from alignment for fit
deviceClass: ophyd_devices.ComputedSignal
deviceConfig:
compute_method: "def just_rand():\n return 42"
enabled: true
readOnly: false
readoutPriority: baseline
############################################################
######### Simulated beamline devices (shutter, DDG) ########
############################################################
ddg1:
description: Simulated main delay generator for triggering
deviceClass: csaxs_bec.devices.sim.simulated_beamline_devices.SimulatedDDG1
enabled: true
deviceConfig:
prefix: 'X12SA-CPCL-DDG1:'
onFailure: raise
readOnly: false
readoutPriority: baseline
softwareTrigger: true
fsh:
description: Simulated fast shutter manual control and readback
deviceClass: csaxs_bec.devices.sim.simulated_beamline_devices.cSAXSSimulatedFastShutter
deviceConfig:
prefix: 'X12SA-ES1-TTL:'
onFailure: raise
enabled: true
readoutPriority: monitored
csaxs_bec/devices/sim/sim_cameras.py
+100
View File
@@ -0,0 +1,100 @@
"""
Simulated flOMNI cameras.
`SimWebcamViewer` and `SimIDSCamera` reuse the real device classes and replace only the
frame source: the webcam viewer's MJPEG HTTP stream and the IDS camera's pyueye backend.
All signals (`preview` / `image`, `roi_signal`, `live_mode_enabled`), the USER_ACCESS
methods (`start_live_mode`, `set_rect_roi`, `get_last_image`, ...) and the ROI/trigger
logic therefore run through the unmodified code paths, so GUIs and the X-ray eye
alignment can be exercised against the simulation.
The synthetic frame is a static test pattern: dark background, centered gaussian blob
and a crosshair, which gives the previews something recognizable and provides contrast
for image analysis.
"""
from __future__ import annotations
import numpy as np
from bec_lib.logger import bec_logger
from csaxs_bec.devices.ids_cameras.ids_camera import IDSCamera
from csaxs_bec.devices.omny.webcam_viewer import WebcamViewer
logger = bec_logger.logger
def make_test_pattern(height: int, width: int, rgb: bool) -> np.ndarray:
"""Create a static synthetic camera frame (uint8), mono or RGB."""
yy, xx = np.mgrid[0:height, 0:width]
cy, cx = height / 2.0, width / 2.0
sigma = min(height, width) / 12.0
blob = 180.0 * np.exp(-(((yy - cy) ** 2) + ((xx - cx) ** 2)) / (2.0 * sigma**2))
frame = 20.0 + blob
frame[int(cy) - 1 : int(cy) + 2, :] = 230.0 # horizontal crosshair line
frame[:, int(cx) - 1 : int(cx) + 2] = 230.0 # vertical crosshair line
frame = frame.clip(0, 255).astype(np.uint8)
if rgb:
frame = np.repeat(frame[:, :, np.newaxis], 3, axis=2)
return frame
class SimWebcamViewer(WebcamViewer):
"""WebcamViewer serving a static synthetic frame instead of an MJPEG stream."""
def __init__(self, url: str, name: str, 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)
def _update_loop(self) -> None:
while not self._shutdown_event.is_set():
self.preview.put(self._sim_frame)
self._shutdown_event.wait(0.2) # 5 Hz
class _SimIDSSensor:
"""Mimics the width/height ctypes attributes of the pyueye camera object."""
class _Value:
def __init__(self, value: int):
self.value = value
def __init__(self, width: int, height: int):
self.width = self._Value(width)
self.height = self._Value(height)
class _SimIDSBackend:
"""Drop-in replacement for `base_integration.camera.Camera` serving synthetic frames."""
def __init__(self, width: int, height: int, rgb: bool):
self.cam = _SimIDSSensor(width, height)
self.force_monochrome = False
self._connected = False
self._rgb = rgb
self._frame = make_test_pattern(height, width, rgb=rgb)
def on_connect(self):
self._connected = True
def on_disconnect(self):
self._connected = False
def get_image_data(self) -> np.ndarray:
frame = self._frame
if self.force_monochrome and frame.ndim == 3:
frame = frame[:, :, 0]
return frame
def set_camera_rate_limiting(self, enabled: bool):
pass
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)
self.cam.force_monochrome = self._force_monochrome
csaxs_bec/devices/sim/sim_galil.py
+513
View File
@@ -0,0 +1,513 @@
"""
Simulated Galil DMC controllers for flOMNI.
Implements the command subset of `fgalil.dmc` (sample/optics/eye/tracker controllers on
ports 8081/8082) and `galil_micos_upr.dmc` (FUPR rotation stage on port 8084) that is used
by the BEC device classes (`FlomniGalilMotor`, `FuprGalilMotor`) and by the flomni
IPython-client scripts. The real controller classes talk to this simulation through
`SimGalilSocket`, injected via `socket_cls`.
Simulated behavior ("working hardware", no error injection):
- constant-velocity motion with linear position interpolation
- thread 0 active while a `#NEWPAR`-initiated move is running
- `#FES` / `#FRM` / `#REFAX` reference an axis (position -> 0, limit switch behavior)
- gripper sample transfer routines `#GRGET` / `#GRPUT` on thread 3, including the
`confirm` handshake (`confirm=-1` -> host sets `confirm=1`; `atconfir=1` autoconfirms)
- `#MNTMODE` / `#POSMODE`, gripper open/close on thread 4, `#Tracker` on thread 5
- `#STOP` aborts motion and clears `mntprgs` / `mntmod`
"""
from __future__ import annotations
import re
import threading
import time
from bec_lib.logger import bec_logger
from csaxs_bec.devices.omny.galil.fgalil_ophyd import FlomniGalilMotor
from csaxs_bec.devices.omny.galil.fupr_ophyd import FuprGalilMotor
from csaxs_bec.devices.sim.sim_socket import SimSocketBase, SimStateRegistry
logger = bec_logger.logger
_ASSIGNMENT_RE = re.compile(r"^([A-Za-z]+)(?:\[(\d+)\])?\s*=\s*(.+)$")
_STPPERMM_EXPR_RE = re.compile(r"^([0-9.]+)\s*\*\s*stppermm\[(\d+)\]$")
class _SimGalilAxis:
"""State of a single simulated Galil axis (positions in microsteps)."""
def __init__(self, index: int):
self.index = index
self.connected = False
self.referenced = True
self.motor_on = True
self.auto_motor_off = True # fgalil #MOTOFFA switches motors off when idle
self.stppermm = 25600.0
self.speed_steps = 5.0 * 25600.0
self.pos_steps = 0.0
self._move_start_steps = 0.0
self._move_target_steps = 0.0
self._move_t0 = 0.0
self._move_t1 = 0.0
self.limit_low_active = False
self.limit_high_active = False
def position(self) -> float:
now = time.time()
if now >= self._move_t1:
return self.pos_steps
frac = (now - self._move_t0) / (self._move_t1 - self._move_t0)
return self._move_start_steps + frac * (self._move_target_steps - self._move_start_steps)
def is_moving(self) -> bool:
return time.time() < self._move_t1
def motor_is_on(self) -> bool:
if self.auto_motor_off:
return self.is_moving()
return self.motor_on
def start_move(self, target_steps: float):
speed = max(self.speed_steps, 1.0)
current = self.position()
duration = abs(target_steps - current) / speed
self._move_start_steps = current
self._move_target_steps = target_steps
self._move_t0 = time.time()
self._move_t1 = self._move_t0 + duration
self.pos_steps = target_steps
self.limit_low_active = False
self.limit_high_active = False
def stop_move(self):
self.pos_steps = self.position()
self._move_t1 = 0.0
def wait_move(self, abort_event: threading.Event | None = None):
while self.is_moving():
if abort_event is not None and abort_event.wait(0.02):
self.stop_move()
return
time.sleep(0.02)
class SimGalilState:
"""Shared state of one simulated Galil controller (all axes, variables, threads)."""
N_AXES = 8
def __init__(self, host=None, port=None):
self.host = host
self.port = port
self.lock = threading.RLock()
self.axes = [_SimGalilAxis(i) for i in range(self.N_AXES)]
self.variables = {
"naxis": 0.0,
"ndir": 0.0,
"ntarget": 0.0,
"movereq": 0.0,
"mntmod": 0.0,
"mntprgs": 0.0,
"confirm": 0.0,
"atconfir": 0.0,
"getaprch": 0.0,
"mntaprch": 0.0,
"mntax": 0.0,
"tracken": 0.0,
"trackyct": 0.0,
"trackzct": 0.0,
"allaxref": 0.0,
"axisref": 0.0,
"frmmv": 0.0,
}
self.digital_inputs = {14: 1.0} # light curtain reads triggered by default
self.digital_outputs = {9: 0.0, 15: 1.0} # gripper closed, lights on
self.analog_inputs = {1: -1.6}
self._thread_deadlines = {} # thread_id -> unix time until which thread is active
self._transfer_thread = None
self._transfer_abort = threading.Event()
self._tracker_running = False
self.confirm_steps = 3 # number of confirm handshakes per transfer routine
# --- axis / thread helpers -------------------------------------------------
def axis(self, index: int) -> _SimGalilAxis:
return self.axes[index]
def configure_axis(self, index: int, **kwargs):
with self.lock:
ax = self.axes[index]
ax.connected = True
if "stppermm" in kwargs and kwargs["stppermm"] is not None:
ax.stppermm = float(kwargs["stppermm"])
if "velocity" in kwargs and kwargs["velocity"] is not None:
ax.speed_steps = float(kwargs["velocity"]) * ax.stppermm
if "referenced" in kwargs and kwargs["referenced"] is not None:
ax.referenced = bool(kwargs["referenced"])
if "initial_position" in kwargs and kwargs["initial_position"] is not None:
ax.pos_steps = float(kwargs["initial_position"]) * ax.stppermm
if "analog_inputs" in kwargs and kwargs["analog_inputs"]:
for chan, val in kwargs["analog_inputs"].items():
self.analog_inputs[int(chan)] = float(val)
if "digital_inputs" in kwargs and kwargs["digital_inputs"]:
for chan, val in kwargs["digital_inputs"].items():
self.digital_inputs[int(chan)] = float(val)
if kwargs.get("limit_low_active") is not None:
ax.limit_low_active = bool(kwargs["limit_low_active"])
if kwargs.get("limit_high_active") is not None:
ax.limit_high_active = bool(kwargs["limit_high_active"])
def thread_active(self, thread_id: int) -> bool:
if thread_id == 0:
if any(ax.is_moving() for ax in self.axes):
return True
if thread_id == 3:
if self._transfer_thread is not None and self._transfer_thread.is_alive():
return True
if thread_id == 5:
return self._tracker_running
return time.time() < self._thread_deadlines.get(thread_id, 0.0)
def set_thread_active(self, thread_id: int, duration: float):
self._thread_deadlines[thread_id] = time.time() + duration
def all_axes_referenced(self) -> bool:
connected = [ax for ax in self.axes if ax.connected]
if not connected:
return False
return all(ax.referenced for ax in connected)
# --- routines ---------------------------------------------------------------
def stop_all(self):
self._transfer_abort.set()
for ax in self.axes:
ax.stop_move()
with self.lock:
self.variables["mntmod"] = 0.0
self.variables["mntprgs"] = 0.0
self._tracker_running = False
def start_move_from_newpar(self):
with self.lock:
if self.variables.get("movereq", 0.0) != 1.0:
return
axis_index = int(self.variables.get("naxis", 0))
target_mm = float(self.variables.get("ntarget", 0.0))
self.variables["movereq"] = 0.0
ax = self.axes[axis_index]
if not (self.all_axes_referenced() and ax.referenced):
return
ax.motor_on = True
ax.start_move(target_mm * ax.stppermm)
def reference_axis(self, axis_index: int, direction: float = 0.0):
"""Simulate #FES / #FRM: short motion, then define zero at the reference."""
ax = self.axes[axis_index]
ax.motor_on = True
ax.start_move(ax.position()) # ensure no stale motion
# emulate the drive to the limit / reference with a short, visible motion
self.set_thread_active(0, 0.5)
deadline = time.time() + 0.5
while time.time() < deadline:
time.sleep(0.02)
ax.pos_steps = 0.0
ax.stop_move()
ax.referenced = True
if direction > 0:
ax.limit_high_active = True
elif direction < 0:
ax.limit_low_active = True
def _confirm_handshake(self):
with self.lock:
if self.variables.get("atconfir", 0.0) == 1.0:
return
self.variables["confirm"] = -1.0
while not self._transfer_abort.is_set():
with self.lock:
if self.variables.get("confirm", 0.0) >= 1.0:
self.variables["confirm"] = 0.0
return
time.sleep(0.05)
def _transfer_routine(self, approach_var: str):
"""Simulate #GRGET / #GRPUT on thread 3."""
try:
with self.lock:
approach_mm = float(self.variables.get(approach_var, 0.0))
mntax = int(self.variables.get("mntax", 0))
ax = self.axes[mntax]
initial_steps = ax.position()
if approach_var == "getaprch":
self._set_gripper(True) # JS#GROPEN before approaching
ax.start_move(approach_mm * ax.stppermm)
ax.wait_move(self._transfer_abort)
for _ in range(self.confirm_steps):
if self._transfer_abort.is_set():
return
self._set_gripper(not self._gripper_open())
self._confirm_handshake()
if self._transfer_abort.is_set():
return
ax.start_move(initial_steps - 0.5 * ax.stppermm)
ax.wait_move(self._transfer_abort)
# final gripper state: closed after #GRPUT, closed around sample after #GRGET
self._set_gripper(False)
finally:
with self.lock:
self.variables["mntprgs"] = 0.0
self.variables[approach_var] = 0.0
def start_transfer(self, approach_var: str):
with self.lock:
if self.variables.get("mntmod", 0.0) != 1.0:
logger.warning("[sim galil] transfer requested but system not in mount mode")
return
if self.variables.get("mntprgs", 0.0) == 1.0:
logger.warning("[sim galil] transfer requested but a transfer is in progress")
return
if self.variables.get(approach_var, 0.0) == 0.0:
logger.warning(f"[sim galil] transfer requested but {approach_var} not set")
return
self.variables["mntprgs"] = 1.0
self._transfer_abort.clear()
self._transfer_thread = threading.Thread(
target=self._transfer_routine, args=(approach_var,), daemon=True
)
self._transfer_thread.start()
def _gripper_open(self) -> bool:
return bool(self.digital_outputs.get(9, 0.0))
def _set_gripper(self, is_open: bool):
self.digital_outputs[9] = 1.0 if is_open else 0.0
class SimGalilSocket(SimSocketBase):
"""Simulated socket implementing the fgalil.dmc / galil_micos_upr.dmc command set."""
state_cls = SimGalilState
def handle_command(self, line: str): # noqa: C901
state: SimGalilState = self.state
if line.startswith("MG"):
return self._handle_mg(line[2:].strip())
if line.startswith("XQ"):
return self._handle_xq(line[2:].strip())
if line.startswith("SB") or line.startswith("CB"):
channel = int(line[2:])
state.digital_outputs[channel] = 1.0 if line.startswith("SB") else 0.0
return ":"
# FUPR: absolute position in counts, e.g. "PAA=25600"
match = re.match(r"^PA([A-H])=(-?[0-9.]+)$", line)
if match:
axis = ord(match.group(1).lower()) - 97
state.variables[f"_fupr_target_{axis}"] = float(match.group(2))
return ":"
# FUPR: begin motion, e.g. "BGA"
match = re.match(r"^BG([A-H])$", line)
if match:
axis = ord(match.group(1).lower()) - 97
target = state.variables.get(f"_fupr_target_{axis}", 0.0)
ax = state.axis(axis)
ax.motor_on = True
ax.start_move(target)
return ":"
# FUPR: position readback in counts, e.g. "TPA"
match = re.match(r"^T[DP]([A-H])$", line)
if match:
axis = ord(match.group(1).lower()) - 97
return f"{state.axis(axis).position():.4f}"
match = _ASSIGNMENT_RE.match(line)
if match:
return self._handle_assignment(*match.groups())
logger.warning(f"[sim galil] {self.host}:{self.port} unhandled command '{line}'")
return ":"
def _handle_assignment(self, name: str, index: str | None, value: str):
state: SimGalilState = self.state
value = value.strip()
expr = _STPPERMM_EXPR_RE.match(value)
with state.lock:
if name == "axspeed" and index is not None:
ax = state.axis(int(index))
if expr:
ax.speed_steps = float(expr.group(1)) * state.axis(int(expr.group(2))).stppermm
else:
try:
ax.speed_steps = float(value)
except ValueError:
logger.warning(f"[sim galil] cannot parse axspeed value '{value}'")
return ":"
try:
parsed = float(value)
except ValueError:
parsed = value
if index is not None:
state.variables[f"{name}[{index}]"] = parsed
else:
state.variables[name] = parsed
return ":"
def _handle_mg(self, arg: str):
state: SimGalilState = self.state
if arg.startswith("_XQ"):
thread_id = int(arg[3:])
return "0.0000" if state.thread_active(thread_id) else "-1.0000"
if arg.startswith("_BG"):
axis = ord(arg[3:].lower()) - 97
return f"{1.0 if state.axis(axis).is_moving() else 0.0:.4f}"
if arg.startswith("bcklact["):
return "0.0000"
if arg.startswith("_MO"):
axis = ord(arg[3:].lower()) - 97
return f"{0.0 if state.axis(axis).motor_is_on() else 1.0:.4f}"
if arg.startswith("_LR") or arg.startswith("_LF"):
# "MG _LRA, _LFA" -> low and high limit switch (active low)
axes = re.findall(r"_L[RF]([A-H])", arg)
parts = []
for token, axis_char in zip(re.findall(r"_L[RF]", arg), axes):
ax = state.axis(ord(axis_char.lower()) - 97)
active = ax.limit_low_active if token == "_LR" else ax.limit_high_active
parts.append(f"{0.0 if active else 1.0:.4f}")
return " ".join(parts)
if arg.startswith("_TD"):
axis = ord(arg[3:].lower()) - 97
return f"{state.axis(axis).position():.4f}"
if arg.startswith("_FL") or arg.startswith("_BL"):
return "0.0000"
if arg.startswith("@IN["):
channel = int(arg[4:].split("]")[0])
return f"{state.digital_inputs.get(channel, 0.0):.4f}"
if arg.startswith("@OUT["):
channel = int(arg[5:].split("]")[0])
return f"{state.digital_outputs.get(channel, 0.0):.4f}"
if arg.startswith("@AN["):
channel = int(arg[4:].split("]")[0])
return f"{state.analog_inputs.get(channel, 0.0):.4f}"
if arg == "allaxref" or arg == "axisref":
return f"{1.0 if state.all_axes_referenced() else 0.0:.4f}"
if arg.startswith("axisref["):
axis = int(arg.split("[")[1].split("]")[0])
return f"{1.0 if state.axis(axis).referenced else 0.0:.4f}"
if arg.startswith("folaxerr["):
return "0.0000"
if arg.startswith("stppermm["):
axis = int(arg.split("[")[1].split("]")[0])
return f"{state.axis(axis).stppermm:.4f}"
with state.lock:
if arg in state.variables:
val = state.variables[arg]
return f"{val:.4f}" if isinstance(val, float) else str(val)
logger.warning(f"[sim galil] {self.host}:{self.port} unhandled MG '{arg}'")
return "0.0000"
def _handle_xq(self, arg: str):
state: SimGalilState = self.state
routine = arg.split(",")[0]
if routine == "#NEWPAR":
state.start_move_from_newpar()
return ":"
if routine in ("#FES", "#FRM"):
with state.lock:
axis_index = int(state.variables.get("naxis", 0))
direction = float(state.variables.get("ndir", 0.0))
threading.Thread(
target=state.reference_axis, args=(axis_index, direction), daemon=True
).start()
return ":"
if routine == "#REFAX": # FUPR
threading.Thread(target=state.reference_axis, args=(0, 0.0), daemon=True).start()
return ":"
if routine == "#MOTON": # FUPR
state.axis(0).motor_on = True
state.axis(0).auto_motor_off = False
return ":"
if routine == "#STOP":
state.stop_all()
state.set_thread_active(1, 0.1)
return ":"
if routine == "#MNTMODE":
with state.lock:
state.variables["mntmod"] = 1.0
return ":"
if routine == "#POSMODE":
with state.lock:
if state.variables.get("mntprgs", 0.0) == 0.0:
state.variables["mntmod"] = 0.0
return ":"
if routine == "#GRGET":
state.start_transfer("getaprch")
return ":"
if routine == "#GRPUT":
state.start_transfer("mntaprch")
return ":"
if routine == "#GROPEN":
state.set_thread_active(4, 1.0)
state._set_gripper(True)
return ":"
if routine == "#GRCLOS":
state.set_thread_active(4, 1.0)
state._set_gripper(False)
return ":"
if routine == "#Tracker":
with state.lock:
if state.variables.get("tracken", 0.0) == 1.0 and state.all_axes_referenced():
state._tracker_running = True
return ":"
logger.warning(f"[sim galil] {self.host}:{self.port} unhandled XQ '{arg}'")
return ":"
def _pop_sim_kwargs(kwargs: dict) -> dict:
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),
}
class SimFlomniGalilMotor(FlomniGalilMotor):
"""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)
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
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)
csaxs_bec/devices/sim/sim_rt_flomni.py
+272
View File
@@ -0,0 +1,272 @@
"""
Simulated flOMNI RT (Orchestra) communication server.
Implements the TCP command set of `flOMNI/FeedbackLoop/tcpip_server/CommunicationServer.cpp`
used by `RtFlomniController`, `RtFlomniMotor` and `RtFlomniFlyer`. Reply formats and, in
particular, which commands are reply-less (`pa`, `sc`, `l0/l1/l3`, `a<val>`, `T0/T1`,
`v0/v1`, emitter setters) follow the real server to keep the command/reply stream in sync.
Simulated behavior:
- interferometer feedback and laser tracker as simple boolean states; tracker reports
on-target beam positions and healthy intensity when enabled
- position scans (`s...`, `sd`, `sr`, `r<i>`) advance with a configurable per-point dwell
time so that scan durations, progress updates and the flyer readout behave realistically
- sampled data rows contain the target positions plus small gaussian noise
"""
from __future__ import annotations
import random
import threading
import time
from bec_lib.logger import bec_logger
from csaxs_bec.devices.omny.rt.rt_flomni_ophyd import RtFlomniFlyer, RtFlomniMotor
from csaxs_bec.devices.sim.sim_socket import SimSocketBase, SimStateRegistry
logger = bec_logger.logger
_POSITION_NOISE_UM = 0.002 # ~2 nm rms positioning noise in the sampled data
class SimRtFlomniState:
"""Shared state of the simulated Orchestra server (one per host:port)."""
def __init__(self, host=None, port=None):
self.host = host
self.port = port
self.lock = threading.RLock()
self.targets = [0.0, 0.0, 0.0]
self.feedback_running = False
self.tracker_enabled = False
self.angle_rad = 0.0
self.pid_x_voltage = -0.062
self.emitter = {"ty": 0.0, "tz": 0.0, "thr_laser": 1000.0, "psd_low": 100.0}
self.ssi_signal = 12000.0
self.tracker_intensity = 30000.0
self.point_dwell_s = 0.02
self.scan_positions = []
self.scan_mode = 0
self.scan_current = 0
self._scan_thread = None
self._scan_abort = threading.Event()
def configure(self, **kwargs):
with self.lock:
if kwargs.get("point_dwell_s") is not None:
self.point_dwell_s = float(kwargs["point_dwell_s"])
if kwargs.get("pid_x_voltage") is not None:
self.pid_x_voltage = float(kwargs["pid_x_voltage"])
if kwargs.get("ssi_signal") is not None:
self.ssi_signal = float(kwargs["ssi_signal"])
# --- scan handling -----------------------------------------------------------
def clear_scan(self):
self._scan_abort.set()
if self._scan_thread is not None and self._scan_thread.is_alive():
self._scan_thread.join(timeout=2)
with self.lock:
self.scan_positions = []
self.scan_mode = 0
self.scan_current = 0
self.targets = [0.0, 0.0, self.targets[2]]
def add_scan_position(self, pos):
with self.lock:
self.scan_positions.append(pos)
return len(self.scan_positions)
def start_scan(self) -> int:
with self.lock:
num_pos = len(self.scan_positions)
if num_pos == 0 or self.scan_mode > 0:
return num_pos
self.scan_mode = 3
self.scan_current = 0
self._scan_abort.clear()
self._scan_thread = threading.Thread(target=self._scan_loop, daemon=True)
self._scan_thread.start()
return num_pos
def _scan_loop(self):
with self.lock:
self.scan_mode = 2
num_pos = len(self.scan_positions)
for index in range(num_pos):
if self._scan_abort.wait(self.point_dwell_s):
return
with self.lock:
self.scan_current = index + 1
self.targets = list(self.scan_positions[index])
with self.lock:
self.scan_mode = 0
def scan_status(self):
with self.lock:
return (self.scan_mode, len(self.scan_positions), self.scan_current)
def sample_row(self, index: int) -> str:
with self.lock:
if 0 <= index < len(self.scan_positions):
target = self.scan_positions[index]
else:
target = self.targets
noise = _POSITION_NOISE_UM
avg_x = target[0] + random.gauss(0.0, noise)
avg_y = target[1] + random.gauss(0.0, noise)
rotz = random.gauss(0.0, 0.1)
return (
f"{index}, 100, {target[0]:.5f}, {avg_x:.5f}, {abs(random.gauss(noise, noise / 4)):.5f},"
f" {target[1]:.5f}, {avg_y:.5f}, {abs(random.gauss(noise, noise / 4)):.5f},"
f" {rotz:.5f}, {abs(random.gauss(0.1, 0.02)):.5f}\n"
)
class SimRtFlomniSocket(SimSocketBase):
"""Simulated socket implementing the flOMNI Orchestra communication server protocol."""
state_cls = SimRtFlomniState
def handle_command(self, line: str): # noqa: C901
state: SimRtFlomniState = self.state
cmd, args = line[0], line[1:]
if cmd == "l":
if args.startswith("0"):
state.feedback_running = False
return None
if args.startswith("1") or args.startswith("3"):
state.feedback_running = True
return None
if args.startswith("2"):
return f"{0 if state.feedback_running else 1}\n"
return None
if cmd == "p":
if args.startswith("r"):
targets = state.targets
return f"{targets[0]:.5f},{targets[1]:.5f},{targets[2]:.5f}\n"
if args.startswith("a"):
axis_str, value_str = args[1:].split(",")
with state.lock:
state.targets[int(float(axis_str))] = float(value_str)
return None
values = [float(val) for val in args.split(",")]
with state.lock:
state.targets = values[:3]
return None
if cmd == "s":
if args.startswith("c"):
state.clear_scan()
return None
if args.startswith("r"):
mode, num_pos, current = state.scan_status()
return f"{mode:.5f},{num_pos:.5f},{current:.5f}\n"
if args.startswith("d"):
num_pos = state.start_scan()
return f"Scan started {num_pos:.0f} positions.\n"
if args.startswith("h"):
state.clear_scan()
return "0.00000, 0.00000, 0.00000\n"
count = state.add_scan_position([float(val) for val in args.split(",")])
return f"{count:.0f}\n"
if cmd == "r":
return state.sample_row(int(args))
if cmd == "a":
if args.startswith("r"):
return f"{state.angle_rad:f}\n"
state.angle_rad = float(args)
return None
if cmd == "T":
if args.startswith("s"):
return self._tracker_status()
state.tracker_enabled = args.startswith("1")
return None
if cmd == "e":
if args[:1] in ("y", "z", "t", "l"):
key = {"y": "ty", "z": "tz", "t": "thr_laser", "l": "psd_low"}[args[0]]
with state.lock:
state.emitter[key] = float(args[1:])
return None
emitter = state.emitter
return (
f"{emitter['ty']:.5f},{emitter['tz']:.5f},"
f"{emitter['thr_laser']:.2f},{emitter['psd_low']:.5f}\n"
)
if cmd == "g":
return f"{state.pid_x_voltage + random.gauss(0.0, 0.005):f}\n"
if cmd == "G":
return f"{random.gauss(0.0, 0.005):f}\n"
if cmd == "y":
return "3.000000\n"
if cmd == "w":
return "32.000000\n"
if cmd == "j":
return f"{state.ssi_signal:f}\n"
if cmd == "k":
axis = int(args)
return f"{state.targets[axis] if axis < 3 else 0.0:f}\n"
if cmd == "v":
return None
if cmd == "d":
return None
if cmd == "o":
# not present in the current real server; provided for API completeness
return "1\n"
if cmd == "t":
return "1"
logger.warning(f"[sim rt] {self.host}:{self.port} unhandled command '{line}'")
return None
def _tracker_status(self) -> str:
state: SimRtFlomniState = self.state
enabled = 1.0 if state.tracker_enabled else 0.0
intensity = state.tracker_intensity
threshold = state.emitter["thr_laser"]
target_z = state.emitter["tz"]
target_y = state.emitter["ty"]
return (
f"{target_z:.2f},{target_z:.2f},{intensity:.2f},{threshold:.2f},{5.0:.2f},"
f"{target_y:.2f},{target_y:.2f},{intensity:.2f},{threshold:.2f},{5.0:.2f},"
f"{enabled:.0f}\n"
)
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."""
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
)
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
)
csaxs_bec/devices/sim/sim_smaract.py
+208
View File
@@ -0,0 +1,208 @@
"""
Simulated Smaract MCS controller for the flOMNI OSA stages (fosax/fosay/fosaz).
Implements the ASCII command subset used by `SmaractController` / `SmaractMotor`.
Positions are handled in nanometers internally, matching the MCS protocol. Motion is
simulated with a constant closed-loop speed; `SCLS` updates it like on the real system.
"""
from __future__ import annotations
import re
import threading
import time
from bec_lib.logger import bec_logger
from csaxs_bec.devices.sim.sim_socket import SimSocketBase, SimStateRegistry
from csaxs_bec.devices.smaract.smaract_ophyd import SmaractMotor
logger = bec_logger.logger
_STATUS_STOPPED = 0
_STATUS_HOLDING = 3
_STATUS_TARGETING = 4
_OPEN_LOOP_NM_PER_STEP = 100.0 # coarse approximation for MST bursts
class _SimSmaractAxis:
def __init__(self, index: int):
self.index = index
self.connected = False
self.referenced = True
self.pos_nm = 0.0
self.speed_nm_s = 1.0e6 # 1 mm/s default
self.low_limit_nm = -1.0e9
self.high_limit_nm = 1.0e9
self._move_start_nm = 0.0
self._move_target_nm = 0.0
self._move_t0 = 0.0
self._move_t1 = 0.0
def position(self) -> float:
now = time.time()
if now >= self._move_t1:
return self.pos_nm
frac = (now - self._move_t0) / (self._move_t1 - self._move_t0)
return self._move_start_nm + frac * (self._move_target_nm - self._move_start_nm)
def is_moving(self) -> bool:
return time.time() < self._move_t1
def start_move(self, target_nm: float):
speed = max(self.speed_nm_s, 1.0)
current = self.position()
duration = abs(target_nm - current) / speed
self._move_start_nm = current
self._move_target_nm = target_nm
self._move_t0 = time.time()
self._move_t1 = self._move_t0 + duration
self.pos_nm = target_nm
def stop_move(self):
self.pos_nm = self.position()
self._move_t1 = 0.0
class SimSmaractState:
"""Shared state of one simulated Smaract MCS controller."""
N_AXES = 9
def __init__(self, host=None, port=None):
self.host = host
self.port = port
self.lock = threading.RLock()
self.axes = [_SimSmaractAxis(i) for i in range(self.N_AXES)]
def axis(self, index: int) -> _SimSmaractAxis:
return self.axes[index]
def configure_axis(self, index: int, **kwargs):
with self.lock:
ax = self.axes[index]
ax.connected = True
if kwargs.get("velocity") is not None:
ax.speed_nm_s = float(kwargs["velocity"]) * 1.0e6 # mm/s -> nm/s
if kwargs.get("referenced") is not None:
ax.referenced = bool(kwargs["referenced"])
if kwargs.get("initial_position") is not None:
ax.pos_nm = float(kwargs["initial_position"]) * 1.0e6 # mm -> nm
class SimSmaractSocket(SimSocketBase):
"""Simulated socket implementing the Smaract MCS ASCII protocol subset."""
state_cls = SimSmaractState
def handle_command(self, line: str): # noqa: C901
state: SimSmaractState = self.state
cmd = line.lstrip(":").strip()
match = re.match(r"^GS(\d+)$", cmd)
if match:
ax = state.axis(int(match.group(1)))
status = _STATUS_TARGETING if ax.is_moving() else _STATUS_HOLDING
return f":S{match.group(1)},{status}\n"
match = re.match(r"^GP(\d+)$", cmd)
if match:
ax = state.axis(int(match.group(1)))
return f":P{match.group(1)},{ax.position():.0f}\n"
match = re.match(r"^GPPK(\d+)$", cmd)
if match:
ax = state.axis(int(match.group(1)))
return f":PPK{match.group(1)},{1 if ax.referenced else 0}\n"
match = re.match(r"^MPA(\d+),(-?\d+),(\d+)$", cmd)
if match:
ax = state.axis(int(match.group(1)))
ax.start_move(float(match.group(2)))
return f":E{match.group(1)},0\n"
match = re.match(r"^MPR(\d+),(-?\d+),(\d+)$", cmd)
if match:
ax = state.axis(int(match.group(1)))
ax.start_move(ax.position() + float(match.group(2)))
return f":E{match.group(1)},0\n"
match = re.match(r"^MST(\d+),(-?\d+),(\d+),(\d+)$", cmd)
if match:
ax = state.axis(int(match.group(1)))
ax.start_move(ax.position() + float(match.group(2)) * _OPEN_LOOP_NM_PER_STEP)
return f":E{match.group(1)},0\n"
match = re.match(r"^FRM(\d+),(-?\d+),(\d+),(\d+)$", cmd)
if match:
ax = state.axis(int(match.group(1)))
ax.stop_move()
if int(match.group(4)): # autoZero
ax.pos_nm = 0.0
ax.referenced = True
return f":E{match.group(1)},0\n"
match = re.match(r"^S(\d+)$", cmd)
if match:
state.axis(int(match.group(1))).stop_move()
return f":E{match.group(1)},0\n"
match = re.match(r"^SCLS(\d+),(\d+)$", cmd)
if match:
state.axis(int(match.group(1))).speed_nm_s = float(match.group(2))
return f":E{match.group(1)},0\n"
match = re.match(r"^GCLS(\d+)$", cmd)
if match:
ax = state.axis(int(match.group(1)))
return f":CLS{match.group(1)},{ax.speed_nm_s:.0f}\n"
match = re.match(r"^GPL(\d+)$", cmd)
if match:
ax = state.axis(int(match.group(1)))
return f":GPL{match.group(1)},{ax.low_limit_nm:.0f},{ax.high_limit_nm:.0f}\n"
match = re.match(r"^SPL(\d+),(-?\d+),(-?\d+)$", cmd)
if match:
ax = state.axis(int(match.group(1)))
ax.low_limit_nm = float(match.group(2))
ax.high_limit_nm = float(match.group(3))
return f":E{match.group(1)},0\n"
match = re.match(r"^GST(\d+)$", cmd)
if match:
return f":ST{match.group(1)},1\n"
if cmd == "GCM":
return ":CM0\n"
if cmd == "GNC":
return f":N{SimSmaractState.N_AXES}\n"
if cmd == "GSI":
return ":ID4242\n"
if cmd == "GIV":
return ":IV1,0,0\n"
if cmd.startswith("SHE"):
return ":E-1,0\n"
if cmd == "R":
return ":E-1,0\n"
logger.warning(f"[sim smaract] {self.host}:{self.port} unhandled command '{cmd}'")
return ":E-1,0\n"
class SimSmaractMotor(SmaractMotor):
"""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)
csaxs_bec/devices/sim/sim_socket.py
+110
View File
@@ -0,0 +1,110 @@
"""
Core infrastructure for simulated controller sockets.
The socket-based controllers in ophyd_devices (`Controller.on`) create their transport via
``self._socket_cls(host=..., port=...)``. Any class exposing the `SocketIO` interface
(`put`, `receive`, `open`, `close`, `host`, `port`, `is_open`) can be injected instead of a
real TCP socket. The simulated sockets in this package implement the wire protocols of the
flOMNI hardware (Galil DMC, Smaract MCS, flOMNI Orchestra communication server) as small
state machines, so that the *real* device and controller classes can be used unchanged.
Commands are dispatched synchronously within `put()`; replies are appended to an internal
queue that `receive()` drains. Commands that do not produce a reply on the real hardware
must not enqueue one here, otherwise the command/reply stream desynchronizes.
Simulation state is shared per (host, port) via a registry, mirroring the singleton
behavior of `ophyd_devices.utils.controller.Controller`. This allows the simulated device
classes to seed axis parameters (initial position, velocity, resolution) at construction
time, before the controller opens its socket.
"""
from __future__ import annotations
import threading
from bec_lib.logger import bec_logger
logger = bec_logger.logger
class SimStateRegistry:
"""Registry of simulation states, shared per (host, port) like the controllers."""
_states = {}
_lock = threading.RLock()
@classmethod
def get(cls, state_cls, host, port):
"""Return the simulation state for (host, port), creating it if needed."""
key = (state_cls, str(host), int(port))
with cls._lock:
if key not in cls._states:
cls._states[key] = state_cls(host=host, port=port)
return cls._states[key]
@classmethod
def reset(cls):
"""Clear all simulation states (mainly for tests)."""
with cls._lock:
cls._states.clear()
class SimSocketBase:
"""
Drop-in replacement for `ophyd_devices.utils.socket.SocketIO`.
Children must set `state_cls` and implement `handle_command(line) -> str | None`.
A returned string is encoded and appended to the reply queue; None means no reply.
"""
state_cls = None
def __init__(self, host, port, socket_timeout: int = 2):
self.host = host
self.port = port
self.socket_timeout = socket_timeout
self.is_open = False
self._recv_buffer = []
self._cmd_buffer = b""
self._lock = threading.RLock()
self.state = SimStateRegistry.get(self.state_cls, host, port)
# --- SocketIO interface -------------------------------------------------
def open(self, timeout: int = 10):
logger.info(f"[sim] Connecting to simulated controller {self.host}:{self.port}.")
self.is_open = True
def connect(self, timeout: int = 10):
self.is_open = True
def close(self):
self.is_open = False
def put(self, msg: bytes):
with self._lock:
self._cmd_buffer += msg
# both \r (Galil) and \n (RT, Smaract) terminate commands
normalized = self._cmd_buffer.replace(b"\r", b"\n")
*lines, rest = normalized.split(b"\n")
self._cmd_buffer = rest
for line in lines:
line = line.decode(errors="replace").strip()
if not line:
continue
try:
reply = self.handle_command(line)
except Exception: # pylint: disable=broad-except
logger.exception(f"[sim] {self.host}:{self.port} failed to handle '{line}'")
reply = None
if reply is not None:
self._recv_buffer.append(reply.encode())
def receive(self, buffer_length=1024):
with self._lock:
if self._recv_buffer:
return self._recv_buffer.pop(0)
return b""
# --- to be implemented by children ---------------------------------------
def handle_command(self, line: str):
raise NotImplementedError