Refactor buffer retrieval methods in BPM, LBLM, PMT, and Wire classes

slac_devices/bpm.py

@@ -57,27 +57,27 @@ def x(self):
         """Get TMIT value"""
         return self.controls_information.PVs.x.get()
 
-    def x_buffer(self, buffer: Buffer):
-        """Retrieve X position data from timing buffer"""
-        return buffer.get(f"{self.controls_information.control_name}:X")
+    def x_buffer(self, buffer: Buffer, **kwargs):
+        """Retrieve X signal data from timing buffer"""
+        return buffer.get(f"{self.controls_information.control_name}:X", **kwargs)
 
     @property
     def y(self):
         """Get TMIT value"""
         return self.controls_information.PVs.y.get()
 
-    def y_buffer(self, buffer: Buffer):
-        """Retrieve Y position data from timing buffer"""
-        return buffer.get(f"{self.controls_information.control_name}:Y")
+    def y_buffer(self, buffer: Buffer, **kwargs):
+        """Retrieve Y signal data from timing buffer"""
+        return buffer.get(f"{self.controls_information.control_name}:Y", **kwargs)
 
     @property
     def tmit(self):
         """Get TMIT value"""
         return self.controls_information.PVs.tmit.get()
 
-    def tmit_buffer(self, buffer: Buffer):
+    def tmit_buffer(self, buffer: Buffer, **kwargs):
         """Retrieve TMIT signal data from timing buffer"""
-        return buffer.get(f"{self.controls_information.control_name}:TMIT")
+        return buffer.get(f"{self.controls_information.control_name}:TMIT", **kwargs)
 
 
 class BPMCollection(BaseModel):
@@ -94,13 +94,14 @@ def validate_bpms(cls, v) -> Dict[str, BPM]:
         return v
 
     def get_buffer_data(
-        self, buffer, suffix: str = "TMIT"
+        self, buffer, suffix: str = "TMIT", **kwargs
     ) -> Dict[str, Optional[list]]:
         """Retrieve buffer data for all BPMs in the collection.
 
         Args:
             buffer: An edef EventDefinition or BSABuffer object.
             suffix: PV suffix to read (e.g. "TMIT", "X", "Y").
+            **kwargs: Passed to buffer.get() (e.g. pad, retries).
 
         Returns:
             Dict mapping BPM name to data array, or None for unreachable BPMs.
@@ -110,7 +111,7 @@ def _yield_buffer_data():
             for name, bpm in self.bpms.items():
                 address = f"{bpm.controls_information.control_name}:{suffix}"
                 try:
-                    data = buffer.get(address)
+                    data = buffer.get(address, **kwargs)
                 except (TypeError, BufferError):
                     data = None
                 yield name, data

slac_devices/lblm.py

@@ -66,9 +66,9 @@ class LBLM(Device):
     def __init__(self, **kwargs):
         super().__init__(**kwargs)
 
-    def fast_buffer(self, buffer: Buffer):
+    def fast_buffer(self, buffer: Buffer, **kwargs):
         """Retrieve fast signal data from timing buffer"""
-        return buffer.get(f"{self.controls_information.control_name}:FAST")
+        return buffer.get(f"{self.controls_information.control_name}:FAST", **kwargs)
 
     @property
     def i0_loss(self):
@@ -106,13 +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: Buffer):
+    def i0_loss_buffer(self, buffer: Buffer, **kwargs):
         """Retrieve I0 Loss data from timing buffer"""
-        return buffer.get(self.controls_information.PVs.i0_loss.pvname)
+        return buffer.get(self.controls_information.PVs.i0_loss.pvname, **kwargs)
 
-    def gated_integral_buffer(self, buffer: Buffer):
+    def gated_integral_buffer(self, buffer: Buffer, **kwargs):
         """Get Gated Integral data from timing buffer"""
-        return buffer.get(self.controls_information.PVs.gated_integral.pvname)
+        return buffer.get(self.controls_information.PVs.gated_integral.pvname, **kwargs)
 
 
 class LBLMCollection(BaseModel):

slac_devices/pmt.py

@@ -46,9 +46,9 @@ class PMT(Device):
     def __init__(self, **kwargs):
         super().__init__(**kwargs)
 
-    def qdcraw_buffer(self, buffer: Buffer):
+    def qdcraw_buffer(self, buffer: Buffer, **kwargs):
         """Retrieve QDCRAW signal data from timing buffer"""
-        return buffer.get(f"{self.controls_information.control_name}:QDCRAW")
+        return buffer.get(f"{self.controls_information.control_name}:QDCRAW", **kwargs)
 
     @property
     def qdcraw(self):

slac_devices/wire.py

@@ -227,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: Buffer):
-        return buffer.get(f"{self.controls_information.control_name}:POSN")
+    def position_buffer(self, buffer: Buffer, **kwargs):
+        return buffer.get(f"{self.controls_information.control_name}:POSN", **kwargs)
 
     def retract(self):
         """Retracts the wire scanner"""