diff --git a/src/aare/devices/mx_lib.py b/src/aare/devices/mx_lib.py index ead6dd85..312880d5 100644 --- a/src/aare/devices/mx_lib.py +++ b/src/aare/devices/mx_lib.py @@ -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)):