Source code for openpathsampling.engines.external_snapshots.snapshot

import time

from openpathsampling.engines import features
from openpathsampling.engines.snapshot import BaseSnapshot
from . import features as ext_features

import logging
logger = logging.getLogger(__name__)


@features.base.attach_features([
    features.engine,
    features.coordinates,
    features.velocities,
    features.box_vectors,
    ext_features.file_info
])
class InternalizedMDSnapshot(BaseSnapshot):
    """
    Internalized version of standard external MD snapshot.

    This can be used, for example, to store intial conditions in an OPS
    storage file.
    """
    pass

[docs]@features.base.attach_features([ features.engine, ext_features.coordinates, ext_features.velocities, ext_features.box_vectors, ext_features.file_info ]) class ExternalMDSnapshot(BaseSnapshot): """ Snapshot for external MD engines Internally, this only stores the file_name and the file_position. All specific details (positions, velocities, box vectors) are loaded from file when requested. Parameters ---------- file_name : str the name of the external file where the positions/velocities/etc. reside file_position : int position within the file; the engine should be able to load data for this specific snapshot based on this number engine : :class:`.DynamicsEngine` the engine associated with this snapshot """
[docs] def __init__(self, file_name=None, file_position=None, engine=None): # these are done in place of calling super self._reversed = None self.__uuid__ = self.get_uuid() # these are the requried attributes self.file_name = file_name self.file_position = file_position self.engine = engine self.velocity_direction = 1 # by default; reversed flips it # these are containers for temporary data self._xyz = None self._velocities = None self._box_vectors = None self._internalized = None
def load_details(self): """Cache coords, velocities, box vectors from the external file""" try: (xyz, vel, box) = self.engine.read_frame_data( self.file_name, self.file_position ) except IndexError: # Out of bounds on buffer access (axis 0) logger.debug("Exception reading from %s[%d]", self.file_name, self.file_position) time.sleep(self.engine.sleep_ms/10000.0) # 1/10 the normal self.load_details() except RecursionError: raise RuntimeError("Unrecoverable error in load_details") else: self._xyz = xyz self._velocities = vel self._box_vectors = box def set_details(self, xyz, velocities, box_vectors): """Set coords, velocities, and box vectors. This is mainly used if OPS must modify/create a snapshot. Parameters ---------- xyz : np.array unitless coordinates velocities : np.array velocities box_vectors : np.array unit cell for the periodic box """ try: self.load_details() except: self._xyz = xyz self._velocities = velocities self._box_vectors = box_vectors else: raise RuntimeError("Can't set details if frame already exists.") def clear_cache(self): """Remove internal details from snapshot. These details should always be accessible later using :meth:`.load_details`. Removing them allows them memory to be freed. """ self._xyz = None self._velocities = None self._box_vectors = None def __repr__(self): num_str = "file_name=" + str(self.file_name) pos_str = "file_position=" + str(self.file_position) eng_str = "engine=" + repr(self.engine) args = ", ".join([num_str, pos_str, eng_str]) return "{cls_str}(".format(cls_str=self.cls) + args + ")" def internalize(self): """Return a version of this snapshot with storable details. This allows these snapshots to be stored internally in OPS storage files, instead of only in external files. This is convenient to avoid the need to transfer files to remote computers. """ if self._internalized is None: self._internalized = self.engine.InternalizedSnapshotClass( coordinates=self.coordinates, velocities=self.velocities, box_vectors=self.box_vectors, file_name=self.file_name, file_position=self.file_position, engine=self.engine.internalized_engine ) return self._internalized