RMSD Mad Service

GitHub Link to Code.

MAD RMSD service implementation.

class mdxplain.analysis.structure.services.rmsd_mad_service.RMSDMadService(pipeline_data: PipelineData | None)

Expose MAD 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 absolute deviation (MAD) metric for outlier-resistant RMSD values.

Examples

>>> service = RMSDMadService(pipeline_data)
>>> service.metric
'mad'

__init__(pipeline_data: PipelineData | None) → None

Store pipeline context with MAD 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 = RMSDMadService(pipeline_data)
>>> service.metric
'mad'

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 MAD metric.

Selects the requested trajectories, extracts the reference frame and delegates the numerical work to the calculator using the median absolute deviation (MAD) metric for maximum outlier resistance.

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 = RMSDMadService(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 MAD metric.

Resolves the requested trajectories and atom selection, then evaluates RMSD values between frames separated by lag using the MAD 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 = RMSDMadService(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 MAD 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 = RMSDMadService(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 MAD 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 = RMSDMadService(pipeline_data)
>>> service.window_frame_to_frame(window_size=4, lag=1)
{'traj_0': array([...])}