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

feat: add SmarAct MCS2 device driver
CI for csaxs_bec / test (push) Successful in 1m54s
Details

Browse Source 
Adds a new driver for the SmarAct MCS2 controller (MCS2-C-0008, 9 channels),
parallel to the existing MCS1 (SCU) implementation. Uses the MCS2's raw
ASCII/SCPI interface over TCP (port 55551), which differs from MCS1 in units
(picometers), message termination (\r\n), and error handling (a polled error
queue instead of inline echoes).

- csaxs_bec/devices/mcs2: Mcs2Controller/Mcs2Motor and errors
- csaxs_bec/devices/sim/sim_mcs2.py: simulated backend for testing
- tests/tests_devices/test_mcs2.py: unit tests + sim end-to-end test
- device_configs: commented example stage in bl_optics_hutch.yaml, and a
  standalone single-axis mcs2_config_test.yaml

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
Mirko Holler
2026-07-29 17:42:15 +02:00
co-authored by Claude Sonnet 5
parent 53f246a8d3
commit a65ab6cbdd
8 changed files with 1481 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
csaxs_bec/device_configs/bl_optics_hutch.yaml
+23
View File
@@ -848,3 +848,26 @@ kbhtrydw:
softwareTrigger: false
deviceTags:
- bl_optics_hutch
# Example (disabled): a single optics-hutch stage on the new SmarAct MCS2 controller
# (MCS2-C-0008, 9 channels). Uncomment and adjust host/axis_Id/limits/sign once the
# controller is racked and reachable on the network. See
# csaxs_bec/devices/mcs2/mcs2_ophyd.py for the full deviceConfig options.
# kbh_mcs2_example:
# description: "Example MCS2 stage (placeholder, not yet installed)"
# deviceClass: csaxs_bec.devices.mcs2.mcs2_ophyd.Mcs2Motor
# deviceConfig:
# axis_Id: A
# host: <mcs2-controller-hostname-or-ip>
# port: 55551
# limits:
# - -5
# - 5
# sign: 1
# enabled: false
# onFailure: raise
# readoutPriority: baseline
# readOnly: false
# softwareTrigger: false
# deviceTags:
# - bl_optics_hutch
csaxs_bec/device_configs/mcs2_config_test.yaml
+18
View File
@@ -0,0 +1,18 @@
mcs2test:
description: "Single-axis MCS2 test config (MCS2-C-0008, channel A)"
deviceClass: csaxs_bec.devices.mcs2.mcs2_ophyd.Mcs2Motor
deviceConfig:
axis_Id: A
host: <mcs2-controller-hostname-or-ip>
port: 55551
limits:
- -5
- 5
sign: 1
enabled: true
onFailure: raise
readoutPriority: baseline
readOnly: false
softwareTrigger: false
deviceTags:
- test_config
csaxs_bec/devices/mcs2/__init__.py
+2
View File
@@ -0,0 +1,2 @@
from .mcs2_controller import Mcs2Controller
from .mcs2_ophyd import Mcs2Motor
csaxs_bec/devices/mcs2/mcs2_controller.py
+443
View File
@@ -0,0 +1,443 @@
import enum
import logging
import time
import numpy as np
from ophyd_devices.utils.controller import Controller, axis_checked, retry_once, threadlocked
from prettytable import PrettyTable
from typeguard import typechecked
from csaxs_bec.devices.mcs2.mcs2_errors import Mcs2CommunicationError, Mcs2ErrorCode
logger = logging.getLogger("mcs2_controller")
class Mcs2ChannelType(enum.Enum):
STICK_SLIP_PIEZO_DRIVER = 1
MAGNETIC_DRIVER = 2
PIEZO_SCANNER_DRIVER = 3
class Mcs2MoveMode(enum.Enum):
CL_ABSOLUTE = 0
CL_RELATIVE = 1
SCAN_ABSOLUTE = 2
SCAN_RELATIVE = 3
STEP = 4
CL_WFG = 5
OL_WFG = 6
class Mcs2ChannelState(enum.IntFlag):
ACTIVELY_MOVING = 0x000001
CLOSED_LOOP_ACTIVE = 0x000002
CALIBRATING = 0x000004
REFERENCING = 0x000008
MOVE_DELAYED = 0x000010
SENSOR_PRESENT = 0x000020
IS_CALIBRATED = 0x000040
IS_REFERENCED = 0x000080
END_STOP_REACHED = 0x000100
RANGE_LIMIT_REACHED = 0x000200
FOLLOWING_LIMIT_REACHED = 0x000400
MOVEMENT_FAILED = 0x000800
IS_STREAMING = 0x001000
POSITIONER_OVERLOAD = 0x002000
OVER_TEMPERATURE = 0x004000
REFERENCE_MARK = 0x008000
IS_PHASED = 0x010000
POSITIONER_FAULT = 0x020000
AMPLIFIER_ENABLED = 0x040000
IN_POSITION = 0x080000
BRAKE_ENABLED = 0x100000
TACTILE_SENSING_TRIGGERED = 0x200000
MODULE_LINK_VALID = 0x400000
MODULE_LINK_CTRL = 0x800000
class Mcs2ReferencingOption(enum.IntFlag):
START_DIR = 0x0000001
REVERSE_DIR = 0x0000002
AUTO_ZERO = 0x0000004
ABORT_ON_ENDSTOP = 0x0000008
CONTINUE_ON_REF_FOUND = 0x0000010
STOP_ON_REF_FOUND = 0x0000020
INVALIDATE = 0x1000000
_MOVING_STATE_MASK = (
Mcs2ChannelState.ACTIVELY_MOVING
| Mcs2ChannelState.REFERENCING
| Mcs2ChannelState.CALIBRATING
| Mcs2ChannelState.MOVE_DELAYED
)
class Mcs2Controller(Controller):
"""Controller for the SmarAct MCS2, communicating via its raw ASCII/SCPI interface
over TCP (fixed port 55551, message termination <CR><LF>). See the MCS2 Programmer's
Guide, chapter 6 "ASCII Interface", for the protocol reference.
Unlike the MCS1 SCU protocol, set/move/action commands on the MCS2 never generate a
direct response. Errors are instead accumulated in a FIFO error queue that must be
polled explicitly (:SYST:ERR:NEXT?). Positions, velocities and accelerations are
exchanged in picometers (pm) rather than MCS1's nanometers.
"""
_axes_per_controller = 9
_initialized = False
PM_PER_MM = 1e9
USER_ACCESS = [
"query",
"command",
"mcs2_show_all",
"move_open_loop_steps",
"find_reference_mark",
"describe",
"axis_is_referenced",
"all_axes_referenced",
"set_closed_loop_move_speed",
"is_axis_moving",
"get_error_count",
"get_next_error",
"get_idn",
"print_command_history",
]
def __init__(
self,
*,
name="Mcs2Controller",
kind=None,
parent=None,
socket_cls=None,
socket_host=None,
socket_port=55551,
device_manager=None,
attr_name="",
labels=None,
):
if not self._initialized:
super().__init__(
name=name,
socket_cls=socket_cls,
socket_host=socket_host,
socket_port=socket_port,
device_manager=device_manager,
attr_name=attr_name,
parent=parent,
labels=labels,
kind=kind,
)
self._initialized = True
@threadlocked
def socket_put(self, cmd: str) -> None:
"""Send a raw command. ``cmd`` must already include its own leading ':' or '*'."""
self.command_history.append(f"[PUT]: {cmd}")
self.sock.put(f"{cmd}\r\n".encode())
@threadlocked
def _read_response(self) -> str:
"""Read a single response, waiting for the <CR><LF> terminator. Returns an
empty string if no response arrived within the timeout (this is expected for
set/move commands, and can also happen for malformed queries)."""
return_val = ""
max_wait_time = 1
elapsed_time = 0
sleep_time = 0.01
while True:
ret = self.socket_get()
return_val += ret
if ret.endswith("\r\n"):
break
time.sleep(sleep_time)
elapsed_time += sleep_time
if elapsed_time > max_wait_time:
break
return self._remove_trailing_characters(return_val)
def _remove_trailing_characters(self, var: str) -> str:
if len(var) > 1:
return var.split("\r\n")[0]
return var
@threadlocked
def _send_and_read(self, cmd: str) -> str:
self.socket_put(cmd)
return self._read_response()
def get_error_count(self) -> int:
"""Return the number of errors currently in the device's error queue."""
return int(self._send_and_read(":SYST:ERR:COUN?"))
def get_next_error(self) -> tuple[int, str]:
"""Pop and return the next (oldest) error from the device's error queue as
(code, message). code == 0 means the queue was empty ("No Error")."""
raw = self._send_and_read(":SYST:ERR:NEXT?")
code_str, _, message = raw.partition(",")
return int(code_str), message.strip().strip('"')
def _raise_if_error(self) -> None:
code, message = self.get_next_error()
if code != 0:
raise Mcs2ErrorCode(code, message=message)
@threadlocked
def command(self, cmd: str, check_error: bool = True) -> None:
"""Send a set/move/action command. These never produce a direct response on the
MCS2, so by default the error queue is polled immediately afterwards to detect
failures (there is no other way to know a command succeeded)."""
self.socket_put(cmd)
if check_error:
self._raise_if_error()
@threadlocked
def query(self, cmd: str) -> str:
"""Send a query (``cmd`` must end with '?') and return the bare value as a
string (quotes stripped for string properties). Numeric callers are expected to
convert the result themselves, e.g. ``int(controller.query(...))``."""
raw = self._send_and_read(cmd)
if raw == "":
code, message = self.get_next_error()
if code != 0:
raise Mcs2ErrorCode(code, message=f"No response received for '{cmd}'. {message}")
raise Mcs2CommunicationError(
f"No response received for '{cmd}' and no error queued. The last "
f"{self._command_history_length} commands were: {list(self.command_history)}"
)
if len(raw) > 1 and raw.startswith('"') and raw.endswith('"'):
return raw[1:-1]
return raw
@retry_once
@axis_checked
def get_position(self, axis_Id_numeric: int) -> float:
"""Return the current position of a channel, in mm."""
return int(self.query(f":CHAN{axis_Id_numeric}:POS?")) / self.PM_PER_MM
@retry_once
@axis_checked
def get_target_position(self, axis_Id_numeric: int) -> float:
"""Return the currently configured target position of a channel, in mm."""
return int(self.query(f":CHAN{axis_Id_numeric}:POS:TARG?")) / self.PM_PER_MM
@retry_once
@axis_checked
@typechecked
def move_axis_to_absolute_position(
self, axis_Id_numeric: int, target_val: float, hold_time: int = 1000
) -> None:
"""Instruct a channel to move to an absolute position.
Args:
axis_Id_numeric (int): Axis number.
target_val (float): Target position in mm.
hold_time (int, optional): Time in ms the position is actively held after
reaching the target. 0 disables the feature. Defaults to 1000.
"""
ax = axis_Id_numeric
self.command(f":CHAN{ax}:MMOD {Mcs2MoveMode.CL_ABSOLUTE.value}")
self.command(f":CHAN{ax}:HOLD {hold_time}")
self.command(f":MOVE{ax} {int(np.round(target_val * self.PM_PER_MM))}")
@retry_once
@axis_checked
@typechecked
def move_axis_to_relative_position(
self, axis_Id_numeric: int, target_val: float, hold_time: int = 1000
) -> None:
"""Instruct a channel to move by a relative distance.
Args:
axis_Id_numeric (int): Axis number.
target_val (float): Relative distance to move in mm.
hold_time (int, optional): Time in ms the position is actively held after
reaching the target. 0 disables the feature. Defaults to 1000.
"""
ax = axis_Id_numeric
self.command(f":CHAN{ax}:MMOD {Mcs2MoveMode.CL_RELATIVE.value}")
self.command(f":CHAN{ax}:HOLD {hold_time}")
self.command(f":MOVE{ax} {int(np.round(target_val * self.PM_PER_MM))}")
@retry_once
@axis_checked
@typechecked
def move_open_loop_steps(
self, axis_Id_numeric: int, steps: int, amplitude: int = 4000, frequency: int = 2000
) -> None:
"""Move open loop steps. Performs a burst of steps with the given parameters.
Args:
axis_Id_numeric (int): Axis number.
steps (int): Number and direction of steps to perform.
amplitude (int): Step amplitude, valid range 0..65535 (0 to 100V). Default: 4000.
frequency (int): Step frequency in Hz, valid range 1..20000. Default: 2000.
"""
ax = axis_Id_numeric
self.command(f":CHAN{ax}:STEP:FREQ {frequency}")
self.command(f":CHAN{ax}:STEP:AMPL {amplitude}")
self.command(f":CHAN{ax}:MMOD {Mcs2MoveMode.STEP.value}")
self.command(f":MOVE{ax} {steps}")
@retry_once
def stop_all_axes(self):
return [
self.command(f":STOP{ax.axis_Id_numeric}") for ax in self._axis if ax is not None
]
@retry_once
@axis_checked
def get_status(self, axis_Id_numeric: int) -> Mcs2ChannelState:
"""Return the current channel state bitmask of a channel."""
return Mcs2ChannelState(int(self.query(f":CHAN{axis_Id_numeric}:STAT?")))
@retry_once
@axis_checked
def is_axis_moving(self, axis_Id_numeric: int) -> bool:
"""Check if a channel is currently moving, referencing, calibrating, or delayed."""
return bool(self.get_status(axis_Id_numeric) & _MOVING_STATE_MASK)
@retry_once
@axis_checked
def axis_is_referenced(self, axis_Id_numeric: int) -> bool:
return bool(self.get_status(axis_Id_numeric) & Mcs2ChannelState.IS_REFERENCED)
def all_axes_referenced(self) -> bool:
return all(
self.axis_is_referenced(ax.axis_Id_numeric) for ax in self._axis if ax is not None
)
@retry_once
@axis_checked
def find_reference_mark(
self,
axis_Id_numeric: int,
direction: int = 0,
autoZero: int = 1,
velocity: float = 1.0,
acceleration: float = 10.0,
) -> None:
"""Start the referencing sequence for a channel.
Args:
axis_Id_numeric (int): Axis number.
direction (int): 0 = start/forward direction, 1 = reverse direction.
autoZero (int): If truthy, the position is set to zero once the reference
mark is found.
velocity (float): Referencing velocity in mm/s.
acceleration (float): Referencing acceleration in mm/s^2.
"""
ax = axis_Id_numeric
opt = Mcs2ReferencingOption(0)
if direction:
opt |= Mcs2ReferencingOption.REVERSE_DIR
if autoZero:
opt |= Mcs2ReferencingOption.AUTO_ZERO
self.command(f":CHAN{ax}:REF:OPT {int(opt)}")
self.command(f":CHAN{ax}:VEL {int(np.round(velocity * self.PM_PER_MM))}")
self.command(f":CHAN{ax}:ACC {int(np.round(acceleration * self.PM_PER_MM))}")
self.command(f":REF{ax}")
@retry_once
@axis_checked
def set_closed_loop_move_speed(self, axis_Id_numeric: int, move_speed: float) -> None:
"""Configure the move velocity used for closed-loop move commands.
Args:
axis_Id_numeric (int): Axis number.
move_speed (float): Movement speed in mm/s. 0 deactivates speed control.
"""
move_speed_in_pm_per_s = int(round(move_speed * self.PM_PER_MM))
if move_speed_in_pm_per_s < 0:
raise ValueError("Move speed must be positive.")
self.command(f":CHAN{axis_Id_numeric}:VEL {move_speed_in_pm_per_s}")
@retry_once
@axis_checked
def get_closed_loop_move_speed(self, axis_Id_numeric: int) -> float:
"""Return the currently configured move velocity, in mm/s."""
return int(self.query(f":CHAN{axis_Id_numeric}:VEL?")) / self.PM_PER_MM
@retry_once
@axis_checked
def get_position_limits(self, axis_Id_numeric: int) -> list:
"""Return the currently configured [low_limit, high_limit] in mm."""
ax = axis_Id_numeric
return [
int(self.query(f":CHAN{ax}:RLIM:MIN?")) / self.PM_PER_MM,
int(self.query(f":CHAN{ax}:RLIM:MAX?")) / self.PM_PER_MM,
]
@retry_once
@axis_checked
def set_position_limits(
self, axis_Id_numeric: int, low_limit: float, high_limit: float
) -> None:
"""Limit the travel range of a channel by software, in mm."""
ax = axis_Id_numeric
self.command(f":CHAN{ax}:RLIM:MIN {int(np.round(low_limit * self.PM_PER_MM))}")
self.command(f":CHAN{ax}:RLIM:MAX {int(np.round(high_limit * self.PM_PER_MM))}")
@retry_once
@axis_checked
def get_positioner_type(self, axis_Id_numeric: int) -> str:
"""Return the human-readable positioner type name reported by the device."""
return self.query(f":CHAN{axis_Id_numeric}:PTYP:NAME?")
@retry_once
@axis_checked
def get_channel_type(self, axis_Id_numeric: int) -> Mcs2ChannelType:
return Mcs2ChannelType(int(self.query(f":CHAN{axis_Id_numeric}:TYPE?")))
@retry_once
def get_number_of_channels(self) -> int:
"""Return the number of channels available on the device."""
return int(self.query(":DEV:NOCH?"))
@retry_once
def get_system_id(self) -> str:
"""Return the device's serial number, e.g. 'MCS2-00001234'."""
return self.query(":DEV:SNUM?")
@retry_once
def get_idn(self) -> str:
"""Return the device identification string (manufacturer;serial;name;firmware date)."""
return self.query("*IDN?")
def factory_reset(self) -> None:
"""Reset the device to factory defaults and restart it.
WARNING: this wipes ALL device configuration (positioner types, calibration,
limits, ...) and requires reconnecting afterwards. Unlike MCS1's convenience
``reset()``, this is deliberately NOT exposed via USER_ACCESS and must only be
called interactively with full awareness of the consequences.
"""
self.command(":SYST:DEF", check_error=False)
def describe(self) -> None:
t = PrettyTable()
t.title = f"{self.__class__.__name__} on {self.sock.host}:{self.sock.port}"
t.field_names = ["Axis", "Name", "Connected", "Referenced", "Closed Loop Speed", "Position"]
for ax in range(self._axes_per_controller):
axis = self._axis[ax]
if axis is not None:
t.add_row(
[
f"{axis.axis_Id_numeric}/{axis.axis_Id}",
axis.name,
axis.connected,
self.axis_is_referenced(axis.axis_Id_numeric),
self.get_closed_loop_move_speed(axis.axis_Id_numeric),
axis.readback.read().get(axis.name).get("value"),
]
)
else:
t.add_row([None for _ in t.field_names])
print(t)
def mcs2_show_all(self) -> None:
for controller in self._controller_instances.values():
if isinstance(controller, Mcs2Controller):
controller.describe()
csaxs_bec/devices/mcs2/mcs2_errors.py
+68
View File
@@ -0,0 +1,68 @@
# SCPI-level errors (negative codes), from the MCS2 Programmer's Guide, table 6.5 "SCPI Error Codes"
SCPI_ERROR_CODES = {
0: "No error",
-101: "Invalid character",
-103: "Invalid separator",
-104: "Data type error",
-108: "Parameter not allowed",
-109: "Missing parameter",
-113: "Undefined header",
-151: "Invalid string data",
-350: "Queue overflow",
-363: "Input buffer overrun",
}
# SmarAct-specific errors (positive codes), from the MCS2 Programmer's Guide, appendix A.1
# "Error Codes". Not exhaustive; limited to the codes relevant to channel motion, referencing,
# calibration, limits, and amplifier/sensor state.
MCS2_ERROR_CODES = {
0x0012: "INVALID_KEY: The given property key could not be resolved.",
0x0013: "INVALID_PARAMETER: The passed parameter is not in the valid range.",
0x0016: "INVALID_DATA_TYPE: The data type of a parameter is invalid.",
0x0020: "INVALID_DEVICE_INDEX: An invalid device index has been passed.",
0x0021: "INVALID_MODULE_INDEX: An invalid module index has been passed.",
0x0022: "INVALID_CHANNEL_INDEX: An invalid channel index has been passed.",
0x0023: "PERMISSION_DENIED: The request cannot be processed due to an access violation.",
0x0025: "MOVEMENT_LOCKED: The command cannot be processed due to movements being locked.",
0x0100: "NO_HM_PRESENT: The command could not be processed because no Hand-Control-Module is present.",
0x0102: "NO_SM_PRESENT: The command could not be processed because no Sensor-Module is present.",
0x0103: "NO_SENSOR_PRESENT: The command could not be processed because no sensor is present.",
0x0104: "SENSOR_DISABLED: The command could not be processed because the sensor is disabled.",
0x0105: "POWER_SUPPLY_DISABLED: The command could not be processed because the power supply is disabled.",
0x0106: "AMPLIFIER_DISABLED: The command could not be processed because the amplifier is disabled.",
0x0110: "MOVEMENT_OVERRIDDEN: A software movement was interrupted by the Hand Control Module or vice versa.",
0x0111: "NOT_CALIBRATED: The command could not be processed because the channel is not calibrated.",
0x0112: "NOT_REFERENCED: The command could not be processed because the channel is not referenced.",
0x0118: "NOT_PHASED: The command could not be processed because the channel is not phased.",
0x0119: "POSITIONER_FAULT: The command could not be processed because the channel detected a positioner fault.",
0x0123: "AMPLIFIER_LOCKED: The command cannot be processed because the amplifier is disabled and locked.",
0x0124: "WRITE_ACCESS_LOCKED: The command cannot be processed because the channel is locked for write access.",
0x0150: "BUSY_MOVING: The command could not be processed because the channel is currently moving.",
0x0151: "BUSY_CALIBRATING: The command could not be processed because the channel is currently calibrating.",
0x0152: "BUSY_REFERENCING: The command could not be processed because the channel is currently referencing.",
0x0200: "END_STOP_REACHED: An endstop was detected.",
0x0201: "FOLLOWING_ERR_LIMIT: The following error exceeded the configured limit.",
0x0202: "RANGE_LIMIT_REACHED: A configured position limit was hit.",
0x0203: "POSITIONER_OVERLOAD: The channel detected an overload condition of the positioner.",
0x0206: "OVER_TEMPERATURE: An over-temperature condition of a driver amplifier was detected.",
}
class Mcs2Error(Exception):
pass
class Mcs2CommunicationError(Mcs2Error):
pass
class Mcs2ErrorCode(Mcs2Error):
def __init__(self, error_code: int, message: str = ""):
self.error_code = error_code
table = SCPI_ERROR_CODES if error_code < 0 else MCS2_ERROR_CODES
self.error_code_message = table.get(error_code, "UNKNOWN ERROR")
self.message = message
super().__init__(self.message)
def __str__(self):
return f"{self.error_code} / {self.error_code_message}. {self.message}"
csaxs_bec/devices/mcs2/mcs2_ophyd.py
+298
View File
@@ -0,0 +1,298 @@
import logging
import threading
import time
import numpy as np
from bec_lib import bec_logger
from ophyd import Component as Cpt
from ophyd import Device, PositionerBase, Signal
from ophyd.status import wait as status_wait
from ophyd.utils import LimitError, ReadOnlyError
from ophyd_devices.utils.controller import threadlocked
from ophyd_devices.utils.socket import SocketIO, SocketSignal, raise_if_disconnected
from csaxs_bec.devices.mcs2.mcs2_controller import Mcs2Controller
from csaxs_bec.devices.mcs2.mcs2_errors import Mcs2Error
logger = bec_logger.logger
class Mcs2SignalBase(SocketSignal):
def __init__(self, signal_name, **kwargs):
self.signal_name = signal_name
super().__init__(**kwargs)
self.controller = self.parent.controller
self.sock = self.parent.controller.sock
class Mcs2SignalRO(Mcs2SignalBase):
def __init__(self, signal_name, **kwargs):
super().__init__(signal_name, **kwargs)
self._metadata["write_access"] = False
@threadlocked
def _socket_set(self, val):
raise ReadOnlyError("Read-only signals cannot be set")
class Mcs2ReadbackSignal(Mcs2SignalRO):
@threadlocked
def _socket_get(self):
return self.controller.get_position(self.parent.axis_Id_numeric) * self.parent.sign
class Mcs2SetpointSignal(Mcs2SignalBase):
def __init__(self, signal_name, **kwargs):
super().__init__(signal_name, **kwargs)
self.setpoint = 0
@threadlocked
def _socket_get(self):
return self.setpoint
@threadlocked
def _socket_set(self, val):
target_val = val * self.parent.sign
self.setpoint = target_val
if self.controller.axis_is_referenced(self.parent.axis_Id_numeric):
self.controller.move_axis_to_absolute_position(self.parent.axis_Id_numeric, target_val)
else:
raise Mcs2Error(f"Axis {self.parent.axis_Id_numeric} is not referenced.")
class Mcs2MotorIsMoving(Mcs2SignalRO):
@threadlocked
def _socket_get(self):
return self.controller.is_axis_moving(self.parent.axis_Id_numeric)
class Mcs2AxisReferenced(Mcs2SignalRO):
@threadlocked
def _socket_get(self):
return self.parent.controller.axis_is_referenced(self.parent.axis_Id_numeric)
class Mcs2Motor(Device, PositionerBase):
USER_ACCESS = ["controller"]
readback = Cpt(Mcs2ReadbackSignal, signal_name="readback", kind="hinted")
user_setpoint = Cpt(Mcs2SetpointSignal, signal_name="setpoint")
motor_is_moving = Cpt(Mcs2MotorIsMoving, signal_name="motor_is_moving", kind="normal")
high_limit_travel = Cpt(Signal, value=0, kind="omitted")
low_limit_travel = Cpt(Signal, value=0, kind="omitted")
SUB_READBACK = "readback"
SUB_CONNECTION_CHANGE = "connection_change"
_default_sub = SUB_READBACK
def __init__(
self,
axis_Id,
prefix="",
*,
name,
kind=None,
read_attrs=None,
configuration_attrs=None,
parent=None,
host,
port=55551,
limits=None,
sign=1,
socket_cls=SocketIO,
device_manager=None,
**kwargs,
):
self.controller = Mcs2Controller(
socket_cls=socket_cls, socket_host=host, socket_port=port, device_manager=device_manager
)
self.axis_Id = axis_Id
self.sign = sign
self.controller.set_axis(axis=self, axis_nr=self.axis_Id_numeric)
self.tolerance = kwargs.pop("tolerance", 0.5)
super().__init__(
prefix,
name=name,
kind=kind,
read_attrs=read_attrs,
configuration_attrs=configuration_attrs,
parent=parent,
**kwargs,
)
self.readback.name = self.name
self.controller.subscribe(
self._update_connection_state, event_type=self.SUB_CONNECTION_CHANGE
)
self._update_connection_state()
if limits is not None:
assert len(limits) == 2
self.low_limit_travel.put(limits[0])
self.high_limit_travel.put(limits[1])
def wait_for_connection(self, timeout: float = 30.0) -> bool:
self.controller.on(timeout=timeout)
self._update_setpoint_from_readback()
def _update_setpoint_from_readback(self):
"""
The setpoint is only stored locally. After a restart,
we need to update it to match the current readback value.
"""
self.user_setpoint.setpoint = self.readback.get()
def destroy(self):
"""Make sure to turn off the controller socket on destroy."""
self.controller.off(update_config=False)
return super().destroy()
@property
def limits(self):
return (self.low_limit_travel.get(), self.high_limit_travel.get())
@property
def low_limit(self):
return self.limits[0]
@property
def high_limit(self):
return self.limits[1]
def check_value(self, pos):
"""Check that the position is within the soft limits"""
low_limit, high_limit = self.limits
if low_limit < high_limit and not (low_limit <= pos <= high_limit):
raise LimitError(f"position={pos} not within limits {self.limits}")
def _update_connection_state(self, **kwargs):
for walk in self.walk_signals():
walk.item._metadata["connected"] = self.controller.connected
@raise_if_disconnected
def move(self, position, wait=True, **kwargs):
"""Move to a specified position, optionally waiting for motion to
complete.
Parameters
----------
position
Position to move to
moved_cb : callable
Call this callback when movement has finished. This callback must
accept one keyword argument: 'obj' which will be set to this
positioner instance.
timeout : float, optional
Maximum time to wait for the motion. If None, the default timeout
for this positioner is used.
Returns
-------
status : MoveStatus
Raises
------
TimeoutError
When motion takes longer than `timeout`
ValueError
On invalid positions
RuntimeError
If motion fails other than timing out
"""
self._started_moving = False
timeout = kwargs.pop("timeout", 4)
status = super().move(position, timeout=timeout, **kwargs)
self.user_setpoint.put(position, wait=False)
def move_and_finish():
while self.motor_is_moving.get():
val = self.readback.read()
self._run_subs(sub_type=self.SUB_READBACK, value=val, timestamp=time.time())
time.sleep(0.1)
val = self.readback.read()
success = np.isclose(val[self.name]["value"], position, atol=self.tolerance)
self._done_moving(success=success)
threading.Thread(target=move_and_finish, daemon=True).start()
try:
if wait:
status_wait(status)
except KeyboardInterrupt:
self.stop()
raise
return status
@property
def axis_Id(self):
return self._axis_Id_alpha
@axis_Id.setter
def axis_Id(self, val):
if isinstance(val, str):
if len(val) != 1:
raise ValueError(f"Only single-character axis_Ids are supported.")
self._axis_Id_alpha = val
self._axis_Id_numeric = ord(val.lower()) - 97
else:
raise TypeError(f"Expected value of type str but received {type(val)}")
@property
def axis_Id_numeric(self):
return self._axis_Id_numeric
@axis_Id_numeric.setter
def axis_Id_numeric(self, val):
if isinstance(val, int):
if val > 26:
raise ValueError(f"Numeric value exceeds supported range.")
self._axis_Id_alpha = val
self._axis_Id_numeric = (chr(val + 97)).capitalize()
else:
raise TypeError(f"Expected value of type int but received {type(val)}")
@property
def egu(self):
"""The engineering units (EGU) for positions"""
return "mm"
def stage(self) -> list[object]:
return super().stage()
def unstage(self) -> list[object]:
return super().unstage()
def stop(self, *, success=False):
self.controller.stop_all_axes()
return super().stop(success=success)
if __name__ == "__main__":
logging.basicConfig(level=logging.DEBUG)
mock = False
if not mock:
lmcs2A = Mcs2Motor("A", name="lmcs2A", host="mpc____.psi.ch", port=55551, sign=1)
lmcs2A.stage()
lmcs2A.read()
lmcs2A.get()
lmcs2A.describe()
lmcs2A.unstage()
lmcs2A.controller.off()
else:
from ophyd_devices.utils.socket import SocketMock
lmcs2A = Mcs2Motor(
"A", name="lmcs2A", host="mpc____.psi.ch", port=55551, sign=1, socket_cls=SocketMock
)
lmcs2A.stage()
lmcs2A.read()
lmcs2A.get()
lmcs2A.describe()
lmcs2A.unstage()
lmcs2A.controller.off()
csaxs_bec/devices/sim/sim_mcs2.py
+312
View File
@@ -0,0 +1,312 @@
"""
Simulated SmarAct MCS2 controller.
Implements the ASCII/SCPI command subset used by `Mcs2Controller` / `Mcs2Motor`.
Positions are handled in picometers internally, matching the MCS2 protocol. Motion is
simulated with a constant closed-loop speed, and (unlike the MCS1 simulator) an error
queue is modeled so that unhandled/invalid commands surface through
`:SYST:ERR:NEXT?`, matching real MCS2 error-handling behavior.
"""
from __future__ import annotations
import re
import threading
import time
from bec_lib.logger import bec_logger
from csaxs_bec.devices.mcs2.mcs2_ophyd import Mcs2Motor
from csaxs_bec.devices.sim.sim_socket import SimSocketBase, SimStateRegistry
logger = bec_logger.logger
_STATE_ACTIVELY_MOVING = 0x000001
_STATE_IS_CALIBRATED = 0x000040
_STATE_IS_REFERENCED = 0x000080
_STATE_SENSOR_PRESENT = 0x000020
_MOVE_MODE_CL_RELATIVE = 1
_MOVE_MODE_STEP = 4
_STEP_PM_PER_STEP = 1.0e5 # coarse approximation for open-loop step bursts
class _SimMcs2Axis:
def __init__(self, index: int):
self.index = index
self.connected = False
self.referenced = True
self.calibrated = True
self.move_mode = 0 # CL_ABSOLUTE
self.pos_pm = 0.0
self.speed_pm_s = 1.0e9 # 1 mm/s default
self.hold_time_ms = 0
self.low_limit_pm = -1.0e12
self.high_limit_pm = 1.0e12
self._move_start_pm = 0.0
self._move_target_pm = 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_pm
frac = (now - self._move_t0) / (self._move_t1 - self._move_t0)
return self._move_start_pm + frac * (self._move_target_pm - self._move_start_pm)
def is_moving(self) -> bool:
return time.time() < self._move_t1
def start_move(self, target_pm: float):
speed = max(self.speed_pm_s, 1.0)
current = self.position()
duration = abs(target_pm - current) / speed
self._move_start_pm = current
self._move_target_pm = target_pm
self._move_t0 = time.time()
self._move_t1 = self._move_t0 + duration
self.pos_pm = target_pm
def stop_move(self):
self.pos_pm = self.position()
self._move_t1 = 0.0
class SimMcs2State:
"""Shared state of one simulated MCS2 controller."""
N_AXES = 9
def __init__(self, host=None, port=None):
self.host = host
self.port = port
self.lock = threading.RLock()
self.axes = [_SimMcs2Axis(i) for i in range(self.N_AXES)]
self.error_queue: list[tuple[int, str]] = []
def axis(self, index: int) -> _SimMcs2Axis:
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_pm_s = float(kwargs["velocity"]) * 1.0e9 # mm/s -> pm/s
if kwargs.get("referenced") is not None:
ax.referenced = bool(kwargs["referenced"])
if kwargs.get("initial_position") is not None:
ax.pos_pm = float(kwargs["initial_position"]) * 1.0e9 # mm -> pm
def push_error(self, code: int, message: str):
with self.lock:
if len(self.error_queue) >= 32:
self.error_queue[-1] = (-350, "Queue overflow")
return
self.error_queue.append((code, message))
def pop_error(self) -> tuple[int, str]:
with self.lock:
if self.error_queue:
return self.error_queue.pop(0)
return (0, "No Error")
class SimMcs2Socket(SimSocketBase):
"""Simulated socket implementing the MCS2 ASCII/SCPI protocol subset."""
state_cls = SimMcs2State
def handle_command(self, line: str): # noqa: C901
state: SimMcs2State = self.state
cmd = line.lstrip(":").strip()
if cmd == "*IDN?":
return '"SmarAct;MCS2-00000000;SimMCS2;01/01/24"\r\n'
if cmd == "DEV:NOCH?":
return f"{state.N_AXES}\r\n"
if cmd == "DEV:SNUM?":
return '"MCS2-00000000"\r\n'
if cmd == "DEV:ITYP?":
return "2\r\n"
if cmd == "DEV:STAT?":
return "0\r\n"
if cmd == "SYST:ERR:COUN?":
return f"{len(state.error_queue)}\r\n"
if cmd == "SYST:ERR:NEXT?":
code, message = state.pop_error()
return f'{code},"{message}"\r\n'
if cmd == "SYST:DEF":
return None
match = re.match(r"^CHAN(\d+):POS\?$", cmd)
if match:
ax = state.axis(int(match.group(1)))
return f"{ax.position():.0f}\r\n"
match = re.match(r"^CHAN(\d+):POS:TARG\?$", cmd)
if match:
ax = state.axis(int(match.group(1)))
return f"{ax._move_target_pm:.0f}\r\n"
match = re.match(r"^CHAN(\d+):STAT\?$", cmd)
if match:
ax = state.axis(int(match.group(1)))
bits = _STATE_SENSOR_PRESENT
if ax.is_moving():
bits |= _STATE_ACTIVELY_MOVING
if ax.referenced:
bits |= _STATE_IS_REFERENCED
if ax.calibrated:
bits |= _STATE_IS_CALIBRATED
return f"{bits}\r\n"
match = re.match(r"^CHAN(\d+):MMOD (\d+)$", cmd)
if match:
state.axis(int(match.group(1))).move_mode = int(match.group(2))
return None
match = re.match(r"^CHAN(\d+):VEL\?$", cmd)
if match:
ax = state.axis(int(match.group(1)))
return f"{ax.speed_pm_s:.0f}\r\n"
match = re.match(r"^CHAN(\d+):VEL (\d+)$", cmd)
if match:
state.axis(int(match.group(1))).speed_pm_s = float(match.group(2))
return None
match = re.match(r"^CHAN(\d+):ACC (\d+)$", cmd)
if match:
return None # acceleration is accepted but not modeled kinematically
match = re.match(r"^CHAN(\d+):HOLD (\d+)$", cmd)
if match:
state.axis(int(match.group(1))).hold_time_ms = int(match.group(2))
return None
match = re.match(r"^CHAN(\d+):RLIM:MIN\?$", cmd)
if match:
ax = state.axis(int(match.group(1)))
return f"{ax.low_limit_pm:.0f}\r\n"
match = re.match(r"^CHAN(\d+):RLIM:MIN (-?\d+)$", cmd)
if match:
state.axis(int(match.group(1))).low_limit_pm = float(match.group(2))
return None
match = re.match(r"^CHAN(\d+):RLIM:MAX\?$", cmd)
if match:
ax = state.axis(int(match.group(1)))
return f"{ax.high_limit_pm:.0f}\r\n"
match = re.match(r"^CHAN(\d+):RLIM:MAX (-?\d+)$", cmd)
if match:
state.axis(int(match.group(1))).high_limit_pm = float(match.group(2))
return None
match = re.match(r"^CHAN(\d+):PTYP:NAME\?$", cmd)
if match:
return '"Simulated Positioner"\r\n'
match = re.match(r"^CHAN(\d+):TYPE\?$", cmd)
if match:
return "1\r\n"
match = re.match(r"^CHAN(\d+):REF:OPT (\d+)$", cmd)
if match:
return None
match = re.match(r"^CHAN(\d+):CAL:OPT (\d+)$", cmd)
if match:
return None
match = re.match(r"^CHAN(\d+):STEP:FREQ (\d+)$", cmd)
if match:
return None
match = re.match(r"^CHAN(\d+):STEP:AMPL (\d+)$", cmd)
if match:
return None
match = re.match(r"^REF(\d+)$", cmd)
if match:
ax = state.axis(int(match.group(1)))
ax.stop_move()
ax.referenced = True
return None
match = re.match(r"^CAL(\d+)$", cmd)
if match:
state.axis(int(match.group(1))).calibrated = True
return None
match = re.match(r"^MOVE(\d+) (-?\d+)$", cmd)
if match:
ax = state.axis(int(match.group(1)))
value = float(match.group(2))
if ax.move_mode == _MOVE_MODE_CL_RELATIVE:
ax.start_move(ax.position() + value)
elif ax.move_mode == _MOVE_MODE_STEP:
ax.start_move(ax.position() + value * _STEP_PM_PER_STEP)
else:
ax.start_move(value)
return None
match = re.match(r"^STOP(\d+)$", cmd)
if match:
state.axis(int(match.group(1))).stop_move()
return None
logger.warning(f"[sim mcs2] {self.host}:{self.port} unhandled command '{cmd}'")
state.push_error(-113, "Undefined header")
return None
class SimMcs2Motor(Mcs2Motor):
"""Mcs2Motor connected to the simulated MCS2 controller.
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="mpc____.psi.ch",
port=55551,
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=SimMcs2Socket,
device_manager=device_manager,
**kwargs,
)
state = SimStateRegistry.get(SimMcs2State, 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/tests_devices/test_mcs2.py
+317
View File
@@ -0,0 +1,317 @@
from unittest import mock
import pytest
from ophyd_devices.tests.utils import SocketMock
from csaxs_bec.devices.mcs2 import Mcs2Controller
from csaxs_bec.devices.mcs2.mcs2_controller import Mcs2ChannelType
from csaxs_bec.devices.mcs2.mcs2_errors import Mcs2CommunicationError, Mcs2ErrorCode
from csaxs_bec.devices.mcs2.mcs2_ophyd import Mcs2Motor
from csaxs_bec.devices.sim.sim_mcs2 import SimMcs2Motor
from csaxs_bec.devices.sim.sim_socket import SimStateRegistry
NO_ERROR = b'0,"No Error"\r\n'
@pytest.fixture
def controller(dm_with_devices):
Mcs2Controller._reset_controller()
controller = Mcs2Controller(
socket_cls=SocketMock, socket_host="dummy", socket_port=55551, device_manager=dm_with_devices
)
controller.on()
controller.sock.flush_buffer()
yield controller
@pytest.fixture
def lmcs2A(dm_with_devices):
Mcs2Controller._reset_controller()
motor_a = Mcs2Motor(
"A",
name="lmcs2A",
host="dummy",
port=55551,
sign=1,
socket_cls=SocketMock,
device_manager=dm_with_devices,
)
motor_a.controller.on()
motor_a.controller.sock.flush_buffer()
motor_a.stage()
yield motor_a
@pytest.mark.parametrize(
"axis,position,get_message,return_msg",
[
(0, 50, b":CHAN0:POS?\r\n", b"50000000000\r\n"),
(1, 0, b":CHAN1:POS?\r\n", b"0\r\n"),
(0, -50, b":CHAN0:POS?\r\n", b"-50000000000\r\n"),
(0, -50.23, b":CHAN0:POS?\r\n", b"-50230000000\r\n"),
],
)
def test_get_position(controller, axis, position, get_message, return_msg):
controller.sock.buffer_recv = return_msg
val = controller.get_position(axis)
assert val == position
assert controller.sock.buffer_put[0] == get_message
@pytest.mark.parametrize(
"axis,is_referenced,get_message,return_msg,exception",
[
(0, True, b":CHAN0:STAT?\r\n", b"128\r\n", None),
(1, True, b":CHAN1:STAT?\r\n", b"129\r\n", None),
(0, False, b":CHAN0:STAT?\r\n", b"0\r\n", None),
(200, False, b":CHAN0:STAT?\r\n", b"0\r\n", ValueError),
],
)
def test_axis_is_referenced(controller, axis, is_referenced, get_message, return_msg, exception):
controller.sock.buffer_recv = return_msg
if exception is not None:
with pytest.raises(exception):
controller.axis_is_referenced(axis)
else:
val = controller.axis_is_referenced(axis)
assert val == is_referenced
assert controller.sock.buffer_put[0] == get_message
@pytest.mark.parametrize(
"is_moving,get_message,return_msg",
[
(False, b":CHAN0:STAT?\r\n", b"0\r\n"),
(True, b":CHAN0:STAT?\r\n", b"1\r\n"), # ACTIVELY_MOVING
(True, b":CHAN0:STAT?\r\n", b"8\r\n"), # REFERENCING
(True, b":CHAN0:STAT?\r\n", b"4\r\n"), # CALIBRATING
(True, b":CHAN0:STAT?\r\n", b"16\r\n"), # MOVE_DELAYED
(False, b":CHAN0:STAT?\r\n", b"32\r\n"), # SENSOR_PRESENT only
(False, b":CHAN0:STAT?\r\n", b"128\r\n"), # IS_REFERENCED only
(True, b":CHAN0:STAT?\r\n", b"129\r\n"), # ACTIVELY_MOVING | IS_REFERENCED
],
)
def test_is_axis_moving(controller, is_moving, get_message, return_msg):
controller.sock.buffer_recv = return_msg
val = controller.is_axis_moving(0)
assert val == is_moving
assert controller.sock.buffer_put[0] == get_message
def test_command_raises_scpi_error_code(controller):
controller.sock.buffer_recv = b'-101,"Invalid character"\r\n'
with pytest.raises(Mcs2ErrorCode) as exc_info:
controller.command(":BOGUS")
assert exc_info.value.error_code == -101
def test_command_raises_mcs2_error_code(controller):
controller.sock.buffer_recv = b'514,"Range limit reached"\r\n'
with pytest.raises(Mcs2ErrorCode) as exc_info:
controller.command(":MOVE0 1000000000000")
assert exc_info.value.error_code == 514
def test_command_no_error_does_not_raise(controller):
controller.sock.buffer_recv = NO_ERROR
controller.command(":STOP0")
assert controller.sock.buffer_put == [b":STOP0\r\n", b":SYST:ERR:NEXT?\r\n"]
def test_query_timeout_without_error_raises_communication_error(controller):
# _send_and_read is patched directly (rather than driving the real 1s socket
# timeout loop via buffer_recv) to keep this test fast and deterministic: the
# first call simulates the original query timing out (empty response), the
# second simulates the subsequent error-queue poll finding nothing queued.
with mock.patch.object(controller, "_send_and_read", side_effect=["", '0,"No Error"']):
with pytest.raises(Mcs2CommunicationError):
controller.query(":CHAN0:POS?")
def test_query_timeout_with_queued_error_raises_error_code(controller):
with mock.patch.object(
controller, "_send_and_read", side_effect=["", '-113,"Undefined header"']
):
with pytest.raises(Mcs2ErrorCode) as exc_info:
controller.query(":CHAN0:BOGUS?")
assert exc_info.value.error_code == -113
@pytest.mark.parametrize(
"pos,axis,hold_time,get_msg",
[
(50, 0, None, [b":CHAN0:MMOD 0\r\n", b":CHAN0:HOLD 1000\r\n", b":MOVE0 50000000000\r\n"]),
(0, 0, 800, [b":CHAN0:MMOD 0\r\n", b":CHAN0:HOLD 800\r\n", b":MOVE0 0\r\n"]),
(20.23, 1, None, [b":CHAN1:MMOD 0\r\n", b":CHAN1:HOLD 1000\r\n", b":MOVE1 20230000000\r\n"]),
],
)
def test_move_axis_to_absolute_position(controller, pos, axis, hold_time, get_msg):
controller.sock.buffer_recv = [NO_ERROR, NO_ERROR, NO_ERROR]
if hold_time is not None:
controller.move_axis_to_absolute_position(axis, pos, hold_time=hold_time)
else:
controller.move_axis_to_absolute_position(axis, pos)
expected = []
for msg in get_msg:
expected.append(msg)
expected.append(b":SYST:ERR:NEXT?\r\n")
assert controller.sock.buffer_put == expected
def test_move_open_loop_steps(controller):
controller.sock.buffer_recv = [NO_ERROR, NO_ERROR, NO_ERROR, NO_ERROR]
controller.move_open_loop_steps(0, 500, amplitude=65535, frequency=1000)
assert controller.sock.buffer_put == [
b":CHAN0:STEP:FREQ 1000\r\n",
b":SYST:ERR:NEXT?\r\n",
b":CHAN0:STEP:AMPL 65535\r\n",
b":SYST:ERR:NEXT?\r\n",
b":CHAN0:MMOD 4\r\n",
b":SYST:ERR:NEXT?\r\n",
b":MOVE0 500\r\n",
b":SYST:ERR:NEXT?\r\n",
]
def test_stop_all_axes(lmcs2A):
controller = lmcs2A.controller
controller.sock.buffer_recv = NO_ERROR
controller.stop_all_axes()
assert controller.sock.buffer_put == [b":STOP0\r\n", b":SYST:ERR:NEXT?\r\n"]
def test_find_reference_mark(controller):
controller.sock.buffer_recv = [NO_ERROR, NO_ERROR, NO_ERROR, NO_ERROR]
controller.find_reference_mark(0, direction=1, autoZero=1, velocity=1.0, acceleration=10.0)
assert controller.sock.buffer_put == [
b":CHAN0:REF:OPT 6\r\n", # REVERSE_DIR (0x2) | AUTO_ZERO (0x4)
b":SYST:ERR:NEXT?\r\n",
b":CHAN0:VEL 1000000000\r\n",
b":SYST:ERR:NEXT?\r\n",
b":CHAN0:ACC 10000000000\r\n",
b":SYST:ERR:NEXT?\r\n",
b":REF0\r\n",
b":SYST:ERR:NEXT?\r\n",
]
@pytest.mark.parametrize(
"move_speed,axis,get_msg",
[(50, 0, b":CHAN0:VEL 50000000000\r\n"), (0, 0, b":CHAN0:VEL 0\r\n"), (20.23, 1, b":CHAN1:VEL 20230000000\r\n")],
)
def test_set_closed_loop_move_speed(controller, move_speed, axis, get_msg):
controller.sock.buffer_recv = NO_ERROR
controller.set_closed_loop_move_speed(axis, move_speed)
assert controller.sock.buffer_put == [get_msg, b":SYST:ERR:NEXT?\r\n"]
def test_get_closed_loop_move_speed(controller):
controller.sock.buffer_recv = b"1000000000\r\n"
val = controller.get_closed_loop_move_speed(0)
assert val == 1.0
assert controller.sock.buffer_put[0] == b":CHAN0:VEL?\r\n"
def test_get_position_limits(controller):
controller.sock.buffer_recv = [b"-1000000000\r\n", b"1000000000\r\n"]
val = controller.get_position_limits(0)
assert val == [-1.0, 1.0]
assert controller.sock.buffer_put == [b":CHAN0:RLIM:MIN?\r\n", b":CHAN0:RLIM:MAX?\r\n"]
def test_set_position_limits(controller):
controller.sock.buffer_recv = [NO_ERROR, NO_ERROR]
controller.set_position_limits(0, -1.0, 1.0)
assert controller.sock.buffer_put == [
b":CHAN0:RLIM:MIN -1000000000\r\n",
b":SYST:ERR:NEXT?\r\n",
b":CHAN0:RLIM:MAX 1000000000\r\n",
b":SYST:ERR:NEXT?\r\n",
]
def test_get_positioner_type(controller):
controller.sock.buffer_recv = b'"SLC-24180-S"\r\n'
val = controller.get_positioner_type(0)
assert val == "SLC-24180-S"
assert controller.sock.buffer_put[0] == b":CHAN0:PTYP:NAME?\r\n"
def test_get_channel_type(controller):
controller.sock.buffer_recv = b"1\r\n"
val = controller.get_channel_type(0)
assert val == Mcs2ChannelType.STICK_SLIP_PIEZO_DRIVER
def test_get_number_of_channels(controller):
controller.sock.buffer_recv = b"9\r\n"
assert controller.get_number_of_channels() == 9
assert controller.sock.buffer_put[0] == b":DEV:NOCH?\r\n"
def test_get_idn(controller):
controller.sock.buffer_recv = b'"SmarAct;MCS2-00001234;TestDevice;09/22/23"\r\n'
assert controller.get_idn() == "SmarAct;MCS2-00001234;TestDevice;09/22/23"
def test_all_axes_referenced(lmcs2A):
controller = lmcs2A.controller
with mock.patch.object(controller, "axis_is_referenced", return_value=True) as mock_is_ref:
val = controller.all_axes_referenced()
assert val
mock_is_ref.assert_called_once_with(0)
def test_move_axis(lmcs2A):
controller = lmcs2A.controller
controller.sock.buffer_recv = [
b"128\r\n", # axis_is_referenced() -> STAT? (IS_REFERENCED only, not moving)
NO_ERROR, # MMOD command error-check
NO_ERROR, # HOLD command error-check
NO_ERROR, # MOVE command error-check
b"128\r\n", # background thread's is_axis_moving() -> STAT? (not moving)
b"50000000000\r\n", # final readback.read() -> POS?
]
lmcs2A.move(50)
assert controller.sock.buffer_put == [
b":CHAN0:STAT?\r\n",
b":CHAN0:MMOD 0\r\n",
b":SYST:ERR:NEXT?\r\n",
b":CHAN0:HOLD 1000\r\n",
b":SYST:ERR:NEXT?\r\n",
b":MOVE0 50000000000\r\n",
b":SYST:ERR:NEXT?\r\n",
b":CHAN0:STAT?\r\n",
b":CHAN0:POS?\r\n",
]
def test_sim_mcs2_end_to_end(dm_with_devices):
"""Exercises the simulated MCS2 backend (SimMcs2Socket/SimMcs2State) end-to-end
through the real Mcs2Motor/Mcs2Controller classes, as a sanity check independent of
the exact-byte-sequence unit tests above."""
Mcs2Controller._reset_controller()
SimStateRegistry.reset()
mot = SimMcs2Motor(
"A",
name="simmcs2a",
host="sim-mcs2",
port=55551,
sign=1,
device_manager=dm_with_devices,
sim_initial_position=1.0,
sim_velocity=5.0,
sim_referenced=True,
)
mot.controller.on()
mot.stage()
try:
assert mot.controller.axis_is_referenced(0) is True
assert abs(mot.read()[mot.name]["value"] - 1.0) < 1e-6
status = mot.move(2.0, wait=True)
assert status.done
assert abs(mot.read()[mot.name]["value"] - 2.0) < 0.5
mot.stop()
finally:
mot.unstage()
mot.controller.off()