127 lines
3.9 KiB
Python
Executable File
127 lines
3.9 KiB
Python
Executable File
from typing import cast
|
|
|
|
from epics import PV
|
|
|
|
from .mx_lib import ValueWaitTimeout, pv_wait
|
|
|
|
|
|
class EnumPv(object):
|
|
def __init__(self, args):
|
|
mandatory_args = {"pv", "readback", "states"}
|
|
supplied_args = set(args.keys())
|
|
if not mandatory_args.issubset(supplied_args):
|
|
raise RuntimeError("Missing arguments: %s" % (", ".join(mandatory_args.difference(supplied_args))))
|
|
|
|
if type(args["states"]) is not dict:
|
|
raise RuntimeError("states must be a dictionary mapping put/readback for waiting purposes")
|
|
|
|
self.val = PV(args["pv"])
|
|
self.rbv = PV(args["readback"])
|
|
|
|
if "name" in args:
|
|
self.device_name = args["name"]
|
|
else:
|
|
self.device_name = self.val.pvname
|
|
|
|
if not hasattr(self.val, "enum_strs"):
|
|
raise RuntimeError("PV %s does not have enum_strs" % self.val.pvname)
|
|
if not hasattr(self.rbv, "enum_strs"):
|
|
raise RuntimeError("PV %s does not have enum_strs" % self.rbv.pvname)
|
|
|
|
self.__target = None
|
|
|
|
p = args.get("move_done_when")
|
|
if p is not None:
|
|
self.__wait_pv = PV(p[0])
|
|
self.__wait_target = p[1]
|
|
else:
|
|
self.__wait_pv = None
|
|
self.__wait_target = None
|
|
|
|
self.states = {}
|
|
for k, v in list(args["states"].items()):
|
|
self.states[k.upper()] = v.upper()
|
|
|
|
self.val_enums = [x.upper() for x in self.val.enum_strs] # type: ignore
|
|
self.rbv_enums = [x.upper() for x in self.rbv.enum_strs] # type: ignore
|
|
|
|
if "timeout" not in args:
|
|
self.__timeout = 60 # seconds
|
|
else:
|
|
self.__timeout = args["timeout"]
|
|
|
|
def __str__(self):
|
|
return f"<{self.device_name} at {self.position}>"
|
|
|
|
def __repr__(self):
|
|
s = (
|
|
f"<{self.device_name} at {self.position} an {self.__class__.__name__} "
|
|
f"instance at {hex(id(self))} positions = {self.positions}>"
|
|
)
|
|
return s
|
|
|
|
def has_position(self, position):
|
|
return position.upper() in self.states
|
|
|
|
@property
|
|
def positions(self):
|
|
return self.rbv_enums
|
|
|
|
def __position(self) -> str:
|
|
return self.rbv.enum_strs[self.rbv.get()] # type: ignore
|
|
|
|
position = property(__position)
|
|
|
|
def position_is(self, position) -> bool:
|
|
if type(position) in (int, float):
|
|
return int(position) == self.get()
|
|
else:
|
|
return str(position).upper() == str(self.__position()).upper()
|
|
|
|
def get(self, req_type=None) -> str | int:
|
|
if req_type:
|
|
res = self.position
|
|
else:
|
|
res = cast(int, self.rbv.get())
|
|
return res
|
|
|
|
def put(self, value, wait=False) -> None:
|
|
self.val.put(value)
|
|
self.__target = self.rbv_enums.index(self.states[self.val_enums[value]])
|
|
if wait:
|
|
self.wait()
|
|
|
|
def move(self, position, wait=False) -> None:
|
|
if type(position) is not int:
|
|
if not self.has_position(position):
|
|
raise ValueError(f"Invalid position {position} for {self.device_name}")
|
|
target = self.val_enums.index(position.upper())
|
|
else:
|
|
target = position
|
|
|
|
self.put(target, wait=wait)
|
|
|
|
def equal_dbr_string(self, val) -> bool:
|
|
value = self.rbv.get(as_string=True)
|
|
return cast(str, value).lower() == val.lower()
|
|
|
|
def equal_dbr_int(self, val) -> bool:
|
|
return val == self.rbv.get()
|
|
|
|
def wait(self, target=None):
|
|
if target is None:
|
|
target = self.__target
|
|
|
|
if self.__wait_pv:
|
|
pv = self.__wait_pv
|
|
else:
|
|
pv = self.rbv
|
|
|
|
if target is None:
|
|
target = self.__wait_target
|
|
|
|
try:
|
|
pv_wait(pv, target, timeout=self.__timeout, verbose=True)
|
|
except ValueWaitTimeout:
|
|
raise ValueWaitTimeout(f"Timeout waiting for device to reach [{target}], currently at [{self.position}]")
|