282 lines
9.0 KiB
Python
282 lines
9.0 KiB
Python
import time
|
|
from typing import Callable, cast
|
|
|
|
from epics import PV, poll
|
|
from epics.ca import pend_io
|
|
|
|
|
|
def is_number(x):
|
|
return isinstance(x, (int, float))
|
|
|
|
|
|
class NonStandard(object):
|
|
def __init__(self, **kwargs):
|
|
mandatory = ["name", "setpv", "getpv"]
|
|
if set(mandatory) != set(kwargs.keys()) & set(mandatory):
|
|
raise RuntimeError(
|
|
"missing mandatory argument(s): {}".format(str(list(set(mandatory) - set(kwargs.keys()))))
|
|
)
|
|
self.name = kwargs.get("name")
|
|
self.__target_pos: float | int | str = 0
|
|
self.device_name = self.name
|
|
self.__setpv = PV(kwargs["setpv"])
|
|
self.__getpv = PV(kwargs["getpv"])
|
|
self.__timeout_margin = kwargs.get("timeout", 2.0) # add this safety margin to timeout calc from speed
|
|
self.__timeout = kwargs.get("timeout", 60.0)
|
|
self.__format = kwargs.get("format")
|
|
|
|
p = kwargs.get("speed")
|
|
if p is not None:
|
|
if is_number(p):
|
|
self.__speed_set, self.__speed_get = p, p
|
|
elif type(p) in [tuple, list] and len(p) == 2:
|
|
self.__speed_set = PV(p[0])
|
|
self.__speed_get = PV(p[1])
|
|
else:
|
|
raise Exception(f"{self.name} => don't know how to parse speed parameter")
|
|
else:
|
|
self.__speed_set, self.__speed_get = None, None
|
|
#
|
|
# move_done_when a tuple, first item the pv to read
|
|
# second item, the int type indicating that it is DONE
|
|
#
|
|
p = kwargs.get("move_done_when")
|
|
if p is not None:
|
|
self.__move_flag = PV(p[0])
|
|
self.__move_flag_done = p[1]
|
|
else:
|
|
self.__move_flag = None
|
|
|
|
#
|
|
# stoppv a tuple, first item the pv to read
|
|
# second item, the int type to put to stop the motor
|
|
#
|
|
p = kwargs.get("stoppv")
|
|
if p is not None:
|
|
self.__stoppv = PV(p[0])
|
|
self.__stoppv_done = p[1]
|
|
else:
|
|
self.__stoppv = None
|
|
|
|
#
|
|
# still_moving_when a tuple, first item the pv to read
|
|
# second item, the int type indicating that it is still moving.
|
|
#
|
|
p = kwargs.get("still_moving_when")
|
|
if p is not None:
|
|
self.__still_moving_flag = PV(p[0])
|
|
self.__still_moving_flag_value = p[1]
|
|
else:
|
|
self.__still_moving_flag = None
|
|
|
|
p = kwargs.get("predefs")
|
|
if p:
|
|
self.__predefs = p
|
|
self.positions = list(p.keys())
|
|
else:
|
|
self.__predefs = {}
|
|
self.positions = []
|
|
|
|
pend_io(5.0)
|
|
|
|
typ = self.__ret_type = self.__getpv.type
|
|
|
|
match str(typ):
|
|
case "double" | "time_double":
|
|
self.still_moving = self.__moving_float
|
|
case "int" | "time_int":
|
|
self.still_moving = self.__moving_int
|
|
case "string" | "time_string":
|
|
self.still_moving = self.__moving_str
|
|
case _:
|
|
self.still_moving = self.__moving_unknown
|
|
|
|
tolerance = kwargs.get("tolerance", None)
|
|
if tolerance is not None and tolerance > 0:
|
|
self.tolerance = tolerance
|
|
elif tolerance is not None and tolerance < 0:
|
|
self.still_moving = self.__moving_unknown
|
|
|
|
if self.__move_flag:
|
|
self.still_moving = self.is_move_not_done
|
|
elif self.__still_moving_flag:
|
|
self.still_moving = self.still_moving_when
|
|
|
|
def __str__(self):
|
|
if self.positions:
|
|
return "<{} at {}({}) Pre-defined positions {}>".format(
|
|
self.name, self.__position(predef=True), self.position, self.positions
|
|
)
|
|
else:
|
|
return "<{} at {}>".format(self.name, self.__position())
|
|
|
|
def __repr__(self):
|
|
if self.positions:
|
|
return "<{} instance at {}: {} at {}({}) Pre-defined positions {}>".format(
|
|
self.__class__.__name__,
|
|
hex(id(self)),
|
|
self.name,
|
|
self.__position(predef=True),
|
|
self.position,
|
|
self.positions,
|
|
)
|
|
else:
|
|
return "<{} instance at {}: {} at {}>".format(
|
|
self.__class__.__name__, hex(id(self)), self.name, self.__position()
|
|
)
|
|
|
|
def still_moving_when(self):
|
|
return self.__still_moving_flag_value == self.__still_moving_flag.get() # type: ignore
|
|
|
|
def is_moving(self):
|
|
return not self.is_move_done()
|
|
|
|
def is_move_done(self):
|
|
return self.__move_flag_done == self.__move_flag.get() # type: ignore
|
|
|
|
def is_move_not_done(self):
|
|
return not self.is_move_done()
|
|
|
|
def __moving_float(self):
|
|
if type(self.__target_pos) is not float:
|
|
raise RuntimeError("target position is not an float")
|
|
current_value = cast(float, self.__getpv.get())
|
|
return self.tolerance < abs(cast(float, self.__target_pos) - current_value)
|
|
|
|
def __moving_int(self):
|
|
if type(self.__target_pos) is not int:
|
|
raise RuntimeError("target position is not an integer")
|
|
return self.__target_pos != self.__getpv.get()
|
|
|
|
def __moving_str(self):
|
|
if type(self.__target_pos) is not str:
|
|
raise RuntimeError("target position is not a string")
|
|
current_value = cast(str, self.__getpv.get())
|
|
return self.__target_pos.upper() != current_value.upper()
|
|
|
|
def __moving_unknown(self):
|
|
return False
|
|
|
|
def __position(self, predef: bool = False):
|
|
"""returns either a predefined position or a readback
|
|
|
|
:rtype: Union[str, float, int]
|
|
"""
|
|
cur = self.readback
|
|
if not predef:
|
|
return cur
|
|
# loop over predefined positions, if one found and
|
|
# readback matches it's value return "position"
|
|
for pos_label, val_or_cbk in self.__predefs.items():
|
|
if self.__ret_type is bytes:
|
|
tst = val_or_cbk == cur
|
|
elif type(val_or_cbk) is tuple:
|
|
# a tuple (Callable, arg-to-Callable)
|
|
val = val_or_cbk[0](*val_or_cbk[1])
|
|
tst = 0.1 > abs(val - cur) # FIXME sooo bad
|
|
else:
|
|
tst = 0.1 > abs(val_or_cbk - cur)
|
|
if tst:
|
|
return pos_label
|
|
return "unknown"
|
|
|
|
def __readback(self):
|
|
return self.__getpv.get()
|
|
|
|
readback = property(__readback)
|
|
position = property(__position)
|
|
|
|
def has_position(self, position: str) -> bool:
|
|
return position.lower() in [p.lower() for p in self.positions]
|
|
|
|
def position_is(self, position, tolerance=0.1):
|
|
pos = self.__predefs.get(position, position)
|
|
cur = self.position
|
|
|
|
print(f"position_is: {pos} == {cur}")
|
|
|
|
if type(pos) is tuple:
|
|
pos = pos[0](*pos[1])
|
|
|
|
print(f"position_is: {pos} == {cur}")
|
|
|
|
if self.__ret_type is bytes:
|
|
print("checking bytes")
|
|
return pos == cur
|
|
else:
|
|
print(f"checking numbers: abs({pos} - {cur}) < {tolerance}")
|
|
return abs(pos - cur) < tolerance
|
|
|
|
def __value(self):
|
|
val = self.__getpv.get()
|
|
if self.__format is not None:
|
|
val = self.__format % (val,)
|
|
return val
|
|
|
|
value = property(__value)
|
|
|
|
def __raw_value(self):
|
|
return self.__getpv.get()
|
|
|
|
raw_value = property(__raw_value)
|
|
|
|
def __char_value(self):
|
|
return str(self.value)
|
|
|
|
char_value = property(__char_value)
|
|
|
|
def set_speed(self, speed):
|
|
if isinstance(self.__speed_set, PV):
|
|
self.__speed_set.put(speed)
|
|
elif is_number(self.__speed_set):
|
|
self.__speed_set = speed
|
|
self.__speed_get = speed
|
|
|
|
def get_speed(self):
|
|
if is_number(self.__speed_get):
|
|
return self.__speed_get
|
|
elif isinstance(self.__speed_get, PV):
|
|
return self.__speed_get.value
|
|
else:
|
|
return None
|
|
|
|
def get_timeout(self):
|
|
if not self.__speed_get:
|
|
timeout = self.__timeout
|
|
else:
|
|
speed = self.get_speed()
|
|
curp = self.position
|
|
timeout = self.__timeout_margin + (abs(curp - self.__target_pos) / speed)
|
|
return timeout
|
|
|
|
def wait(self):
|
|
pos = self.__target_pos
|
|
timeout = time.time() + self.get_timeout()
|
|
|
|
while time.time() < timeout and self.still_moving():
|
|
poll(0.1)
|
|
if time.time() > timeout:
|
|
msg = "timeout occurred when moving %s to %s" % (self.name, pos)
|
|
print(msg)
|
|
raise RuntimeWarning(msg)
|
|
|
|
def move(self, pos, relative=False, wait=False):
|
|
pos = self.__predefs.get(pos, pos)
|
|
if type(pos) is tuple:
|
|
pos = pos[0](*pos[1])
|
|
|
|
if relative:
|
|
pos += self.__getpv.get()
|
|
|
|
self.__target_pos = pos
|
|
self.__setpv.put(pos)
|
|
if wait:
|
|
time.sleep(0.1)
|
|
self.wait()
|
|
|
|
def stop(self):
|
|
if self.__stoppv:
|
|
self.__stoppv.put(self.__stoppv_done)
|
|
else:
|
|
raise RuntimeError("stop pv not configured for this motor")
|