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

fix(controller): deprecate get_device_manager() in favor of dm

Browse Source
This commit is contained in:
appel_c
2025-12-05 15:55:27 +01:00
parent ac8177a132
commit ec45bb4c33
8 changed files with 54 additions and 74 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
csaxs_bec/devices/omny/galil/fgalil_ophyd.py
View File

@@ -212,7 +212,7 @@ class FlomniGalilMotor(Device, PositionerBase):
self.low_limit_travel.put(limits[0])
self.high_limit_travel.put(limits[1])
def wait_for_connection(self, timeout: int = 10, **kwargs) -> None:
def wait_for_connection(self, timeout: int = 30, **kwargs) -> None:
"""Wait for the device to be connected."""
self.controller.on(timeout=timeout)

2
csaxs_bec/devices/omny/galil/fupr_ophyd.py
View File

@@ -185,7 +185,7 @@ class FuprGalilMotor(Device, PositionerBase):
self.low_limit_travel.put(limits[0])
self.high_limit_travel.put(limits[1])
def wait_for_connection(self, timeout: int = 10, **kwargs) -> None:
def wait_for_connection(self, timeout: int = 30, **kwargs) -> None:
"""Wait for the device to be connected."""
self.controller.on(timeout=timeout)

2
csaxs_bec/devices/omny/galil/lgalil_ophyd.py
View File

@@ -170,7 +170,7 @@ class LamniGalilMotor(Device, PositionerBase):
self.low_limit_travel.put(limits[0])
self.high_limit_travel.put(limits[1])
def wait_for_connection(self, timeout: int = 10, **kwargs) -> None:
def wait_for_connection(self, timeout: int = 30, **kwargs) -> None:
"""Wait for the device to be connected."""
self.controller.on(timeout=timeout)

2
csaxs_bec/devices/omny/galil/ogalil_ophyd.py
View File

@@ -324,7 +324,7 @@ class OMNYGalilMotor(Device, PositionerBase):
self.low_limit_travel.put(limits[0])
self.high_limit_travel.put(limits[1])
def wait_for_connection(self, timeout: int = 10, **kwargs) -> None:
def wait_for_connection(self, timeout: int = 30, **kwargs) -> None:
"""Wait for the device to be connected."""
self.controller.on(timeout=timeout)

2
csaxs_bec/devices/omny/galil/sgalil_ophyd.py
View File

@@ -530,7 +530,7 @@ class SGalilMotor(Device, PositionerBase):
self.low_limit_travel.put(limits[0])
self.high_limit_travel.put(limits[1])
def wait_for_connection(self, timeout: int = 10, **kwargs) -> None:
def wait_for_connection(self, timeout: int = 30, **kwargs) -> None:
"""Wait for the device to be connected."""
self.controller.on(timeout=timeout)

40
csaxs_bec/devices/omny/rt/rt_flomni_ophyd.py
View File

@@ -128,15 +128,15 @@ class RtFlomniController(Controller):
while not self.slew_rate_limiters_on_target() or np.abs(self.pid_y()) > 0.1:
time.sleep(0.05)
self.get_device_manager().devices.rty.update_user_parameter({"tomo_additional_offsety": 0})
self.device_manager.devices.rty.update_user_parameter({"tomo_additional_offsety": 0})
self.clear_trajectory_generator()
self.laser_tracker_on()
# move to 0. FUPR will set the rotation angle during readout
self.get_device_manager().devices.fsamroy.obj.move(0, wait=True)
self.device_manager.devices.fsamroy.obj.move(0, wait=True)
fsamx = self.get_device_manager().devices.fsamx
fsamx = self.device_manager.devices.fsamx
fsamx.obj.pid_x_correction = 0
fsamx.obj.controller.socket_put_confirmed("axspeed[4]=0.1*stppermm[4]")
@@ -166,7 +166,7 @@ class RtFlomniController(Controller):
self.show_cyclic_error_compensation()
self.rt_pid_voltage = self.get_pid_x()
rtx = self.get_device_manager().devices.rtx
rtx = self.device_manager.devices.rtx
rtx.update_user_parameter({"rt_pid_voltage": self.rt_pid_voltage})
self.set_device_read_write("fsamx", False)
@@ -177,7 +177,7 @@ class RtFlomniController(Controller):
def move_samx_to_scan_region(self, fovx: float, cenx: float):
time.sleep(0.05)
if self.rt_pid_voltage is None:
rtx = self.get_device_manager().devices.rtx
rtx = self.device_manager.devices.rtx
self.rt_pid_voltage = rtx.user_parameter.get("rt_pid_voltage")
if self.rt_pid_voltage is None:
raise RtError(
@@ -194,7 +194,7 @@ class RtFlomniController(Controller):
break
wait_on_exit = True
self.socket_put("v0")
fsamx = self.get_device_manager().devices.fsamx
fsamx = self.device_manager.devices.fsamx
fsamx.read_only = False
fsamx.obj.controller.socket_put_confirmed("axspeed[4]=0.1*stppermm[4]")
fsamx.obj.pid_x_correction -= (self.get_pid_x() - expected_voltage) * 0.007
@@ -240,7 +240,7 @@ class RtFlomniController(Controller):
self.set_device_read_write("foptx", True)
self.set_device_read_write("fopty", True)
fsamx = self.get_device_manager().devices.fsamx
fsamx = self.device_manager.devices.fsamx
fsamx.obj.controller.socket_put_confirmed("axspeed[4]=025*stppermm[4]")
print("rt feedback is now disalbed.")
@@ -291,12 +291,8 @@ class RtFlomniController(Controller):
self.socket_put("T1")
time.sleep(0.5)
self.get_device_manager().devices.ftrackz.obj.controller.socket_put_confirmed(
"trackyct=0"
)
self.get_device_manager().devices.ftrackz.obj.controller.socket_put_confirmed(
"trackzct=0"
)
self.device_manager.devices.ftrackz.obj.controller.socket_put_confirmed("trackyct=0")
self.device_manager.devices.ftrackz.obj.controller.socket_put_confirmed("trackzct=0")
self.laser_tracker_wait_on_target()
logger.info("Laser tracker running!")
@@ -343,7 +339,7 @@ class RtFlomniController(Controller):
}
def laser_tracker_galil_enable(self):
ftrackz_con = self.get_device_manager().devices.ftrackz.obj.controller
ftrackz_con = self.device_manager.devices.ftrackz.obj.controller
ftrackz_con.socket_put_confirmed("tracken=1")
ftrackz_con.socket_put_confirmed("trackyct=0")
ftrackz_con.socket_put_confirmed("trackzct=0")
@@ -391,7 +387,7 @@ class RtFlomniController(Controller):
self.laser_tracker_wait_on_target()
signal = self.read_ssi_interferometer(1)
rtx = self.get_device_manager().devices.rtx
rtx = self.device_manager.devices.rtx
min_signal = rtx.user_parameter.get("min_signal")
low_signal = rtx.user_parameter.get("low_signal")
print(f"low signal: {low_signal}")
@@ -480,12 +476,6 @@ class RtFlomniController(Controller):
current_position_in_scan = int(float(return_table[2]))
return (mode, number_of_positions_planned, current_position_in_scan)
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 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
@@ -500,7 +490,7 @@ class RtFlomniController(Controller):
# if not (mode==2 or mode==3):
# error
self.get_device_manager().connector.set(
self.device_manager.connector.set(
MessageEndpoints.device_status("rt_scan"),
messages.DeviceStatusMessage(
device="rt_scan", status=1, metadata=self.readout_metadata
@@ -535,7 +525,7 @@ class RtFlomniController(Controller):
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(
self.device_manager.connector.set(
MessageEndpoints.device_status("rt_scan"),
messages.DeviceStatusMessage(
device="rt_scan", status=0, metadata=self.readout_metadata
@@ -549,7 +539,7 @@ class RtFlomniController(Controller):
)
def publish_device_data(self, signals, point_id):
self.get_device_manager().connector.set_and_publish(
self.device_manager.connector.set_and_publish(
MessageEndpoints.device_read("rt_flomni"),
messages.DeviceMessage(
signals=signals, metadata={"point_id": point_id, **self.readout_metadata}
@@ -688,7 +678,7 @@ class RtFlomniMotor(Device, PositionerBase):
self.low_limit_travel.put(limits[0])
self.high_limit_travel.put(limits[1])
def wait_for_connection(self, timeout: int = 10, **kwargs) -> None:
def wait_for_connection(self, timeout: int = 30, **kwargs) -> None:
"""Wait for the device to be connected."""
self.controller.on(timeout=timeout)

24
csaxs_bec/devices/omny/rt/rt_lamni_ophyd.py
View File

@@ -152,8 +152,8 @@ class RtLamniController(Controller):
# set these as closed loop target position
self.socket_put(f"pa0,{x_curr:.4f}")
self.socket_put(f"pa1,{y_curr:.4f}")
self.dm.devices.rtx.obj.user_setpoint.set_with_feedback_disabled(x_curr)
self.dm.devices.rty.obj.user_setpoint.set_with_feedback_disabled(y_curr)
self.device_manager.devices.rtx.obj.user_setpoint.set_with_feedback_disabled(x_curr)
self.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_read_write("lsamx", False)
@@ -286,7 +286,7 @@ class RtLamniController(Controller):
# if not (mode==2 or mode==3):
# error
self.get_device_manager().connector.set(
self.device_manager.connector.set(
MessageEndpoints.device_status("rt_scan"),
messages.DeviceStatusMessage(
device="rt_scan", status=1, metadata=self.readout_metadata
@@ -321,7 +321,7 @@ class RtLamniController(Controller):
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(
self.device_manager.connector.set(
MessageEndpoints.device_status("rt_scan"),
messages.DeviceStatusMessage(
device="rt_scan", status=0, metadata=self.readout_metadata
@@ -333,7 +333,7 @@ class RtLamniController(Controller):
)
def publish_device_data(self, signals, point_id):
self.get_device_manager().connector.set_and_publish(
self.device_manager.connector.set_and_publish(
MessageEndpoints.device_read("rt_lamni"),
messages.DeviceMessage(
signals=signals, metadata={"point_id": point_id, **self.readout_metadata}
@@ -368,10 +368,10 @@ class RtLamniController(Controller):
) # 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)
self.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()
self.device_manager.devices.lsamx.obj.controller.lgalil_is_air_off_and_orchestra_enabled()
)
if galil_controller_rt_status == 0:
@@ -384,16 +384,16 @@ class RtLamniController(Controller):
time.sleep(0.03)
lsamx_user_params = self.get_device_manager().devices.lsamx.user_parameter
lsamx_user_params = self.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
lsamy_user_params = self.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.device_manager.devices.lsamx.obj.move(lsamx_center, wait=True)
self.device_manager.devices.lsamy.obj.move(lsamy_center, wait=True)
self.socket_put("J1")
_waitforfeedbackctr = 0
@@ -588,7 +588,7 @@ class RtLamniMotor(Device, PositionerBase):
self.low_limit_travel.put(limits[0])
self.high_limit_travel.put(limits[1])
def wait_for_connection(self, timeout: int = 10, **kwargs) -> None:
def wait_for_connection(self, timeout: int = 30, **kwargs) -> None:
"""Wait for the device to be connected."""
self.controller.on(timeout=timeout)

54
csaxs_bec/devices/omny/rt/rt_omny_ophyd.py
View File

@@ -270,7 +270,7 @@ class RtOMNYController(Controller):
return self.mirror_parameters[channel]
def laser_tracker_check_and_wait_for_signalstrength(self):
self.get_device_manager().connector.send_client_info(
self.device_manager.connector.send_client_info(
"Checking laser tracker...", scope="", show_asap=True
)
if not self.laser_tracker_check_enabled():
@@ -282,12 +282,12 @@ class RtOMNYController(Controller):
self.laser_tracker_wait_on_target()
# when on target, check interferometer signal
signal = self._omny_interferometer_get_signalsample("ssi_4", 0.1)
rtx = self.get_device_manager().devices.rtx
rtx = self.device_manager.devices.rtx
min_signal = rtx.user_parameter.get("min_signal")
low_signal = rtx.user_parameter.get("low_signal")
wait_counter = 0
while signal < min_signal and wait_counter < 10:
self.get_device_manager().connector.send_client_info(
self.device_manager.connector.send_client_info(
f"The signal of the tracker {signal} is below the minimum required signal of {min_signal}. Waiting...",
scope="laser_tracker_check_and_wait_for_signalstrength",
show_asap=True,
@@ -298,14 +298,14 @@ class RtOMNYController(Controller):
signal = self._omny_interferometer_get_signalsample("ssi_4", 0.1)
if signal < low_signal:
self.get_device_manager().connector.send_client_info(
self.device_manager.connector.send_client_info(
f"\x1b[91mThe signal of the tracker {signal} is below the low limit of {low_signal}. Auto readjustment...\x1b[0m",
scope="laser_tracker_check_and_wait_for_signalstrength",
show_asap=True,
)
self.omny_interferometer_align_tracking()
self.get_device_manager().connector.send_client_info(
self.device_manager.connector.send_client_info(
"Checking laser tracker completed.", scope="", show_asap=True
)
@@ -364,7 +364,7 @@ class RtOMNYController(Controller):
elif amplitude < 10:
amplitude = 10
oshield = self.get_device_manager().devices.oshield
oshield = self.device_manager.devices.oshield
oshield.obj.controller.move_open_loop_steps(
channel, steps, amplitude=amplitude, frequency=500
@@ -444,7 +444,7 @@ class RtOMNYController(Controller):
info_str = f"\rAuto aligning Channel {mirror_channel}, {opt_mirrorname}, Current signal: {current_sample:.0f}"
message = info_str + " (q)uit \r"
self.get_device_manager().connector.send_client_info(
self.device_manager.connector.send_client_info(
message, scope="_omny_interferometer_optimize", show_asap=True
)
@@ -504,7 +504,7 @@ class RtOMNYController(Controller):
return True
def feedback_enable_with_reset(self):
self.get_device_manager().connector.send_client_info(
self.device_manager.connector.send_client_info(
f"Enabling the feedback...", scope="", show_asap=True
)
@@ -525,16 +525,16 @@ class RtOMNYController(Controller):
self.laser_tracker_on()
time.sleep(0.01)
osamroy = self.get_device_manager().devices.osamroy
osamroy = self.device_manager.devices.osamroy
# the following read will also upate the angle in rt during readout
readback = osamroy.obj.readback.get()
if np.fabs(readback) > 0.1:
self.get_device_manager().connector.send_client_info(
self.device_manager.connector.send_client_info(
f"Rotating to zero", scope="feedback enable", show_asap=True
)
osamroy.obj.move(0, wait=True)
osamx = self.get_device_manager().devices.osamx
osamx = self.device_manager.devices.osamx
osamx_in = osamx.user_parameter.get("in")
if not np.isclose(osamx.obj.readback.get(), osamx_in, atol=0.01):
@@ -622,7 +622,7 @@ class RtOMNYController(Controller):
# update variables and enable if not yet active
if not self.laser_tracker_check_enabled():
logger.info("Enabling the laser tracker. Please wait...")
self.get_device_manager().connector.send_client_info(
self.device_manager.connector.send_client_info(
f"Enabling the laser tracker. Please wait...", scope="", show_asap=True
)
@@ -639,12 +639,8 @@ class RtOMNYController(Controller):
self.socket_put("T1")
time.sleep(0.5)
self.get_device_manager().devices.otracky.obj.controller.socket_put_confirmed(
"trackyct=0"
)
self.get_device_manager().devices.otracky.obj.controller.socket_put_confirmed(
"trackzct=0"
)
self.device_manager.devices.otracky.obj.controller.socket_put_confirmed("trackyct=0")
self.device_manager.devices.otracky.obj.controller.socket_put_confirmed("trackzct=0")
self.laser_tracker_wait_on_target()
logger.info("Laser tracker running!")
@@ -718,17 +714,17 @@ class RtOMNYController(Controller):
self.laser_tracker_galil_status()
def laser_tracker_galil_enable(self):
otracky_con = self.get_device_manager().devices.otracky.obj.controller
otracky_con = self.device_manager.devices.otracky.obj.controller
otracky_con.socket_put_confirmed("tracken=1")
otracky_con.socket_put_confirmed("trackyct=0")
otracky_con.socket_put_confirmed("trackzct=0")
def laser_tracker_galil_disable(self):
otracky_con = self.get_device_manager().devices.otracky.obj.controller
otracky_con = self.device_manager.devices.otracky.obj.controller
otracky_con.socket_put_confirmed("tracken=0")
def laser_tracker_galil_status(self):
otracky_con = self.get_device_manager().devices.otracky.obj.controller
otracky_con = self.device_manager.devices.otracky.obj.controller
if bool(float(otracky_con.socket_put_and_receive("MGtracken").strip())) == 0:
print(self.red + "Tracking in the Galil Controller is disabled." + self.white)
print("Use dev.rtx.controller.laser_tracker_galil_enable to enable.\n")
@@ -914,12 +910,6 @@ class RtOMNYController(Controller):
current_position_in_scan = int(float(return_table[2]))
return (mode, number_of_positions_planned, current_position_in_scan)
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 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
@@ -933,7 +923,7 @@ class RtOMNYController(Controller):
# if not (mode==2 or mode==3):
# error
self.get_device_manager().connector.set(
self.device_manager.connector.set(
MessageEndpoints.device_status("rt_scan"),
messages.DeviceStatusMessage(
device="rt_scan", status=1, metadata=self.readout_metadata
@@ -968,7 +958,7 @@ class RtOMNYController(Controller):
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(
self.device_manager.connector.set(
MessageEndpoints.device_status("rt_scan"),
messages.DeviceStatusMessage(
device="rt_scan", status=0, metadata=self.readout_metadata
@@ -981,7 +971,7 @@ class RtOMNYController(Controller):
f" {self.average_stdeviations_y_st_fzp/read_counter*1000:.1f}."
)
self.get_device_manager().connector.send_client_info(
self.device_manager.connector.send_client_info(
"OMNY statistics: Average of all standard deviations [nm]: x"
f" {self.average_stdeviations_x_st_fzp/read_counter*1000:.1f}, y"
f" {self.average_stdeviations_y_st_fzp/read_counter*1000:.1f}.",
@@ -990,7 +980,7 @@ class RtOMNYController(Controller):
)
def publish_device_data(self, signals, point_id):
self.get_device_manager().connector.set_and_publish(
self.device_manager.connector.set_and_publish(
MessageEndpoints.device_read("rt_omny"),
messages.DeviceMessage(
signals=signals, metadata={"point_id": point_id, **self.readout_metadata}
@@ -1129,7 +1119,7 @@ class RtOMNYMotor(Device, PositionerBase):
self.low_limit_travel.put(limits[0])
self.high_limit_travel.put(limits[1])
def wait_for_connection(self, timeout: int = 10, **kwargs) -> None:
def wait_for_connection(self, timeout: int = 30, **kwargs) -> None:
"""Wait for the device to be connected."""
self.controller.on(timeout=timeout)