Support per-frame analysis by AnalysisBase-based classes for streamed trajectories

[PardhavMaradani]

Is your feature request related to a problem?

AnalysisBase-based analysis classes currently cannot be used for streamed trajectories (as clarified by @orbeckst on Discord).

To illustrate, here is a streaming trajectory setup based on the IMD workshop for which DSSP analysis is needed:

import MDAnalysis as mda
from MDAnalysis.analysis.dssp import DSSP

top = "/tmp/imd-workshop-2025/workshop/sample_simulation/GROMACS/input/start.gro"
u = mda.Universe(top, "imd://localhost:8889")

We cannot run the full analysis because there is no concept of frame length for streamed trajectories, so

run = DSSP(u).run()

throws a RuntimeError: IMDReader: n_frames is unknown exception as expected.

We also cannot use the start and stop params because we can't access arbitrary fames. So, something like this

run = DSSP(u).run(start=0, stop=1)`

throws a ValueError: IMDReader: Cannot expect a start index from a stream, 'start' must be None exception.

The only other option available to call run(...) is to pass a frames array. Naturally, we cannot pass arbitrary frames, but even passing the current frame as follows

run = DSSP(u).run(frames=[u.trajectory.frame])

throws a TypeError: Streamed Trajectories must be an indexed using a slice exception as only slicing using a step ([::n]) or all frames sequentially ([:]) is currently supported.

This limits the choices to use any standalone functions for these analysis types (if they exist) or create a separate universe per frame to run the analysis on, both of which are not ideal.

Describe the solution you'd like

I'd like to be able to do something like the following:

run = DSSP(u).run(frames=[u.trajectory.frame])

where the current streamed trajectory frame is passed and get the analysis results for this current frame.

Describe alternatives you've considered

For DSSP analysis, there is a standalone assign function, but that requires re-inventing quite bit of what is already done in the DSSP class setup. Another alternative suggested by @marinegor is to create a new universe for the current frame and run the analysis class on it. Though this is simpler than using standalone functions (if they exist), it comes with quite a bit of overhead.

Additional context

My current use case is to show secondary structures computed by the DSSP analysis class for streamed trajectories in realtime in MolecularNodes (cc: @BradyAJohnston ). However, the ability to run any other AnalysisBase-based analysis classes would probably benefit a lot of streamed trajectory use cases in the future.

I will create a PR that supports the approach described in the "solution you'd like" section above. Thanks

[marinegor]

hi, thanks for raising it!

Though this is simpler than using standalone functions (if they exist), it comes with quite a bit of overhead.

Do you have any estimates on what's exactly the overhead here (e.g. some benchmarks that you did)? I would naively suggest that if topology is already provided, parsing a single frame of coordinates that are already in memory shouldn't take that much time, but I'm happy to be proven wrong.

However, the ability to run any other AnalysisBase-based analysis classes would probably benefit a lot of streamed trajectory use cases in the future

That looks tempting indeed. After some quick thinking, I'd suggest looking at _prepare_sliced_trajectory:

    def _prepare_sliced_trajectory(self, slicer: Union[slice, np.ndarray]):
        """Prepares sliced trajectory for use in subsequent parallel computations:
        namely, assigns self._sliced_trajectory and its appropriate attributes,
        self.n_frames, self.frames and self.times.

        Parameters
        ----------
        slicer : Union[slice, np.ndarray]
            appropriate slicer for the trajectory


        .. versionadded:: 2.8.0
        """
        self._sliced_trajectory = self._trajectory[slicer]
        self.n_frames = len(self._sliced_trajectory)
        self.frames = np.zeros(self.n_frames, dtype=int)
        self.times = np.zeros(self.n_frames)

and perhaps splitting it if the slicer is an ndarray -- in this case, you actually know the number of frames in advance, and can just self.n_frames = len(slicer) instead. I'd suggest checking this part in more detail. And this should work exclusively for serial backend, since I guess it doesn't make sense to parallelize it, right?

[PardhavMaradani]

Do you have any estimates on what's exactly the overhead here (e.g. some benchmarks that you did)?

Hi Egor, I don't have detailed benchmarks, but have some coarse numbers from MN tests though - you can see them in this comment of MN PR#1072. One number that isn't there is the one with DSSP initialized and run every frame - this came to around 70s on average (which would be similar to this workaround of creating a new universe per frame and running DSSP). More than the creating universe part, it was the DSSP setup cost that led me to explore if there is a possible fix before committing to one of the workarounds.

After some quick thinking, I'd suggest looking at _prepare_sliced_trajectory: and perhaps splitting it if the slicer is an ndarray

Thanks for the pointer. Yes, I initially started off with the slicer and went from there. But given that the _sliced_trajectory has to be eventually iterated in AnalysisBase's _compute, it felt cleaner and simpler to create a separate iterator specific to the streaming case (given this already works for regular trajectories) so that the changes are contained. Could you please take a look at the corresponding PR #5184 when you get a chance. Thanks