RMSD Median Service

GitHub Link to Code.

Median RMSD service implementation.

class mdxplain.analysis.structure.services.rmsd_median_service.RMSDMedianService(pipeline_data: PipelineData | None)

Expose median RMSD workflows.

The service mediates between the pipeline-facing API and the numerical RMSDCalculator. It resolves trajectory selections and atom filters, ensures consistent reference handling, and delegates the computation using the median metric for outlier-resistant RMSD values.

Examples

>>> service = RMSDMedianService(pipeline_data)
>>> service.metric
'median'

__init__(pipeline_data: PipelineData | None) → None

Store pipeline context with median metric.

Ensures that valid pipeline data is supplied before storing configuration values for later use.

Parameters

pipeline_dataPipelineData | None: Pipeline configuration controlling chunk size and memmap usage. Must not be None.

Returns

None: This initializer returns None.

Examples

>>> service = RMSDMedianService(pipeline_data)
>>> service.metric
'median'

to_reference(reference_traj: int = 0, reference_frame: int = 0, traj_selection: int | str | List[int | str] | all = 'all', atom_selection: str = 'all') → Dict[str, ndarray]

Calculate RMSD against a fixed reference frame using median metric.

Selects the requested trajectories, extracts the reference frame and delegates the numerical work to the calculator using the median metric for outlier-resistant values.

Parameters

reference_trajint, optional: Index of the trajectory containing the reference frame. Defaults to 0.
reference_frameint, optional: Frame index within the reference trajectory. Defaults to 0.
traj_selectionUnion[int, str, list[Union[int, str]], ‘all’], optional: Selection describing the trajectories to process. Defaults to "all".
atom_selectionstr, optional: MDTraj atom selection string forwarded to the calculation. Defaults to "all".

Returns

dict[str, np.ndarray]: Mapping of trajectory names to RMSD arrays.

Examples

>>> service = RMSDMedianService(pipeline_data)
>>> service.to_reference()
{'traj_0': array([...])}

frame_to_frame(lag: int = 1, traj_selection: int | str | List[int | str] | all = 'all', atom_selection: str = 'all') → Dict[str, ndarray]

Calculate RMSD between lag-separated frames using median metric.

Resolves the requested trajectories and atom selection, then evaluates RMSD values between frames separated by lag using the median metric.

Parameters

lagint, optional: Distance between frames. Must be positive. Defaults to 1.
traj_selectionUnion[int, str, list[Union[int, str]], ‘all’], optional: Selection describing the trajectories to process. Defaults to "all".
atom_selectionstr, optional: MDTraj atom selection string. Defaults to "all".

Returns

dict[str, np.ndarray]: Mapping of trajectory names to RMSD arrays computed for the requested lag.

Examples

>>> service = RMSDMedianService(pipeline_data)
>>> service.frame_to_frame(lag=1)
{'traj_0': array([...])}

Calculate window-wise RMSD using first frame as reference.

Sliding windows are extracted per trajectory, aligned to the window’s first frame and condensed via the median metric into a single RMSD value per window.

Parameters

window_sizeint: Number of frames within each window.
strideint, optional: Step size between windows. Defaults to window_size when omitted.
traj_selectionUnion[int, str, list[Union[int, str]], ‘all’], optional: Selection describing the trajectories to process. Defaults to "all".
atom_selectionstr, optional: MDTraj atom selection string. Defaults to "all".

Returns

dict[str, np.ndarray]: Mapping of trajectory names to window-wise RMSD arrays.

Examples

>>> service = RMSDMedianService(pipeline_data)
>>> service.window_frame_to_start(window_size=3)
{'traj_0': array([...])}

Calculate window-wise RMSD between lag-separated frames.

Sliding windows are processed with lag-separated frame pairs and condensed using the median metric.

Parameters

window_sizeint: Number of frames within each window.
strideint, optional: Step size between windows. Defaults to window_size when omitted.
lagint, optional: Lag between frames inside each window. Defaults to 1.
traj_selectionUnion[int, str, list[Union[int, str]], ‘all’], optional: Selection describing the trajectories to process. Defaults to "all".
atom_selectionstr, optional: MDTraj atom selection string. Defaults to "all".

Returns

dict[str, np.ndarray]: Mapping of trajectory names to window-wise RMSD arrays.

Examples

>>> service = RMSDMedianService(pipeline_data)
>>> service.window_frame_to_frame(window_size=4, lag=1)
{'traj_0': array([...])}