From d05a60d49584147cc4003c6fa2e1e1c5715433fd Mon Sep 17 00:00:00 2001 From: appleb_m Date: Thu, 22 Jan 2026 10:13:54 +0100 Subject: [PATCH] Devices: non_standard rewrite, WIP --- src/aare/devices/non_standard.py | 340 +++++++------------------------ 1 file changed, 70 insertions(+), 270 deletions(-) diff --git a/src/aare/devices/non_standard.py b/src/aare/devices/non_standard.py index a5b64ae2..a6b198c4 100644 --- a/src/aare/devices/non_standard.py +++ b/src/aare/devices/non_standard.py @@ -1,285 +1,85 @@ import time -from typing import Callable, cast - +from typing import Any, Dict, Optional, Union from epics import PV, poll -from epics.ca import pend_io +from aare.devices.mx_lib import pv_wait -def is_number(x): - return isinstance(x, (int, float)) +class NonStandard: + """ + A wrapper for EPICS PVs that have separate setpoint and readback PVs. + Similar to EnumPv but supports arbitrary numeric or string types and predefined positions. + """ + def __init__( + self, + name: str, + setpv: str, + getpv: str, + timeout: float = 60.0, + tolerance: Optional[float] = None, + predefs: Optional[Dict[str, Any]] = None, + **kwargs + ): + self.name = name + self.device_name = name + self.setpoint_pv = PV(setpv) + self.readback_pv = PV(getpv) + self.default_timeout = timeout + self.tolerance = tolerance -class NonStandard(object): - def __init__(self, **kwargs): - mandatory = ["name", "setpv", "getpv"] - if set(mandatory) != set(kwargs.keys()) & set(mandatory): - raise RuntimeError( - "missing mandatory argument(s): {}".format(str(list(set(mandatory) - set(kwargs.keys())))) - ) - self.name = kwargs.get("name") - self.__target_pos: float | int | str = 0 - self.device_name = self.name - self.__setpv = PV(kwargs["setpv"]) - self.__getpv = PV(kwargs["getpv"]) - self.__timeout_margin = kwargs.get("timeout", 2.0) # add this safety margin to timeout calc from speed - self.__timeout = kwargs.get("timeout", 60.0) - self.__format = kwargs.get("format") + self._predefs = predefs or {} + self.positions = list(self._predefs.keys()) + self._target_pos: Optional[Any] = None + if kwargs.get("move_done_when"): + self.__move_done_when = PV(kwargs.get("move_done_when")) - p = kwargs.get("speed") - if p is not None: - if is_number(p): - self.__speed_set, self.__speed_get = p, p - elif type(p) in [tuple, list] and len(p) == 2: - self.__speed_set = PV(p[0]) - self.__speed_get = PV(p[1]) - else: - raise Exception(f"{self.name} => don't know how to parse speed parameter") - else: - self.__speed_set, self.__speed_get = None, None - # - # move_done_when a tuple, first item the pv to read - # second item, the int type indicating that it is DONE - # - p = kwargs.get("move_done_when") - if p is not None: - self.__move_flag = PV(p[0]) - self.__move_flag_done = p[1] - else: - self.__move_flag = None + @property + def value(self) -> Any: + """Alias for position to maintain compatibility.""" + if not self.readback_pv.connected: + return None + return self.readback_pv.get() - # - # stoppv a tuple, first item the pv to read - # second item, the int type to put to stop the motor - # - p = kwargs.get("stoppv") - if p is not None: - self.__stoppv = PV(p[0]) - self.__stoppv_done = p[1] - else: - self.__stoppv = None + def get_predefined_name(self) -> str: + """Returns the name of the predefined position if current position matches one.""" + current = self.value + for name, val in self._predefs.items(): + # Handle callable predefs (like offsets) + target = val[0](*val[1]) if isinstance(val, tuple) and callable(val[0]) else val - # - # still_moving_when a tuple, first item the pv to read - # second item, the int type indicating that it is still moving. - # - p = kwargs.get("still_moving_when") - if p is not None: - self.__still_moving_flag = PV(p[0]) - self.__still_moving_flag_value = p[1] - else: - self.__still_moving_flag = None - - p = kwargs.get("predefs") - if p: - self.__predefs = p - self.positions = list(p.keys()) - else: - self.__predefs = {} - self.positions = [] - - pend_io(5.0) - - typ = self.__ret_type = self.__getpv.type - - match str(typ): - case "double" | "time_double": - self.still_moving = self.__moving_float - case "int" | "time_int": - self.still_moving = self.__moving_int - case "string" | "time_string": - self.still_moving = self.__moving_str - case _: - self.still_moving = self.__moving_unknown - - tolerance = kwargs.get("tolerance", None) - if tolerance is not None and tolerance > 0: - self.tolerance = tolerance - elif tolerance is not None and tolerance < 0: - self.still_moving = self.__moving_unknown - - if self.__move_flag: - self.still_moving = self.is_move_not_done - elif self.__still_moving_flag: - self.still_moving = self.still_moving_when - - def __str__(self): - if self.positions: - return "<{} at {}({}) Pre-defined positions {}>".format( - self.name, self.__position(predef=True), self.position, self.positions - ) - else: - return "<{} at {}>".format(self.name, self.__position()) - - def __repr__(self): - if self.positions: - return "<{} instance at {}: {} at {}({}) Pre-defined positions {}>".format( - self.__class__.__name__, - hex(id(self)), - self.name, - self.__position(predef=True), - self.position, - self.positions, - ) - else: - return "<{} instance at {}: {} at {}>".format( - self.__class__.__name__, hex(id(self)), self.name, self.__position() - ) - - def still_moving_when(self): - return self.__still_moving_flag_value == self.__still_moving_flag.get() # type: ignore - - def is_moving(self): - return not self.is_move_done() - - def is_move_done(self): - return self.__move_flag_done == self.__move_flag.get() # type: ignore - - def is_move_not_done(self): - return not self.is_move_done() - - def __moving_float(self): - if type(self.__target_pos) is not float: - raise RuntimeError("target position is not an float") - current_value = cast(float, self.__getpv.get()) - return self.tolerance < abs(cast(float, self.__target_pos) - current_value) - - def __moving_int(self): - if type(self.__target_pos) is not int: - raise RuntimeError("target position is not an integer") - return self.__target_pos != self.__getpv.get() - - def __moving_str(self): - if type(self.__target_pos) is not str: - raise RuntimeError("target position is not a string") - current_value = cast(str, self.__getpv.get()) - return self.__target_pos.upper() != current_value.upper() - - def __moving_unknown(self): - return False - - def __position(self, predef: bool = False): - """returns either a predefined position or a readback - - :rtype: Union[str, float, int] - """ - cur = self.readback - if not predef: - return cur - # loop over predefined positions, if one found and - # readback matches it's value return "position" - for pos_label, val_or_cbk in self.__predefs.items(): - if self.__ret_type is bytes: - tst = val_or_cbk == cur - elif type(val_or_cbk) is tuple: - # a tuple (Callable, arg-to-Callable) - val = val_or_cbk[0](*val_or_cbk[1]) - tst = 0.1 > abs(val - cur) # FIXME sooo bad - else: - tst = 0.1 > abs(val_or_cbk - cur) - if tst: - return pos_label + # Use a small tolerance for floats if not provided + tol = self.tolerance if self.tolerance is not None else 0.1 + try: + if abs(current - target) < tol: + return name + except (TypeError, ValueError): + if current == target: + return name return "unknown" - def __readback(self): - return self.__getpv.get() + def move(self, pos: Union[str, float, int], wait: bool = False, timeout: Optional[float] = None): + """ + Move to a position. + 'pos' can be a raw value or a key from the predefs dictionary. + """ + # Resolve predefined position + target = self._predefs.get(pos, pos) + if isinstance(target, tuple) and callable(target[0]): + target = target[0](*target[1]) - readback = property(__readback) - position = property(__position) + self._target_pos = target + self.setpoint_pv.put(target) - def has_position(self, position: str) -> bool: - return position.lower() in [p.lower() for p in self.positions] - - def position_is(self, position, tolerance=0.1): - pos = self.__predefs.get(position, position) - cur = self.position - - print(f"position_is: {pos} == {cur}") - - if type(pos) is tuple: - pos = pos[0](*pos[1]) - - print(f"position_is: {pos} == {cur}") - - if self.__ret_type is bytes: - print("checking bytes") - return pos == cur - else: - print(f"checking numbers: abs({pos} - {cur}) < {tolerance}") - return abs(pos - cur) < tolerance - - def __value(self): - val = self.__getpv.get() - if self.__format is not None: - val = self.__format % (val,) - return val - - value = property(__value) - - def __raw_value(self): - return self.__getpv.get() - - raw_value = property(__raw_value) - - def __char_value(self): - return str(self.value) - - char_value = property(__char_value) - - def set_speed(self, speed): - if isinstance(self.__speed_set, PV): - self.__speed_set.put(speed) - elif is_number(self.__speed_set): - self.__speed_set = speed - self.__speed_get = speed - - def get_speed(self): - if is_number(self.__speed_get): - return self.__speed_get - elif isinstance(self.__speed_get, PV): - return self.__speed_get.value - else: - return None - - def get_timeout(self): - if not self.__speed_get: - timeout = self.__timeout - else: - speed = self.get_speed() - curp = self.position - if curp is None or self.__target_pos is None or speed is None: - timeout = self.__timeout_margin - raise RuntimeWarning("can't calculate timeout, ABR may be disconnected") - else: - timeout = self.__timeout_margin + (abs(curp - self.__target_pos) / speed) - return timeout - - def wait(self): - pos = self.__target_pos - timeout = time.time() + self.get_timeout() - - while time.time() < timeout and self.still_moving(): - poll(0.1) - if time.time() > timeout: - msg = "timeout occurred when moving %s to %s" % (self.name, pos) - print(msg) - raise RuntimeWarning(msg) - - def move(self, pos, relative=False, wait=False): - pos = self.__predefs.get(pos, pos) - if type(pos) is tuple: - pos = pos[0](*pos[1]) - - if relative: - pos += self.__getpv.get() - - self.__target_pos = pos - self.__setpv.put(pos) if wait: - time.sleep(0.1) - self.wait() + self.wait(timeout=timeout) - def stop(self): - if self.__stoppv: - self.__stoppv.put(self.__stoppv_done) - else: - raise RuntimeError("stop pv not configured for this motor") + def wait(self, timeout: Optional[float] = None): + """Wait for the readback to match the last commanded target position.""" + if self._target_pos is None: + return + + tout = timeout or self.default_timeout + pv_wait(self.readback_pv, self._target_pos, timeout=tout, tolerance=self.tolerance) + + def __repr__(self): + return f"<{self.__class__.__name__} '{self.name}' at {self.value}>"