__author__ = 'Jan-Hendrik Prinz'
import logging
import openpathsampling as paths
from openpathsampling.netcdfplus import StorableObject, lazy_loading_attributes
from openpathsampling.netcdfplus import DelayedLoader
from .treelogic import TreeMixin
logger = logging.getLogger(__name__)
# @lazy_loading_attributes('details')
[docs]
class MoveChange(TreeMixin, StorableObject):
'''
A class that described the concrete realization of a PathMove.
Attributes
----------
mover : PathMover
The mover that generated this MoveChange
samples : list of Sample
A list of newly generated samples by this particular move.
Only used by node movers like RepEx or Shooters
subchanges : list of MoveChanges
the MoveChanges created by submovers
details : Details
an object that contains MoveType specific attributes and information.
E.g. for a RandomChoiceMover which Mover was selected.
'''
details = DelayedLoader()
[docs]
def __init__(self, subchanges=None, samples=None, mover=None,
details=None, input_samples=None):
StorableObject.__init__(self)
self._lazy = {}
self._len = None
self._collapsed = None
self._results = None
self._trials = None
self._accepted = None
self.mover = mover
if subchanges is None:
self.subchanges = []
else:
self.subchanges = subchanges
if samples is None:
self.samples = []
else:
self.samples = samples
if input_samples is None:
self.input_samples = []
else:
self.input_samples = input_samples
self.details = details
def __getattr__(self, item):
# try to get attributes from details dict
try:
return getattr(self.details, item)
except AttributeError as e:
msg = "{0} not found in change's details".format(str(item))
if not e.args:
e.args = [msg]
else:
e.args = tuple([e.args[0] + "; " + msg] + list(e.args[1:]))
raise
def to_dict(self):
return {
'mover': self.mover,
'details': self.details,
'samples': self.samples,
'input_samples': self.input_samples,
'subchanges': self.subchanges,
'cls': self.__class__.__name__
}
# hook for TreeMixin
@property
def _subnodes(self):
return self.subchanges
@property
def submovers(self):
return [ch.mover for ch in self.subchanges]
@property
def subchange(self):
"""
Return the single/only sub-movechange if there is only one.
Returns
-------
MoveChange
"""
if len(self.subchanges) == 1:
return self.subchanges[0]
else:
# TODO: might raise exception
return None
@staticmethod
def _default_match(original, test):
if isinstance(test, paths.MoveChange):
return original is test
elif isinstance(test, paths.PathMover):
return original.mover is test
elif issubclass(test, paths.PathMover):
return original.mover.__class__ is test
else:
return False
def movetree(self):
"""
Return a tree with the movers of each node
Notes
-----
This is equivalent to
`tree.map_tree(lambda x : x.mover)`
"""
return self.map_tree(lambda x: x.mover)
@property
def identifier(self):
return self.mover
@property
def collapsed_samples(self):
"""
Return a collapsed set of samples with non used samples removed
This is the minimum required set of samples to keep the `MoveChange`
correct and allow to target sampleset to be correctly created.
These are the samples used by `.closed`
Examples
--------
Assume that you run 3 shooting moves for replica #1. Then only the
last of the three really matters for the target sample_set since #1
will be replaced by #2 which will be replaced by #3. So this function
will return only the last sample.
"""
if self._collapsed is None:
s = paths.SampleSet([]).apply_samples(self.results)
# keep order just for being thorough
self._collapsed = [
samp for samp in self.results
if samp in s
]
return self._collapsed
@property
def accepted(self):
"""
Returns if this particular move was accepted.
Mainly used for rejected samples.
Notes
-----
Acceptance is determined from the number of resulting samples. If at
least one sample is returned then this move will change the sampleset
and is considered an accepted change.
"""
if self._accepted is None:
self._accepted = len(self.results) > 0
return self._accepted
def __add__(self, other):
"""
This allows to use `+` to create SequentialPMCs
Notes
-----
You can also use this to apply several changes
>>> new_sset = old_sset + change1 + change2
>>> new_sset = old_sset + (change1 + change2)
"""
if isinstance(other, MoveChange):
return SequentialMoveChange([self, other])
else:
raise ValueError('Only MoveChanges can be combined')
@property
def results(self):
"""
Returns a list of all samples that are accepted in this move
This contains unnecessary, but accepted samples, too.
Returns
-------
list of Samples
the list of samples that should be applied to the SampleSet
"""
if self._results is None:
self._results = self._get_results()
return self._results
def _get_results(self):
"""
Determines all relevant accepted samples for this move
Includes all accepted samples also from subchanges
Returns
-------
list of Sample
the list of accepted samples for this move
"""
return []
@property
def trials(self):
"""
Returns a list of all samples generated during the PathMove.
This includes all accepted and rejected samples (which does NOT
include hidden samples yet)
"""
if self._trials is None:
self._trials = self._get_trials()
return self._trials
def _get_trials(self):
"""
Determines all samples for this move
Includes all samples also from subchanges
Returns
-------
list of Sample
the list of all samples generated for this move
Notes
-----
This function needs to be implemented for custom changes
"""
return []
def __str__(self):
if self.accepted:
return 'SampleMove : %s : %s : %d samples' % (self.mover.cls, self.accepted, len(self.trials)) + ' ' + str(self.trials) + ''
else:
return 'SampleMove : %s : %s :[]' % (self.mover.cls, self.accepted)
@property
def canonical(self):
"""
Return the first non single-subchange
Notes
-----
Usually a mover that returns a single subchange is for deciding what to
do rather than describing what is actually happening. This property
returns the first mover that is not one of these delegating movers and
contains information of what has been done in this move.
What you are usually interested in is `.canonical.mover` to get the
relevant mover.
Examples
--------
>>> a = OnewayShootingMover()
>>> change = a.move(sset)
>>> change.canonical.mover # returns either Forward or Backward
"""
pmc = self
while pmc.subchange is not None:
if pmc.mover.is_canonical is True:
return pmc
pmc = pmc.subchange
return pmc
@property
def description(self):
"""
Return a compact representation of the change
"""
subs = self.subchanges
if len(subs) == 0:
return str(self.mover)
elif len(subs) == 1:
return subs[0].description
else:
return ':'.join([sub.description for sub in subs])
class EmptyMoveChange(MoveChange):
"""
A MoveChange representing no changes
"""
def __init__(self, mover=None, details=None):
super(EmptyMoveChange, self).__init__(mover=mover, details=details)
def __str__(self):
return ''
def _get_trials(self):
return []
def _get_results(self):
return []
class SampleMoveChange(MoveChange):
"""
A MoveChange representing the application of samples.
This is the most common MoveChange and all other moves use this
as leaves and on the lowest level consist only of `SampleMoveChange`
"""
def __init__(self, samples, mover=None, details=None, input_samples=None):
"""
Parameters
----------
samples : list of Samples
a list of trial samples that are used in this change
mover : PathMover
the generating PathMover
details : Details
a details object containing specifics about the change
Attributes
----------
samples
mover
details
"""
super(SampleMoveChange, self).__init__(
mover=mover,
details=details,
input_samples=input_samples
)
if samples.__class__ is paths.Sample:
samples = [samples]
self.samples = samples
def _get_results(self):
return []
def _get_trials(self):
return self.samples
class AcceptedSampleMoveChange(SampleMoveChange):
"""
Represents an accepted SamplePMC
This will return the trial samples also as its result, hence it is
accepted.
"""
def _get_trials(self):
return self.samples
def _get_results(self):
return self.samples
class RejectedSampleMoveChange(SampleMoveChange):
"""
Represents an rejected SamplePMC
This will return no samples also as its result, hence it is
rejected.
"""
class RejectedNaNSampleMoveChange(RejectedSampleMoveChange):
"""
Represents an rejected SamplePMC because of occurance of NaN
This will return no samples as its result, hence it is rejected.
"""
pass
class RejectedMaxLengthSampleMoveChange(RejectedSampleMoveChange):
"""
Represents an rejected SamplePMC because of hitting the max length limit
This will return no samples as its result, hence it is rejected.
"""
pass
class SequentialMoveChange(MoveChange):
"""
SequentialMoveChange has no own samples, only inferred Sampled from the
underlying MovePaths
"""
def __init__(self, subchanges, mover=None, details=None):
"""
Parameters
----------
subchanges : list of MoveChanges
a list of MoveChanges to be applied in sequence
mover
details
Attributes
----------
subchanges
mover
details
"""
super(SequentialMoveChange, self).__init__(mover=mover, details=details)
self.subchanges = subchanges
def _get_results(self):
samples = []
for subchange in self.subchanges:
samples = samples + subchange.results
return samples
def _get_trials(self):
samples = []
for subchange in self.subchanges:
samples = samples + subchange.trials
return samples
def __str__(self):
return 'SequentialMove : %s : %d samples\n' % \
(self.accepted, len(self.results)) + \
MoveChange._indent('\n'.join(map(str, self.subchanges)))
class PartialAcceptanceSequentialMoveChange(SequentialMoveChange):
"""
PartialAcceptanceSequentialMovePath has no own samples, only inferred
Sampled from the underlying MovePaths
"""
def _get_results(self):
changes = []
for subchange in self.subchanges:
if subchange.accepted:
changes.extend(subchange.results)
else:
break
return changes
def __str__(self):
return 'PartialAcceptanceMove : %s : %d samples\n' % \
(self.accepted, len(self.results)) + \
MoveChange._indent('\n'.join(map(str, self.subchanges)))
class ConditionalSequentialMoveChange(SequentialMoveChange):
"""
ConditionalSequentialMovePath has no own samples, only inferred Samples
from the underlying MovePaths
"""
def _get_results(self):
changes = []
for subchange in self.subchanges:
if subchange.accepted:
changes.extend(subchange.results)
else:
return []
return changes
def __str__(self):
return 'ConditionalSequentialMove : %s : %d samples\n' % \
(self.accepted, len(self.results)) + \
MoveChange._indent('\n'.join(map(str, self.subchanges)))
class NonCanonicalConditionalSequentialMoveChange(
ConditionalSequentialMoveChange):
""" Special move change for reactive flux and S-shooting simulation.
This move change inherits from :class:`.ConditionalSequentialMoveChange`
and returns the outcome of the last subchange.
"""
@property
def canonical(self):
return self.subchanges[-1]
class SubMoveChange(MoveChange):
"""
A helper MoveChange that represents the application of a submover.
The raw implementation delegates all to the subchange
"""
def __init__(self, subchange, mover=None, details=None):
"""
Parameters
----------
subchange : MoveChange
the actual subchange used by this wrapper PMC
mover
details
Attributes
----------
subchange
mover
details
"""
super(SubMoveChange, self).__init__(mover=mover, details=details)
self.subchanges = [subchange]
def _get_results(self):
return self.subchange.results
def _get_trials(self):
return self.subchange.trials
def __str__(self):
# Defaults to use the name of the used mover
return self.mover.__class__.__name__[:-5] + ' :\n' + MoveChange._indent(str(self.subchange))
class RandomChoiceMoveChange(SubMoveChange):
"""
A MoveChange that represents the application of a mover chosen randomly
"""
# This class is empty since all of the decision is specified by the mover
# and it requires no additional logic to decide if it is accepted.
class FilterByEnsembleMoveChange(SubMoveChange):
"""
A MoveChange that filters out all samples not in specified ensembles
"""
# TODO: Question: filter out also trials not in the ensembles? I think so,
# because we are only interested in trials that could be relevant, right?
def _get_results(self):
all_samples = self.subchange.results
filtered_samples = list(filter(
lambda s: s.ensemble in self.mover.ensembles,
all_samples
))
return filtered_samples
def _get_trials(self):
all_samples = self.subchange.trials
filtered_samples = list(filter(
lambda s: s.ensemble in self.mover.ensembles,
all_samples
))
return filtered_samples
def __str__(self):
return 'FilterMove : allow only ensembles [%s] from sub moves : %s : %d samples\n' % \
(str(self.mover.ensembles), self.accepted, len(self.results)) + \
MoveChange._indent(str(self.subchange))
class FilterSamplesMoveChange(SubMoveChange):
"""
A MoveChange that keeps a selection of the underlying samples
"""
def _get_results(self):
sample_set = self.subchange.results
# allow for negative indices to be picked, e.g. -1 is the last sample
samples = [idx % len(sample_set) for idx in self.mover.selected_samples]
return samples
def __str__(self):
return 'FilterMove : pick samples [%s] from sub moves : %s : %d samples\n' % \
(str(self.mover.selected_samples), self.accepted, len(self.results)) + \
MoveChange._indent(str(self.subchange))
class KeepLastSampleMoveChange(SubMoveChange):
"""
A MoveChange that only keeps the last generated sample.
This is different from using `.reduced` which will only change the
level of detail that is stored. This MoveChange will actually remove
potential relevant samples and thus affect the outcome of the new
SampleSet. To really remove samples also from storage you can use
this MoveChange in combination with `.closed` or `.reduced`
Notes
-----
Does the same as `FilterSamplesMoveChange(subchange, [-1], False)`
I think we should try to not use this. It would be better to make submoves
and finally filter by relevant ensembles. Much like running a function
with local variables/local ensembles.
"""
def _get_results(self):
samples = self.subchange.results
if len(samples) > 1:
samples = [samples[-1]]
return samples
def __str__(self):
return 'Restrict to last sample : %s : %d samples\n' % \
(self.accepted, len(self.results)) + \
MoveChange._indent(str(self.subchange))
class PathSimulatorMoveChange(SubMoveChange):
"""
A MoveChange that just wraps a subchange and references a PathSimulator
"""
def __str__(self):
return 'PathSimulatorStep : %s : Step # %d with %d samples\n' % \
(str(self.mover.pathsimulator.cls), self.details.step, len(self.results)) + \
MoveChange._indent(str(self.subchange))