mx_lib: tidied up wait functions and removed redundant code.
This commit is contained in:
+53
-55
@@ -1,44 +1,9 @@
|
||||
import datetime
|
||||
import re
|
||||
import time
|
||||
from typing import Callable, Union
|
||||
from typing import Callable, Union, Any
|
||||
|
||||
from epics import PV, Motor, poll
|
||||
|
||||
|
||||
def timestamp():
|
||||
"""
|
||||
Returns a fixed width string (15 characters) with a timestamp
|
||||
in the format (24Hour:Minute:Second.Microsecond).
|
||||
|
||||
Example: 13:09:43.009508
|
||||
"""
|
||||
x = datetime.datetime(1, 1, 1).now()
|
||||
return "%4d-%02d-%02d %02d:%02d:%02d,%03d" % (
|
||||
x.year,
|
||||
x.month,
|
||||
x.day,
|
||||
x.hour,
|
||||
x.minute,
|
||||
x.second,
|
||||
x.microsecond / 1000,
|
||||
)
|
||||
|
||||
|
||||
def itoa(x, base=10):
|
||||
is_negative = x < 0
|
||||
if is_negative:
|
||||
x = -x
|
||||
digits = []
|
||||
while x > 0:
|
||||
x, last_digit = divmod(x, base)
|
||||
digits.append("0123456789abcdefghijklmnopqrstuvwxyz"[last_digit])
|
||||
if is_negative:
|
||||
digits.append("-")
|
||||
digits.reverse()
|
||||
return "".join(digits)
|
||||
|
||||
|
||||
def wait_for_movement_to_finish(*motors):
|
||||
"""
|
||||
Wait for all {motors} passed in argument to finish movement.
|
||||
@@ -51,10 +16,6 @@ def wait_for_movement_to_finish(*motors):
|
||||
|
||||
Returns: nothing
|
||||
"""
|
||||
# timeout = 1.5 * max([abs(motor.get_position(readback=True) -
|
||||
# motor.get_position()) / motor.slew_speed
|
||||
# for motor in motors])
|
||||
#
|
||||
poll(0.3)
|
||||
longest = 0.0
|
||||
for motor in motors:
|
||||
@@ -77,15 +38,31 @@ class ValueWaitTimeout(Exception):
|
||||
pass
|
||||
|
||||
|
||||
def pv_wait(pv, value, *, timeout=30.0, polling=0.2, tolerance=None, verbose=False):
|
||||
if not (isinstance(pv, PV) or isinstance(pv, Motor)):
|
||||
raise ValueError("wait what!? what pv?!")
|
||||
|
||||
def pv_wait(pv: PV | Motor, value: Any, *, timeout: float = 60.0,
|
||||
polling: float = 0.1, tolerance: float | None=None, verbose: bool =False):
|
||||
"""wait until an epics.PV reaches a value
|
||||
pv: epics.PV | epics.Motor
|
||||
the PV on which you want to wait on
|
||||
value/target: any
|
||||
this value depends on the PV type: str, enum, double, ...
|
||||
:timeout: float default = 60.0
|
||||
timeout in seconds
|
||||
:polling: float default = 0.1
|
||||
polling interval in seconds
|
||||
tolerance: float or None
|
||||
provide a tolerance to accept when comparing values, currently None by default, however certain PVs and motors
|
||||
have inbuilt tolerances that can be used instead
|
||||
verbose: bool default = False
|
||||
NotImplemented
|
||||
"""
|
||||
if isinstance(pv, Motor):
|
||||
|
||||
def checker(m, target=None, tolerance=None):
|
||||
if tolerance is None:
|
||||
tolerance = m.get("RDBD")
|
||||
if tolerance is None:
|
||||
print(f"WARNING: motor {m._prefix[:-1]} has no RDBD, using 0.001")
|
||||
tolerance = 0.001 #FIXME
|
||||
if target is None:
|
||||
target = m.drive
|
||||
|
||||
@@ -107,14 +84,25 @@ def pv_wait(pv, value, *, timeout=30.0, polling=0.2, tolerance=None, verbose=Fal
|
||||
|
||||
|
||||
def is_epics_type(pv: PV, pv_type: str) -> bool:
|
||||
"""Check to see if a PV is of a certain type such as double, enum, string, ..."""
|
||||
if isinstance(pv_type, type):
|
||||
pv_type = pv_type.__name__
|
||||
|
||||
return pv_type == pv.type
|
||||
|
||||
|
||||
def wait_string_condition(pv, target: Union[str, re.Pattern], *, timeout=60.0, polling=0.1):
|
||||
if not (isinstance(pv, PV) or "string" not in pv.type):
|
||||
def wait_string_condition(pv: PV, target: Union[str, re.Pattern], *, timeout: float = 60.0, polling: float = 0.1):
|
||||
"""wait until an epics.PV of type string reaches target
|
||||
:pv: epics.PV
|
||||
PV should be of type string
|
||||
:target: str or re.Pattern
|
||||
:timeout: float default = 60.0
|
||||
timeout in seconds
|
||||
:polling: float default = 0.1
|
||||
polling interval in seconds
|
||||
"""
|
||||
|
||||
if not (isinstance(pv, PV) and "string" not in pv.type):
|
||||
raise AttributeError("argument 'pv' must be an epics.PV of type string")
|
||||
|
||||
if not isinstance(target, re.Pattern):
|
||||
@@ -131,7 +119,8 @@ def wait_string_condition(pv, target: Union[str, re.Pattern], *, timeout=60.0, p
|
||||
raise TimeoutError(f"timeout waiting for string {pv.pvname} == {target}; actual value == {pv.char_value}")
|
||||
|
||||
|
||||
def wait_float_condition(pv: PV, value, *, timeout: float = 60.0, **kwargs):
|
||||
def wait_float_condition(pv: PV, value, *, timeout: float = 60.0,
|
||||
polling: float = 0.1, tolerance: float| None = None):
|
||||
"""wait until an epics.PV of type double reaches value
|
||||
pv: epics.PV
|
||||
the PV enum on which you want to wait on
|
||||
@@ -139,19 +128,22 @@ def wait_float_condition(pv: PV, value, *, timeout: float = 60.0, **kwargs):
|
||||
value: float
|
||||
the target value
|
||||
|
||||
polling: float
|
||||
how often pv is checked during wait loop
|
||||
|
||||
tolerance: float or None
|
||||
the tolerance to accept when comparing values, if None (default) we
|
||||
try to figure an appropriate value
|
||||
|
||||
|
||||
timeout: double
|
||||
timeout: float default = 60.0
|
||||
a timeout in seconds
|
||||
|
||||
return: nothing
|
||||
|
||||
raises: TimeoutError if a timeout occurs
|
||||
"""
|
||||
if not (isinstance(pv, PV) or "double" not in pv.type):
|
||||
if not (isinstance(pv, PV) and "double" not in pv.type):
|
||||
raise AttributeError("argument 'pv' must be an epics.PV of type double")
|
||||
|
||||
try:
|
||||
@@ -159,10 +151,16 @@ def wait_float_condition(pv: PV, value, *, timeout: float = 60.0, **kwargs):
|
||||
except ValueError:
|
||||
raise AttributeError("argument 'value' must be a number")
|
||||
|
||||
tolerance = kwargs.get("tolerance", pow(10, -(pv.precision - 1))) # type: ignore
|
||||
polling = kwargs.get("polling", 0.1)
|
||||
if tolerance is None:
|
||||
# If pv.precision is missing/None, fall back to a default
|
||||
precision = getattr(pv, "precision", None)
|
||||
if precision is None:
|
||||
tolerance = 1e-6
|
||||
else:
|
||||
tolerance = pow(10, -(precision - 1))
|
||||
|
||||
tout = time.time() + timeout
|
||||
|
||||
while time.time() < tout:
|
||||
if abs(pv.value - value) < tolerance:
|
||||
return
|
||||
@@ -190,13 +188,13 @@ def wait_motor_position(motor: Motor, tester: Callable, *, timeout: float = 50.0
|
||||
|
||||
raises: TimeoutError if a timeout occurs
|
||||
"""
|
||||
if not isinstance(tester, Callable):
|
||||
if not callable(tester):
|
||||
raise RuntimeError("argument 'tester' must be a function")
|
||||
|
||||
try:
|
||||
move_time = abs(motor.drive - motor.readback) / motor.speed
|
||||
except Exception:
|
||||
move_time = 1.0 # in case we're dealing with weird motor record
|
||||
move_time = 1.0 # in case of unusual motor record
|
||||
|
||||
tout = move_time + time.time() + timeout
|
||||
|
||||
@@ -209,7 +207,7 @@ def wait_motor_position(motor: Motor, tester: Callable, *, timeout: float = 50.0
|
||||
raise TimeoutError(f"timeout waiting for a condition on {motor} {motor.drive} != {motor.readback}")
|
||||
|
||||
|
||||
def wait_enum_condition(pv: PV, value: Union[str, int, re.Pattern], *, timeout: float = 60.0, polling=0.1, **kwargs):
|
||||
def wait_enum_condition(pv: PV, value: Union[str, int, re.Pattern], *, timeout: float = 60.0, polling=0.1):
|
||||
"""wait until an epics.PV enum reaches value
|
||||
pv: epics.PV
|
||||
the PV enum on which you want to wait on
|
||||
@@ -224,7 +222,7 @@ def wait_enum_condition(pv: PV, value: Union[str, int, re.Pattern], *, timeout:
|
||||
|
||||
raises: TimeoutError if a timeout occurs
|
||||
"""
|
||||
if not (isinstance(pv, PV) or "enum" != pv.type[-4:].lower()):
|
||||
if not (isinstance(pv, PV) and pv.type.lower().endswith("enum")):
|
||||
raise AttributeError("argument 'pv' must be an epics.PV of type enum")
|
||||
|
||||
if not (isinstance(value, str) or isinstance(value, int) or isinstance(value, re.Pattern)):
|
||||
|
||||
Reference in New Issue
Block a user