Source code for openpathsampling.netcdfplus.stores.dict

from .named import NamedObjectStore

from future.utils import iterkeys

import logging

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

[docs]class DictStore(NamedObjectStore):
[docs] def __init__(self): super(DictStore, self).__init__( None, json='json' )
def to_dict(self): return {} def load(self, idx): """ Returns an object from the storage. Parameters ---------- idx : str a string (name) of the objects. This will always return the last object found with the specified name. If immutable is true for the store it assures that there is only a single object per name Returns ------- :class:`openpathsampling.netcdfplus.base.StorableObject` the loaded object """ if type(idx) is str: n_idx = -1 # we want to load by name and it was not in cache. if idx not in self.name_idx: logger.debug('Name "%s" not found in the storage!' % idx) raise KeyError('str "' + idx + '" not found in storage') if idx in self.name_idx: if len(self.name_idx[idx]) > 1: logger.debug(( 'Found name "%s" multiple (%d) times in storage! ' 'Loading last!') % ( idx, len(self.name_idx[idx]))) n_idx = sorted(list(self.name_idx[idx]))[-1] logger.debug( 'Found name "%s" in storage! Loading IDX %d!' % (idx, n_idx)) elif type(idx) is int: n_idx = idx else: raise ValueError( 'Unsupported index type (only str and int allowed).') # turn into python int if it was a numpy int (in some rare cases!) n_idx = int(n_idx) logger.debug( 'Calling load object of type `%s` and @ IDX #%d' % (str(self.content_class), n_idx)) if n_idx >= len(self): logger.warning( 'Trying to load from IDX #' + str(n_idx) + ' > number of objects ' + str(len(self)) ) raise RuntimeError( 'Loading of too large int should be attempted. ' 'Problem in name cache. This should never happen!') elif n_idx < 0: logger.warning( 'Trying to load negative IDX #' + str(n_idx) + ' < 0. ' 'This should never happen!!!' ) raise RuntimeError( 'Loading of negative int should result in no object. ' 'This should never happen!' ) else: logger.debug('Loading named object from index IDX # %d' % n_idx) obj = self._load(n_idx) logger.debug( 'Loading named object from index IDX # %d.. DONE' % n_idx) return obj def restore(self): self.update_name_cache() def save(self, obj, idx=None): """ Saves an object to the storage. Parameters ---------- obj : :py:class:`openpathsampling.netcdfplus.base.StorableObject` the object to be stored idx : string or `None` the string index to be used for storing. Objects will not be replaced but stored again with the same name. When loading the last stored object under the idx is retrieved. Effectively mimicking a mutual dict with versioning. We usually encourage for most cases to use the immutual dict class :class:`openpathsampling.netcdfplus.ImmutableDictStore` instead to avoid ambiguity in stored objects. See Also -------- :class:`openpathsampling.netcdfplus.ImmutableDictStore` """ if idx is None: # a DictStore needs a specific name raise ValueError( 'Saving in a DictStore without specifying a ' 'string key is not allowed. ') if type(idx) is not str: # key needs to be a string raise ValueError( 'Index `%s` for DictStore needs to be a string! ' % idx) n_idx = int( # make sure in nested saving that an IDX is not used twice! # self.reserve_idx(n_idx) logger.debug( 'Saving `%s` with name `%s` @ IDX #%d' % (str(obj.__class__), idx, n_idx)) self._save(obj, n_idx)[self.prefix + '_name'][n_idx] = idx self._update_name_in_cache(idx, n_idx) return n_idx def keys(self): return self.name_idx.keys() # def iterkeys(self): # return self.name_idx.keys() def __iter__(self): return iter(iterkeys(self.name_idx)) def iteritems(self): for name in self: yield name, self[name] def get(self, idx, default=None): try: return self.load(idx) except KeyError: return default
[docs]class ImmutableDictStore(DictStore): def save(self, obj, idx=None): """ Saves an object to the storage. Parameters ---------- obj : :py:class:`openpathsampling.netcdfplus.base.StorableObject` the object to be stored idx : int or string or `None` the index to be used for storing. This is highly discouraged since it changes an immutable object (at least in the storage). It is better to store also the new object and just ignore the previously stored one. """ if idx in self.name_idx: # immutable means no duplicates, so quit raise RuntimeWarning( 'Cannot re-save existing key "%s" in ' 'immutable dict store.' % idx ) return super(ImmutableDictStore, self).save(obj, idx)