"""
RMSD Analysis
"""
import logging
from collections import deque
from typing import ClassVar
import matplotlib.pyplot as plt
from IPython.display import display
from joblib import delayed
from MDAnalysis.analysis import rms
from mdadash.backend.widgets.base import WidgetBase
logger = logging.getLogger(__name__)
[docs]
class RMSD(WidgetBase):
"""
**RMSD Analysis**
This widget uses `MDAnalysis.analysis.rms.rmsd`_ to calculate RMSD of a
selection. The reference positions used by this widget are the initial
positions of the selection when the widget instance is created or the
initial positions whenever the selection is updated.
.. _MDAnalysis.analysis.rms.rmsd: https://docs.mdanalysis.org/stable/
documentation_pages/analysis/rms.html#MDAnalysis.analysis.rms.rmsd
"""
name = "RMSD"
description = "RMSD of a selection"
_notes = (
"If simulations are performed under periodic boundary conditions "
"then you must make your molecules whole before performing RMSD "
"calculations so that the centers of mass of the mobile and reference "
"structure are properly superimposed. You can add custom transformations "
"to the universe in the Universe Configuration section in the Settings page.\n\n"
"Note: The reference positions used by this widget are the initial positions "
"of the selection when the widget instance is created or the initial positions "
"whenever the selection is updated."
)
_inputs: ClassVar = [
{
"attribute": "_run_frequency",
"name": "Run frequency",
"description": "The frequency with which the widget is run",
"type": "select",
"items": [
"every-frame",
"batch",
],
},
{
"attribute": "_run_mode",
"name": "Run mode",
"description": "The mode in which the widget is run",
"type": "select",
"items": [
"serial",
"parallel",
],
},
{
"attribute": "selection",
"name": "Selection",
"description": "MDAnalysis selection phrase",
"type": "str",
"validations": ["required"],
},
{
"attribute": "center",
"name": "Center",
"description": "Subtract center of geometry before calculation",
"type": "bool",
},
{
"attribute": "superposition",
"name": "Superposition",
"description": (
"Perform a rotational and translational superposition with the fast QCP algorithm"
),
"type": "bool",
},
{
"attribute": "custom_title",
"name": "Custom title",
"description": "Custom title for the plot",
"type": "str",
},
{
"attribute": "maxlen",
"name": "Max values",
"description": "Max values to show in plot",
"type": "int",
},
{
"attribute": "x_type",
"name": "X-axis",
"type": "toggle",
"options": [
{"name": "Time", "value": "time"},
{"name": "Step", "value": "step"},
],
},
]
def __init__(self):
super().__init__()
self.selection = "protein"
self.center = False
self.superposition = False
self.ag = None
self.reference_positions = None
self.title = "RMSD"
self.custom_title = None
self.default_maxlen = 100
self.maxlen = self.default_maxlen
self.x_type = "time"
self.x_values = None
self._setup_plot()
self._reset_plot_values()
def _setup_plot(self):
"""Setup matplotlib plot"""
self.fig, self.ax = plt.subplots()
(self.plot,) = self.ax.plot([], [])
self.ax.set_ylabel("RMSD (Å)")
self.ax.grid(True)
self._set_title()
def _reset_plot_values(self):
"""Reset plot values"""
self.steps = deque(maxlen=self.maxlen)
self.times = deque(maxlen=self.maxlen)
self.y_values = deque(maxlen=self.maxlen)
self._set_x_values()
def _set_title(self):
"""Set plot title"""
self.ax.set_title(
self.custom_title.replace("\\n", "\n") if self.custom_title else self.title
)
def _set_x_values(self):
"""Set the values for the x-axis"""
if self.x_type == "step":
x_label = "Step"
self.x_values = self.steps
else:
x_label = "Time (ps)"
self.x_values = self.times
self.ax.set_xlabel(x_label)
def _update_selection(self):
"""Update atom groups when selection phrase changes"""
self.ag = self.u.select_atoms(self.selection)
self.reference_positions = self.ag.positions.copy()
self.title = f"RMSD of '{self.selection}'"
self._set_title()
self._update_plot(self._compute_current_frame())
[docs]
def on_post_create(self):
"""on_post_create handler"""
self._set_title()
self._reset_plot_values()
[docs]
def on_post_connect(self):
"""on_post_connect handler"""
self._update_selection()
def _compute_current_frame(self):
"""Compute values for current frame"""
rmsd_value = rms.rmsd(
self.ag.positions,
self.reference_positions,
center=self.center,
superposition=self.superposition,
)
return (
self.u.trajectory.ts.data["step"],
self.u.trajectory.ts.data["time"],
rmsd_value,
)
def _compute_batch(self):
"""Compute values for current batch"""
values = []
for i in range(self.u.trajectory.buffer_size):
_ = self.u.trajectory[i]
values.append(self._compute_current_frame())
return values
def _update_plot(self, values):
"""Append values and update plot"""
if isinstance(values, tuple):
values = [values]
# update plot points
for value in values:
(steps, times, v) = value
self.steps.append(steps)
self.times.append(times)
self.y_values.append(v)
# update plot
self.plot.set_data(self.x_values, self.y_values)
self.ax.relim()
self.ax.autoscale_view()
self.fig.canvas.draw()
display(self.fig)
[docs]
def run_every_frame(self):
"""every-frame run handler"""
self._update_plot(self._compute_current_frame())
[docs]
def run_batch(self):
"""batch run handler"""
self._update_plot(self._compute_batch())
[docs]
def get_parallel_job(self):
"""get parallel job handler"""
if self._run_frequency == "batch":
return delayed(self._compute_batch)()
return delayed(self._compute_current_frame)()
[docs]
def apply_parallel_results(self, values):
"""apply parallel results handler"""
self._update_plot(values)