252 lines
7.6 KiB
Python
252 lines
7.6 KiB
Python
import datetime
|
|
import re
|
|
import time
|
|
from typing import Callable, Union
|
|
|
|
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.
|
|
|
|
Requirements: epics.Motor -- each {motor} is an instance of this
|
|
class.
|
|
|
|
The overall timeout is the maximum of (.RBV - .VAL)/.VELO for each
|
|
concerned {motor}.
|
|
|
|
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:
|
|
time_to_target = motor.readback / motor.slew_speed
|
|
longest = time_to_target if time_to_target > longest else longest
|
|
|
|
timeout = time.time() + 1.5 * longest
|
|
done = False
|
|
while not done and time.time() < timeout:
|
|
done = all([m.done_moving for m in motors])
|
|
|
|
if time.time() > timeout:
|
|
print("TIMEOUT waiting for motors to be done moving; current motor positions:")
|
|
for m in motors:
|
|
print(f"...{m._prefix[:-1]} ({m.units}) VAL={m.drive:.1f}; RBV={m.readback:.1f}")
|
|
raise TimeoutError("timeout waiting for movement to finish.")
|
|
|
|
|
|
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?!")
|
|
|
|
if isinstance(pv, Motor):
|
|
|
|
def checker(m, target=None, tolerance=None):
|
|
if tolerance is None:
|
|
tolerance = m.get("RDBD")
|
|
if target is None:
|
|
target = m.drive
|
|
|
|
diff = abs(target - m.readback)
|
|
return m.done_moving and diff < tolerance
|
|
|
|
wait_motor_position(pv, checker, target=value, tolerance=tolerance, timeout=timeout, polling=polling)
|
|
|
|
elif is_epics_type(pv, "double") or is_epics_type(pv, "time_double"):
|
|
wait_float_condition(pv, value, tolerance=tolerance, timeout=timeout, polling=polling)
|
|
|
|
elif is_epics_type(pv, "enum") or is_epics_type(pv, "time_enum"):
|
|
wait_enum_condition(pv, value, timeout=timeout, polling=polling)
|
|
|
|
elif is_epics_type(pv, "string") or is_epics_type(pv, "time_string"):
|
|
wait_string_condition(pv, value, timeout=timeout, polling=polling)
|
|
else:
|
|
raise ValueError(f"pv_wait() don't know how to handle {pv.pvname} of type {pv.type}")
|
|
|
|
|
|
def is_epics_type(pv: PV, pv_type: str) -> bool:
|
|
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):
|
|
raise AttributeError("argument 'pv' must be an epics.PV of type string")
|
|
|
|
if not isinstance(target, re.Pattern):
|
|
target = re.compile(f"^{target}$", re.IGNORECASE)
|
|
|
|
tout = time.time() + timeout
|
|
|
|
while time.time() < tout:
|
|
if target.match(pv.char_value):
|
|
return
|
|
poll(polling)
|
|
|
|
if time.time() > tout:
|
|
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):
|
|
"""wait until an epics.PV of type double reaches value
|
|
pv: epics.PV
|
|
the PV enum on which you want to wait on
|
|
|
|
value: float
|
|
the target value
|
|
|
|
tolerance: float or None
|
|
the tolerance to accept when comparing values, if None (default) we
|
|
try to figure an appropriate value
|
|
|
|
|
|
timeout: double
|
|
a timeout in seconds
|
|
|
|
return: nothing
|
|
|
|
raises: TimeoutError if a timeout occurs
|
|
"""
|
|
if not (isinstance(pv, PV) or "double" not in pv.type):
|
|
raise AttributeError("argument 'pv' must be an epics.PV of type double")
|
|
|
|
try:
|
|
value = float(value)
|
|
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)
|
|
|
|
tout = time.time() + timeout
|
|
while time.time() < tout:
|
|
if abs(pv.value - value) < tolerance:
|
|
return
|
|
poll(polling)
|
|
|
|
if time.time() > tout:
|
|
raise TimeoutError(f"timeout waiting for enum {pv.pvname} == {value}")
|
|
|
|
|
|
def wait_motor_position(motor: Motor, tester: Callable, *, timeout: float = 50.0, polling: float = 0.1, **kwargs):
|
|
"""wait until an epics.Motor reaches a condition
|
|
motor: epics.Motor
|
|
the motor on which you want to wait on
|
|
|
|
tester: Callable
|
|
a Callable which is passed motor and **kwargs
|
|
=> Callable(motor, **kwargs)
|
|
|
|
timeout: double
|
|
a timeout in seconds
|
|
|
|
**kwargs: extra arguments to tester
|
|
|
|
return: nothing
|
|
|
|
raises: TimeoutError if a timeout occurs
|
|
"""
|
|
if not isinstance(tester, Callable):
|
|
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
|
|
|
|
tout = move_time + time.time() + timeout
|
|
|
|
while time.time() < tout:
|
|
if tester(motor, **kwargs):
|
|
return
|
|
poll(polling)
|
|
|
|
if time.time() > tout:
|
|
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):
|
|
"""wait until an epics.PV enum reaches value
|
|
pv: epics.PV
|
|
the PV enum on which you want to wait on
|
|
|
|
value: int or str
|
|
if int check against index of current value
|
|
|
|
timeout: double
|
|
a timeout in seconds
|
|
|
|
return: nothing
|
|
|
|
raises: TimeoutError if a timeout occurs
|
|
"""
|
|
if not (isinstance(pv, PV) or "enum" != pv.type[-4:].lower()):
|
|
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)):
|
|
raise AttributeError("argument 'value' must be either an int, str, or re.Pattern")
|
|
|
|
if type(value) is int:
|
|
tester = lambda pv: value == pv.get() # noqa: E731
|
|
elif type(value) is str:
|
|
tester = lambda pv: str(value) == pv.get(as_string=True).lower() # noqa: E731
|
|
value = str(value).lower() # it's already a str :-/
|
|
elif isinstance(value, re.Pattern):
|
|
tester = lambda pv: value.match(pv.get(as_string=True)) # noqa: E731
|
|
else:
|
|
raise AttributeError("argument 'value' must be either an int, str, or re.Pattern")
|
|
|
|
tout = time.time() + timeout
|
|
|
|
while time.time() < tout:
|
|
if tester(pv):
|
|
return
|
|
poll(polling)
|
|
|
|
if time.time() > tout:
|
|
raise TimeoutError(f"timeout waiting for enum {pv.pvname} == {value}")
|