refactor: added galil base class and moved lamni-specific galil components to subclass
This commit is contained in:
@@ -1,4 +1,3 @@
|
||||
import functools
|
||||
import threading
|
||||
import time
|
||||
|
||||
@@ -7,15 +6,13 @@ 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 Controller, threadlocked
|
||||
from ophyd_devices.utils.socket import SocketIO, SocketSignal, raise_if_disconnected
|
||||
from prettytable import PrettyTable
|
||||
from ophyd.utils import LimitError
|
||||
from ophyd_devices.utils.controller import threadlocked
|
||||
from ophyd_devices.utils.socket import SocketIO, raise_if_disconnected
|
||||
|
||||
from csaxs_bec.devices.galil.galil_ophyd import (
|
||||
BECConfigError,
|
||||
GalilAxesReferenced,
|
||||
GalilCommunicationError,
|
||||
GalilController,
|
||||
GalilError,
|
||||
GalilMotorIsMoving,
|
||||
|
||||
@@ -1,15 +1,14 @@
|
||||
"""
|
||||
This module contains the base class for Galil controllers as well as the signals used for Galil devices.
|
||||
"""
|
||||
|
||||
import functools
|
||||
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.utils import ReadOnlyError
|
||||
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 SocketSignal
|
||||
from prettytable import PrettyTable
|
||||
|
||||
logger = bec_logger.logger
|
||||
@@ -42,6 +41,10 @@ def retry_once(fcn):
|
||||
|
||||
|
||||
class GalilController(Controller):
|
||||
"""
|
||||
Base class for Galil controllers. This class provides the basic functionality for Galil controllers and should be subclassed for specific devices.
|
||||
"""
|
||||
|
||||
_axes_per_controller = 8
|
||||
USER_ACCESS = [
|
||||
"describe",
|
||||
@@ -49,7 +52,6 @@ class GalilController(Controller):
|
||||
"galil_show_all",
|
||||
"socket_put_and_receive",
|
||||
"socket_put_confirmed",
|
||||
"lgalil_is_air_off_and_orchestra_enabled",
|
||||
"drive_axis_to_limit",
|
||||
"find_reference",
|
||||
"get_motor_limit_switch",
|
||||
@@ -61,18 +63,6 @@ class GalilController(Controller):
|
||||
def socket_put(self, val: str) -> None:
|
||||
self.sock.put(f"{val}\r".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()
|
||||
|
||||
@retry_once
|
||||
def socket_put_confirmed(self, val: str) -> None:
|
||||
"""Send message to controller and ensure that it is received by checking that the socket receives a colon.
|
||||
@@ -107,19 +97,8 @@ class GalilController(Controller):
|
||||
return False
|
||||
return True
|
||||
|
||||
def _remove_trailing_characters(self, var) -> str:
|
||||
if len(var) > 1:
|
||||
return var.split("\r\n")[0]
|
||||
return var
|
||||
|
||||
def stop_all_axes(self) -> str:
|
||||
return self.socket_put_and_receive(f"XQ#STOP,1")
|
||||
|
||||
def lgalil_is_air_off_and_orchestra_enabled(self) -> bool:
|
||||
# TODO: move this to the LamNI-specific controller
|
||||
rt_not_blocked_by_galil = bool(self.socket_put_and_receive(f"MG@OUT[9]"))
|
||||
air_off = bool(self.socket_put_and_receive(f"MG@OUT[13]"))
|
||||
return rt_not_blocked_by_galil and air_off
|
||||
return self.socket_put_and_receive("XQ#STOP,1")
|
||||
|
||||
def get_digital_input(self, channel):
|
||||
return bool(float(self.socket_put_and_receive(f"MG @IN[{channel}]").strip()))
|
||||
@@ -250,40 +229,17 @@ class GalilController(Controller):
|
||||
self.show_running_threads()
|
||||
self.show_status_other()
|
||||
|
||||
def show_status_other(self):
|
||||
# Todo: move to lgalil specific section
|
||||
if self.get_digital_input(5):
|
||||
print("Emergency stop is not pushed.")
|
||||
else:
|
||||
print("Emergency stop is pushed.")
|
||||
if self.get_digital_input(6):
|
||||
print("Driver axis 2 error.")
|
||||
if self.get_digital_input(13):
|
||||
print("No air pressure at inner rotation.")
|
||||
else:
|
||||
print("There is air pressure at the inner rotation.")
|
||||
if self.get_digital_input(14):
|
||||
print("No air pressure at outer rotation axial.")
|
||||
else:
|
||||
print("There is air pressure at the outer rotation axial.")
|
||||
if self.get_digital_input(15):
|
||||
print("No air pressure at outer rotation radial.")
|
||||
else:
|
||||
print("There is air pressure at the outer rotation radial.")
|
||||
swver = float(self.socket_put_and_receive("MGswver"))
|
||||
print(f"Lgalil LAMNI firmware version {swver:2.0f}.")
|
||||
def show_status_other(self) -> None:
|
||||
"""
|
||||
Show additional device-specific status information.
|
||||
Override in subclasses.
|
||||
"""
|
||||
|
||||
def galil_show_all(self) -> None:
|
||||
for controller in self._controller_instances.values():
|
||||
if isinstance(controller, GalilController):
|
||||
controller.describe()
|
||||
|
||||
def lamni_lights_off(self):
|
||||
self.socket_put_confirmed("SB1")
|
||||
|
||||
def lamni_lights_on(self):
|
||||
self.socket_put_confirmed("CB1")
|
||||
|
||||
@staticmethod
|
||||
def axis_Id_to_numeric(axis_Id: str) -> int:
|
||||
return ord(axis_Id.lower()) - 97
|
||||
@@ -419,229 +375,3 @@ class GalilAxesReferenced(GalilSignalRO):
|
||||
@threadlocked
|
||||
def _socket_get(self):
|
||||
return self.controller.all_axes_referenced()
|
||||
|
||||
|
||||
class GalilMotor(Device, PositionerBase):
|
||||
USER_ACCESS = ["controller"]
|
||||
readback = Cpt(GalilReadbackSignal, signal_name="readback", kind="hinted")
|
||||
user_setpoint = Cpt(GalilSetpointSignal, signal_name="setpoint")
|
||||
motor_resolution = Cpt(GalilMotorResolution, signal_name="resolution", kind="config")
|
||||
motor_is_moving = Cpt(GalilMotorIsMoving, signal_name="motor_is_moving", kind="normal")
|
||||
all_axes_referenced = Cpt(GalilAxesReferenced, signal_name="all_axes_referenced", kind="config")
|
||||
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="mpc2680.psi.ch",
|
||||
port=8081,
|
||||
limits=None,
|
||||
sign=1,
|
||||
socket_cls=SocketIO,
|
||||
device_manager=None,
|
||||
**kwargs,
|
||||
):
|
||||
self.controller = GalilController(socket_cls=socket_cls, socket_host=host, socket_port=port)
|
||||
self.axis_Id = axis_Id
|
||||
self.controller.set_axis(axis=self, axis_nr=self.axis_Id_numeric)
|
||||
self.sign = sign
|
||||
self.tolerance = kwargs.pop("tolerance", 0.5)
|
||||
self.device_mapping = kwargs.pop("device_mapping", {})
|
||||
self.device_manager = device_manager
|
||||
|
||||
if len(self.device_mapping) > 0 and self.device_manager is None:
|
||||
raise BECConfigError(
|
||||
"device_mapping has been specified but the device_manager cannot be accessed."
|
||||
)
|
||||
self.rt = self.device_mapping.get("rt")
|
||||
|
||||
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()
|
||||
# self.readback.subscribe(self._forward_readback, event_type=self.readback.SUB_VALUE)
|
||||
|
||||
if limits is not None:
|
||||
assert len(limits) == 2
|
||||
self.low_limit_travel.put(limits[0])
|
||||
self.high_limit_travel.put(limits[1])
|
||||
|
||||
@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
|
||||
|
||||
def _forward_readback(self, **kwargs):
|
||||
kwargs.pop("sub_type")
|
||||
self._run_subs(sub_type="readback", **kwargs)
|
||||
|
||||
@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", 100)
|
||||
status = super().move(position, timeout=timeout, **kwargs)
|
||||
self.user_setpoint.put(position, wait=False)
|
||||
|
||||
def move_and_finish():
|
||||
while self.motor_is_moving.get():
|
||||
logger.info("motor is moving")
|
||||
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)
|
||||
|
||||
if not success:
|
||||
print(" stop")
|
||||
self._done_moving(success=success)
|
||||
logger.info("Move finished")
|
||||
|
||||
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 = self.controller.axis_Id_to_numeric(val)
|
||||
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 = self.controller.axis_Id_numeric_to_alpha(val)
|
||||
self._axis_Id_numeric = val
|
||||
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__":
|
||||
# pytest: skip-file
|
||||
mock = False
|
||||
if not mock:
|
||||
leyey = GalilMotor("H", name="leyey", host="mpc2680.psi.ch", port=8081, sign=-1)
|
||||
leyey.stage()
|
||||
status = leyey.move(0, wait=True)
|
||||
status = leyey.move(10, wait=True)
|
||||
leyey.read()
|
||||
|
||||
leyey.get()
|
||||
leyey.describe()
|
||||
|
||||
leyey.unstage()
|
||||
else:
|
||||
from ophyd_devices.utils.socket import SocketMock
|
||||
|
||||
leyex = GalilMotor(
|
||||
"G", name="leyex", host="mpc2680.psi.ch", port=8081, socket_cls=SocketMock
|
||||
)
|
||||
leyey = GalilMotor(
|
||||
"H", name="leyey", host="mpc2680.psi.ch", port=8081, socket_cls=SocketMock
|
||||
)
|
||||
leyex.stage()
|
||||
# leyey.stage()
|
||||
|
||||
leyex.controller.galil_show_all()
|
||||
|
||||
@@ -0,0 +1,263 @@
|
||||
import threading
|
||||
import time
|
||||
|
||||
import numpy as np
|
||||
from bec_lib.logger 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
|
||||
from ophyd_devices.utils.socket import SocketIO, raise_if_disconnected
|
||||
|
||||
from .galil_ophyd import (
|
||||
BECConfigError,
|
||||
GalilAxesReferenced,
|
||||
GalilController,
|
||||
GalilMotorIsMoving,
|
||||
GalilMotorResolution,
|
||||
GalilReadbackSignal,
|
||||
GalilSetpointSignal,
|
||||
)
|
||||
|
||||
logger = bec_logger.logger
|
||||
|
||||
|
||||
class LamniGalilController(GalilController):
|
||||
USER_ACCESS = [
|
||||
"describe",
|
||||
"show_running_threads",
|
||||
"galil_show_all",
|
||||
"socket_put_and_receive",
|
||||
"socket_put_confirmed",
|
||||
"lgalil_is_air_off_and_orchestra_enabled",
|
||||
"drive_axis_to_limit",
|
||||
"find_reference",
|
||||
"get_motor_limit_switch",
|
||||
"is_motor_on",
|
||||
"all_axes_referenced",
|
||||
]
|
||||
|
||||
def show_status_other(self):
|
||||
if self.get_digital_input(5):
|
||||
print("Emergency stop is not pushed.")
|
||||
else:
|
||||
print("Emergency stop is pushed.")
|
||||
if self.get_digital_input(6):
|
||||
print("Driver axis 2 error.")
|
||||
if self.get_digital_input(13):
|
||||
print("No air pressure at inner rotation.")
|
||||
else:
|
||||
print("There is air pressure at the inner rotation.")
|
||||
if self.get_digital_input(14):
|
||||
print("No air pressure at outer rotation axial.")
|
||||
else:
|
||||
print("There is air pressure at the outer rotation axial.")
|
||||
if self.get_digital_input(15):
|
||||
print("No air pressure at outer rotation radial.")
|
||||
else:
|
||||
print("There is air pressure at the outer rotation radial.")
|
||||
swver = float(self.socket_put_and_receive("MGswver"))
|
||||
print(f"Lgalil LAMNI firmware version {swver:2.0f}.")
|
||||
|
||||
def lamni_lights_off(self):
|
||||
self.socket_put_confirmed("SB1")
|
||||
|
||||
def lamni_lights_on(self):
|
||||
self.socket_put_confirmed("CB1")
|
||||
|
||||
def lgalil_is_air_off_and_orchestra_enabled(self) -> bool:
|
||||
# TODO: move this to the LamNI-specific controller
|
||||
rt_not_blocked_by_galil = bool(self.socket_put_and_receive("MG@OUT[9]"))
|
||||
air_off = bool(self.socket_put_and_receive("MG@OUT[13]"))
|
||||
return rt_not_blocked_by_galil and air_off
|
||||
|
||||
|
||||
class LamniGalilMotor(Device, PositionerBase):
|
||||
USER_ACCESS = ["controller"]
|
||||
readback = Cpt(GalilReadbackSignal, signal_name="readback", kind="hinted")
|
||||
user_setpoint = Cpt(GalilSetpointSignal, signal_name="setpoint")
|
||||
motor_resolution = Cpt(GalilMotorResolution, signal_name="resolution", kind="config")
|
||||
motor_is_moving = Cpt(GalilMotorIsMoving, signal_name="motor_is_moving", kind="normal")
|
||||
all_axes_referenced = Cpt(GalilAxesReferenced, signal_name="all_axes_referenced", kind="config")
|
||||
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="mpc2680.psi.ch",
|
||||
port=8081,
|
||||
limits=None,
|
||||
sign=1,
|
||||
socket_cls=SocketIO,
|
||||
device_manager=None,
|
||||
**kwargs,
|
||||
):
|
||||
self.controller = GalilController(socket_cls=socket_cls, socket_host=host, socket_port=port)
|
||||
self.axis_Id = axis_Id
|
||||
self.controller.set_axis(axis=self, axis_nr=self.axis_Id_numeric)
|
||||
self.sign = sign
|
||||
self.tolerance = kwargs.pop("tolerance", 0.5)
|
||||
self.device_mapping = kwargs.pop("device_mapping", {})
|
||||
self.device_manager = device_manager
|
||||
|
||||
if len(self.device_mapping) > 0 and self.device_manager is None:
|
||||
raise BECConfigError(
|
||||
"device_mapping has been specified but the device_manager cannot be accessed."
|
||||
)
|
||||
self.rt = self.device_mapping.get("rt")
|
||||
|
||||
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()
|
||||
# self.readback.subscribe(self._forward_readback, event_type=self.readback.SUB_VALUE)
|
||||
|
||||
if limits is not None:
|
||||
assert len(limits) == 2
|
||||
self.low_limit_travel.put(limits[0])
|
||||
self.high_limit_travel.put(limits[1])
|
||||
|
||||
@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
|
||||
|
||||
def _forward_readback(self, **kwargs):
|
||||
kwargs.pop("sub_type")
|
||||
self._run_subs(sub_type="readback", **kwargs)
|
||||
|
||||
@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", 100)
|
||||
status = super().move(position, timeout=timeout, **kwargs)
|
||||
self.user_setpoint.put(position, wait=False)
|
||||
|
||||
def move_and_finish():
|
||||
while self.motor_is_moving.get():
|
||||
logger.info("motor is moving")
|
||||
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)
|
||||
|
||||
if not success:
|
||||
print(" stop")
|
||||
self._done_moving(success=success)
|
||||
logger.info("Move finished")
|
||||
|
||||
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("Only single-character axis_Ids are supported.")
|
||||
self._axis_Id_alpha = val
|
||||
self._axis_Id_numeric = self.controller.axis_Id_to_numeric(val)
|
||||
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("Numeric value exceeds supported range.")
|
||||
self._axis_Id_alpha = self.controller.axis_Id_numeric_to_alpha(val)
|
||||
self._axis_Id_numeric = val
|
||||
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 stop(self, *, success=False):
|
||||
self.controller.stop_all_axes()
|
||||
return super().stop(success=success)
|
||||
Reference in New Issue
Block a user