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

refactor: merged controller methods

Browse Source
This commit is contained in:
wakonig_k
2024-05-07 17:58:14 +02:00
parent 8410b3ceec
commit 73ce61dd96
5 changed files with 76 additions and 661 deletions
Download Patch File Download Diff File Expand all files Collapse all files
csaxs_bec/devices/rt_lamni/rt_flomni_ophyd.py
+16 -7
View File
@@ -9,14 +9,13 @@ 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
from ophyd_devices.utils.controller import threadlocked
from ophyd_devices.utils.controller import Controller, threadlocked
from ophyd_devices.utils.socket import SocketIO, raise_if_disconnected
from prettytable import PrettyTable
from csaxs_bec.devices.rt_lamni.rt_ophyd import (
BECConfigError,
RtCommunicationError,
RtController,
RtError,
RtReadbackSignal,
RtSetpointSignal,
@@ -27,7 +26,8 @@ from csaxs_bec.devices.rt_lamni.rt_ophyd import (
logger = bec_logger.logger
class RtFlomniController(RtController):
class RtFlomniController(Controller):
_axes_per_controller = 3
USER_ACCESS = [
"socket_put_and_receive",
"set_rotation_angle",
@@ -52,7 +52,7 @@ class RtFlomniController(RtController):
def __init__(
self,
*,
name=None,
name="RtFlomniController",
socket_cls=None,
socket_host=None,
socket_port=None,
@@ -75,6 +75,15 @@ class RtFlomniController(RtController):
self._min_scan_buffer_reached = False
self.rt_pid_voltage = None
def is_axis_moving(self, axis_Id) -> bool:
# this checks that axis is on target
axis_is_on_target = bool(float(self.socket_put_and_receive("o")))
return not axis_is_on_target
@threadlocked
def stop_all_axes(self):
self.socket_put("sc")
def add_pos_to_scan(self, positions) -> None:
def send_positions(parent, positions):
parent._min_scan_buffer_reached = False
@@ -489,7 +498,7 @@ class RtFlomniController(RtController):
MessageEndpoints.device_status("rt_scan"),
messages.DeviceStatusMessage(
device="rt_scan", status=1, metadata=self.readout_metadata
).dumps(),
),
)
# while scan is running
while mode > 0:
@@ -524,7 +533,7 @@ class RtFlomniController(RtController):
MessageEndpoints.device_status("rt_scan"),
messages.DeviceStatusMessage(
device="rt_scan", status=0, metadata=self.readout_metadata
).dumps(),
),
)
logger.info(
@@ -538,7 +547,7 @@ class RtFlomniController(RtController):
MessageEndpoints.device_read("rt_flomni"),
messages.DeviceMessage(
signals=signals, metadata={"point_id": point_id, **self.readout_metadata}
).dumps(),
),
)
def start_readout(self):
csaxs_bec/devices/rt_lamni/rt_lamni_ophyd.py
+48 -158
View File
@@ -12,6 +12,8 @@ from ophyd.utils import LimitError, ReadOnlyError
from ophyd_devices.utils.controller import Controller, threadlocked
from ophyd_devices.utils.socket import SocketIO, SocketSignal, raise_if_disconnected
from csaxs_bec.devices.rt_lamni.rt_ophyd import RtCommunicationError, RtError
logger = bec_logger.logger
@@ -42,6 +44,11 @@ def retry_once(fcn):
class RtLamniController(Controller):
"""
RT-Lamni controller class for all rt devices.
"""
_axes_per_controller = 3
USER_ACCESS = [
"socket_put_and_receive",
"set_rotation_angle",
@@ -61,110 +68,25 @@ class RtLamniController(Controller):
self,
*,
name="RtLamniController",
kind=None,
parent=None,
socket=None,
socket_cls=None,
socket_host=None,
socket_port=None,
attr_name="",
parent=None,
labels=None,
kind=None,
):
if not hasattr(self, "_initialized") or not self._initialized:
self._rtlamni_axis_per_controller = 3
self._axis = [None for axis_num in range(self._rtlamni_axis_per_controller)]
self._min_scan_buffer_reached = False
super().__init__(
name=name,
socket=socket,
attr_name=attr_name,
parent=parent,
labels=labels,
kind=kind,
)
self.readout_metadata = {}
def on(self, controller_num=0) -> None:
"""Open a new socket connection to the controller"""
if not self.connected:
try:
self.sock.open()
# discuss - after disconnect takes a while for the server to be ready again
max_retries = 10
tries = 0
while not self.connected:
try:
welcome_message = self.sock.receive()
self.connected = True
except ConnectionResetError as conn_reset:
if tries > max_retries:
raise conn_reset
tries += 1
time.sleep(2)
except ConnectionRefusedError as conn_error:
logger.error("Failed to open a connection to RTLamNI.")
raise RtLamniCommunicationError from conn_error
else:
logger.info("The connection has already been established.")
# warnings.warn(f"The connection has already been established.", stacklevel=2)
self._update_flyer_device_info()
def off(self) -> None:
"""Close the socket connection to the controller"""
if self.connected:
self.sock.close()
self.connected = False
else:
logger.info("The connection is already closed.")
def set_axis(self, axis: Device, axis_nr: int) -> None:
"""Assign an axis to a device instance.
Args:
axis (Device): Device instance (e.g. GalilMotor)
axis_nr (int): Controller axis number
"""
self._axis[axis_nr] = axis
@threadlocked
def socket_put(self, val: str) -> None:
self.sock.put(f"{val}\n".encode())
@threadlocked
def socket_get(self) -> str:
return self.sock.receive().decode()
@retry_once
@threadlocked
def socket_put_and_receive(self, val: str, remove_trailing_chars=True) -> str:
self.socket_put(val)
if remove_trailing_chars:
return self._remove_trailing_characters(self.sock.receive().decode())
return self.socket_get()
def is_axis_moving(self, axis_Id) -> bool:
# this checks that axis is on target
axis_is_on_target = bool(float(self.socket_put_and_receive(f"o")))
return not axis_is_on_target
# def is_thread_active(self, thread_id: int) -> bool:
# val = float(self.socket_put_and_receive(f"MG_XQ{thread_id}"))
# if val == -1:
# return False
# return True
def _remove_trailing_characters(self, var) -> str:
if len(var) > 1:
return var.split("\r\n")[0]
return var
@threadlocked
def set_rotation_angle(self, val: float):
self.socket_put(f"a{(val-300+30.538)/180*np.pi}")
@threadlocked
def stop_all_axes(self):
self.socket_put("sc")
super().__init__(
name=name,
socket_cls=socket_cls,
socket_host=socket_host,
socket_port=socket_port,
attr_name=attr_name,
parent=parent,
labels=labels,
kind=kind,
)
self._min_scan_buffer_reached = False
@threadlocked
def feedback_disable(self):
@@ -176,6 +98,19 @@ class RtLamniController(Controller):
self.set_device_enabled("lopty", True)
self.set_device_enabled("loptz", True)
def is_axis_moving(self, axis_Id) -> bool:
# this checks that axis is on target
axis_is_on_target = bool(float(self.socket_put_and_receive("o")))
return not axis_is_on_target
@threadlocked
def stop_all_axes(self):
self.socket_put("sc")
@threadlocked
def set_rotation_angle(self, val: float):
self.socket_put(f"a{(val-300+30.538)/180*np.pi}")
@threadlocked
def _set_axis_velocity(self, um_per_s):
self.socket_put(f"V{um_per_s}")
@@ -235,19 +170,6 @@ class RtLamniController(Controller):
self.set_device_enabled("lopty", True)
self.set_device_enabled("loptz", True)
def get_device_manager(self):
for axis in self._axis:
if hasattr(axis, "device_manager") and axis.device_manager:
return axis.device_manager
raise BECConfigError("Could not access the device_manager")
def get_axis_by_name(self, name):
for axis in self._axis:
if axis:
if axis.name == name:
return axis
raise RuntimeError(f"Could not find an axis with name {name}")
@threadlocked
def clear_trajectory_generator(self):
self.socket_put("sc")
@@ -257,7 +179,7 @@ class RtLamniController(Controller):
def send_positions(parent, positions):
parent._min_scan_buffer_reached = False
for pos_index, pos in enumerate(positions):
parent.socket_put_and_receive(f"s{pos[0]:.05f},{pos[1]:05f},0")
parent.socket_put_and_receive(f"s{pos[0]:.05f},{pos[1]:.05f},0")
if pos_index > 100:
parent._min_scan_buffer_reached = True
parent._min_scan_buffer_reached = True
@@ -269,7 +191,7 @@ class RtLamniController(Controller):
def get_scan_status(self):
return_table = (self.socket_put_and_receive(f"sr")).split(",")
if len(return_table) != 3:
raise RtLamniCommunicationError(
raise RtCommunicationError(
f"Expected to receive 3 return values. Instead received {return_table}"
)
mode = int(return_table[0])
@@ -289,7 +211,7 @@ class RtLamniController(Controller):
logger.error(
"Cannot start scan because feedback loop is not running or there is an interferometer error."
)
raise RtLamniError(
raise RtError(
"Cannot start scan because feedback loop is not running or there is an interferometer error."
)
# here exception
@@ -297,7 +219,7 @@ class RtLamniController(Controller):
if number_of_positions_planned == 0:
logger.error("Cannot start scan because no target positions are planned.")
raise RtLamniError("Cannot start scan because no target positions are planned.")
raise RtError("Cannot start scan because no target positions are planned.")
# hier exception
# start a point-by-point scan (for cont scan in flomni it would be "sa")
self.socket_put_and_receive("sd")
@@ -306,29 +228,6 @@ class RtLamniController(Controller):
readout = threading.Thread(target=self.read_positions_from_sampler)
readout.start()
def _update_flyer_device_info(self):
flyer_info = self._get_flyer_device_info()
self.get_device_manager().connector.set(
MessageEndpoints.device_info("rt_scan"),
messages.DeviceInfoMessage(device="rt_scan", info=flyer_info),
)
def _get_flyer_device_info(self) -> dict:
return {
"device_name": self.name,
"device_attr_name": getattr(self, "attr_name", ""),
"device_dotted_name": getattr(self, "dotted_name", ""),
"device_info": {
"device_base_class": "ophydobject",
"signals": [],
"hints": {"fields": ["average_x_st_fzp", "average_y_st_fzp"]},
"describe": {},
"describe_configuration": {},
"sub_devices": [],
"custom_user_access": [],
},
}
def kickoff(self, metadata):
self.readout_metadata = metadata
while not self._min_scan_buffer_reached:
@@ -440,7 +339,7 @@ class RtLamniController(Controller):
)
def feedback_status_angle_lamni(self) -> bool:
return_table = (self.socket_put_and_receive(f"J7")).split(",")
return_table = (self.socket_put_and_receive("J7")).split(",")
logger.debug(
f"LamNI angle interferomter status {bool(return_table[0])}, position {float(return_table[1])}, signal {float(return_table[2])}"
)
@@ -449,21 +348,21 @@ class RtLamniController(Controller):
def feedback_enable_with_reset(self):
if not self.feedback_status_angle_lamni():
self.feedback_disable_and_even_reset_lamni_angle_interferometer()
logger.info(f"LamNI resetting interferometer inclusive angular interferomter.")
logger.info("LamNI resetting interferometer inclusive angular interferomter.")
else:
self.feedback_disable()
logger.info(
f"LamNI resetting interferomter except angular interferometer which is already running."
"LamNI resetting interferomter except angular interferometer which is already running."
)
# set these as closed loop target position
self.socket_put(f"pa0,0")
self.socket_put("pa0,0")
self.get_axis_by_name("rtx").user_setpoint.setpoint = 0
self.socket_put(f"pa1,0")
self.socket_put("pa1,0")
self.get_axis_by_name("rty").user_setpoint.setpoint = 0
self.socket_put(
f"pa2,0"
"pa2,0"
) # we set all three outputs of the traj. gen. although in LamNI case only 0,1 are used
self.clear_trajectory_generator()
@@ -477,7 +376,7 @@ class RtLamniController(Controller):
logger.error(
"Cannot enable feedback. The small rotation air is on and/or orchestra disabled by the motor controller."
)
raise RtLamniError(
raise RtError(
"Cannot enable feedback. The small rotation air is on and/or orchestra disabled by the motor controller."
)
@@ -516,7 +415,7 @@ class RtLamniController(Controller):
logger.error(
"Cannot start scan because feedback loop is not running or there is an interferometer error."
)
raise RtLamniError(
raise RtError(
"Cannot start scan because feedback loop is not running or there is an interferometer error."
)
@@ -524,15 +423,6 @@ class RtLamniController(Controller):
# ptychography_alignment_done = 0
def set_device_enabled(self, device_name: str, enabled: bool) -> None:
"""enable / disable a device"""
if device_name not in self.get_device_manager().devices:
logger.warning(
f"Device {device_name} is not configured and cannot be enabled/disabled."
)
return
self.get_device_manager().devices[device_name].read_only = not enabled
class RtLamniSignalBase(SocketSignal):
def __init__(self, signal_name, **kwargs):
csaxs_bec/devices/rt_lamni/rt_ophyd.py
+3 -489
View File
@@ -10,17 +10,17 @@ 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 Controller, threadlocked
from ophyd_devices.utils.controller import Controller, ControllerCommunicationError, threadlocked
from ophyd_devices.utils.socket import SocketIO, SocketSignal, raise_if_disconnected
logger = bec_logger.logger
class RtCommunicationError(Exception):
class RtCommunicationError(ControllerCommunicationError):
pass
class RtError(Exception):
class RtError(ControllerCommunicationError):
pass
@@ -42,468 +42,6 @@ def retry_once(fcn):
return wrapper
class RtController(Controller):
_axes_per_controller = 3
USER_ACCESS = [
"socket_put_and_receive",
"set_rotation_angle",
"feedback_disable",
"feedback_enable_without_reset",
"feedback_disable_and_even_reset_lamni_angle_interferometer",
"feedback_enable_with_reset",
"add_pos_to_scan",
"clear_trajectory_generator",
"_set_axis_velocity",
"_set_axis_velocity_maximum_speed",
"_position_sampling_single_read",
"_position_sampling_single_reset_and_start_sampling",
]
def on(self, controller_num=0) -> None:
"""Open a new socket connection to the controller"""
# if not self.connected:
# try:
# self.sock.open()
# # discuss - after disconnect takes a while for the server to be ready again
# max_retries = 10
# tries = 0
# while not self.connected:
# try:
# welcome_message = self.sock.receive()
# self.connected = True
# except ConnectionResetError as conn_reset:
# if tries > max_retries:
# raise conn_reset
# tries += 1
# time.sleep(2)
# except ConnectionRefusedError as conn_error:
# logger.error("Failed to open a connection to RTLamNI.")
# raise RtCommunicationError from conn_error
# else:
# logger.info("The connection has already been established.")
# # warnings.warn(f"The connection has already been established.", stacklevel=2)
super().on()
# self._update_flyer_device_info()
def set_axis(self, axis: Device, axis_nr: int) -> None:
"""Assign an axis to a device instance.
Args:
axis (Device): Device instance (e.g. GalilMotor)
axis_nr (int): Controller axis number
"""
self._axis[axis_nr] = axis
@threadlocked
def socket_put(self, val: str) -> None:
self.sock.put(f"{val}\n".encode())
@threadlocked
def socket_get(self) -> str:
return self.sock.receive().decode()
@retry_once
@threadlocked
def socket_put_and_receive(self, val: str, remove_trailing_chars=True) -> str:
self.socket_put(val)
if remove_trailing_chars:
return self._remove_trailing_characters(self.sock.receive().decode())
return self.socket_get()
def is_axis_moving(self, axis_Id) -> bool:
# this checks that axis is on target
axis_is_on_target = bool(float(self.socket_put_and_receive(f"o")))
return not axis_is_on_target
# def is_thread_active(self, thread_id: int) -> bool:
# val = float(self.socket_put_and_receive(f"MG_XQ{thread_id}"))
# if val == -1:
# return False
# return True
def _remove_trailing_characters(self, var) -> str:
if len(var) > 1:
return var.split("\r\n")[0]
return var
@threadlocked
def set_rotation_angle(self, val: float):
self.socket_put(f"a{(val-300+30.538)/180*np.pi}")
@threadlocked
def stop_all_axes(self):
self.socket_put("sc")
@threadlocked
def feedback_disable(self):
self.socket_put("J0")
logger.info("LamNI Feedback disabled.")
self.set_device_enabled("lsamx", True)
self.set_device_enabled("lsamy", True)
self.set_device_enabled("loptx", True)
self.set_device_enabled("lopty", True)
self.set_device_enabled("loptz", True)
@threadlocked
def _set_axis_velocity(self, um_per_s):
self.socket_put(f"V{um_per_s}")
@threadlocked
def _set_axis_velocity_maximum_speed(self):
self.socket_put(f"V0")
# for developement of soft continuous scanning
@threadlocked
def _position_sampling_single_reset_and_start_sampling(self):
self.socket_put(f"Ss")
@threadlocked
def _position_sampling_single_read(self):
(number_of_samples, sum0, sum0_2, sum1, sum1_2, sum2, sum2_2) = self.socket_put_and_receive(
f"Sr"
).split(",")
avg_x = float(sum1) / int(number_of_samples)
avg_y = float(sum0) / int(number_of_samples)
stdev_x = np.sqrt(
float(sum1_2) / int(number_of_samples)
- np.power(float(sum1) / int(number_of_samples), 2)
)
stdev_y = np.sqrt(
float(sum0_2) / int(number_of_samples)
- np.power(float(sum0) / int(number_of_samples), 2)
)
return (avg_x, avg_y, stdev_x, stdev_y)
@threadlocked
def feedback_enable_without_reset(self):
# read current interferometer position
return_table = (self.socket_put_and_receive(f"J4")).split(",")
x_curr = float(return_table[2])
y_curr = float(return_table[1])
# set these as closed loop target position
self.socket_put(f"pa0,{x_curr:.4f}")
self.socket_put(f"pa1,{y_curr:.4f}")
self.get_device_manager().devices.rtx.obj.user_setpoint.set_with_feedback_disabled(x_curr)
self.get_device_manager().devices.rty.obj.user_setpoint.set_with_feedback_disabled(y_curr)
self.socket_put("J5")
logger.info("LamNI Feedback enabled (without reset).")
self.set_device_enabled("lsamx", False)
self.set_device_enabled("lsamy", False)
self.set_device_enabled("loptx", False)
self.set_device_enabled("lopty", False)
self.set_device_enabled("loptz", False)
@threadlocked
def feedback_disable_and_even_reset_lamni_angle_interferometer(self):
self.socket_put("J6")
logger.info("LamNI Feedback disabled including the angular interferometer.")
self.set_device_enabled("lsamx", True)
self.set_device_enabled("lsamy", True)
self.set_device_enabled("loptx", True)
self.set_device_enabled("lopty", True)
self.set_device_enabled("loptz", True)
def get_device_manager(self):
for axis in self._axis:
if hasattr(axis, "device_manager") and axis.device_manager:
return axis.device_manager
raise BECConfigError("Could not access the device_manager")
def get_axis_by_name(self, name):
for axis in self._axis:
if axis:
if axis.name == name:
return axis
raise RuntimeError(f"Could not find an axis with name {name}")
@threadlocked
def clear_trajectory_generator(self):
self.socket_put("sc")
logger.info("LamNI scan stopped and deleted, moving to start position")
def add_pos_to_scan(self, positions) -> None:
def send_positions(parent, positions):
parent._min_scan_buffer_reached = False
for pos_index, pos in enumerate(positions):
parent.socket_put_and_receive(f"s{pos[0]},{pos[1]},0")
if pos_index > 100:
parent._min_scan_buffer_reached = True
parent._min_scan_buffer_reached = True
threading.Thread(target=send_positions, args=(self, positions), daemon=True).start()
@retry_once
@threadlocked
def get_scan_status(self):
return_table = (self.socket_put_and_receive(f"sr")).split(",")
if len(return_table) != 3:
raise RtCommunicationError(
f"Expected to receive 3 return values. Instead received {return_table}"
)
mode = int(return_table[0])
# mode 0: direct positioning
# mode 1: running internal timer (not tested/used anymore)
# mode 2: rt point scan running
# mode 3: rt point scan starting
# mode 5/6: rt continuous scanning (not available in LamNI)
number_of_positions_planned = int(return_table[1])
current_position_in_scan = int(return_table[2])
return (mode, number_of_positions_planned, current_position_in_scan)
@threadlocked
def start_scan(self):
interferometer_feedback_not_running = int((self.socket_put_and_receive("J2")).split(",")[0])
if interferometer_feedback_not_running == 1:
logger.error(
"Cannot start scan because feedback loop is not running or there is an interferometer error."
)
raise RtError(
"Cannot start scan because feedback loop is not running or there is an interferometer error."
)
# here exception
(mode, number_of_positions_planned, current_position_in_scan) = self.get_scan_status()
if number_of_positions_planned == 0:
logger.error("Cannot start scan because no target positions are planned.")
raise RtError("Cannot start scan because no target positions are planned.")
# hier exception
# start a point-by-point scan (for cont scan in flomni it would be "sa")
self.socket_put_and_receive("sd")
def start_readout(self):
readout = threading.Thread(target=self.read_positions_from_sampler)
readout.start()
def _update_flyer_device_info(self):
flyer_info = self._get_flyer_device_info()
self.get_device_manager().connector.set(
MessageEndpoints.device_info("rt_scan"),
messages.DeviceInfoMessage(device="rt_scan", info=flyer_info).dumps(),
)
def _get_flyer_device_info(self) -> dict:
return {
"device_name": self.name,
"device_attr_name": getattr(self, "attr_name", ""),
"device_dotted_name": getattr(self, "dotted_name", ""),
"device_info": {
"device_base_class": "ophydobject",
"signals": [],
"hints": {"fields": ["average_x_st_fzp", "average_y_st_fzp"]},
"describe": {},
"describe_configuration": {},
"sub_devices": [],
"custom_user_access": [],
},
}
def kickoff(self, metadata):
self.readout_metadata = metadata
while not self._min_scan_buffer_reached:
time.sleep(0.001)
self.start_scan()
time.sleep(0.1)
self.start_readout()
def _get_signals_from_table(self, return_table) -> dict:
self.average_stdeviations_x_st_fzp += float(return_table[5])
self.average_stdeviations_y_st_fzp += float(return_table[8])
self.average_lamni_angle += float(return_table[19])
signals = {
"target_x": {"value": float(return_table[3])},
"average_x_st_fzp": {"value": float(return_table[4])},
"stdev_x_st_fzp": {"value": float(return_table[5])},
"target_y": {"value": float(return_table[6])},
"average_y_st_fzp": {"value": float(return_table[7])},
"stdev_y_st_fzp": {"value": float(return_table[8])},
"average_cap1": {"value": float(return_table[9])},
"stdev_cap1": {"value": float(return_table[10])},
"average_cap2": {"value": float(return_table[11])},
"stdev_cap2": {"value": float(return_table[12])},
"average_cap3": {"value": float(return_table[13])},
"stdev_cap3": {"value": float(return_table[14])},
"average_cap4": {"value": float(return_table[15])},
"stdev_cap4": {"value": float(return_table[16])},
"average_cap5": {"value": float(return_table[17])},
"stdev_cap5": {"value": float(return_table[18])},
"average_angle_interf_ST": {"value": float(return_table[19])},
"stdev_angle_interf_ST": {"value": float(return_table[20])},
"average_stdeviations_x_st_fzp": {
"value": self.average_stdeviations_x_st_fzp / (int(return_table[0]) + 1)
},
"average_stdeviations_y_st_fzp": {
"value": self.average_stdeviations_y_st_fzp / (int(return_table[0]) + 1)
},
"average_lamni_angle": {"value": self.average_lamni_angle / (int(return_table[0]) + 1)},
}
return signals
def read_positions_from_sampler(self):
# this was for reading after the scan completed
number_of_samples_to_read = 1 # self.get_scan_status()[1] #number of valid samples, will be updated upon first data read
read_counter = 0
previous_point_in_scan = 0
self.average_stdeviations_x_st_fzp = 0
self.average_stdeviations_y_st_fzp = 0
self.average_lamni_angle = 0
mode, number_of_positions_planned, current_position_in_scan = self.get_scan_status()
# if not (mode==2 or mode==3):
# error
self.get_device_manager().connector.set(
MessageEndpoints.device_status("rt_scan"),
messages.DeviceStatusMessage(
device="rt_scan", status=1, metadata=self.readout_metadata
).dumps(),
)
# while scan is running
while mode > 0:
# logger.info(f"Current scan position {current_position_in_scan} out of {number_of_positions_planned}")
mode, number_of_positions_planned, current_position_in_scan = self.get_scan_status()
time.sleep(0.01)
if current_position_in_scan > 5:
while current_position_in_scan > read_counter + 1:
return_table = (self.socket_put_and_receive(f"r{read_counter}")).split(",")
# logger.info(f"{return_table}")
logger.info(f"Read {read_counter} out of {number_of_positions_planned}")
read_counter = read_counter + 1
signals = self._get_signals_from_table(return_table)
self.publish_device_data(signals=signals, point_id=int(return_table[0]))
time.sleep(0.05)
# read the last samples even though scan is finished already
while number_of_positions_planned > read_counter:
return_table = (self.socket_put_and_receive(f"r{read_counter}")).split(",")
logger.info(f"Read {read_counter} out of {number_of_positions_planned}")
# logger.info(f"{return_table}")
read_counter = read_counter + 1
signals = self._get_signals_from_table(return_table)
self.publish_device_data(signals=signals, point_id=int(return_table[0]))
self.get_device_manager().connector.set(
MessageEndpoints.device_status("rt_scan"),
messages.DeviceStatusMessage(
device="rt_scan", status=0, metadata=self.readout_metadata
).dumps(),
)
logger.info(
f"LamNI statistics: Average of all standard deviations: x {self.average_stdeviations_x_st_fzp/number_of_samples_to_read}, y {self.average_stdeviations_y_st_fzp/number_of_samples_to_read}, angle {self.average_lamni_angle/number_of_samples_to_read}."
)
def publish_device_data(self, signals, point_id):
self.get_device_manager().connector.set_and_publish(
MessageEndpoints.device_read("rt_lamni"),
messages.DeviceMessage(
signals=signals, metadata={"point_id": point_id, **self.readout_metadata}
).dumps(),
)
def feedback_status_angle_lamni(self) -> bool:
return_table = (self.socket_put_and_receive(f"J7")).split(",")
logger.debug(
f"LamNI angle interferomter status {bool(return_table[0])}, position {float(return_table[1])}, signal {float(return_table[2])}"
)
return bool(return_table[0])
def feedback_enable_with_reset(self):
if not self.feedback_status_angle_lamni():
self.feedback_disable_and_even_reset_lamni_angle_interferometer()
logger.info(f"LamNI resetting interferometer inclusive angular interferomter.")
else:
self.feedback_disable()
logger.info(
f"LamNI resetting interferomter except angular interferometer which is already running."
)
# set these as closed loop target position
self.socket_put(f"pa0,0")
self.get_axis_by_name("rtx").user_setpoint.setpoint = 0
self.socket_put(f"pa1,0")
self.get_axis_by_name("rty").user_setpoint.setpoint = 0
self.socket_put(
f"pa2,0"
) # we set all three outputs of the traj. gen. although in LamNI case only 0,1 are used
self.clear_trajectory_generator()
self.get_device_manager().devices.lsamrot.obj.move(0, wait=True)
galil_controller_rt_status = (
self.get_device_manager().devices.lsamx.obj.controller.lgalil_is_air_off_and_orchestra_enabled()
)
if galil_controller_rt_status == 0:
logger.error(
"Cannot enable feedback. The small rotation air is on and/or orchestra disabled by the motor controller."
)
raise RtError(
"Cannot enable feedback. The small rotation air is on and/or orchestra disabled by the motor controller."
)
time.sleep(0.03)
lsamx_user_params = self.get_device_manager().devices.lsamx.user_parameter
if lsamx_user_params is None or lsamx_user_params.get("center") is None:
raise RuntimeError("lsamx center is not defined")
lsamy_user_params = self.get_device_manager().devices.lsamy.user_parameter
if lsamy_user_params is None or lsamy_user_params.get("center") is None:
raise RuntimeError("lsamy center is not defined")
lsamx_center = lsamx_user_params.get("center")
lsamy_center = lsamy_user_params.get("center")
self.get_device_manager().devices.lsamx.obj.move(lsamx_center, wait=True)
self.get_device_manager().devices.lsamy.obj.move(lsamy_center, wait=True)
self.socket_put("J1")
_waitforfeedbackctr = 0
interferometer_feedback_not_running = int((self.socket_put_and_receive("J2")).split(",")[0])
while interferometer_feedback_not_running == 1 and _waitforfeedbackctr < 100:
time.sleep(0.01)
_waitforfeedbackctr = _waitforfeedbackctr + 1
interferometer_feedback_not_running = int(
(self.socket_put_and_receive("J2")).split(",")[0]
)
self.set_device_enabled("lsamx", False)
self.set_device_enabled("lsamy", False)
self.set_device_enabled("loptx", False)
self.set_device_enabled("lopty", False)
self.set_device_enabled("loptz", False)
if interferometer_feedback_not_running == 1:
logger.error(
"Cannot start scan because feedback loop is not running or there is an interferometer error."
)
raise RtError(
"Cannot start scan because feedback loop is not running or there is an interferometer error."
)
time.sleep(0.01)
# ptychography_alignment_done = 0
def set_device_enabled(self, device_name: str, enabled: bool) -> None:
"""enable / disable a device"""
if device_name not in self.get_device_manager().devices:
logger.warning(
f"Device {device_name} is not configured and cannot be enabled/disabled."
)
return
self.get_device_manager().devices[device_name].read_only = not enabled
class RtSignalBase(SocketSignal):
def __init__(self, signal_name, **kwargs):
self.signal_name = signal_name
@@ -792,27 +330,3 @@ class RtMotor(Device, PositionerBase):
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:
rty = RtLamniMotor("B", name="rty", host="mpc2680.psi.ch", port=3333, sign=1)
rty.stage()
status = rty.move(0, wait=True)
status = rty.move(10, wait=True)
rty.read()
rty.get()
rty.describe()
rty.unstage()
else:
from ophyd_devices.utils.socket import SocketMock
rtx = RtLamniMotor("A", name="rtx", host="mpc2680.psi.ch", port=3333, socket_cls=SocketMock)
rty = RtLamniMotor("B", name="rty", host="mpc2680.psi.ch", port=3333, socket_cls=SocketMock)
rtx.stage()
# rty.stage()
csaxs_bec/devices/smaract/smaract_controller.py
-4
View File
@@ -111,10 +111,6 @@ class SmaractController(Controller):
def socket_put(self, val: str):
self.sock.put(f":{val}\n".encode())
@threadlocked
def socket_get(self):
return self.sock.receive().decode()
@threadlocked
def socket_put_and_receive(
self, val: str, remove_trailing_chars=True, check_for_errors=True, raise_if_not_status=False
tests/tests_devices/test_rt_flomni.py
+9 -3
View File
@@ -16,13 +16,19 @@ def rt_flomni():
with mock.patch.object(rt_flomni, "sock"):
rtx = mock.MagicMock(spec=RtFlomniMotor)
rtx.name = "rtx"
rtx.axis_Id = "A"
rtx.axis_Id_numeric = 0
rty = mock.MagicMock(spec=RtFlomniMotor)
rty.name = "rty"
rty.axis_Id = "B"
rty.axis_Id_numeric = 1
rtz = mock.MagicMock(spec=RtFlomniMotor)
rtz.name = "rtz"
rt_flomni.set_axis(rtx, 0)
rt_flomni.set_axis(rty, 1)
rt_flomni.set_axis(rtz, 2)
rtz.axis_Id = "C"
rtz.axis_Id_numeric = 2
rt_flomni.set_axis(axis=rtx, axis_nr=0)
rt_flomni.set_axis(axis=rty, axis_nr=1)
rt_flomni.set_axis(axis=rtz, axis_nr=2)
yield rt_flomni