mx/AareDAQ
0
1
Fork 0
Code Issues 52 Pull Requests 2 Actions Packages Projects Releases 77 Wiki Activity

DAQ: re added enum_pv and hopefully fixed front and back light settings

Browse Source
This commit is contained in:
appleb_m
2026-01-16 11:52:54 +01:00
parent 199f8dc57c
commit dc8d7799b1
2 changed files with 53 additions and 81 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/aare/daq/devices.py
+13 -14
View File
@@ -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
src/aare/devices/enum_pv.py
+40 -67
View File
@@ -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