255 lines
8.3 KiB
Python
255 lines
8.3 KiB
Python
import re
|
|
import time
|
|
from typing import Callable, Union, Any
|
|
|
|
from epics import PV, Motor, poll
|
|
|
|
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
|
|
"""
|
|
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: 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
|
|
|
|
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:
|
|
"""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: 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):
|
|
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:float, *, 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
|
|
|
|
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: float default = 60.0
|
|
a timeout in seconds
|
|
|
|
return: nothing
|
|
|
|
raises: TimeoutError if a timeout occurs
|
|
"""
|
|
|
|
try:
|
|
value = float(value)
|
|
except ValueError:
|
|
raise AttributeError("argument 'value' must be a number")
|
|
|
|
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
|
|
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 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 of unusual 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):
|
|
"""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) 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)):
|
|
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},"
|
|
f"current value is {pv.get(as_string=True)}")
|
|
|
|
def clean_filename(self, filename: str) -> str:
|
|
cleaned = re.sub(r"[^A-Za-z0-9._-]", "_", filename.strip())
|
|
cleaned = cleaned.strip("._-")
|
|
if not cleaned:
|
|
raise ValueError("Filename is empty after sanitization.")
|
|
return cleaned |