Devices: non_standard rewrite, WIP
This commit is contained in:
@@ -1,285 +1,85 @@
|
||||
import time
|
||||
from typing import Callable, cast
|
||||
|
||||
from typing import Any, Dict, Optional, Union
|
||||
from epics import PV, poll
|
||||
from epics.ca import pend_io
|
||||
from aare.devices.mx_lib import pv_wait
|
||||
|
||||
|
||||
def is_number(x):
|
||||
return isinstance(x, (int, float))
|
||||
class NonStandard:
|
||||
"""
|
||||
A wrapper for EPICS PVs that have separate setpoint and readback PVs.
|
||||
Similar to EnumPv but supports arbitrary numeric or string types and predefined positions.
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
name: str,
|
||||
setpv: str,
|
||||
getpv: str,
|
||||
timeout: float = 60.0,
|
||||
tolerance: Optional[float] = None,
|
||||
predefs: Optional[Dict[str, Any]] = None,
|
||||
**kwargs
|
||||
):
|
||||
self.name = name
|
||||
self.device_name = name
|
||||
self.setpoint_pv = PV(setpv)
|
||||
self.readback_pv = PV(getpv)
|
||||
self.default_timeout = timeout
|
||||
self.tolerance = tolerance
|
||||
|
||||
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")
|
||||
self._predefs = predefs or {}
|
||||
self.positions = list(self._predefs.keys())
|
||||
self._target_pos: Optional[Any] = None
|
||||
if kwargs.get("move_done_when"):
|
||||
self.__move_done_when = PV(kwargs.get("move_done_when"))
|
||||
|
||||
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
|
||||
@property
|
||||
def value(self) -> Any:
|
||||
"""Alias for position to maintain compatibility."""
|
||||
if not self.readback_pv.connected:
|
||||
return None
|
||||
return self.readback_pv.get()
|
||||
|
||||
#
|
||||
# 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
|
||||
def get_predefined_name(self) -> str:
|
||||
"""Returns the name of the predefined position if current position matches one."""
|
||||
current = self.value
|
||||
for name, val in self._predefs.items():
|
||||
# Handle callable predefs (like offsets)
|
||||
target = val[0](*val[1]) if isinstance(val, tuple) and callable(val[0]) else val
|
||||
|
||||
#
|
||||
# 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
|
||||
# Use a small tolerance for floats if not provided
|
||||
tol = self.tolerance if self.tolerance is not None else 0.1
|
||||
try:
|
||||
if abs(current - target) < tol:
|
||||
return name
|
||||
except (TypeError, ValueError):
|
||||
if current == target:
|
||||
return name
|
||||
return "unknown"
|
||||
|
||||
def __readback(self):
|
||||
return self.__getpv.get()
|
||||
def move(self, pos: Union[str, float, int], wait: bool = False, timeout: Optional[float] = None):
|
||||
"""
|
||||
Move to a position.
|
||||
'pos' can be a raw value or a key from the predefs dictionary.
|
||||
"""
|
||||
# Resolve predefined position
|
||||
target = self._predefs.get(pos, pos)
|
||||
if isinstance(target, tuple) and callable(target[0]):
|
||||
target = target[0](*target[1])
|
||||
|
||||
readback = property(__readback)
|
||||
position = property(__position)
|
||||
self._target_pos = target
|
||||
self.setpoint_pv.put(target)
|
||||
|
||||
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
|
||||
if curp is None or self.__target_pos is None or speed is None:
|
||||
timeout = self.__timeout_margin
|
||||
raise RuntimeWarning("can't calculate timeout, ABR may be disconnected")
|
||||
else:
|
||||
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()
|
||||
self.wait(timeout=timeout)
|
||||
|
||||
def stop(self):
|
||||
if self.__stoppv:
|
||||
self.__stoppv.put(self.__stoppv_done)
|
||||
else:
|
||||
raise RuntimeError("stop pv not configured for this motor")
|
||||
def wait(self, timeout: Optional[float] = None):
|
||||
"""Wait for the readback to match the last commanded target position."""
|
||||
if self._target_pos is None:
|
||||
return
|
||||
|
||||
tout = timeout or self.default_timeout
|
||||
pv_wait(self.readback_pv, self._target_pos, timeout=tout, tolerance=self.tolerance)
|
||||
|
||||
def __repr__(self):
|
||||
return f"<{self.__class__.__name__} '{self.name}' at {self.value}>"
|
||||
|
||||
Reference in New Issue
Block a user