From ddec076fcf84cdb690ef7ab6f2d37d5a1257e4f0 Mon Sep 17 00:00:00 2001 From: appleb_m Date: Thu, 29 Jan 2026 16:48:56 +0100 Subject: [PATCH] DAQ: updates to enum_pc and set_get_pv classes, WIP --- src/aare/devices/enum_pv.py | 9 +++- .../{non_standard.py => set_get_pv.py} | 53 +++++++++++-------- 2 files changed, 37 insertions(+), 25 deletions(-) rename src/aare/devices/{non_standard.py => set_get_pv.py} (76%) diff --git a/src/aare/devices/enum_pv.py b/src/aare/devices/enum_pv.py index 4253b1ff..1eb0aac2 100755 --- a/src/aare/devices/enum_pv.py +++ b/src/aare/devices/enum_pv.py @@ -3,17 +3,22 @@ from enum import Enum from typing import Union, Optional from epics import PV, Motor, poll +from aare.devices.set_get_pv import SetGetPV + + class ValueWaitTimeout(Exception): """Raised when a PV fails to reach a target value within the timeout period.""" pass -class EnumPv: +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): + 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. diff --git a/src/aare/devices/non_standard.py b/src/aare/devices/set_get_pv.py similarity index 76% rename from src/aare/devices/non_standard.py rename to src/aare/devices/set_get_pv.py index a6b198c4..1e31cc7e 100644 --- a/src/aare/devices/non_standard.py +++ b/src/aare/devices/set_get_pv.py @@ -1,13 +1,16 @@ import time from typing import Any, Dict, Optional, Union from epics import PV, poll + +from aare.common.beamline import MXBeamline from aare.devices.mx_lib import pv_wait -class NonStandard: +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__( @@ -15,34 +18,41 @@ class NonStandard: 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 - - 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")) @property - def value(self) -> Any: - """Alias for position to maintain compatibility.""" - if not self.readback_pv.connected: - return None + def value(self): 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.value + 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 @@ -79,7 +89,4 @@ class NonStandard: 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}>" + pv_wait(self.readback_pv, self._target_pos, timeout=tout, tolerance=self.tolerance) \ No newline at end of file