Add a current frame iterator for streamed trajectories

package/AUTHORS

@@ -265,6 +265,7 @@ Chronological list of authors
   - Raúl Lois-Cuns  
   - Pranay Pelapkar
   - Shreejan Dolai
+  - Pardhav Maradani
 
 External code
 -------------

package/CHANGELOG

@@ -15,7 +15,7 @@ The rules for this file:
 
 -------------------------------------------------------------------------------
 ??/??/?? IAlibay, orbeckst, marinegor, tylerjereddy, ljwoods2, marinegor,
-         spyke7, talagayev
+         spyke7, talagayev, PardhavMaradani
 
  * 2.11.0
 
@@ -30,6 +30,9 @@ Fixes
    DSSP by porting upstream PyDSSP 0.9.1 fix (Issue #4913)
 
 Enhancements
+ * Added a current frame iterator `StreamFrameIteratorCurrent` for streamed
+   trajectories to enable `AnalysisBase.run` on per-frame streamed data
+   (Issue #5183, PR #5184)
  * Enables parallelization for analysis.diffusionmap.DistanceMatrix
    (Issue #4679, PR #4745)
 

package/MDAnalysis/coordinates/IMD.py

@@ -286,6 +286,9 @@ def __init__(
             raise RuntimeError(f"IMDReader: Read error: {e}") from e
 
     def _read_frame(self, frame):
+        if frame == self._frame:
+            logger.debug("IMDReader: Using current frame %d", self._frame)
+            return self.ts
 
         imdf = self._imdclient.get_imdframe()
 

package/MDAnalysis/coordinates/base.py

@@ -52,6 +52,8 @@
 
 .. autoclass:: StreamFrameIteratorSliced
 
+.. autoclass:: StreamFrameIteratorCurrent
+
 .. _ReadersBase:
 
 Readers
@@ -1882,10 +1884,11 @@ class StreamReaderBase(ReaderBase):
     See Also
     --------
     StreamFrameIteratorSliced : Iterator for stepped streaming access
+    StreamFrameIteratorCurrent : Iterator for current frame streaming access
     ReaderBase : Base class for standard trajectory readers
 
 
-    .. versionadded:: 2.10.0 
+    .. versionadded:: 2.10.0
     """
 
     def __init__(self, filename, convert_units=True, **kwargs):
@@ -2005,7 +2008,9 @@ def _reopen(self):
             raise RuntimeError(
                 "{}: Cannot reopen stream".format(self.__class__.__name__)
             )
-        self._frame = -1
+        if self._frame == 0:
+            # only reset when stream hasn't been iterated using next
+            self._frame = -1
         self._reopen_called = True
 
     def timeseries(self, **kwargs):
@@ -2040,7 +2045,7 @@ def __getitem__(self, frame):
 
         Returns
         -------
-        FrameIteratorAll or StreamFrameIteratorSliced
+        FrameIteratorAll or StreamFrameIteratorSliced or StreamFrameIteratorCurrent
             Iterator for the requested slice.
 
         Raises
@@ -2060,15 +2065,30 @@ def __getitem__(self, frame):
         See Also
         --------
         StreamFrameIteratorSliced
+        StreamFrameIteratorCurrent
         """
-        if isinstance(frame, slice):
+        if isinstance(frame, numbers.Integral):
+            if frame == self.trajectory.frame:
+                return self._read_frame(frame)
+            else:
+                raise ValueError(
+                    "Streamed trajectories must specify current frame value"
+                )
+        elif isinstance(frame, slice):
             _, _, step = self.check_slice_indices(
                 frame.start, frame.stop, frame.step
             )
             if step is None:
                 return FrameIteratorAll(self)
             else:
                 return StreamFrameIteratorSliced(self, step)
+        elif isinstance(frame, (list, np.ndarray)):
+            if len(frame) == 1 and frame[0] == self.trajectory.frame:
+                return StreamFrameIteratorCurrent(self)
+            else:
+                raise ValueError(
+                    "Streamed trajectories must have single current frame value"
+                )
         else:
             raise TypeError(
                 "Streamed trajectories must be an indexed using a slice"
@@ -2310,4 +2330,39 @@ def step(self):
             Step size for iteration. Always a positive integer greater than 0.
 
         """
-        return self._step
+        return self._step
+
+
+class StreamFrameIteratorCurrent(FrameIteratorBase):
+    """Iterator for current frame access in a streamed trajectory.
+
+    Created when an array with a single current frame value is passed.
+
+    Parameters
+    ----------
+    trajectory : StreamReaderBase
+        The streaming trajectory reader to iterate over. Must be a
+        stream-based reader that supports continuous data reading.
+
+    See Also
+    --------
+    StreamReaderBase
+    FrameIteratorBase
+
+    .. versionadded:: 2.11.0
+    """
+
+    def __init__(self, trajectory):
+        super(StreamFrameIteratorCurrent, self).__init__(trajectory)
+
+    def __len__(self):
+        return 1
+
+    def __iter__(self):
+        yield self.trajectory._read_frame(self.trajectory.frame)
+
+    def __next__(self):
+        raise StopIteration from None
+
+    def __getitem__(self, frame):
+        raise RuntimeError("Current frame iterator does not support indexing")