bec/ophyd_devices
0
0
Fork 0
mirror of https://github.com/bec-project/ophyd_devices.git synced 2025-06-02 10:00:40 +02:00
Code Issues Packages Projects Releases Wiki Activity

black; added push events to rt_lamni

Browse Source
This commit is contained in:
wakonig_k 2022-07-25 19:03:53 +02:00
parent 069887c5cb
commit 98a1426a98
5 changed files with 204 additions and 157 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
ophyd_devices/galil/galil_ophyd.py
View File

@ -1,5 +1,4 @@
import functools import functools
import logging
import threading import threading
import time import time
from typing import List from typing import List
@ -12,9 +11,9 @@ from ophyd.utils import ReadOnlyError, LimitError
from ophyd_devices.utils.controller import Controller, threadlocked from ophyd_devices.utils.controller import Controller, threadlocked
from ophyd_devices.utils.socket import SocketIO, SocketSignal, raise_if_disconnected from ophyd_devices.utils.socket import SocketIO, SocketSignal, raise_if_disconnected
from prettytable import PrettyTable from prettytable import PrettyTable
from bec_utils import bec_logger
logger = logging.getLogger("galil") logger = bec_logger.logger
class GalilCommunicationError(Exception): class GalilCommunicationError(Exception):
pass pass
@ -546,7 +545,6 @@ class GalilMotor(Device, PositionerBase):
if __name__ == "__main__": if __name__ == "__main__":
logging.basicConfig(level=logging.DEBUG)
mock = False mock = False
if not mock: if not mock:

5
ophyd_devices/galil/galil_ophyd_controller.py
View File

@ -1,4 +1,3 @@
import logging
import threading import threading
import time import time
from typing import List from typing import List
@ -10,8 +9,9 @@ from ophyd.utils import ReadOnlyError
from ophyd_devices.utils.controller import Controller from ophyd_devices.utils.controller import Controller
from ophyd_devices.utils.socket import SocketIO, SocketSignal, raise_if_disconnected from ophyd_devices.utils.socket import SocketIO, SocketSignal, raise_if_disconnected
from prettytable import PrettyTable from prettytable import PrettyTable
from bec_utils import bec_logger
logger = logging.getLogger("galil") logger = bec_logger.logger
class GalilCommunicationError(Exception): class GalilCommunicationError(Exception):
@ -460,7 +460,6 @@ class GalilMotor(Device, PositionerBase):
if __name__ == "__main__": if __name__ == "__main__":
logging.basicConfig(level=logging.DEBUG)
mock = True mock = True
if not mock: if not mock:

188
ophyd_devices/rt_lamni/rt_lamni_ophyd.py
View File

@ -1,24 +1,21 @@
import functools import functools
import threading import threading
import time import time
import numpy as np
from typing import List from typing import List
from ophyd import PositionerBase, Device, Component as Cpt
from prettytable import PrettyTable
from ophyd_devices.utils.controller import Controller
from ophyd_devices.utils.socket import (
SocketIO,
raise_if_disconnected,
SocketSignal,
)
from bec_utils import bec_logger
from ophyd.utils import ReadOnlyError,LimitError
from ophyd.status import wait as status_wait
from ophyd import Signal
import numpy as np
from bec_utils import BECMessage, MessageEndpoints, 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 Controller
from ophyd_devices.utils.socket import SocketIO, SocketSignal, raise_if_disconnected
from prettytable import PrettyTable
logger = bec_logger.logger logger = bec_logger.logger
def threadlocked(fcn): def threadlocked(fcn):
"""Ensure that thread acquires and releases the lock.""" """Ensure that thread acquires and releases the lock."""
@ -29,6 +26,7 @@ def threadlocked(fcn):
return wrapper return wrapper
class RtLamniCommunicationError(Exception): class RtLamniCommunicationError(Exception):
pass pass
@ -36,9 +34,11 @@ class RtLamniCommunicationError(Exception):
class RtLamniError(Exception): class RtLamniError(Exception):
pass pass
class BECConfigError(Exception): class BECConfigError(Exception):
pass pass
def retry_once(fcn): def retry_once(fcn):
"""Decorator to rerun a function in case a CommunicationError was raised. This may happen if the buffer was not empty.""" """Decorator to rerun a function in case a CommunicationError was raised. This may happen if the buffer was not empty."""
@ -66,8 +66,9 @@ class RtLamniController(Controller):
"_set_axis_velocity", "_set_axis_velocity",
"_set_axis_velocity_maximum_speed", "_set_axis_velocity_maximum_speed",
"_position_sampling_single_read", "_position_sampling_single_read",
"_position_sampling_single_reset_and_start_sampling" "_position_sampling_single_reset_and_start_sampling",
] ]
def __init__( def __init__(
self, self,
*, *,
@ -90,7 +91,7 @@ class RtLamniController(Controller):
labels=labels, labels=labels,
kind=kind, kind=kind,
) )
self.readout_metadata = {}
def on(self, controller_num=0) -> None: def on(self, controller_num=0) -> None:
"""Open a new socket connection to the controller""" """Open a new socket connection to the controller"""
@ -181,14 +182,21 @@ class RtLamniController(Controller):
self.socket_put(f"Ss") self.socket_put(f"Ss")
def _position_sampling_single_read(self): 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(",") (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_x = float(sum1) / int(number_of_samples)
avg_y = float(sum0) / 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_x = np.sqrt(
stdev_y = np.sqrt(float(sum0_2)/int(number_of_samples)-np.power(float(sum0)/int(number_of_samples),2)) 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) return (avg_x, avg_y, stdev_x, stdev_y)
def feedback_enable_without_reset(self): def feedback_enable_without_reset(self):
# read current interferometer position # read current interferometer position
return_table = (self.socket_put_and_receive(f"J4")).split(",") return_table = (self.socket_put_and_receive(f"J4")).split(",")
@ -217,16 +225,16 @@ class RtLamniController(Controller):
def get_device_manager(self): def get_device_manager(self):
for axis in self._axis: for axis in self._axis:
if hasattr(axis, 'device_manager') and axis.device_manager: if hasattr(axis, "device_manager") and axis.device_manager:
return axis.device_manager return axis.device_manager
raise BECConfigError('Could not access the device_manager') raise BECConfigError("Could not access the device_manager")
def get_axis_by_name(self, name): def get_axis_by_name(self, name):
for axis in self._axis: for axis in self._axis:
if axis: if axis:
if axis.name == name: if axis.name == name:
return axis return axis
raise RuntimeError(f'Could not find an axis with name {name}') raise RuntimeError(f"Could not find an axis with name {name}")
def clear_trajectory_generator(self): def clear_trajectory_generator(self):
self.socket_put("sc") self.socket_put("sc")
@ -240,13 +248,16 @@ class RtLamniController(Controller):
if pos_index > 100: if pos_index > 100:
parent._min_scan_buffer_reached = True parent._min_scan_buffer_reached = True
parent._min_scan_buffer_reached = True parent._min_scan_buffer_reached = True
threading.Thread(target=send_positions, args=(self, positions), daemon=True).start() threading.Thread(target=send_positions, args=(self, positions), daemon=True).start()
@retry_once @retry_once
def get_scan_status(self): def get_scan_status(self):
return_table = (self.socket_put_and_receive(f"sr")).split(",") return_table = (self.socket_put_and_receive(f"sr")).split(",")
if len(return_table) != 3: if len(return_table) != 3:
raise RtLamniCommunicationError(f"Expected to receive 3 return values. Instead received {return_table}") raise RtLamniCommunicationError(
f"Expected to receive 3 return values. Instead received {return_table}"
)
mode = int(return_table[0]) mode = int(return_table[0])
# mode 0: direct positioning # mode 0: direct positioning
# mode 1: running internal timer (not tested/used anymore) # mode 1: running internal timer (not tested/used anymore)
@ -260,8 +271,12 @@ class RtLamniController(Controller):
def start_scan(self): def start_scan(self):
interferometer_feedback_not_running = int((self.socket_put_and_receive("J2")).split(",")[0]) interferometer_feedback_not_running = int((self.socket_put_and_receive("J2")).split(",")[0])
if interferometer_feedback_not_running == 1: if interferometer_feedback_not_running == 1:
logger.error("Cannot start scan because feedback loop is not running or there is an interferometer error.") logger.error(
raise RtLamniError("Cannot start scan because feedback loop is not running or there is an interferometer error.") "Cannot start scan because feedback loop is not running or there is an interferometer error."
)
raise RtLamniError(
"Cannot start scan because feedback loop is not running or there is an interferometer error."
)
# here exception # here exception
(mode, number_of_positions_planned, current_position_in_scan) = self.get_scan_status() (mode, number_of_positions_planned, current_position_in_scan) = self.get_scan_status()
@ -276,15 +291,14 @@ class RtLamniController(Controller):
readout = threading.Thread(target=self.read_positions_from_sampler) readout = threading.Thread(target=self.read_positions_from_sampler)
readout.start() readout.start()
def kickoff(self, metadata):
def kickoff(self): self.readout_metadata = metadata
while not self._min_scan_buffer_reached: while not self._min_scan_buffer_reached:
time.sleep(0.001) time.sleep(0.001)
self.start_scan() self.start_scan()
time.sleep(0.1) time.sleep(0.1)
self.start_readout() self.start_readout()
def read_positions_from_sampler(self): def read_positions_from_sampler(self):
# this was for reading after the scan completed # 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 number_of_samples_to_read = 1 # self.get_scan_status()[1] #number of valid samples, will be updated upon first data read
@ -305,19 +319,32 @@ class RtLamniController(Controller):
while mode > 0: while mode > 0:
# logger.info(f"Current scan position {current_position_in_scan} out of {number_of_positions_planned}") # 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() mode, number_of_positions_planned, current_position_in_scan = self.get_scan_status()
time.sleep(.01) time.sleep(0.01)
if (current_position_in_scan>5): if current_position_in_scan > 5:
while current_position_in_scan > read_counter: while current_position_in_scan > read_counter:
return_table = (self.socket_put_and_receive(f"r{read_counter}")).split(",") return_table = (self.socket_put_and_receive(f"r{read_counter}")).split(",")
# logger.info(f"{return_table}") # logger.info(f"{return_table}")
logger.info(f"Read {read_counter} out of {number_of_positions_planned}") logger.info(f"Read {read_counter} out of {number_of_positions_planned}")
read_counter = read_counter + 1 read_counter = read_counter + 1
average_stdeviations_x_st_fzp=average_stdeviations_x_st_fzp+float(return_table[5]) average_stdeviations_x_st_fzp = average_stdeviations_x_st_fzp + float(
average_stdeviations_y_st_fzp=average_stdeviations_y_st_fzp+float(return_table[8]) return_table[5]
)
average_stdeviations_y_st_fzp = average_stdeviations_y_st_fzp + float(
return_table[8]
)
average_lamni_angle = average_lamni_angle + float(return_table[19]) average_lamni_angle = average_lamni_angle + float(return_table[19])
#DataPoint TotalPoints Target_x Average_x_st_fzp Stdev_x_st_fzp Target_y Average_y_st_fzp Stdev_y_st_fzp Average_cap1 Stdev_cap1 Average_cap2 Stdev_cap2 Average_cap3 Stdev_cap3 Average_cap4 Stdev_cap4 Average_cap5 Stdev_cap5 Average_angle_interf_ST Stdev_angle_interf_ST
# TODO!!
signals = {
"target_x":
"average_stdeviations_x_st_fzp": {"value": average_stdeviations_x_st_fzp},
"average_stdeviations_y_st_fzp": {"value": average_stdeviations_y_st_fzp},
"average_lamni_angle": {"value": average_lamni_angle},
}
self.publish_device_data(signals=signals, pointID=current_position_in_scan)
time.sleep(.05) time.sleep(0.05)
# read the last samples even though scan is finished already # read the last samples even though scan is finished already
while number_of_positions_planned > read_counter: while number_of_positions_planned > read_counter:
@ -329,13 +356,24 @@ class RtLamniController(Controller):
average_stdeviations_y_st_fzp = average_stdeviations_y_st_fzp + float(return_table[8]) average_stdeviations_y_st_fzp = average_stdeviations_y_st_fzp + float(return_table[8])
average_lamni_angle = average_lamni_angle + float(return_table[19]) average_lamni_angle = average_lamni_angle + float(return_table[19])
logger.info(f"LamNI statistics: Average of all standard deviations: x {average_stdeviations_x_st_fzp/number_of_samples_to_read}, y {average_stdeviations_y_st_fzp/number_of_samples_to_read}, angle {average_lamni_angle/number_of_samples_to_read}.") logger.info(
f"LamNI statistics: Average of all standard deviations: x {average_stdeviations_x_st_fzp/number_of_samples_to_read}, y {average_stdeviations_y_st_fzp/number_of_samples_to_read}, angle {average_lamni_angle/number_of_samples_to_read}."
)
def publish_device_data(self, signals, pointID):
self.get_device_manager().producer.send(
MessageEndpoints.device_read("rt_lamni"),
BECMessage.DeviceMessage(
signals=signals,
metadata={"pointID": pointID, **self.readout_metadata},
).dumps(),
)
def feedback_status_angle_lamni(self) -> bool: def feedback_status_angle_lamni(self) -> bool:
return_table = (self.socket_put_and_receive(f"J7")).split(",") 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])}") 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]) return bool(return_table[0])
def feedback_enable_with_reset(self): def feedback_enable_with_reset(self):
@ -344,24 +382,34 @@ class RtLamniController(Controller):
logger.info(f"LamNI resetting interferometer inclusive angular interferomter.") logger.info(f"LamNI resetting interferometer inclusive angular interferomter.")
else: else:
self.feedback_disable() self.feedback_disable()
logger.info(f"LamNI resetting interferomter except angular interferometer which is already running.") logger.info(
f"LamNI resetting interferomter except angular interferometer which is already running."
)
# set these as closed loop target position # set these as closed loop target position
self.socket_put(f"pa0,0") self.socket_put(f"pa0,0")
self.get_axis_by_name('rtx').user_setpoint.setpoint = 0 self.get_axis_by_name("rtx").user_setpoint.setpoint = 0
self.socket_put(f"pa1,0") self.socket_put(f"pa1,0")
self.get_axis_by_name('rty').user_setpoint.setpoint = 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.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.clear_trajectory_generator()
self.get_device_manager().devices.lsamrot.obj.move(0, wait=True) 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() 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: 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.") logger.error(
raise RtLamniError("Cannot enable feedback. The small rotation air is on and/or orchestra disabled by the motor controller.") "Cannot enable feedback. The small rotation air is on and/or orchestra disabled by the motor controller."
)
raise RtLamniError(
"Cannot enable feedback. The small rotation air is on and/or orchestra disabled by the motor controller."
)
time.sleep(0.03) time.sleep(0.03)
# needs to be changed to configurable center psoition # needs to be changed to configurable center psoition
@ -376,7 +424,9 @@ class RtLamniController(Controller):
while interferometer_feedback_not_running == 1 and _waitforfeedbackctr < 100: while interferometer_feedback_not_running == 1 and _waitforfeedbackctr < 100:
time.sleep(0.01) time.sleep(0.01)
_waitforfeedbackctr = _waitforfeedbackctr + 1 _waitforfeedbackctr = _waitforfeedbackctr + 1
interferometer_feedback_not_running = int((self.socket_put_and_receive("J2")).split(",")[0]) interferometer_feedback_not_running = int(
(self.socket_put_and_receive("J2")).split(",")[0]
)
# motor_par("lsamx","disable",1) # motor_par("lsamx","disable",1)
# motor_par("lsamy","disable",1) # motor_par("lsamy","disable",1)
@ -384,15 +434,18 @@ class RtLamniController(Controller):
# motor_par("lopty","disable",1) # motor_par("lopty","disable",1)
# motor_par("loptz","disable",1) # motor_par("loptz","disable",1)
if interferometer_feedback_not_running == 1: if interferometer_feedback_not_running == 1:
logger.error("Cannot start scan because feedback loop is not running or there is an interferometer error.") logger.error(
raise RtLamniError("Cannot start scan because feedback loop is not running or there is an interferometer error.") "Cannot start scan because feedback loop is not running or there is an interferometer error."
)
raise RtLamniError(
"Cannot start scan because feedback loop is not running or there is an interferometer error."
)
time.sleep(0.01) time.sleep(0.01)
# ptychography_alignment_done = 0 # ptychography_alignment_done = 0
class RtLamniSignalBase(SocketSignal): class RtLamniSignalBase(SocketSignal):
def __init__(self, signal_name, **kwargs): def __init__(self, signal_name, **kwargs):
self.signal_name = signal_name self.signal_name = signal_name
@ -442,10 +495,6 @@ class RtLamniSetpointSignal(RtLamniSignalBase):
Returns: Returns:
float: setpoint / target value float: setpoint / target value
""" """
return self.setpoint return self.setpoint
@ -463,11 +512,15 @@ class RtLamniSetpointSignal(RtLamniSignalBase):
""" """
target_val = val * self.parent.sign target_val = val * self.parent.sign
self.setpoint = target_val self.setpoint = target_val
interferometer_feedback_not_running = int((self.controller.socket_put_and_receive("J2")).split(",")[0]) interferometer_feedback_not_running = int(
(self.controller.socket_put_and_receive("J2")).split(",")[0]
)
if interferometer_feedback_not_running == 0: if interferometer_feedback_not_running == 0:
self.controller.socket_put(f"pa{self.parent.axis_Id_numeric},{target_val:.4f}") self.controller.socket_put(f"pa{self.parent.axis_Id_numeric},{target_val:.4f}")
else: else:
raise RtLamniError("The interferometer feedback is not running. Either it is turned off or and interferometer error occured.") raise RtLamniError(
"The interferometer feedback is not running. Either it is turned off or and interferometer error occured."
)
class RtLamniMotorIsMoving(RtLamniSignalRO): class RtLamniMotorIsMoving(RtLamniSignalRO):
@ -477,7 +530,8 @@ class RtLamniMotorIsMoving(RtLamniSignalRO):
def get(self): def get(self):
val = super().get() val = super().get()
if val is not None: if val is not None:
self._run_subs(sub_type=self.SUB_VALUE, self._run_subs(
sub_type=self.SUB_VALUE,
value=val, value=val,
timestamp=time.time(), timestamp=time.time(),
) )
@ -491,10 +545,9 @@ class RtLamniFeedbackRunning(RtLamniSignalRO):
else: else:
return 0 return 0
class RtLamniMotor(Device, PositionerBase): class RtLamniMotor(Device, PositionerBase):
USER_ACCESS = [ USER_ACCESS = ["controller"]
"controller"
]
readback = Cpt( readback = Cpt(
RtLamniReadbackSignal, RtLamniReadbackSignal,
signal_name="readback", signal_name="readback",
@ -502,9 +555,7 @@ class RtLamniMotor(Device, PositionerBase):
) )
user_setpoint = Cpt(RtLamniSetpointSignal, signal_name="setpoint") user_setpoint = Cpt(RtLamniSetpointSignal, signal_name="setpoint")
motor_is_moving = Cpt( motor_is_moving = Cpt(RtLamniMotorIsMoving, signal_name="motor_is_moving", kind="normal")
RtLamniMotorIsMoving, signal_name="motor_is_moving", kind="normal"
)
high_limit_travel = Cpt(Signal, value=0, kind="omitted") high_limit_travel = Cpt(Signal, value=0, kind="omitted")
low_limit_travel = Cpt(Signal, value=0, kind="omitted") low_limit_travel = Cpt(Signal, value=0, kind="omitted")
@ -581,12 +632,10 @@ class RtLamniMotor(Device, PositionerBase):
for walk in self.walk_signals(): for walk in self.walk_signals():
walk.item._metadata["connected"] = self.controller.connected walk.item._metadata["connected"] = self.controller.connected
def _forward_readback(self, **kwargs): def _forward_readback(self, **kwargs):
kwargs.pop("sub_type") kwargs.pop("sub_type")
self._run_subs(sub_type="readback", **kwargs) self._run_subs(sub_type="readback", **kwargs)
@raise_if_disconnected @raise_if_disconnected
def move(self, position, wait=True, **kwargs): def move(self, position, wait=True, **kwargs):
"""Move to a specified position, optionally waiting for motion to """Move to a specified position, optionally waiting for motion to
@ -626,7 +675,8 @@ class RtLamniMotor(Device, PositionerBase):
while self.motor_is_moving.get(): while self.motor_is_moving.get():
print("motor is moving") print("motor is moving")
val = self.readback.read() val = self.readback.read()
self._run_subs(sub_type=self.SUB_READBACK, self._run_subs(
sub_type=self.SUB_READBACK,
value=val, value=val,
timestamp=time.time(), timestamp=time.time(),
) )
@ -672,6 +722,9 @@ class RtLamniMotor(Device, PositionerBase):
else: else:
raise TypeError(f"Expected value of type int but received {type(val)}") raise TypeError(f"Expected value of type int but received {type(val)}")
def kickoff(self, metadata) -> None:
self.controller.kickoff(metadata)
@property @property
def egu(self): def egu(self):
"""The engineering units (EGU) for positions""" """The engineering units (EGU) for positions"""
@ -710,12 +763,7 @@ if __name__ == "__main__":
else: else:
from ophyd_devices.utils.socket import SocketMock from ophyd_devices.utils.socket import SocketMock
rtx = RtLamniMotor( rtx = RtLamniMotor("A", name="rtx", host="mpc2680.psi.ch", port=3333, socket_cls=SocketMock)
"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)
)
rty = RtLamniMotor(
"B", name="rty", host="mpc2680.psi.ch", port=3333, socket_cls=SocketMock
)
rtx.stage() rtx.stage()
# rty.stage() # rty.stage()

3
ophyd_devices/sim/sim.py
View File

@ -3,12 +3,13 @@ import time as ttime
import warnings import warnings
import numpy as np import numpy as np
from bec_utils import BECMessage, MessageEndpoints from bec_utils import BECMessage, MessageEndpoints, bec_logger
from ophyd import Component as Cpt from ophyd import Component as Cpt
from ophyd import Device, DeviceStatus, PositionerBase, Signal from ophyd import Device, DeviceStatus, PositionerBase, Signal
from ophyd.sim import _ReadbackSignal, _SetpointSignal from ophyd.sim import _ReadbackSignal, _SetpointSignal
from ophyd.utils import LimitError, ReadOnlyError from ophyd.utils import LimitError, ReadOnlyError
logger = bec_logger.logger
class DeviceStop(Exception): class DeviceStop(Exception):
pass pass

3
ophyd_devices/smaract/smaract_ophyd.py
View File

@ -11,8 +11,9 @@ from ophyd.utils import ReadOnlyError, LimitError
from ophyd_devices.smaract.smaract_controller import SmaractController from ophyd_devices.smaract.smaract_controller import SmaractController
from ophyd_devices.smaract.smaract_errors import SmaractCommunicationError, SmaractError from ophyd_devices.smaract.smaract_errors import SmaractCommunicationError, SmaractError
from ophyd_devices.utils.socket import SocketIO, SocketSignal, raise_if_disconnected from ophyd_devices.utils.socket import SocketIO, SocketSignal, raise_if_disconnected
from bec_utils import bec_logger
logger = logging.getLogger("smaract") logger = bec_logger.logger
class SmaractSignalBase(SocketSignal): class SmaractSignalBase(SocketSignal):