mx/AareDAQ
0
1
Fork 0
Code Issues 49 Pull Requests Actions Packages Projects Releases 84 Wiki Activity

DAQ: refactord enum pv and setgetpv, added prederfined PV

Browse Source
This commit is contained in:
appleb_m
2026-02-06 15:59:11 +01:00
parent 7fb1c61677
commit 6b47ceec5b
4 changed files with 127 additions and 178 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/aare/daq/devices.py
+52 -21
View File
@@ -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
# 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
src/aare/devices/enum_pv.py
+25 -82
View File
@@ -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
raise TypeError(f"Unsupported enum command type: {type(x).__name__}")
src/aare/devices/mx_lib.py
+2 -1
View File
@@ -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)}")
src/aare/devices/set_get_pv.py
+48 -74
View File
@@ -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)
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)