Source code for openpathsampling.pathsimulators.path_simulator

import abc
import sys
import logging
from future.utils import with_metaclass

import openpathsampling as paths
from openpathsampling.netcdfplus import StorableNamedObject, StorableObject

from ..ops_logging import initialization_logging


logger = logging.getLogger(__name__)
init_log = logging.getLogger('openpathsampling.initialization')

[docs]class MCStep(StorableObject): """ A monte-carlo step in the main PathSimulation loop It references all objects created and used in a MC step. The used mover, and simulator as well as the initial and final sampleset, the step number and the generated movechange. Attributes ---------- simulation : PathSimulation the running pathsimulation responsible for generating the step mccycle : int the step number counting from the root sampleset previous : SampleSet the initial (pre) sampleset active : SampleSet the final (post) sampleset change : MoveChange the movechange describing the transition from pre to post """
[docs] def __init__(self, simulation=None, mccycle=-1, previous=None, active=None, change=None): super(MCStep, self).__init__() self.simulation = simulation self.previous = previous self.active = active self.change = change self.mccycle = mccycle
[docs]class PathSimulator(with_metaclass(abc.ABCMeta, StorableNamedObject)): """Abstract class for the "main" function of a simulation. Parameters ---------- storage : :class:`.Storage` Storage file for results Attributes ---------- save_frequency : int Results should be sync'd (saved to disk) after every ``save_frequency`` steps. Note: subclasses must directly implement this, the attribute is just a placeholder. output_stream : file Subclasses should write output to this, allowing a standard way to redirect any output. allow_refresh : bool Whether to allow the output to refresh an ipynb cell; default True. This is likely to be overridden when a pathsimulator is wrapped in another simulation. """ calc_name = "PathSimulator" _excluded_attr = ['sample_set', 'step', 'save_frequency', 'output_stream'] hook_names = ['before_simulation', 'before_step', 'after_step', 'after_simulation']
[docs] def __init__(self, storage): super(PathSimulator, self).__init__() self.storage = storage # self.engine = engine self.save_frequency = 1 self.step = 0 initialization_logging( logger=init_log, obj=self, entries=['storage']#, 'engine'] ) self.sample_set = None self.output_stream = sys.stdout # user can change to file handler self.allow_refresh = True self.hooks = self.empty_hooks() self.attach_default_hooks()
def sync_storage(self): """ Will sync all collective variables and the storage to disk """ if self.storage is not None: self.storage.sync_all() def attach_default_hooks(self): pass def empty_hooks(self): """Return a hook dictionary with no hooks.""" return {k: [] for k in self.hook_names} def attach_hook(self, hook, hook_for=None): """Attach a hook class or method to this simulation. Parameters ---------- hook : :class:`.PathSimulatorHook` or method Hook to add hook_for : str or None If None (default) then the ``hook`` must be a class with methods named to match the hook names in this simulator. If ``hook`` is a method, then ``hook_for`` must be the name of the hook it represents """ def add_hook_method(hook_method, hook_name): try: self.hooks[hook_name].append(hook_method) except KeyError: raise TypeError("No hook '" + hook_name + "' in " + str(self.__class__.__name__)) if hook_for is None: for hook_name in hook.implemented_for: hook_method = getattr(hook, hook_name) add_hook_method(hook_method, hook_name) else: add_hook_method(hook, hook_for) def run_hooks(self, hook_name, **kwargs): """Run the hooks for the given ``hook_name``""" hook_name_state = {} for hook in self.hooks[hook_name]: result = hook(**kwargs) if result is not None: hook_name_state[hook] = result if hook_name_state != {}: return hook_name_state @abc.abstractmethod def run(self, n_steps): """ Run the simulator for a number of steps Parameters ---------- n_steps : int number of step to be run """ raise NotImplementedError() def save_initial_step(self): """ Save the initial state as an MCStep to the storage """ mcstep = MCStep( simulation=self, mccycle=self.step, active=self.sample_set, change=paths.AcceptedSampleMoveChange(self.sample_set.samples) ) if self.storage is not None: self.storage.steps.save(mcstep) self.storage.sync_all()