DAQ: updates to enum_pc and set_get_pv classes, WIP

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.

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)