From 6b47ceec5b980eed36c17f6fd4abc6c2e5756324 Mon Sep 17 00:00:00 2001 From: appleb_m Date: Fri, 6 Feb 2026 15:59:11 +0100 Subject: [PATCH] DAQ: refactord enum pv and setgetpv, added prederfined PV --- src/aare/daq/devices.py | 73 ++++++++++++++------ src/aare/devices/enum_pv.py | 107 +++++++---------------------- src/aare/devices/mx_lib.py | 3 +- src/aare/devices/set_get_pv.py | 122 +++++++++++++-------------------- 4 files changed, 127 insertions(+), 178 deletions(-) diff --git a/src/aare/daq/devices.py b/src/aare/daq/devices.py index 7fd1f071..201e36c8 100644 --- a/src/aare/daq/devices.py +++ b/src/aare/daq/devices.py @@ -17,10 +17,10 @@ from aare.common.coordinate import Coordinate, SmargonCoordinate from aare.common.models import SampleCameraSettings, StagePositionEnum from aare.devices import smargon, aerotech from aare.devices.area_detector import epicsAD, AutoEnum -from aare.devices.enum_pv import EnumPv +from aare.devices.enum_pv import EnumPV from aare.devices.my_motor import MyMotor -from aare.devices.set_get_pv import SetGetPV +from aare.devices.set_get_pv import SetGetPV, PredefinedPV from aare.devices.tell_client import TellClient class BeamlineDevices: @@ -28,6 +28,7 @@ class BeamlineDevices: BEAMLINE = beamline.value.upper() self.tell = TellClient(beamline) self.__aerotech = aerotech.AerotechControllerEpics(beamline) + self.aerotech = aerotech.AerotechController(controller_ip="129.129.118.96") self.__smargon = smargon.Smargon(beamline) self.__ring_current_pv = PV(f"ARS07-DPCT-0100:CURR") @@ -36,19 +37,35 @@ class BeamlineDevices: self.__sample_cam = epicsAD(f"{BEAMLINE}-SAMCAM:") - self.__front_light = PV(f"{BEAMLINE}-ES-FL:SET") - self.__back_light = PV(f"{BEAMLINE}-ES-BL:SET") + self.__front_light = PredefinedPV(name='front_light', + setpv="{BEAMLINE}-ES-FL:SET", + getpv="{BEAMLINE}-ES-FL:GET", + predefs={"off":1.49, + 'half':2.0, + 'max':3.0}, + timeout=10.0 + ) + self.__back_light = PredefinedPV(name='back_light', + setpv ="{BEAMLINE}-ES-BL:SET", + getpv="{BEAMLINE}-ES-BL:GET", + predefs={"off": 0, + 'half': 0.98, + 'max': 1.2}, + timeout=10.0 + ) - self.__back_light_pos = EnumPv(pv_name = f"{BEAMLINE}-ES-BL:POS-SET", - rbv_name = f"{BEAMLINE}-ES-BL:POS-GET", + self.__back_light_pos = EnumPV(name = "back_light_pos", + setpv = f"{BEAMLINE}-ES-BL:POS-SET", + getpv = f"{BEAMLINE}-ES-BL:POS-GET", timeout = 10.0) self.__collimator_pos = PV(f"{BEAMLINE}-ES-COL:TRY") # HOW TO HANDLE!!! self.__scintillator_pos = PV(f"{BEAMLINE}-ES-SCL:TRY") # how to handle!!! - self.__beamstop_pos = EnumPv(pv_name = f"{BEAMLINE}-ES-BS:POS-SET", - rbv_name = f"{BEAMLINE}-ES-BS:POS-GET", + self.__beamstop_pos = EnumPV(name = "beamstop_pos", + setpv = f"{BEAMLINE}-ES-BS:POS-SET", + getpv = f"{BEAMLINE}-ES-BS:POS-GET", timeout = 10.0) @@ -57,8 +74,9 @@ class BeamlineDevices: setpv = f"{BEAMLINE}-ES-SAMCAM:ZOOM.VAL", getpv = f"{BEAMLINE}-ES-SAMCAM:ZOOM.RBV") - self.__cryojet_pos = EnumPv(pv_name = f"{BEAMLINE}-ES-CJ:POS-SET", - rbv_name = f"{BEAMLINE}-ES-CJ:POS-GET", + self.__cryojet_pos = EnumPV(name='cryojet_pos', + setpv = f"{BEAMLINE}-ES-CJ:POS-SET", + getpv = f"{BEAMLINE}-ES-CJ:POS-GET", timeout = 10.0) @@ -77,26 +95,26 @@ class BeamlineDevices: # Lamp light @property def lamp_light(self) -> float: - return self.__front_light.get() + return self.__front_light.value @lamp_light.setter def lamp_light(self, v: float): self.set_front_light(v, wait=False) def set_front_light(self, v: float, /, wait: bool = True): - self.__front_light.put(v, wait=wait) + self.__front_light.move(v, wait=wait) # Back light @property def back_light(self) -> float: - return self.__back_light.get() + return self.__back_light.value @back_light.setter def back_light(self, v: float): self.set_back_light(v, wait=False) def set_back_light(self, v: float, /, wait: bool = True): - self.__back_light.put(v, wait=wait) + self.__back_light.move(v, wait=wait) # Zoom @property @@ -108,7 +126,7 @@ class BeamlineDevices: self.set_zoom(value, wait=True) def set_zoom(self, value: float, /, wait: bool = True): - self.__zoom.put(value, wait=wait) + self.__zoom.move(value, wait=wait) # Collimator @property @@ -120,7 +138,7 @@ class BeamlineDevices: self.set_collimator(value, wait=True) def set_collimator(self, value: float, /, wait: bool = True): - self.__collimator_pos.put(value, wait=wait) + self.__collimator_pos.move(value, wait=wait) # Scintillator @property @@ -144,7 +162,7 @@ class BeamlineDevices: self.set_reflector_up(value, wait=True) def set_reflector_up(self, value: StagePositionEnum, /, wait: bool = True): - self.__back_light_pos.put(value, wait=wait) + self.__back_light_pos.move(value, wait=wait) # Beamstop @property @@ -197,7 +215,7 @@ class BeamlineDevices: self.cryojet_pos_setter(value, wait=True) def cryojet_pos_setter(self, value: StagePositionEnum, wait:bool=False): - self.__cryojet_pos.put(value, wait=wait) + self.__cryojet_pos.move(value, wait=wait) # Shutter @property @@ -226,6 +244,14 @@ class BeamlineDevices: def samcam_auto(self, state: AutoEnum): self.__sample_cam.set_auto(state) + def samcam_frame_id(self) -> int: + """ + Camera UniqueId for the last produced frame (monotonic counter from AreaDetector). + """ + return int(self.__sample_cam.uid.get()) + + + # Detector Z @property def dtz(self) -> float: @@ -302,11 +328,16 @@ class BeamlineDevices: def smargon_move_home(self): self.__smargon.move_home(wait=True) + def smargon_aerotech_wait(self): + self.__smargon.wait_aerotech(timeout=10.0) + if __name__ == "__main__": from aare.common.beamline import mx_beamline beamline = mx_beamline() devs = BeamlineDevices(beamline) print(devs.aerotech_pos) - devs.aerotech_pos = Coordinate(x=124.0, y=1.0, z=1.0) - print(devs.aerotech_pos) - devs.aerotech_omega = 0.0 \ No newline at end of file + # devs.aerotech_pos = Coordinate(x=124.0, y=1.0, z=1.0) + # print(devs.aerotech_pos) + # devs.aerotech_omega = 0.0 + print(devs.reflector_up) + devs.reflector_up = StagePositionEnum.MEASURE \ No newline at end of file diff --git a/src/aare/devices/enum_pv.py b/src/aare/devices/enum_pv.py index 1eb0aac2..118fe26f 100755 --- a/src/aare/devices/enum_pv.py +++ b/src/aare/devices/enum_pv.py @@ -1,92 +1,35 @@ -import time from enum import Enum -from typing import Union, Optional -from epics import PV, Motor, poll +from typing import Any -from aare.devices.set_get_pv import SetGetPV +from aare.devices.set_get_pv import SetGetPV, MoveResult - -class ValueWaitTimeout(Exception): - """Raised when a PV fails to reach a target value within the timeout period.""" - pass - -class EnumPv(SetGetPV): - """create a class that combines a setter PV and readback PV into a single object. - i.e. self.__back_light_pos = EnumPv(pv_name = f"{BEAMLINE}-ES-BL:POS-SET", - rbv_name=f"{BEAMLINE}-ES-BL:POS-GET", - timeout=10.0) - """ - def __init__(self, pv_name: str, - rbv_name: Optional[str] = None, - timeout: float = 60.0): - """Initialize the EnumPv object. - :param pv_name: The name of the control PV. - :param rbv_name: The name of the readback PV. If not provided, defaults to the control PV. - :param timeout: The timeout for waiting for the readback to reach a target value. - """ - self.control = PV(pv_name) - # If no readback is provided, assume the control PV reflects its own state - self.readback = PV(rbv_name) if rbv_name else self.control - self.timeout = timeout - - # Verify these are actually enums - if not self.control.enum_strs: - raise RuntimeError(f"{pv_name} is not an ENUM PV") +class EnumPV(SetGetPV): + def __init__(self, name: str, setpv: str, getpv: str, **kwargs): + super().__init__(name, setpv, getpv, **kwargs) + if not self.setpoint_pv.enum_strs: + raise RuntimeError(f"{setpv} is not an ENUM PV") @property def position(self) -> str: - """Returns the current string value of the readback.""" - return self.readback.get(as_string=True) + return self.readback_pv.get(as_string=True) - def get(self) -> int: - """Returns the current integer index of the readback.""" - return int(self.readback.get()) + def _resolve(self, x: Any) -> MoveResult: + # accept Enum member + if isinstance(x, Enum): + x = x.name - def put(self, value: Union[str, int, Enum], wait: bool = False, - timeout: float = 30.0, polling: float = 0.1, tolerance: float | None = None): - """ - Set the PV value. - Accepts string names, integer indices, or Python Enum members. - """ - # Extract value if a Python Enum was passed - name = value.name if isinstance(value, Enum) else value + # accept index + if isinstance(x, int): + try: + return MoveResult(target=self.setpoint_pv.enum_strs[x], name=None) + except Exception as e: + raise ValueError(f"Bad enum index {x}") from e - try: - val_to_put = next(s for s in self.control.enum_strs - if s.upper() == name.upper()) - except StopIteration: - raise ValueError(f"'{name}' not found in PV enums: {self.control.enum_strs}") + # accept name -> match against enum strings (case-insensitive) + if isinstance(x, str): + for s in self.setpoint_pv.enum_strs: + if s.strip().lower() == x.strip().lower(): + return MoveResult(target=s, name=s) + raise ValueError(f"'{x}' not in {list(self.setpoint_pv.enum_strs)}") - self.control.put(val_to_put) - - if wait: - self.wait(val_to_put) - - def wait(self, target: Union[str, int, Enum],timeout: float = 30.0, - polling: float = 0.1, tolerance: float | None = None): - """Blocks until the readback matches the target value.""" - # Convert Enum or name to the format pv_wait expects - wait_val = target.name if isinstance(target, Enum) else target - - try: - start_time = time.monotonic() - end_time = start_time + timeout - while time.monotonic() < end_time: - current = self.readback.get(as_string=True) - if current is not None: - if isinstance(target, str): - if str(current).strip().lower() == target.strip().lower(): - return - elif isinstance(target, (int, float)): - if tolerance is None: - # Default tolerance based on PV precision - prec = getattr(self.readback, 'precision', 3) or 3 - tolerance = 10 ** -prec - if abs(float(current) - float(target)) <= tolerance: - return - - poll(polling) - final_val = self.readback.get(as_string=isinstance(target, str)) - raise ValueWaitTimeout(f"PV {self.readback.pvname} timeout. Target: {target}, Current: {final_val}") - except Exception as e: - raise ValueWaitTimeout(f"Timed out waiting for {self.readback.pvname} to reach {wait_val}") from e \ No newline at end of file + raise TypeError(f"Unsupported enum command type: {type(x).__name__}") \ No newline at end of file diff --git a/src/aare/devices/mx_lib.py b/src/aare/devices/mx_lib.py index 9a0e30cc..ead6dd85 100644 --- a/src/aare/devices/mx_lib.py +++ b/src/aare/devices/mx_lib.py @@ -248,4 +248,5 @@ def wait_enum_condition(pv: PV, value: Union[str, int, re.Pattern], *, timeout: poll(polling) if time.time() > tout: - raise TimeoutError(f"timeout waiting for enum {pv.pvname} == {value}") + raise TimeoutError(f"timeout waiting for enum {pv.pvname} == {value}," + f"current value is {pv.get(as_string=True)}") diff --git a/src/aare/devices/set_get_pv.py b/src/aare/devices/set_get_pv.py index 1e31cc7e..c3784fe2 100644 --- a/src/aare/devices/set_get_pv.py +++ b/src/aare/devices/set_get_pv.py @@ -1,92 +1,66 @@ -import time -from typing import Any, Dict, Optional, Union -from epics import PV, poll +from __future__ import annotations -from aare.common.beamline import MXBeamline +from dataclasses import dataclass +from typing import Any, Mapping, Optional, Union, Callable + +from epics import PV, poll from aare.devices.mx_lib import pv_wait +RawValue = Union[str, float, int] +ResolverValue = Union[ + RawValue, + tuple[Callable[..., RawValue], tuple[Any, ...]], # (func, args) pattern you already use +] + +@dataclass +class MoveResult: + target: RawValue + name: Optional[str] = None + class SetGetPV: - """ - 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, - ): + def __init__(self, name: str, setpv: str, getpv: str, *, timeout: float = 60.0, tolerance: float | None = None): self.name = name self.setpoint_pv = PV(setpv) self.readback_pv = PV(getpv) + self.default_timeout = timeout + self.tolerance = tolerance + self._last_target: RawValue | None = None @property - def value(self): + def value(self) -> Any: return self.readback_pv.get() - @property - def position(self): - return self.value - - def put(self, value: Union[str, float], wait : bool=False): - self.setpoint_pv.put(value, wait=wait) - - def __repr__(self): - return f"<{self.__class__.__name__} '{self.name}' at {self.value}>" - -class PredefinedPV(SetGetPV): - - def __init__(self, name: str, setpv: str, getpv: str, predefined_values: Dict[str, Any], - timeout: float = 60.0, tolerance: Optional[float] = None): - - super().__init__(name=name, setpv=setpv,getpv=getpv) - - self._predefs = predefined_values or {} - self.positions = list(self._predefs.keys()) - self._target_pos: Optional[Any] = None - self.tolerance = tolerance - self.default_timeout = timeout - - def get_predefined_name(self) -> str: - """Returns the name of the predefined position if current position matches one.""" - current = self.position - 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 - - # 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 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]) - - self._target_pos = target - self.setpoint_pv.put(target) + def _resolve(self, x: Any) -> MoveResult: + # Default: treat input as raw value + return MoveResult(target=x, name=None) + def move(self, x: Any, *, wait: bool = False, timeout: float | None = None) -> MoveResult: + res = self._resolve(x) + self._last_target = res.target + self.setpoint_pv.put(res.target) if wait: self.wait(timeout=timeout) + return res - def wait(self, timeout: Optional[float] = None): - """Wait for the readback to match the last commanded target position.""" - if self._target_pos is None: + def wait(self, *, timeout: float | None = None): + if self._last_target is None: return + pv_wait(self.readback_pv, self._last_target, timeout=timeout or self.default_timeout, tolerance=self.tolerance) - tout = timeout or self.default_timeout - pv_wait(self.readback_pv, self._target_pos, timeout=tout, tolerance=self.tolerance) \ No newline at end of file +class PredefinedPV(SetGetPV): + def __init__(self, name: str, setpv: str, getpv: str, predefs: Mapping[str, ResolverValue], **kwargs): + super().__init__(name, setpv, getpv, **kwargs) + self._predefs = dict(predefs) + + @property + def positions(self) -> list[str]: + return list(self._predefs.keys()) + + def _resolve(self, x: Any) -> MoveResult: + if isinstance(x, str) and x in self._predefs: + v = self._predefs[x] + if isinstance(v, tuple) and callable(v[0]): + v = v[0](*v[1]) + return MoveResult(target=v, name=x) + return MoveResult(target=x, name=None)