From 7c424d15ad2a2ce61a6824d9d7f9cdacfa94119a Mon Sep 17 00:00:00 2001 From: David Perl Date: Thu, 6 Aug 2026 14:34:14 +0200 Subject: [PATCH] WIP remove pv modules --- src/aare/devices/enum_pv.py | 36 -------------- src/aare/devices/set_get_pv.py | 85 ---------------------------------- 2 files changed, 121 deletions(-) delete mode 100644 src/aare/devices/enum_pv.py delete mode 100644 src/aare/devices/set_get_pv.py diff --git a/src/aare/devices/enum_pv.py b/src/aare/devices/enum_pv.py deleted file mode 100644 index fd725e17..00000000 --- a/src/aare/devices/enum_pv.py +++ /dev/null @@ -1,36 +0,0 @@ -from enum import Enum -from typing import Any - -from aare.devices.set_get_pv import MoveResult, SetGetPV - - -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: - return self.readback_pv.get(as_string=True) - - def _resolve(self, x: Any) -> MoveResult: - # accept Enum member - if isinstance(x, Enum): - x = x.name - - # 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 - - # 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)}") - - raise TypeError(f"Unsupported enum command type: {type(x).__name__}") diff --git a/src/aare/devices/set_get_pv.py b/src/aare/devices/set_get_pv.py deleted file mode 100644 index fa849073..00000000 --- a/src/aare/devices/set_get_pv.py +++ /dev/null @@ -1,85 +0,0 @@ -from __future__ import annotations - -from collections.abc import Callable, Mapping -from dataclasses import dataclass -from typing import Any - -from epics import PV - -from aare.devices.mx_lib import pv_wait - -RawValue = str | float | int -ResolverValue = ( - RawValue - | tuple[Callable[..., RawValue], tuple[Any, ...]] # (func, args) pattern you already use -) - - -@dataclass -class MoveResult: - target: RawValue - name: str | None = None - - -class SetGetPV: - 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) -> Any: - return self.readback_pv.get() - - 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: 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, - ) - - -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)