Add dependencies and refactor buffer methods in multiple classes

pyproject.toml

@@ -16,9 +16,11 @@ dependencies = [
     "pyyaml",
     "pyepics",
     "slac-db @ git+https://github.com/slaclab/slac-db.git@main",
+    "slac-tools @ git+https://github.com/slaclab/slac-tools.git@main",
+    "slac-timing @ git+https://github.com/slaclab/slac-timing.git@master",
 ]
 description = "Tools to support high level application development at LCLS using Python"
 dynamic = ["version"]
 name = "slac-devices"
 readme = {file = "README.md", content-type = "text/markdown"}
-requires-python = ">=3.10"
+requires-python = ">=3.10"

slac_devices/__init__.py

@@ -3,3 +3,6 @@
 
 class BaseModel(BaseModel):
     model_config = ConfigDict(arbitrary_types_allowed=True, extra="forbid")
+
+
+from slac_devices.device import Device as Device, DeviceCollection as DeviceCollection  # noqa: E402

slac_devices/bpm.py

@@ -15,6 +15,7 @@
     Metadata,
     PVSet,
 )
+from slac_timing import Buffer
 from epics import PV
 
 EPICS_ERROR_MESSAGE = "Unable to connect to EPICS."
@@ -56,36 +57,27 @@ def x(self):
         """Get TMIT value"""
         return self.controls_information.PVs.x.get()
 
-    def x_buffer(self, buffer):
-        """Retrieve TMIT signal data from timing buffer"""
-        data = buffer.get_data_buffer(f"{self.controls_information.control_name}:X")
-        if data is None:
-            raise BufferError("No data in buffer or PV not found")
-        return data
+    def x_buffer(self, buffer: Buffer):
+        """Retrieve X position data from timing buffer"""
+        return buffer.get(f"{self.controls_information.control_name}:X")
 
     @property
     def y(self):
         """Get TMIT value"""
         return self.controls_information.PVs.y.get()
 
-    def y_buffer(self, buffer):
-        """Retrieve TMIT signal data from timing buffer"""
-        data = buffer.get_data_buffer(f"{self.controls_information.control_name}:Y")
-        if data is None:
-            raise BufferError("No data in buffer or PV not found")
-        return data
+    def y_buffer(self, buffer: Buffer):
+        """Retrieve Y position data from timing buffer"""
+        return buffer.get(f"{self.controls_information.control_name}:Y")
 
     @property
     def tmit(self):
         """Get TMIT value"""
         return self.controls_information.PVs.tmit.get()
 
-    def tmit_buffer(self, buffer):
+    def tmit_buffer(self, buffer: Buffer):
         """Retrieve TMIT signal data from timing buffer"""
-        data = buffer.get_data_buffer(f"{self.controls_information.control_name}:TMIT")
-        if data is None:
-            raise BufferError("No data in buffer or PV not found")
-        return data
+        return buffer.get(f"{self.controls_information.control_name}:TMIT")
 
 
 class BPMCollection(BaseModel):
@@ -118,7 +110,7 @@ def _yield_buffer_data():
             for name, bpm in self.bpms.items():
                 address = f"{bpm.controls_information.control_name}:{suffix}"
                 try:
-                    data = buffer.get_data_buffer(address)
+                    data = buffer.get(address)
                 except (TypeError, BufferError):
                     data = None
                 yield name, data

slac_devices/device.py

@@ -2,50 +2,7 @@
 from pydantic import SerializeAsAny, ConfigDict, field_validator, field_serializer
 from typing import List, Union, Callable, Dict, Optional
 from epics import PV
-
-
-class LazyPV:
-    """Wraps an EPICS PV name and defers connection until first use."""
-
-    def __init__(self, pvname: str):
-        self._pvname = pvname
-        self._pv = None
-
-    @property
-    def pvname(self) -> str:
-        return self._pvname
-
-    def _ensure_connected(self) -> PV:
-        if self._pv is None:
-            self._pv = PV(self._pvname)
-        return self._pv
-
-    def get(self, *args, **kwargs):
-        return self._ensure_connected().get(*args, **kwargs)
-
-    def put(self, *args, **kwargs):
-        return self._ensure_connected().put(*args, **kwargs)
-
-    def get_ctrlvars(self, *args, **kwargs):
-        return self._ensure_connected().get_ctrlvars(*args, **kwargs)
-
-    def add_callback(self, *args, **kwargs):
-        return self._ensure_connected().add_callback(*args, **kwargs)
-
-    def remove_callback(self, *args, **kwargs):
-        return self._ensure_connected().remove_callback(*args, **kwargs)
-
-    def __eq__(self, other):
-        if isinstance(other, LazyPV):
-            return self._pvname == other._pvname
-        return NotImplemented
-
-    def __hash__(self):
-        return hash(self._pvname)
-
-    def __repr__(self):
-        connected = self._pv.connected if self._pv else False
-        return f"LazyPV({self._pvname!r}, connected={connected})"
+from slac_tools import LazyPV
 
 
 class PVSet(slac_devices.BaseModel):

slac_devices/lblm.py

@@ -18,6 +18,7 @@
     Metadata,
     PVSet,
 )
+from slac_timing import Buffer
 from epics import PV
 
 EPICS_ERROR_MESSAGE = "Unable to connect to EPICS."
@@ -65,12 +66,9 @@ class LBLM(Device):
     def __init__(self, **kwargs):
         super().__init__(**kwargs)
 
-    def fast_buffer(self, buffer):
+    def fast_buffer(self, buffer: Buffer):
         """Retrieve fast signal data from timing buffer"""
-        data = buffer.get_data_buffer(f"{self.controls_information.control_name}:FAST")
-        if data is None:
-            raise BufferError("No data in buffer or PV not found")
-        return data
+        return buffer.get(f"{self.controls_information.control_name}:FAST")
 
     @property
     def i0_loss(self):
@@ -108,21 +106,13 @@ def bypass(self, val: bool) -> None:
         except ValidationError as e:
             print("Bypass must be a boolean:", e)
 
-    def i0_loss_buffer(self, buffer):
+    def i0_loss_buffer(self, buffer: Buffer):
         """Retrieve I0 Loss data from timing buffer"""
-        data = buffer.get_data_buffer(self.controls_information.PVs.i0_loss.pvname)
-        if data is None:
-            raise BufferError("No data in buffer or PV not found")
-        return data
+        return buffer.get(self.controls_information.PVs.i0_loss.pvname)
 
-    def gated_integral_buffer(self, buffer):
+    def gated_integral_buffer(self, buffer: Buffer):
         """Get Gated Integral data from timing buffer"""
-        data = buffer.get_data_buffer(
-            self.controls_information.PVs.gated_integral.pvname
-        )
-        if data is None:
-            raise BufferError("No data in buffer or PV not found")
-        return data
+        return buffer.get(self.controls_information.PVs.gated_integral.pvname)
 
 
 class LBLMCollection(BaseModel):

slac_devices/pmt.py

@@ -13,6 +13,7 @@
     Metadata,
     PVSet,
 )
+from slac_timing import Buffer
 from epics import PV
 
 EPICS_ERROR_MESSAGE = "Unable to connect to EPICS."
@@ -45,14 +46,9 @@ class PMT(Device):
     def __init__(self, **kwargs):
         super().__init__(**kwargs)
 
-    def qdcraw_buffer(self, buffer):
+    def qdcraw_buffer(self, buffer: Buffer):
         """Retrieve QDCRAW signal data from timing buffer"""
-        data = buffer.get_data_buffer(
-            f"{self.controls_information.control_name}:QDCRAW"
-        )
-        if data is None:
-            raise BufferError("No data in buffer or PV not found")
-        return data
+        return buffer.get(f"{self.controls_information.control_name}:QDCRAW")
 
     @property
     def qdcraw(self):

slac_devices/wire.py

@@ -15,6 +15,7 @@
     Metadata,
     PVSet,
 )
+from slac_timing import Buffer
 from epics import PV
 
 EPICS_ERROR_MESSAGE = "Unable to connect to EPICS."
@@ -226,8 +227,8 @@ def on_status(self):
         """Returns the on status of the wire scanner."""
         return self.controls_information.PVs.on_status.get()
 
-    def position_buffer(self, buffer):
-        return buffer.get_data_buffer(f"{self.controls_information.control_name}:POSN")
+    def position_buffer(self, buffer: Buffer):
+        return buffer.get(f"{self.controls_information.control_name}:POSN")
 
     def retract(self):
         """Retracts the wire scanner"""

tests/test_wire.py

@@ -96,9 +96,7 @@ def setUp(self) -> None:
         }
 
         # 2) Patch the PV class before wire construction so all PVs are mocks
-        self.pv_class_patch = patch(
-            "slac_devices.device.PV", autospec=True
-        )
+        self.pv_class_patch = patch("slac_tools.lazy_pv.PV", autospec=True)
         self.mock_pv_class = self.pv_class_patch.start()
 
         # The instance every PV() call returns