WIP remove pv modules
CI / lint (push) Skipped
CI / test (3.11) (push) Skipped
CI / test (3.12) (push) Skipped
CI / test (3.13) (push) Skipped
CI / test-with-beamline-plugins (pxi_bec) (push) Skipped
CI / test-with-beamline-plugins (pxii_bec) (push) Skipped
CI / test-with-beamline-plugins (pxiii_bec) (push) Skipped
CI / lint (pull_request) Failing after 49s
CI / test (3.12) (pull_request) Failing after 45s
CI / test (3.11) (pull_request) Failing after 2m45s
CI / test (3.13) (pull_request) Canceled after 2m7s
CI / test-with-beamline-plugins (pxii_bec) (pull_request) Canceled after 55s
CI / test-with-beamline-plugins (pxi_bec) (pull_request) Canceled after 1m26s
CI / test-with-beamline-plugins (pxiii_bec) (pull_request) Canceled after 41s
CI / test-with-coverage (pull_request) Canceled after 36s
CI / coverage-analysis (pull_request) Canceled after 0s
CI / lint (push) Skipped
CI / test (3.11) (push) Skipped
CI / test (3.12) (push) Skipped
CI / test (3.13) (push) Skipped
CI / test-with-beamline-plugins (pxi_bec) (push) Skipped
CI / test-with-beamline-plugins (pxii_bec) (push) Skipped
CI / test-with-beamline-plugins (pxiii_bec) (push) Skipped
CI / lint (pull_request) Failing after 49s
CI / test (3.12) (pull_request) Failing after 45s
CI / test (3.11) (pull_request) Failing after 2m45s
CI / test (3.13) (pull_request) Canceled after 2m7s
CI / test-with-beamline-plugins (pxii_bec) (pull_request) Canceled after 55s
CI / test-with-beamline-plugins (pxi_bec) (pull_request) Canceled after 1m26s
CI / test-with-beamline-plugins (pxiii_bec) (pull_request) Canceled after 41s
CI / test-with-coverage (pull_request) Canceled after 36s
CI / coverage-analysis (pull_request) Canceled after 0s
This commit is contained in:
@@ -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__}")
|
||||
@@ -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)
|
||||
Reference in New Issue
Block a user