Source code for openpathsampling.pathmover

"""
Created on 19.07.2014

@author: Jan-Hendrik Prinz
@author: David W. H. Swenson
"""
import abc
import logging
import numpy as np
import random

import openpathsampling as paths
from openpathsampling.netcdfplus import StorableNamedObject, StorableObject
from openpathsampling.pathmover_inout import InOutSet, InOut
from openpathsampling.rng import default_rng
from .ops_logging import initialization_logging
from .treelogic import TreeMixin

from openpathsampling.deprecations import deprecate, has_deprecations
from openpathsampling.deprecations import (SAMPLE_DETAILS, MOVE_DETAILS,
                                           NEW_SNAPSHOT_KWARG_SELECTOR)

from future.utils import with_metaclass

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


# TODO: Remove if really not used anymore
# otherwise might move to utils or tools
def make_list_of_pairs(inlist):
    """
    Converts input from several possible formats into a list of pairs: used
    to clean input for swap-like moves.

    Allowed input formats:
    * flat list of length 2N
    * list of pairs
    * None (returns None)

    Anything else will lead to a ValueError or AssertionError

    Parameters
    ----------
    inlist : list
        input list, either flat list of length 2N, a list of pairs or None

    Returns
    -------
    list of pairs
    """
    if inlist is None:
        return None

    _ = len(inlist)  # raises TypeError, avoids everything else

    # based on first element, decide whether this should be a list of lists
    # or a flat list
    try:
        _ = len(inlist[0])
        list_of_lists = True
    except TypeError:
        list_of_lists = False

    if list_of_lists:
        for elem in inlist:
            assert len(elem) == 2, "List of lists: inner list length != 2"
        outlist = inlist
    else:
        assert len(inlist) % 2 == 0, \
            "Flattened list: length not divisible by 2"
        outlist = [
            [a, b] for (a, b) in zip(
                inlist[slice(0, None, 2)],
                inlist[slice(1, None, 2)])
        ]
    # Note that one thing we don't check is whether the items are of the
    # same type. That might be worth doing someday; for now, we trust that
    # part to work.
    return outlist


class SampleNaNError(Exception):
    def __init__(self, message, trial_sample, details):
        super(SampleNaNError, self).__init__(message)
        self.trial_sample = trial_sample
        self.details = details


class SampleMaxLengthError(Exception):
    def __init__(self, message, trial_sample, details):
        super(SampleMaxLengthError, self).__init__(message)
        self.trial_sample = trial_sample
        self.details = details


class MoveChangeNaNError(Exception):
    pass


[docs]class PathMover(with_metaclass(abc.ABCMeta, TreeMixin, StorableNamedObject)): """ A PathMover is the description of a move in replica space. Notes ----- A pathmover takes a SampleSet() and returns MoveChange() that is used to change the old SampleSet() to the new one. SampleSet1 + MoveChange1 => SampleSet2 A MoveChange is effectively a list of Samples. The change acts upon a SampleSet by replacing existing Samples in the same ensemble sequentially. SampleSet({samp1(ens1), samp2(ens2), samp3(ens3)}) + MoveChange([samp4(ens2)]) => SampleSet({samp1(ens1), samp4(ens2), samp3(ens3)}) Note, that a SampleSet is an unordered list (or a set). Hence the ordering in the example is arbitrary. Potential future change: `engine` is not needed for all PathMovers (replica exchange, ensemble hopping, path reversal, and moves which combine these [state swap] have no need for the engine). Maybe that should be moved into only the ensembles that need it? ~~~DWHS Also, I agree with the separating trial and acceptance. We might choose to use a different acceptance criterion than Metropolis. For example, the "waste recycling" approach recently re-discovered by Frenkel (see also work by Athenes, Jourdain, and old work by Kalos) might be interesting. I think the best way to do this is to keep the acceptance in the PathMover, but have it be a separate class ~~~DWHS """ # __metaclass__ = abc.ABCMeta
[docs] def __init__(self): StorableNamedObject.__init__(self) self._rng = default_rng() self._in_ensembles = None self._out_ensembles = None self._len = None self._inout = None self._trust_candidate = False
# initialization_logging(logger=init_log, obj=self, # entries=['ensembles']) _is_ensemble_change_mover = None @property def is_ensemble_change_mover(self): if self._is_ensemble_change_mover is None: return False else: return self._is_ensemble_change_mover _is_canonical = None @property def is_canonical(self): return self._is_canonical @property def default_name(self): return self.__class__.__name__[:-5] # +------------------------------------------------------------------------- # | tree implementation overrides # +------------------------------------------------------------------------- @property def _subnodes(self): return self.submovers @property def identifier(self): return self @staticmethod def _default_match(original, test): if isinstance(test, paths.PathMover): return original is test elif issubclass(test, paths.PathMover): return original.__class__ is test else: return False @property def submovers(self): """ Returns a list of submovers Returns ------- list of openpathsampling.PathMover the list of sub-movers """ return [] @staticmethod def _flatten(ensembles): if type(ensembles) is list: return [s for ens in ensembles for s in PathMover._flatten(ens)] else: return [ensembles] # +------------------------------------------------------------------------- # | analyze effects of sample sets # +------------------------------------------------------------------------- def move_replica_state(self, replica_states): return self.in_out.move(replica_states) def sub_replica_state(self, replica_states): """ Return set of replica states that a submover might be called with Parameters ---------- replica_states : set of `openpathsampling.pathmover_inout.ReplicaState` Returns ------- list of set of `ReplicaState` """ return [replica_states] * len(self.submovers) def _generate_in_out(self): if len(self.output_ensembles) == 0: return { InOutSet([]) } elif (len(self.input_ensembles) == len(self.output_ensembles) == 1): in_ens = self.input_ensembles[0] out_ens = self.output_ensembles[0] return InOutSet([InOut([((in_ens, out_ens, 0), 1)])]) else: # Fallback could be all possibilities, but for now we ask the user! raise NotImplementedError( 'Please implement the in-out-matrix for this mover.') @property def in_out(self): """ List the input -> output relation for ensembles A mover will pick one or more replicas from specific ensembles. Alter them (or not) and place these (or additional ones) in specific ensembles. This relation can be visualized as a mapping of input to output ensembles. Like ReplicaExchange ens1 -> ens2 ens2 -> ens1 EnsembleHop (A sample in ens1 will disappear and appear in ens2) ens1 -> ens2 DuplicateMover (create a copy with a new replica number) Not used yet! ens1 -> ens1 None -> ens1 Returns ------- list of list of tuple : (:obj:`openpathsampling.Ensemble`, :obj:`openpathsampling.Ensemble`) a list of possible lists of tuples of ensembles. Notes ----- The default implementation will (1) in case of a single input and output connect the two, (2) return nothing if there are no out_ensembles and (3) for more then two require implementation """ if self._inout is None: self._inout = self._generate_in_out() return self._inout def _ensemble_signature(self, as_set=False): """Return tuple form of (input_ensembles, output_ensembles). Useful for MoveScheme, e.g., identifying which movers should be removed as part of a replacement. """ inp = tuple(self.input_ensembles) out = tuple(self.output_ensembles) if as_set: inp = set(inp) out = set(out) return inp, out @property def ensemble_signature(self): return self._ensemble_signature() @property def ensemble_signature_set(self): return self._ensemble_signature(as_set=True) @property def input_ensembles(self): """Return a list of possible used ensembles for this mover This list contains all Ensembles from which this mover might pick samples. This is very useful to determine on which ensembles a mover acts for analysis and sanity checking. Returns ------- list of :class:`openpathsampling.Ensemble` the list of input ensembles """ if self._in_ensembles is None: ensembles = self._get_in_ensembles() self._in_ensembles = list(set(self._flatten(ensembles))) return self._in_ensembles @property def output_ensembles(self): """Return a list of possible returned ensembles for this mover This list contains all Ensembles for which this mover might return samples. This is very useful to determine on which ensembles a mover affects in later steps for analysis and sanity checking. Returns ------- list of Ensemble the list of output ensembles """ if self._out_ensembles is None: ensembles = self._get_out_ensembles() self._out_ensembles = list(set(self._flatten(ensembles))) return self._out_ensembles def _get_in_ensembles(self): """Function that computes the list of input ensembles """ return [] def _get_out_ensembles(self): """Function that computes the list of output ensembles Default is the same as in_ensembles """ return self._get_in_ensembles() @staticmethod def legal_sample_set(sample_set, ensembles=None, replicas='all'): """ This returns all the samples from sample_set which are in both self.replicas and the parameter ensembles. If ensembles is None, we use self.ensembles. If you want all ensembles allowed, pass ensembles='all'. Parameters ---------- sample_set : `openpathsampling.SampleSet` the sampleset from which to pick specific samples matching certain criteria ensembles : list of `openpathsampling.Ensembles` the ensembles to pick from replicas : list of int or `all` the replicas to pick or `'all'` for all """ mover_replicas = sample_set.replica_list() if replicas == 'all': selected_replicas = sample_set.replica_list() else: selected_replicas = replicas reps = list(set(mover_replicas) & set(selected_replicas)) rep_samples = [] for rep in reps: rep_samples.extend(sample_set.all_from_replica(rep)) # logger.debug("ensembles = " + str([ensembles])) # logger.debug("self.ensembles = " + str(self.ensembles)) if ensembles is None: ensembles = 'all' if ensembles == 'all': legal_samples = rep_samples else: ens_samples = [] if type(ensembles) is not list: ensembles = [ensembles] for ens in ensembles: # try: # ens_samples.extend(sample_set.all_from_ensemble(ens[0])) # except TypeError: ens_samples.extend(sample_set.all_from_ensemble(ens)) legal_samples = list(set(rep_samples) & set(ens_samples)) return legal_samples @staticmethod def select_sample(sample_set, ensembles=None, replicas=None): """ Returns one of the legal samples given self.replica and the ensemble set in ensembles. Parameters ---------- sample_set : `openpathsampling.SampleSet` the sampleset from which to pick specific samples matching certain criteria ensembles : list of `openpathsampling.Ensembles` or `None` the ensembles to pick from or `None` for all replicas : list of int or None the replicas to pick or `None` for all """ if replicas is None: replicas = 'all' logger.debug( "replicas: " + str(replicas) + " ensembles: " + repr(ensembles)) legal = PathMover.legal_sample_set(sample_set, ensembles, replicas) for sample in legal: logger.debug( "legal: (" + str(sample.replica) + "," + str(sample.trajectory) + "," + repr(sample.ensemble) + ")") # TODO: This can't go through numpy.random as Samples unwrap to # Snapshots when cast to arrays selected = random.choice(legal) logger.debug( "selected sample: (" + str(selected.replica) + "," + str(selected.trajectory) + "," + repr(selected.ensemble) + ")") return selected @abc.abstractmethod def move(self, sample_set): """ Run the generation starting with the initial sample_set specified. Parameters ---------- sample_set : SampleSet the initially used sampleset Returns ------- samples : MoveChange the MoveChange instance describing the change from the old to the new SampleSet """ return paths.EmptyMoveChange() # pragma: no cover def __str__(self): if self.name == self.__class__.__name__: return self.__repr__() else: return self.name
class IdentityPathMover(PathMover): """ The simplest Mover that does nothing ! Notes ----- Since is does nothing it is considered rejected everytime! It can be used to test function of PathMover Parameters ---------- counts_as_trial : bool Whether this mover should count as a trial or not. If `True`, the `EmptyMoveChange` returned includes this mover, which means it gets counted as a trial in analysis of acceptance. If `False` (default), the mover for the returned move change is `None`, which does not get counted as a trial. """ def __init__(self, counts_as_trial=False): super(IdentityPathMover, self).__init__() self.counts_as_trial = counts_as_trial def move(self, sample_set): mover = self if self.counts_as_trial else None return paths.EmptyMoveChange(mover=mover) ############################################################################### # GENERATORS ###############################################################################
[docs]class SampleMover(PathMover):
[docs] def __init__(self): super(SampleMover, self).__init__()
def metropolis(self, trials): """Implements the Metropolis acceptance for a list of trial samples The Metropolis uses the .bias for each sample and checks of samples are valid - are in the proposed ensemble. This will give an acceptance probability for all samples. If the product is smaller than a random number the change will be accepted. Parameters ---------- trials : list of openpathsampling.Sample the list of all samples to be applied in a change. Returns ------- bool True if the trial is accepted, False otherwise details : openpathsampling.Details Returns a Details object that contains information about the decision, i.e. total acceptance and random number """ shoot_str = "MC in {cls} using samples {trials}" logger.info(shoot_str.format(cls=self.__class__.__name__, trials=trials)) trial_dict = dict() for trial in trials: trial_dict[trial.ensemble] = trial accepted = True probability = 1.0 # TODO: This isn't right. `bias` should be associated with the # change; not with each individual sample. ~~~DWHS for ens, sample in trial_dict.items(): valid = ens(sample.trajectory, candidate=self._trust_candidate) if not valid: # one sample not valid reject accepted = False probability = 0.0 break else: probability *= sample.bias rand = self._rng.random() if rand > probability: # rejected accepted = False details = { 'metropolis_acceptance': probability, 'metropolis_random': rand } if accepted: result_str = "accepted" else: result_str = ("rejected. Acceptance probabilty " + str(probability)) logger.info("Trial was " + result_str) return accepted, details @property def submovers(self): # Movers do not have submovers! return [] def _called_ensembles(self): """Function to determine which ensembles to pick samples from Returns ------- list of Ensemble the list of ensembles. Samples can then be selected using PathMover.select_sample """ # Default is that the list of ensembles is in self.ensembles return [] def get_samples_from_sample_set(self, sample_set): """ Select samples to use as input to the move core. See Also -------- move_core move Parameters ---------- sample_set : :class:`.SampleSet` current samples to use as potential input Returns ------- list of :class:`.Sample` samples to use as input to the move core """ ensembles = self._called_ensembles() samples = [self.select_sample(sample_set, ens) for ens in ensembles] return samples def move(self, sample_set): samples = self.get_samples_from_sample_set(sample_set) change = self.move_core(samples) return change def move_core(self, samples): """Core of the Monte Carlo move. Includes acceptance. See Also -------- move Parameters ---------- samples : list of :class:`.Sample` input samples from the correct ensembles of this object Returns ------- :class:`.MoveChange` result MoveChange for this move """ # this is separated out for reuse and remove dependence core MC move # dependence on the entire sample set (for parallelization) try: # pass these samples to the trial move which might throw # engine-specific exceptions if something goes wrong. # Most common should be `EngineNaNError` if nan is detected and # `EngineMaxLengthError` trials, call_details = self(*samples) except SampleNaNError as e: e.details.update({'rejection_reason': 'nan'}) return paths.RejectedNaNSampleMoveChange( samples=e.trial_sample, mover=self, input_samples=samples, details=paths.Details(**e.details) ) except SampleMaxLengthError as e: e.details.update({'rejection_reason': 'max_length'}) return paths.RejectedMaxLengthSampleMoveChange( samples=e.trial_sample, mover=self, input_samples=samples, details=paths.Details(**e.details) ) accepted, acceptance_details = self._accept(trials) # update details kwargs = {} kwargs.update(call_details) kwargs.update(acceptance_details) details = Details(**kwargs) # return change if accepted: return paths.AcceptedSampleMoveChange( samples=trials, mover=self, input_samples=samples, details=details ) else: return paths.RejectedSampleMoveChange( samples=trials, mover=self, input_samples=samples, details=details ) @abc.abstractmethod def __call__(self, *args): """Generate trial samples directly PathMovers can also be called directly with a list of samples that are then used to generate new samples. If the Mover is used as a move the move will first determine the input samples and then pass these to this function """ # Default is that the original samples are returned return args def _accept(self, trials): """Function to determine the acceptance of a trial Defaults to calling the Metropolis acceptance criterion for all returned trial samples. Means all samples most be valid and accepted. """ return self.metropolis(trials)
############################################################################### # SHOOTING GENERATORS ###############################################################################
[docs]class EngineMover(SampleMover): """Baseclass for Movers that use an engine Notes ----- A few comments for developers working with subclasses of ``EngineMover``: This class is intended to do most of the grunt work for a wide range of possible engine-based needs. Remember that your ``selector`` can select first or final points, e.g., to extend a move. In order to help you find your way through the ``EngineMover`` code, here is an overview of what various private methods do: * ``__call__``: Creates the trial. Two steps: (1) make the trajectory; (2) assemble a sample to return * ``_build_sample``: assembles the final sample * ``_make_forward_trajectory``/``_make_backward_trajectory``: creates the actual trajectory, using :class:`.PrefixTrajectoryEnsemble` or :class:`.SuffixTrajectoryEnsemble` to ensure reasonable behavior (see below for further discussion) * ``._run``: this is what is called by ``__call__``, and it in turn calls the functions to make the trajectories (depending on the nature of the mover). Frequently, this is the only thing to override (two-way shooting, shifting). """ default_engine = None reject_max_length = True # this will store the engine attribute for all subclasses as well _included_attr = ['_engine']
[docs] def __init__(self, ensemble, target_ensemble, selector, engine=None, modifier=None): super(EngineMover, self).__init__() self.selector = selector self.ensemble = ensemble self.target_ensemble = target_ensemble self._engine = engine self.modifier = modifier or paths.NoModification(as_copy=False) self._trust_candidate = True # can I safely do that?
# I think that is safe. Note for future: if we come across a bug # based on this, an alternative would be to have the move strategy # set _trust_candidate when it builds the movers; that is likely to # be a little safer (although I think we can trust all candidates # from engine movers to actually be candidates) def to_dict(self): dct = super(EngineMover, self).to_dict() dct['engine'] = self.engine return dct @property def engine(self): if self._engine is not None: return self._engine else: return self.default_engine @engine.setter def engine(self, engine): self._engine = engine def _called_ensembles(self): return [self.ensemble] def _get_in_ensembles(self): return [self.ensemble] def _get_out_ensembles(self): return [self.target_ensemble] def __call__(self, input_sample): initial_trajectory = input_sample.trajectory shooting_index = self.selector.pick(initial_trajectory) try: trial_trajectory, run_details = self._run(initial_trajectory, shooting_index) except paths.engines.EngineNaNError as e: trial, details = self._build_sample( input_sample, shooting_index, e.last_trajectory, 'nan') raise SampleNaNError('Sample with NaN', trial, details) except paths.engines.EngineMaxLengthError as e: trial, details = self._build_sample( input_sample, shooting_index, e.last_trajectory, 'max_length') if EngineMover.reject_max_length: raise SampleMaxLengthError('Sample with MaxLength', trial, details) else: trial, details = self._build_sample( input_sample, shooting_index, trial_trajectory, run_details=run_details) trials = [trial] details.update(run_details) return trials, details def _build_sample( self, input_sample, shooting_index, trial_trajectory, stopping_reason=None, run_details={} ): # TODO OPS 2.0: the passing of run_details is a hack and should be # properly refactored initial_trajectory = input_sample.trajectory if stopping_reason is None: bias = 1.0 old_snapshot = initial_trajectory[shooting_index] new_snapshot = run_details.get("modified_shooting_snapshot", old_snapshot) # Selector bias try: bias *= self.selector.probability_ratio( initial_trajectory[shooting_index], initial_trajectory, trial_trajectory, new_snapshot=new_snapshot ) except TypeError: bias *= self.selector.probability_ratio( initial_trajectory[shooting_index], initial_trajectory, trial_trajectory) NEW_SNAPSHOT_KWARG_SELECTOR.warn() # Modifier bias bias *= self.modifier.probability_ratio(old_snapshot, new_snapshot) else: bias = 0.0 # temporary test to make sure nothing went weird # old_bias = initial_point.sum_bias / trial_point.sum_bias # assert(abs(bias - old_bias) < 10e-6) # assert(initial_trajectory[shooting_index] in trial_trajectory) # we need to save the initial trial_details = { 'initial_trajectory': initial_trajectory, 'shooting_snapshot': initial_trajectory[shooting_index] } if stopping_reason is not None: trial_details['stopping_reason'] = stopping_reason trial = paths.Sample( replica=input_sample.replica, trajectory=trial_trajectory, ensemble=self.target_ensemble, parent=input_sample, mover=self, bias=bias ) return trial, trial_details def _make_forward_trajectory(self, trajectory, shooting_index): initial_snapshot = trajectory[shooting_index] # .copy() run_f = paths.PrefixTrajectoryEnsemble(self.target_ensemble, trajectory[0:shooting_index] ).can_append partial_trajectory = self.engine.generate(initial_snapshot, running=[run_f]) trial_trajectory = (trajectory[0:shooting_index] + partial_trajectory) # TODO: this should check for overshoot; only works now if ensemble # doesn't overshoot return trial_trajectory def _make_backward_trajectory(self, trajectory, shooting_index): initial_snapshot = trajectory[shooting_index].reversed # _copy() run_f = paths.SuffixTrajectoryEnsemble(self.target_ensemble, trajectory[shooting_index + 1:] ).can_prepend partial_trajectory = self.engine.generate(initial_snapshot, running=[run_f]) trial_trajectory = (partial_trajectory.reversed + trajectory[shooting_index + 1:]) # TODO: this should check for overshoot; only works now if ensemble # doesn't overshoot return trial_trajectory # direction is an abstract property to disallow instantiation # of the EngineMover unless we use a concrete subclass that sets this. # This is not super elegant but is the way to do it with abstract classes @abc.abstractproperty def direction(self): return 'unknown' def _run(self, trajectory, shooting_index): """Takes initial trajectory and shooting point; return trial trajectory""" shoot_str = "Running {sh_dir} from frame {fnum} in [0:{maxt}]" logger.info(shoot_str.format( fnum=shooting_index, maxt=len(trajectory) - 1, sh_dir=self.direction )) if self.direction == "forward": trial_trajectory = self._make_forward_trajectory( trajectory, shooting_index ) elif self.direction == "backward": trial_trajectory = self._make_backward_trajectory( trajectory, shooting_index ) else: raise RuntimeError("Unknown direction: " + str(self.direction)) return trial_trajectory, {}
[docs]class ForwardShootMover(EngineMover): """A forward shooting sample generator """
[docs] def __init__(self, ensemble, selector, engine=None): super(ForwardShootMover, self).__init__( ensemble=ensemble, target_ensemble=ensemble, selector=selector, engine=engine )
@property def direction(self): return 'forward'
[docs]class BackwardShootMover(EngineMover): """A Backward shooting generator """
[docs] def __init__(self, ensemble, selector, engine=None): super(BackwardShootMover, self).__init__( ensemble=ensemble, target_ensemble=ensemble, selector=selector, engine=engine )
@property def direction(self): return 'backward'
class ForwardExtendMover(EngineMover): """ A Sample Mover implementing Forward Extension """ _direction = "forward" def __init__(self, ensemble, target_ensemble, engine=None): super(ForwardExtendMover, self).__init__( ensemble=ensemble, target_ensemble=target_ensemble, selector=paths.FinalFrameSelector(), engine=engine ) @property def direction(self): return 'forward'
[docs]class BackwardExtendMover(EngineMover): """ A Sample Mover implementing Backward Extension """ _direction = "backward"
[docs] def __init__(self, ensemble, target_ensemble, engine=None): super(BackwardExtendMover, self).__init__( ensemble=ensemble, target_ensemble=target_ensemble, selector=paths.FirstFrameSelector(), engine=engine )
@property def direction(self): return 'backward'
############################################################################### # REPLICA EXCHANGE GENERATORS ###############################################################################
[docs]class ReplicaExchangeMover(SampleMover): """ A Sample Mover implementing a standard Replica Exchange """ _is_ensemble_change_mover = True
[docs] def __init__(self, ensemble1, ensemble2, bias=None): """ Parameters ---------- ensemble1 : openpathsampling.Ensemble one of the ensemble between to make the repex move ensemble2 : openpathsampling.Ensemble one of the ensemble between to make the repex move bias : list of float bias is not used yet """ # either replicas or ensembles must be a list of pairs; more # complicated filtering can be done with a wrapper class super(ReplicaExchangeMover, self).__init__() # TODO: add support for bias; cf EnsembleHopMover self.bias = bias self.ensemble1 = ensemble1 self.ensemble2 = ensemble2 self._trust_candidate = True initialization_logging(logger=init_log, obj=self, entries=['bias', 'ensemble1', 'ensemble2'])
def _called_ensembles(self): return [self.ensemble1, self.ensemble2] def _get_in_ensembles(self): return [self.ensemble1, self.ensemble2] def _get_out_ensembles(self): return [self.ensemble1, self.ensemble2] def _generate_in_out(self): ens1 = self.ensemble1 ens2 = self.ensemble2 move_type = 1 * InOut._use_move_type return InOutSet([ InOut([((ens1, ens2, move_type), 1), ((ens2, ens1, move_type), 1)]) ]) def __call__(self, sample1, sample2): # convert sample to the language used here before trajectory1 = sample1.trajectory trajectory2 = sample2.trajectory ensemble1 = sample1.ensemble ensemble2 = sample2.ensemble replica1 = sample1.replica replica2 = sample2.replica from1to2 = ensemble2(trajectory1, candidate=self._trust_candidate) logger.debug("trajectory " + repr(trajectory1) + " into ensemble " + repr(ensemble2) + " : " + str(from1to2)) from2to1 = ensemble1(trajectory2, candidate=self._trust_candidate) logger.debug("trajectory " + repr(trajectory2) + " into ensemble " + repr(ensemble1) + " : " + str(from2to1)) trial1 = paths.Sample( replica=replica1, trajectory=trajectory1, ensemble=ensemble2, parent=sample1, mover=self ) trial2 = paths.Sample( replica=replica2, trajectory=trajectory2, ensemble=ensemble1, parent=sample2, mover=self ) return [trial1, trial2], {}
[docs]class StateSwapMover(SampleMover):
[docs] def __init__(self, ensemble1, ensemble2, bias=None): """ A move to swap states for state changing samples This does a replica exchange with prededing PathReversal and will only succeed if initial and final state are different Parameters ---------- ensemble1 : openpathsampling.Ensemble one of the ensemble between to make the swap move ensemble2 : openpathsampling.Ensemble one of the ensemble between to make the swap move bias : list of float bias is not used yet Notes ----- So, if ensemble1 goes from A to B, then ensemble2 must go from B to A. """ # either replicas or ensembles must be a list of pairs; more # complicated filtering can be done with a wrapper class super(StateSwapMover, self).__init__() self.bias = bias self.ensemble1 = ensemble1 self.ensemble2 = ensemble2 initialization_logging(logger=init_log, obj=self, entries=['bias', 'ensemble1', 'ensemble2'])
def _called_ensembles(self): return [self.ensemble1, self.ensemble2] def _get_in_ensembles(self): return [self.ensemble1, self.ensemble2] def _get_out_ensembles(self): return [self.ensemble1, self.ensemble2] def _generate_in_out(self): ens1 = self.ensemble1 ens2 = self.ensemble2 move_type = -1 * InOut._use_move_type return InOutSet([ InOut([((ens1, ens2, move_type), 1), ((ens2, ens1, move_type), 1)]) ]) def __call__(self, sample1, sample2): # convert sample to the language used here before # it is almost a RepEx move but the two trajectories are reversed trajectory1 = sample1.trajectory.reversed trajectory2 = sample2.trajectory.reversed ensemble1 = sample1.ensemble ensemble2 = sample2.ensemble replica1 = sample1.replica replica2 = sample2.replica from1to2 = ensemble2(trajectory1) logger.debug("trajectory " + repr(trajectory1) + " into ensemble " + repr(ensemble2) + " : " + str(from1to2)) from2to1 = ensemble1(trajectory2) logger.debug("trajectory " + repr(trajectory2) + " into ensemble " + repr(ensemble1) + " : " + str(from2to1)) trial1 = paths.Sample( replica=replica1, trajectory=trajectory1, ensemble=ensemble2, parent=sample1, mover=self ) trial2 = paths.Sample( replica=replica2, trajectory=trajectory2, ensemble=ensemble1, parent=sample2, mover=self ) return [trial1, trial2], {}
############################################################################### # SUBTRAJECTORY GENERATORS ###############################################################################
[docs]class SubtrajectorySelectMover(SampleMover): """ Picks a subtrajectory satisfying the given subensemble. If there are no subtrajectories which satisfy the subensemble, this returns the zero-length trajectory. Attributes ---------- ensemble : openpathsampling.Ensemble the set of allows samples to chose from sub_ensemble : openpathsampling.Ensemble the subensemble to be searched for n_l : int or None the number of subtrajectories that need to be found. If `None` every number of subtrajectories > 0 is okay. Otherwise the move is only accepted if exactly n_l subtrajectories are found. """ _is_ensemble_change_mover = True
[docs] def __init__(self, ensemble, sub_ensemble, n_l=None): super(SubtrajectorySelectMover, self).__init__( ) self.n_l = n_l self.ensemble = ensemble self.sub_ensemble = sub_ensemble
def _called_ensembles(self): return [self.ensemble] def _get_in_ensembles(self): return [self.ensemble] def _get_out_ensembles(self): return [self.sub_ensemble] @abc.abstractmethod def _choose(self, trajectory_list): return [], {} def __call__(self, trial): initial_trajectory = trial.trajectory replica = trial.replica logger.debug( "Working with replica " + str(replica) + " (" + str(initial_trajectory) + ")") subtrajs = self.sub_ensemble.split(initial_trajectory) logger.debug("Found " + str(len(subtrajs)) + " subtrajectories.") if (self.n_l is None and len(subtrajs) > 0) or \ (self.n_l is not None and len(subtrajs) == self.n_l): subtraj, selection_details = self._choose(subtrajs) bias = 1.0 trial = paths.Sample( replica=replica, trajectory=subtraj, ensemble=self.sub_ensemble, parent=trial, mover=self, bias=bias ) trials = [trial] else: trials = [] selection_details = {} return trials, selection_details
[docs]class RandomSubtrajectorySelectMover(SubtrajectorySelectMover): """ Samples a random subtrajectory satisfying the given subensemble. If there are no subtrajectories which satisfy the subensemble, this returns the zero-length trajectory. Attributes ---------- ensemble : openpathsampling.Ensemble the set of allows samples to chose from sub_ensemble : openpathsampling.Ensemble the subensemble to be searched for n_l : int or None the number of subtrajectories that need to be found. If `None` every number of subtrajectories > 0 is okay. Otherwise the move is only accepted if exactly n_l subtrajectories are found. """ def _choose(self, trajectory_list): # Needed to prevent ragged nested sequence DeprWarning in np 1.20 idx = self._rng.choice(len(trajectory_list)) return trajectory_list[idx], {}
[docs]class FirstSubtrajectorySelectMover(SubtrajectorySelectMover): """ Samples the first subtrajectory satifying the given subensemble. If there are no subtrajectories which satisfy the ensemble, this returns the zero-length trajectory. Attributes ---------- ensemble : openpathsampling.Ensemble the set of allows samples to chose from sub_ensemble : openpathsampling.Ensemble the subensemble to be searched for n_l : int or None the number of subtrajectories that need to be found. If `None` every number of subtrajectories > 0 is okay. Otherwise the move is only accepted if exactly n_l subtrajectories are found. """ def _choose(self, trajectory_list): return trajectory_list[0], {}
[docs]class FinalSubtrajectorySelectMover(SubtrajectorySelectMover): """ Samples the final subtrajectory satifying the given subensemble. If there are no subtrajectories which satisfy the ensemble, this returns the zero-length trajectory. Attributes ---------- ensemble : openpathsampling.Ensemble the set of allows samples to chose from sub_ensemble : openpathsampling.Ensemble the subensemble to be searched for n_l : int or None the number of subtrajectories that need to be found. If `None` every number of subtrajectories > 0 is okay. Otherwise the move is only accepted if exactly n_l subtrajectories are found. """ def _choose(self, trajectory_list): return trajectory_list[-1], {}
############################################################################### # REVERSAL GENERATOR ###############################################################################
[docs]class PathReversalMover(SampleMover):
[docs] def __init__(self, ensemble): """ Parameters ---------- ensemble : openpathsampling.Ensemble the specific ensemble to be reversed in """ super(PathReversalMover, self).__init__() self.ensemble = ensemble self._trust_candidate = True
def _called_ensembles(self): return [self.ensemble] def _get_in_ensembles(self): return [self.ensemble] def _generate_in_out(self): return InOutSet([ InOut([ ((self.ensemble, self.ensemble, -1 * InOut._use_move_type), 1) ]) ]) def __call__(self, trial): trajectory = trial.trajectory ensemble = trial.ensemble replica = trial.replica reversed_trajectory = trajectory.reversed valid = ensemble(reversed_trajectory, candidate=self._trust_candidate) logger.info("PathReversal move accepted: " + str(valid)) bias = 1.0 trial = paths.Sample( replica=replica, trajectory=reversed_trajectory, ensemble=ensemble, mover=self, parent=trial, bias=bias ) return [trial], {}
[docs]class EnsembleHopMover(SampleMover): _is_ensemble_change_mover = True
[docs] def __init__( self, ensemble, target_ensemble, change_replica=None, bias=None): """ A Mover that allows the change between ensembles. Parameters ---------- ensemble : openpathsampling.Ensemble the initial ensemble to be jumped from target_ensemble : openpathsampling.Ensemble the final ensemble to be jumped to change_replica : int of None if None the replica id of the chosen sample will not be changed. Otherwise the replica id will be set to change_replica. This is useful when hoping to ensembles to create a new replica. bias : float, dict or None (default) gives the bias of accepting (not proposing) a hop. A float will be the acceptance for all possible attempts. If a dict is given, then it contains a list of ensembles and a matrix. None means no bias Notes ----- The bias dict has the following form : .. code-block:: python { 'ensembles' : [ens_1, ens_2, ens_n], 'values' : np.array((n,n)) } The numpy array contains all the acceptance probabilties. If possible a HopMover should (as all movers) be used for only a specific hop and not multiple ones. """ super(EnsembleHopMover, self).__init__() # ensembles -- another version might take a value for each ensemble, # and use the ratio; this latter is better for CITIS self.ensemble = ensemble self.target_ensemble = target_ensemble self.bias = bias self.change_replica = change_replica initialization_logging( logger=init_log, obj=self, entries=['bias'] )
def _called_ensembles(self): return [self.ensemble] @property def submovers(self): return [] def _get_in_ensembles(self): return [self.ensemble] def _get_out_ensembles(self): return [self.target_ensemble] def __call__(self, rep_sample): ens_from = self.ensemble ens_to = self.target_ensemble logger.debug("Selected sample: " + repr(rep_sample)) replica = rep_sample.replica if self.change_replica is not None: replica = self.change_replica logger.info( "Attempting ensemble hop from {e1} to {e2} " "replica ID {rid}".format( e1=repr(ens_from), e2=repr(ens_to), rid=repr(replica))) trajectory = rep_sample.trajectory logger.debug(" selected replica: " + str(replica)) logger.debug(" initial ensemble: " + repr(rep_sample.ensemble)) logger.info("Hop starts from legal ensemble: " + str(ens_from(trajectory))) logger.info("Hop ends in legal ensemble: " + str(ens_to(trajectory))) # TODO: remove this and generalize!!! if type(self.bias) is float: bias = self.bias logger.info("Using fixed bias " + str(bias)) elif type(self.bias) is dict: # special dict ens = self.bias['ensembles'] e1 = ens.index(ens_from) e2 = ens.index(ens_to) bias = float(self.bias['values'][e1, e2]) logger.info("Using dict bias " + str(bias)) else: bias = 1.0 logger.info("Using default bias: self.bias == " + str(self.bias)) trial = paths.Sample( replica=replica, trajectory=trajectory, ensemble=ens_to, mover=self, parent=rep_sample, bias=bias ) details = { 'initial_ensemble': ens_from, 'trial_ensemble': ens_to, 'bias': bias } return [trial], details
# **************************************************************************** # SELECTION MOVERS # ****************************************************************************
[docs]class SelectionMover(PathMover): """ A general mover that selects a single mover from a set of possibilities This is a basic class for all sorts of selectors, like RandomChoice, RandomAllowedChoice. The way it works is to generate a list of weights and pick a random one using the weights. This is as general as possible and is chosen because it also allows to store the possibilities in a general way for better comparison Attributes ---------- movers : list of openpathsampling.PathMover the PathMovers to choose from """
[docs] def __init__(self, movers): super(SelectionMover, self).__init__() self.movers = movers initialization_logging(init_log, self, entries=['movers'])
@property def submovers(self): return self.movers @property def is_ensemble_change_mover(self): if self._is_ensemble_change_mover is not None: return self._is_ensemble_change_mover sub_change = False for mover in self.movers: if mover.is_ensemble_change_mover: sub_change = True break return sub_change def _generate_in_out(self): return InOutSet(set.union(*[sub.in_out for sub in self.submovers])) def _get_in_ensembles(self): return [sub.input_ensembles for sub in self.submovers] def _get_out_ensembles(self): return [sub.output_ensembles for sub in self.submovers] @abc.abstractmethod def _selector(self, sample_set): pass def select_mover(self, weights): p = np.array(weights) p /= sum(weights) logger.debug(self.name + " " + str(weights)) idx = self._rng.choice(len(self.movers), p=p) logger_str = "{name} ({cls}) selecting {mtype} (index {idx})" logger.info(logger_str.format( name=self.name, cls=self.__class__.__name__, idx=idx, mtype=self.movers[idx].name )) mover = self.movers[idx] kwargs = { 'choice': idx, 'chosen_mover': mover, 'probability': weights[idx] / sum(weights), 'weights': weights } details = Details(**kwargs) return mover, details def move(self, sample_set): weights = self._selector(sample_set) mover, details = self.select_mover(weights) subchange = mover.move(sample_set) path = paths.RandomChoiceMoveChange( subchange=subchange, mover=self, details=details ) return path
[docs]class RandomChoiceMover(SelectionMover): """ Chooses a random mover from its movers list, and runs that move. Returns the number of samples the submove return. For example, this would be used to select a specific replica exchange such that each replica exchange is its own move, and which swap is selected at random. Attributes ---------- movers : list of PathMover the PathMovers to choose from weights : list of floats the relative weight of each PathMover (does not need to be normalized) """
[docs] def __init__(self, movers, weights=None): super(RandomChoiceMover, self).__init__(movers) if weights is None: weights = [1.0] * len(movers) self.movers = movers self.weights = weights initialization_logging(init_log, self, entries=['weights'])
def _selector(self, sample_set): return self.weights
class RandomAllowedChoiceMover(RandomChoiceMover): """ Chooses a random mover from its movers which have existing samples. This is different from random choice moves in that this mover only picks from sub movers that actually can succeed because they have samples in all required input_ensembles Attributes ---------- movers : list of PathMover the PathMovers to choose from weights : list of floats the relative weight of each PathMover (does not need to be normalized) """ def _selector(self, sample_set): if self.weights is None: weights = [1.0] * len(self.movers) else: weights = list(self.weights) # make a copy # this is implemented by setting all weights locally to zero that # correspond to movers that will potentially fail since the required # input ensembles are not present in the sample_set present_ensembles = sample_set.ensembles for idx, mover in enumerate(self.movers): for ens in mover.input_ensembles: if ens not in present_ensembles: # ens might be required but is not present weights[idx] = 0.0 return weights
[docs]class FirstAllowedMover(SelectionMover): """ Chooses a first mover that has samples in all required ensembles. A mover can only safely be run, if all inputs can be satisfied. This will pick the first mover from the list where all ensembles from input_ensembles are found. Attributes ---------- movers : list of PathMover the PathMovers to choose from weights : list of floats the relative weight of each PathMover (does not need to be normalized) """ def _selector(self, sample_set): weights = [1.0] * len(self.movers) present_ensembles = sample_set.ensembles found = False for idx, mover in enumerate(self.movers): if not found: for ens in mover.input_ensembles: if ens not in present_ensembles: # ens might be required but is not present weights[idx] = 0.0 if weights[idx] > 0.0: found = True else: weights[idx] = 0.0 return weights
[docs]class LastAllowedMover(SelectionMover): """ Chooses the last mover that has samples in all required ensembles. A mover can only safely be run, if all inputs can be satisfied. This will pick the last mover from the list where all ensembles from input_ensembles are found. Attributes ---------- movers : list of PathMover the PathMovers to choose from weights : list of floats the relative weight of each PathMover (does not need to be normalized) """ def _selector(self, sample_set): weights = [1.0] * len(self.movers) present_ensembles = sample_set.ensembles found = False for idx, mover in reversed(list(enumerate(self.movers))): if not found: for ens in mover.input_ensembles: if ens not in present_ensembles: # ens might be required but is not present weights[idx] = 0.0 if weights[idx] > 0.0: found = True else: weights[idx] = 0.0 return weights
class ConditionalMover(PathMover): """ An if-then-else structure for PathMovers. Returns a SequentialMoveChange of the if_move movepath and the then_move movepath (if if_move is accepted) or the else_move movepath (if if_move is rejected). """ def __init__(self, if_mover, then_mover, else_mover): """ Parameters ---------- if_mover : openpathsampling.PathMover then_mover : openpathsampling.PathMover else_mover : openpathsampling.PathMover """ super(ConditionalMover, self).__init__() self.if_mover = if_mover self.then_mover = then_mover self.else_mover = else_mover initialization_logging(init_log, self, ['if_mover', 'then_mover', 'else_mover']) def _generate_in_out(self): return InOutSet({ self.if_mover.in_out + self.then_mover.in_out } | { self.if_mover.in_out + self.else_mover.in_out }) def sub_replica_state(self, replica_states): if_replica_states = self.if_mover.in_out.move(replica_states) return [ if_replica_states, self.then_mover.in_out.move(if_replica_states), self.else_mover.in_out.move(if_replica_states), ] @property def submovers(self): return [self.if_mover, self.then_mover, self.else_mover] def _get_in_ensembles(self): return [sub.input_ensembles for sub in self.submovers] def _get_out_ensembles(self): return [sub.output_ensembles for sub in self.submovers] def move(self, sample_set): subglobal = sample_set ifclause = self.if_mover.move(subglobal) samples = ifclause.results subglobal = subglobal.apply_samples(samples) if ifclause.accepted: if self.then_mover is not None: resultclause = self.then_mover.move(subglobal) else: resultclause = paths.EmptyMoveChange() else: if self.else_mover is not None: resultclause = self.else_mover.move(subglobal) else: resultclause = paths.EmptyMoveChange() return paths.SequentialMoveChange([ifclause, resultclause], mover=self)
[docs]class SequentialMover(PathMover): """ Performs each of the moves in its movers list. Returns all samples generated, in the order of the mover list. For example, this would be used to create a move that does a sequence of replica exchanges in a given order, regardless of whether the moves succeed or fail. """
[docs] def __init__(self, movers): """ Parameters ---------- movers : list of openpathsampling.PathMover the list of pathmovers to be run in sequence """ super(SequentialMover, self).__init__() self.movers = movers initialization_logging(init_log, self, ['movers'])
@property def submovers(self): return self.movers @property def is_ensemble_change_mover(self): if self._is_ensemble_change_mover is not None: return self._is_ensemble_change_mover sub_change = False for mover in self.movers: if mover.is_ensemble_change_mover: sub_change = True break return sub_change def _generate_in_out(self): total = self.submovers[0].in_out for pp in range(1, len(self.submovers)): new_pos = self.submovers[pp].in_out total = InOutSet(set.union(*[ total, InOutSet( sum([total, new_pos], InOutSet()) ), new_pos ])) return total def sub_replica_state(self, replica_states): ret = list() ret.append(replica_states) for sub in self.submovers[:-1]: replica_states = sub.in_out.move(replica_states) ret.append(replica_states) return ret def _get_in_ensembles(self): return [sub.input_ensembles for sub in self.submovers] def _get_out_ensembles(self): return [sub.output_ensembles for sub in self.submovers] def move(self, sample_set): logger.debug("Starting sequential move") subglobal = sample_set movechanges = [] for mover in self.movers: logger.debug("Starting sequential move step " + str(mover)) # Run the sub mover movepath = mover.move(subglobal) samples = movepath.results subglobal = subglobal.apply_samples(samples) movechanges.append(movepath) return paths.SequentialMoveChange(movechanges, mover=self)
[docs]class PartialAcceptanceSequentialMover(SequentialMover): """ Performs each move in its movers list until complete or until one is not accepted. If any move is not accepted, further moves are not attempted, but the previous accepted samples remain accepted. For example, this would be used to create a bootstrap promotion move, which starts with a shooting move, followed by an EnsembleHop/Replica promotion ConditionalSequentialMover. Even if the EnsembleHop fails, the accepted shooting move should be accepted. """ def _generate_in_out(self): total = self.submovers[0].in_out for pp in range(1, len(self.submovers)): new_pos = self.submovers[pp].in_out total = InOutSet(set.union(*[ total, InOutSet( sum([total, new_pos], InOutSet()) ) ])) return total def move(self, sample_set): logger.debug("==== BEGINNING " + self.name + " ====") subglobal = paths.SampleSet(sample_set) movechanges = [] for mover in self.movers: logger.info(str(self.name) + " starting mover index " + str(self.movers.index(mover)) + " (" + mover.name + ")" ) # Run the sub mover movepath = mover.move(subglobal) samples = movepath.results subglobal = subglobal.apply_samples(samples) movechanges.append(movepath) if not movepath.accepted: break logger.debug("==== FINISHING " + self.name + " ====") return paths.PartialAcceptanceSequentialMoveChange( movechanges, mover=self)
[docs]class ConditionalSequentialMover(SequentialMover): """ Performs each move in its movers list until complete or until one is not accepted. If any move in not accepted, all previous samples are updated to have set their acceptance to False. For example, this would be used to create a minus move, which consists of first a replica exchange and then a shooting (extension) move. If the replica exchange fails, the move is aborted before doing the dynamics. ConditionalSequentialMover only works if there is a *single* active sample per replica. """ def _generate_in_out(self): return InOutSet( sum([sub.in_out for sub in self.submovers], InOutSet()) ) def move(self, sample_set): logger.debug("Starting conditional sequential move") subglobal = sample_set movechanges = [] for mover in self.movers: logger.debug("Starting sequential move step " + str(mover)) # Run the sub mover movepath = mover.move(subglobal) samples = movepath.results subglobal = subglobal.apply_samples(samples) movechanges.append(movepath) if not movepath.accepted: break return paths.ConditionalSequentialMoveChange( movechanges, mover=self)
class NonCanonicalConditionalSequentialMover(ConditionalSequentialMover): """ Special mover for reactive flux and S-shooting simulation. This mover inherits from :class:`.ConditionalSequentialMover` and alters only the `move` method to return the output of the corresponding :class:`.NonCanonicalConditionalSequentialMoveChange`. """ _is_canonical = False def move(self, sample_set): change = super(NonCanonicalConditionalSequentialMover, self).move(sample_set) return paths.NonCanonicalConditionalSequentialMoveChange( subchanges=change.subchanges, mover=change.mover, details=change.details ) # class ReplicaIDChangeMover(PathMover): # """ # Changes the replica ID for a path. # """ # # def __init__(self, replica_pair): # super(ReplicaIDChangeMover, self).__init__() # self.replica_pair = replica_pair # initialization_logging(logger=init_log, obj=self, # entries=['replica_pairs']) # # def move(self, sample_set): # rep_from = self.replica_pair[0] # rep_to = self.replica_pair[1] # rep_sample = self.select_sample(sample_set, # replicas=rep_from) # # logger.info( # "Creating new sample from replica ID " + str(rep_from) + # " and putting it in replica ID " + str(rep_to)) # # # note: currently this clones into a new replica ID. We might later # # want to kill the old replica ID (and possibly rename this mover). # # new_sample = paths.Sample( # replica=rep_to, # ensemble=rep_sample.ensemble, # trajectory=rep_sample.trajectory, # parent=rep_sample, # mover=self # ) # # # Can be used to remove the old sample. Not used yet! # # kill_sample = paths.Sample( # # replica=rep_from, # # trajectory=None, # # ensemble=rep_sample.ensemble, # # parent=None, # # mover=self # # ) # # kwargs = { # 'rep_from': rep_from, # 'rep_to': rep_to # } # # details = Details(**kwargs) # # return paths.AcceptedSampleMoveChange( # samples=[new_sample], # mover=self, # input_samples=[rep_sample], # details=details # )
[docs]class SubPathMover(PathMover): """Mover that delegates to a single submover """
[docs] def __init__(self, mover): """ Parameters ---------- mover : :class:`openpathsampling.PathMover` the submover to be delegated to """ super(SubPathMover, self).__init__() self.mover = mover
@property def submovers(self): return [self.mover] @property def is_ensemble_change_mover(self): return self.mover.is_ensemble_change_mover def _get_in_ensembles(self): return self.mover.input_ensembles def _get_out_ensembles(self): return self.mover.output_ensembles def _generate_in_out(self): return self.mover.in_out def sub_replica_state(self, replica_states): return [replica_states] def move(self, sample_set): subchange = self.mover.move(sample_set) change = paths.SubMoveChange( subchange=subchange, mover=self ) return change
[docs]class EnsembleFilterMover(SubPathMover): """Mover that return only samples from specified ensembles """
[docs] def __init__(self, mover, ensembles): """ Parameters ---------- mover : :class:`openpathsampling.PathMover` the submover to be delegated to ensembles : nested list of :class:`openpathsampling.Ensemble` or None the ensemble specification """ super(EnsembleFilterMover, self).__init__(mover) self.ensembles = ensembles if not set(self.mover.output_ensembles) & set(self.ensembles): # little sanity check, if the underlying move will be removed by # the filter throw a warning raise ValueError( 'Your filter removes the underlying move completely. ' + 'Please check your ensembles and submovers!')
def move(self, sample_set): # TODO: This will only pass filtered samples. We might split # this into an separate input and output filter if only one # side is needed filtered_globalstate = paths.SampleSet([ samp for samp in sample_set if samp.ensemble in self.ensembles ]) subchange = self.mover.move(filtered_globalstate) change = paths.FilterByEnsembleMoveChange( subchange=subchange, mover=self ) return change def _get_in_ensembles(self): # only filter the output, not the input # return self.mover.input_ensembles return self.ensembles def _get_out_ensembles(self): return self.ensembles def _generate_in_out(self): return self.mover.in_out.filter(self.ensembles) def sub_replica_state(self, replica_states): return [{rs.filter(self.ensembles) for rs in replica_states}]
class SpecializedRandomChoiceMover(RandomChoiceMover): """ Superclass for movers that are random choice between two SampleMovers This requires that all submovers accept the same list of samples. """ @classmethod def from_dict(cls, dct): mover = cls.__new__(cls) super(cls, mover).__init__(movers=dct['movers']) return mover def to_dict(self): dct = super(SpecializedRandomChoiceMover, self).to_dict() dct['movers'] = self.movers return dct def move_core(self, samples): weights = self.weights mover, details = self.select_mover(weights) subchange = mover.move_core(samples) change = paths.RandomChoiceMoveChange( subchange=subchange, mover=self, details=details ) return change
[docs]class OneWayShootingMover(SpecializedRandomChoiceMover): """ One-way (stochastic) shooting mover OneWayShootingMover is a special case of a RandomChoiceMover which combines gives a 50/50 chance of selecting either a ForwardShootMover or a BackwardShootMover. Both submovers use the same shooting point selector, and both apply to the same ensembles and replicas. Attributes ---------- selector : :class:`openpathsampling.ShootingPointSelector` The shooting point selection scheme ensemble : :class:`openpathsampling.Ensemble` Ensemble for this shooting mover """
[docs] def __init__(self, ensemble, selector, engine=None): movers = [ ForwardShootMover(ensemble=ensemble, selector=selector, engine=engine), BackwardShootMover(ensemble=ensemble, selector=selector, engine=engine) ] super(OneWayShootingMover, self).__init__(movers=movers)
@classmethod def from_dict(cls, dct): mover = cls.__new__(cls) # override with stored movers and use the init of the super class # this assumes that the super class has movers as its signature super(cls, mover).__init__(movers=dct['movers']) return mover @property def ensemble(self): return self.movers[0].ensemble @property def selector(self): return self.movers[0].selector @property def engine(self): return self.movers[0].engine
[docs]class OneWayExtendMover(SpecializedRandomChoiceMover): """ OneWayShootingMover is a special case of a RandomChoiceMover which gives a 50/50 chance of selecting either a ForwardExtendMover or a BackwardExtendMover. Both submovers use the same same ensembles and replicas. Attributes ---------- ensemble : :class:`openpathsampling.Ensemble` valid ensemble """
[docs] def __init__(self, ensemble, target_ensemble, engine=None): movers = [ ForwardExtendMover(ensemble=ensemble, target_ensemble=target_ensemble, engine=engine), BackwardExtendMover(ensemble=ensemble, target_ensemble=target_ensemble, engine=engine) ] super(OneWayExtendMover, self).__init__(movers=movers)
@property def engine(self): return self.movers[0].engine
class AbstractTwoWayShootingMover(EngineMover): def __init__(self, ensemble, selector, modifier, engine=None): super(AbstractTwoWayShootingMover, self).__init__( ensemble=ensemble, target_ensemble=ensemble, selector=selector, engine=engine, modifier=modifier ) # TODO OPS 2.0: This init signature should be aligned with EngineMover # required for concrete class; not really used @property def direction(self): # pragma: no cover return 'bidrectional' def _make_forward_trajectory(self, trajectory, initial_snapshot, shooting_index): fwd_ens = paths.PrefixTrajectoryEnsemble( self.target_ensemble, trajectory[0:shooting_index] ) fwd_partial = self.engine.generate(initial_snapshot, running=[fwd_ens.can_append]) return fwd_partial def _make_backward_trajectory(self, trajectory, initial_snapshot, shooting_index): # run backward bkwd_ens = paths.SuffixTrajectoryEnsemble( self.target_ensemble, trajectory[shooting_index + 1:] ) bkwd_partial = self.engine.generate(initial_snapshot.reversed, running=[bkwd_ens.can_prepend]) return bkwd_partial def _run(self, trajectory, shooting_index): # to override the default implementation in EngineMover raise NotImplementedError class ForwardFirstTwoWayShootingMover(AbstractTwoWayShootingMover): def _run(self, trajectory, shooting_index): """ The actual shooting process (after shooting point is chosen). Parameters ---------- trajectory : :class:`.Trajectory` input trajectory shooting_index : int index of the shooting point within `trajectory` Returns ------- trial_trajectory : :class:`.Trajectory` the resulting trial trajectory details : dict details dictionary (includes modified shooting point and modification bias) """ shoot_str = "Running {sh_dir} from frame {fnum} in [0:{maxt}]" logger.info(shoot_str.format( fnum=shooting_index, maxt=len(trajectory) - 1, sh_dir="Forward-first" )) original = trajectory[shooting_index] # TODO OPS 2.0: Modification+bias should be done in engine mover modified = self.modifier(original) fwd_partial = self._make_forward_trajectory(trajectory, modified, shooting_index) # TODO: come up with a test that shows why you need mid_traj here; # should be a SeqEns with OptionalEnsembles. Exact example is hard! mid_traj = trajectory[0:shooting_index] + fwd_partial bkwd_partial = self._make_backward_trajectory(mid_traj, modified, shooting_index) # join the two trial_trajectory = bkwd_partial.reversed + fwd_partial[1:] details = {'modified_shooting_snapshot': modified} return trial_trajectory, details class BackwardFirstTwoWayShootingMover(AbstractTwoWayShootingMover): def _run(self, trajectory, shooting_index): """ The actual shooting process (after shooting point is chosen). Parameters ---------- trajectory : :class:`.Trajectory` input trajectory shooting_index : int index of the shooting point within `trajectory` Returns ------- trial_trajectory : :class:`.Trajectory` the resulting trial trajectory details : dict details dictionary (includes modified shooting point and modification bias) """ shoot_str = "Running {sh_dir} from frame {fnum} in [0:{maxt}]" logger.info(shoot_str.format( fnum=shooting_index, maxt=len(trajectory) - 1, sh_dir="Backward-first" )) original = trajectory[shooting_index] # TODO OPS 2.0: Modification+bias should be done in engine mover modified = self.modifier(original) bkwd_partial = self._make_backward_trajectory(trajectory, modified, shooting_index) # logger.info("Complete backward shot (length " + # str(len(bkwd_partial)) + ")") # TODO: come up with a test that shows why you need mid_traj here; # should be a SeqEns with OptionalEnsembles. Exact example is hard! mid_traj = bkwd_partial.reversed + trajectory[shooting_index + 1:] mid_traj_shoot_idx = len(bkwd_partial) - 1 fwd_partial = self._make_forward_trajectory(mid_traj, modified, mid_traj_shoot_idx) # logger.info("Complete forward shot (length " + # str(len(fwd_partial)) + ")") # join the two trial_trajectory = bkwd_partial.reversed + fwd_partial[1:] details = {'modified_shooting_snapshot': modified} return trial_trajectory, details
[docs]class TwoWayShootingMover(SpecializedRandomChoiceMover):
[docs] def __init__(self, ensemble, selector, modifier, engine=None): movers = [ ForwardFirstTwoWayShootingMover( ensemble=ensemble, selector=selector, modifier=modifier, engine=engine ), BackwardFirstTwoWayShootingMover( ensemble=ensemble, selector=selector, modifier=modifier, engine=engine ) ] super(TwoWayShootingMover, self).__init__(movers=movers)
@property def ensemble(self): return self.movers[0].ensemble @property def selector(self): return self.movers[0].selector @property def modifier(self): return self.movers[0].modifier
[docs]class MinusMover(SubPathMover): """ Instance of a MinusMover. The minus move combines a replica exchange with path extension to swap paths between the innermost regular TIS interface ensemble and the minus interface ensemble. This is particularly useful for improving sampling of path space. """ _is_canonical = True
[docs] def __init__(self, minus_ensemble, innermost_ensembles, engine=None): try: innermost_ensembles = list(innermost_ensembles) except TypeError: innermost_ensembles = [innermost_ensembles] segment = minus_ensemble._segment_ensemble sub_trajectory_selector = RandomChoiceMover([ FirstSubtrajectorySelectMover( ensemble=minus_ensemble, sub_ensemble=segment, n_l=minus_ensemble.n_l ), FinalSubtrajectorySelectMover( ensemble=minus_ensemble, sub_ensemble=segment, n_l=minus_ensemble.n_l ), ]) sub_trajectory_selector.named("MinusSubtrajectoryChooser") repexs = [ReplicaExchangeMover( ensemble1=segment, ensemble2=inner ) for inner in innermost_ensembles] repex_chooser = RandomChoiceMover(repexs) repex_chooser.named("InterfaceSetChooser") extension_mover = RandomChoiceMover([ ForwardExtendMover( ensemble=segment, target_ensemble=minus_ensemble, engine=engine ), BackwardExtendMover( ensemble=segment, target_ensemble=minus_ensemble, engine=engine ) ]) extension_mover.named("MinusExtensionDirectionChooser") self.engine = extension_mover.movers[0].engine if self.engine is not extension_mover.movers[1].engine: raise RuntimeWarning("Forward and backward engines differ?!?!") mover = \ EnsembleFilterMover( ConditionalSequentialMover([ sub_trajectory_selector, repex_chooser, extension_mover ]), ensembles=[minus_ensemble] + innermost_ensembles ) self.minus_ensemble = minus_ensemble self.innermost_ensembles = innermost_ensembles initialization_logging(init_log, self, ['minus_ensemble', 'innermost_ensembles']) super(MinusMover, self).__init__(mover)
def move(self, sample_set): change = super(MinusMover, self).move(sample_set) cond_seq_changes = change.subchanges[0].subchanges[0].subchanges seg_swap = None if len(cond_seq_changes) >= 2: seg_swap = cond_seq_changes[1].subchanges[0].trials ext_traj = None if len(cond_seq_changes) >= 3: ext_traj = cond_seq_changes[2].subchanges[0].trials[0].trajectory details = Details(segment_swap_samples=seg_swap, extension_trajectory=ext_traj) if change.details is None: change.details = details return change
[docs]class SingleReplicaMinusMover(MinusMover): """ Minus mover for single replica TIS. In SRTIS, the minus mover doesn't actually keep an active sample in the minus interface. Instead, it just puts the newly generated segment into the innermost ensemble. """
[docs] def __init__(self, minus_ensemble, innermost_ensembles, bias=None, engine=None): try: innermost_ensembles = list(innermost_ensembles) except TypeError: innermost_ensembles = [innermost_ensembles] if bias is None: bias = "" # TODO temp for storage until real bias self.bias = bias self.minus_ensemble = minus_ensemble self.innermost_ensembles = innermost_ensembles # TODO: Until we have automated detailed balance calculations, I # think this will only be valid in the case of only one innermost # ensemble. But I think you only want to use it in the case of only # one innermost ensemble anyway. The following warns us: if len(innermost_ensembles) > 1: logger.warning( "Probably shouldn't use SingleReplicaMinusMover with MISTIS") segment = minus_ensemble._segment_ensemble hop_innermost_to_segment = RandomAllowedChoiceMover([ EnsembleHopMover(innermost, segment, bias=bias) for innermost in innermost_ensembles ]) # TODO: again, works for single interface set, but there has to be a # smarter way to do this in the MISTIS case hop_segment_to_innermost = RandomChoiceMover([ EnsembleHopMover(segment, innermost, bias=bias) for innermost in innermost_ensembles ]) forward_minus = ConditionalSequentialMover([ hop_innermost_to_segment, ForwardExtendMover(segment, minus_ensemble, engine=engine), FinalSubtrajectorySelectMover(minus_ensemble, segment), hop_segment_to_innermost ]) backward_minus = ConditionalSequentialMover([ hop_innermost_to_segment, BackwardExtendMover(segment, minus_ensemble, engine=engine), FirstSubtrajectorySelectMover(minus_ensemble, segment), hop_segment_to_innermost ]) mover = EnsembleFilterMover(RandomChoiceMover([backward_minus, forward_minus]), ensembles=innermost_ensembles) # we skip MinusMover's init and go to the grandparent super(MinusMover, self).__init__(mover)
def move(self, sample_set): # skip the MinusMover's implementation return super(MinusMover, self).move(sample_set)
class PathSimulatorMover(SubPathMover): """ This just wraps a mover and references the used pathsimulator """ def __init__(self, mover, pathsimulator): super(PathSimulatorMover, self).__init__(mover) self.pathsimulator = pathsimulator def move(self, sample_set, step=-1): details = Details( step=step ) return paths.PathSimulatorMoveChange( self.mover.move(sample_set), mover=self, details=details ) def PathReversalSet(ensembles): return list(map(PathReversalMover, ensembles))
[docs]class Details(StorableObject): """Details of an object. Can contain any data """
[docs] def __init__(self, **kwargs): super(Details, self).__init__() for key, value in kwargs.items(): setattr(self, key, value)
_print_repr_types = [paths.Ensemble] _print_nothing_keys = ["__uuid__"] def __str__(self): # primarily for debugging/interactive use mystr = "" for key in self.__dict__.keys(): obj = self.__dict__[key] if key in self._print_nothing_keys: pass # do nothing! elif any([isinstance(obj, tt) for tt in self._print_repr_types]): mystr += str(key) + " = " + repr(obj) + '\n' else: mystr += str(key) + " = " + str(self.__dict__[key]) + '\n' return mystr
[docs]@has_deprecations @deprecate(MOVE_DETAILS) class MoveDetails(Details): """Details of the move as applied to a given replica Specific move types may have add several other attributes for each MoveDetails object. For example, shooting moves will also include information about the shooting point selection, etc. """
[docs] def __init__(self, **kwargs): super(MoveDetails, self).__init__(**kwargs)
# leave this for potential backwards compatibility
[docs]@has_deprecations @deprecate(SAMPLE_DETAILS) class SampleDetails(Details): """Details of a sample """
[docs] def __init__(self, **kwargs): super(SampleDetails, self).__init__(**kwargs)