From dc8d7799b1cfa4f31e256d839c4553576a273fcb Mon Sep 17 00:00:00 2001 From: appleb_m Date: Fri, 16 Jan 2026 11:52:54 +0100 Subject: [PATCH] DAQ: re added enum_pv and hopefully fixed front and back light settings --- src/aare/daq/devices.py | 27 +++++---- src/aare/devices/enum_pv.py | 107 ++++++++++++++---------------------- 2 files changed, 53 insertions(+), 81 deletions(-) diff --git a/src/aare/daq/devices.py b/src/aare/daq/devices.py index 27aab358..a3effcc8 100644 --- a/src/aare/daq/devices.py +++ b/src/aare/daq/devices.py @@ -32,12 +32,12 @@ class BeamlineDevices: self.__dtz = MyMotor(f"{BEAMLINE}-ES-DET:TRZ") self.__sample_cam = epicsAD(f"{BEAMLINE}-SAMCAM:") - self.__back_light_pos_set = PV(f"{BEAMLINE}-ES-BL:POS-SET") - self.__back_light_pos_get = PV(f"{BEAMLINE}-ES-BL:POS-GET") - self.__back_light_bright_set = PV(f"{BEAMLINE}-ES-BL:SET") - self.__back_light_bright_get = PV(f"{BEAMLINE}-ES-BL:GET") - self.__front_light_bright_set = PV(f"{BEAMLINE}-ES-BL:SET") - self.__front_light_bright_get = PV(f"{BEAMLINE}-ES-BL:GET") + self.__back_light_pos = EnumPv(pv_name = f"{BEAMLINE}-ES-BL:POS-SET", + rbv_name=f"{BEAMLINE}-ES-BL:POS-GET", + timeout=10.0) + + self.__back_light_bright= PV(f"{BEAMLINE}-ES-BL:SET") + self.__front_light_bright = PV(f"{BEAMLINE}-ES-FL:SET") self.__zoom_set = PV(f"{BEAMLINE}-ES-SAMCAM:ZOOM.VAL") self.__zoom_get = PV(f"{BEAMLINE}-ES-SAMCAM:ZOOM.RBV") @@ -57,26 +57,26 @@ class BeamlineDevices: # Lamp light @property def lamp_light(self) -> float: - return self.__front_light_bright_get.get() + return self.__front_light_bright.get() @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_bright_set.put(v, wait=wait) + self.__front_light_bright.put(v, wait=wait) # Back light @property def back_light(self) -> float: - return self.__back_light_bright_get.get() + return self.__back_light_bright.get() - @lamp_light.setter + @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_bright_set.put(v, wait=wait) + self.__back_light_bright.put(v, wait=wait) # Zoom @property @@ -117,15 +117,14 @@ class BeamlineDevices: # Reflector (backlight?) @property def reflector_up(self) -> bool: - return self.__back_light_pos_get.get() == BacklightPositionEnum.MEASURE.value + return self.__back_light_pos.position.upper() == StagePositionEnum.MEASURE.name @reflector_up.setter def reflector_up(self, value: StagePositionEnum): self.set_reflector_up(value, wait=True) def set_reflector_up(self, value: StagePositionEnum, /, wait: bool = True): - self.__back_light_pos_set.put(BacklightPositionEnum.MEASURE.value if value else BacklightPositionEnum.PARK.value, - wait=wait) + self.__back_light_pos.put(value, wait=wait) # Beamstop @property diff --git a/src/aare/devices/enum_pv.py b/src/aare/devices/enum_pv.py index 4e9ae9f0..4253b1ff 100755 --- a/src/aare/devices/enum_pv.py +++ b/src/aare/devices/enum_pv.py @@ -7,71 +7,18 @@ class ValueWaitTimeout(Exception): """Raised when a PV fails to reach a target value within the timeout period.""" pass - -def pv_wait(pv: Union[PV, Motor], target: Union[str, int, float], timeout: float = 30.0, - polling: float = 0.1, tolerance: float | None = None): - """ - Unified wait function for Motors and PVs (Strings, Enums, Floats). - """ - start_time = time.monotonic() - end_time = start_time + timeout - - # Handle Motors - if isinstance(pv, Motor): - if tolerance is None: - # Try to get the motor resolution/deadband - tolerance = pv.get("RDBD") or 0.01 - - while time.monotonic() < end_time: - if pv.done_moving and abs(pv.readback - target) <= tolerance: - return - poll(polling) - - raise ValueWaitTimeout( - f"Motor {pv.name} timeout. Target: {target}, Current: {pv.readback}, Done: {pv.done_moving}" - ) - - # Handle PVs (Enums, Strings, Doubles) - pv_type = (pv.type or "").lower() - is_enum = "enum" in pv_type - is_string = "string" in pv_type - - while time.monotonic() < end_time: - # Get current value: as string for Enums/Strings, as raw value for others - as_string = isinstance(target, str) or is_enum or is_string - current = pv.get(as_string=as_string) - - if current is not None: - # Comparison logic - 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(pv, 'precision', 3) or 3 - tolerance = 10 ** -prec - - if abs(float(current) - float(target)) <= tolerance: - return - - poll(polling) - - # If we exit the loop, we timed out - final_val = pv.get(as_string=isinstance(target, str)) - raise ValueWaitTimeout(f"PV {pv.pvname} timeout. Target: {target}, Current: {final_val}") - -def wait_for_movement_to_finish(*motors: Motor, timeout: float = 60.0): - """Wait for a group of motors to stop moving.""" - start_time = time.time() - while not all(m.done_moving for m in motors): - if time.time() - start_time > timeout: - moving = [m.name for m in motors if not m.done_moving] - raise TimeoutError(f"Timeout waiting for motors: {', '.join(moving)}") - poll(0.1) - class EnumPv: + """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 @@ -90,25 +37,51 @@ class EnumPv: """Returns the current integer index of the readback.""" return int(self.readback.get()) - def put(self, value: Union[str, int, Enum], wait: bool = False): + 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 - val_to_put = value.value if isinstance(value, Enum) else value + name = value.name if isinstance(value, Enum) else value + + 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}") self.control.put(val_to_put) if wait: self.wait(val_to_put) - def wait(self, target: Union[str, int, Enum]): + 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: - pv_wait(self.readback, wait_val, timeout=self.timeout) + 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