Source code for ccpn.core.lib.DataStore


Code to:
    - implement path redirections $DATA, $ALONGSIDE, $INSIDE
    - implement a DataStore object to handle Spectrum file paths properly
    - thereby wraps the silly dataStore and dataUrl data structures

    - core.lib.util.expandDollarFilePath
    - cor.lib.util._fetchDataUrl

# Licence, Reference and Credits
__copyright__ = "Copyright (C) CCPN project ( 2014 - 2022"
__credits__ = ("Ed Brooksbank, Joanna Fox, Victoria A Higman, Luca Mureddu, Eliza Płoskoń",
               "Timothy J Ragan, Brian O Smith, Gary S Thompson & Geerten W Vuister")
__licence__ = ("CCPN licence. See")
__reference__ = ("Skinner, S.P., Fogh, R.H., Boucher, W., Ragan, T.J., Mureddu, L.G., & Vuister, G.W.",
                 "CcpNmr AnalysisAssign: a flexible platform for integrated NMR analysis",
                 "J.Biomol.Nmr (2016), 66, 111-124,")
# Last code modification
__modifiedBy__ = "$modifiedBy: Ed Brooksbank $"
__dateModified__ = "$dateModified: 2022-03-29 10:55:51 +0100 (Tue, March 29, 2022) $"
__version__ = "$Revision: 3.1.0 $"
# Created
__author__ = "$Author: gvuister $"
__date__ = "$Date: 2017-04-07 10:28:48 +0000 (Fri, April 07, 2017) $"
# Start of code

import os

from ccpn.util.Path import aPath, Path
from ccpn.framework import constants
from ccpn.util.traits.CcpNmrJson import CcpNmrJson
from ccpn.util.traits.CcpNmrTraits import Unicode, Any, CPath, Bool, Dict, CString

from ccpn.util.decorators import singleton

from ccpn.framework.Application import getApplication
from ccpn.util.Logging import getLogger

# Redirections

[docs]class RedirectionABC(CcpNmrJson): """ Base class for maintaining a single redirection """ apiName = None # to be subclassed identifier = None # to be subclassed expand = False # expand to handle None, zero-length and '.' _application = Any(default_value=None, allow_none=True) _path = CPath(default_value=None, allow_none=True) def __init__(self): super().__init__() self._application = getApplication() @property def path(self): if self.expand: return self.expandPath() else: return self._path
[docs] def expandPath(self): """expand path to handle None, zero-length and '.'""" if self._path is None or len(self._path) == 0: self._path = Path.home() elif self._path == '.': self._path = Path.cwd() return self._path
def __str__(self): return '<%s: %s>' % (self.__class__.__name__, self.identifier) __repr__ = __str__
[docs]@singleton class DataRedirection(RedirectionABC): identifier = '$DATA' apiName = 'remoteData' expand = True @property def path(self): if self._path is None: self._path = aPath(self._application.preferences.general.dataPath) return super().path @path.setter def path(self, path): self._path = aPath(path) self._application.preferences.general.dataPath = str(self._path)
[docs]@singleton class InsideRedirection(RedirectionABC): identifier = '$INSIDE' apiName = 'insideData' expand = False @property def path(self): self._path = aPath(self._application.project.path) return super().path
[docs]@singleton class AlongsideRedirection(RedirectionABC): identifier = '$ALONGSIDE' apiName = 'alongsideData' expand = False @property def path(self): self._path = aPath(self._application.project.path).parent return super().path
[docs]@singleton class PathRedirections(list): """ Class to maintain the path redirections $DATA, $ALONGSIDE, $INSIDE """ def __init__(self): super().__init__() self.append(DataRedirection()) self.append(AlongsideRedirection()) self.append(InsideRedirection()) @property def dataPath(self): return self[0].path @dataPath.setter def dataPath(self, path): self[0].path = path @property def alongsidePath(self): return self[1].path @property def insidePath(self): return self[2].path
[docs] def getPaths(self): """Return a list of (identifier, path) tuples""" # paths = [(r.identifier, r.path) for r in self] paths = [(r.identifier, getattr(r, 'path', None)) for r in self] return paths
[docs] def getApiMappings(self): """Return a list with (apiName, indentifier) tuples""" pairs = [(r.apiName, r.identifier) for r in self] return pairs
def _getPathFromApiStore(self, storeName): """Return path associated with api storeName; for backward compatibility purposes """ apiNames = [r.apiName for r in self] if storeName not in apiNames: raise ValueError('_getPathFromApiStore: invalid storeName "%s"' % storeName) project = getApplication().project if project is None: raise RuntimeError('_getPathFromApiStore: undefined project') dataUrl = project._apiNmrProject.root.findFirstDataLocationStore( name='standard').findFirstDataUrl(name=storeName) path = dataUrl.url.dataLocation return path
#========================================================================================= # DataStores #=========================================================================================
[docs]class DataStore(CcpNmrJson): """ This class wraps the implementation of $DATA, $ALONGSIDE, $INSIDE redirections """ # For old Spectrum instances, its parses the api insideData, remoteData, alongSideData etc. # api dataStores # # Once linked to a Spectrum, it stores the path and other relevant info as json-encoded string # in the spectrum instance internal parameter storage classVersion = 1.0 _path = CPath(allow_none=True, default_value=None).tag(saveToJson=True) dataFormat = CString(allow_none=True, default_value=None).tag(saveToJson=True) useBuffer = Bool(default_value=False).tag(saveToJson=True, info="""Flag to indicate if spectrum should be opened in buffered mode""" ) spectrum = Any(allow_none=True, default_value=None).tag(saveToJson=False) autoVersioning = Bool(default_value=True).tag(saveToJson=False) autoRedirect = Bool(default_value=False).tag(saveToJson=False) # api dataStore apiDataStore = Any(allow_none=True, default_value=None).tag(saveToJson=False) apiDataStoreName = Unicode(allow_none=True, default_value=None).tag(saveToJson=True) apiDataStoreDir = Unicode(allow_none=True, default_value=None).tag(saveToJson=True) apiDataStorePath = Unicode(allow_none=True, default_value=None).tag(saveToJson=True) # Dict with redirection paths; for reference pathRedirections = Dict(default_value={}).tag(saveToJson=True) def __init__(self, spectrum=None, autoRedirect=False, autoVersioning=False): """ autoRedirect: optionally try to redefine path into $DATA, $ALONGSIDE, $INSIDE redirections autoVersioning: optionally add a versioning identifier (e.g. for new spectra) """ super().__init__() self.spectrum = spectrum self.autoRedirect = autoRedirect self.autoVersioning = autoVersioning self._getPathRedirections() @property def path(self): """Return a Path representation of self, optionally encoded with $DATA, $ALONGSIDE, $INSIDE redirections """ if self._path is None: return Path(constants.UNDEFINED_STRING) return Path(self._path) @path.setter def path(self, value): """Set path to value; optionally auto versioning or redirecting None makes it undefined """ self._path = value while self._path is not None and self.autoVersioning and self.exists(): self._path = self.path.incrementVersion().asString() if self._path is not None and self.autoRedirect: self._path = self.redirectPath(self._path) if self.spectrum is not None: self._saveInternal()
[docs] @classmethod def newFromPath(cls, path, autoRedirect=False, autoVersioning=False, appendToName=None, withSuffix=None, dataFormat=None): """Create and return a new instance from path; optionally append to name and set suffix """ _p = Path(path) suffix = _p.suffix if withSuffix is None else withSuffix if appendToName is not None: _p = _p.parent / _p.basename + appendToName if len(suffix) > 0: _p = _p.withSuffix(suffix) instance = cls(autoRedirect=autoRedirect, autoVersioning=autoVersioning) instance.dataFormat = dataFormat instance.path = _p return instance
[docs] def hasPathDefined(self): """Return True if path has been defined """ return self._path is not None
[docs] def expandPath(self, path=None): """return path decoded for $DATA, $ALONGSIDE, $INSIDE redirections returns Path instance """ if path is None: _path = Path(self._path) else: _path = Path(path) for d, p in self._getPathRedirections(): if _path.startswith(d): _path = p / Path._from_parts([1:]) # Using undocumented private method! break return _path
[docs] def redirectPath(self, path=None): """Redefine path into $DATA, $ALONGSIDE, $INSIDE redirections return Path instance """ if path is None: _path = Path(self._path) else: _path = Path(path) # check in reverse order, prioritising $INSIDE, then $ALONGSIDE, then $DATA for d, p in self._getPathRedirections()[::-1]: if str(path).startswith(str(p)): _path = Path(d) / _path.relative_to(p) break return _path
[docs] def aPath(self): """Return aPath instance of self, decoded for $DATA, $ALONGSIDE, $INSIDE redirections """ if self._path is None: return aPath(constants.UNDEFINED_STRING) # expand any $DATA, $INSIDE $ALONGSIDE _path = self.expandPath(self._path) return aPath(_path)
[docs] def exists(self): """Return True if self.aPath() (i.e. expanded) exists """ if self._path is None: return False else: return self.aPath().exists()
[docs] def warningMessage(self): """Error message displayed on logger """ getLogger().warning(self._message())
[docs] def errorMessage(self): """Error message displayed on logger """ getLogger().error(self._message())
#========================================================================================= # Implementation #========================================================================================= def _importFromSpectrum(self, spectrum): """Restore state from spectrum, either from internal parameter store or from api-dataStores Returns self """ if spectrum is None: raise ValueError('Invalid spectrum "%s"' % spectrum) if spectrum._hasInternalParameter(spectrum._DATASTORE_KEY): self.spectrum = spectrum self._restoreInternal() else: self._restoreFromApiSpectrum(spectrum._wrappedData) self.spectrum = spectrum self._saveInternal() return self def _saveInternal(self): """Save into spectrum internal parameter store CCPNINTERNAL: e.g. _newSpectrum """ if self.spectrum is None: raise RuntimeError('%s._saveInternal: spectrum not defined' % self.__class__.__name__) jsonData = self.toJson() self.spectrum._setInternalParameter(self.spectrum._DATASTORE_KEY, jsonData) def _restoreInternal(self): """Restore from spectrum internal parameter store CCPNINTERNAL: e.g. _newSpectrum """ if self.spectrum is None: raise RuntimeError('%s._restoreInternal: spectrum not defined' % self.__class__.__name__) jsonData = self.spectrum._getInternalParameter(self.spectrum._DATASTORE_KEY) if jsonData is None or len(jsonData) == 0: raise RuntimeError('DataStore._restoreInternal: json data appear to be corrupted') self.fromJson(jsonData) def _restoreFromApiSpectrum(self, apiSpectrum): """This routine implements the extraction from the old storage mechanism (using api DataStores). Returns self """ from ccpn.core.lib.SpectrumDataSources.SpectrumDataSourceABC import SpectrumDataSourceABC from ccpn.core.lib.SpectrumDataSources.EmptySpectrumDataSource import EmptySpectrumDataSource if apiSpectrum is None: raise ValueError('Invalid spectrum "%s"' % apiSpectrum) self.apiDataStore = apiSpectrum.dataStore if self.apiDataStore is not None: self.apiDataStoreName = self.apiDataStoreDir = self.apiDataStore.dataUrl.url.path self.apiDataStorePath = self.apiDataStore.path # Encode the different dataStores pDict = dict(PathRedirections().getApiMappings()) if self.apiDataStoreName in pDict: self._path = os.path.join(pDict[self.apiDataStoreName], self.apiDataStorePath) else: self._path = os.path.join(self.apiDataStoreDir, self.apiDataStorePath) # Convert the (potentially old) dataFormat specifier _tmp = self.apiDataStore.fileType if (_dataFormat := SpectrumDataSourceABC._dataFormatDict.get(_tmp)) is None: getLogger().warning(f'Restore dataFormat from older definitions: dataFormat "{_tmp}": not recognised') self.dataFormat = _dataFormat else: # This happens for dummy spectra self._path = None self.dataFormat = EmptySpectrumDataSource.dataFormat return self def _getPathRedirections(self): """Get the redirection paths; return list of items of the pathRedirection dict """ redirections = PathRedirections().getPaths() # Convert and store the redirections for future reference self.pathRedirections = dict( [(r, str(p)) for r, p in redirections] ) return redirections def _message(self): """return message to be displayed on logger """ text = 'path "%s" is invalid ' % self.path for d, p in self._getPathRedirections(): if self.path.startswith(d): if p.exists(): text += ' (check %s)' % d else: text += ' (%s "%s" does not exist)' % (d, p) break return text def __eq__(self, other): return self._path == other._path def __ne__(self, other): return not self.__eq__(other) def __str__(self): return '<%s: %s (dataFormat=%r, useBuffer=%s)>' % \ (self.__class__.__name__, self.path, self.dataFormat, self.useBuffer) __repr__ = __str__
DataStore.register() from ccpn.util.traits.CcpNmrTraits import Instance from ccpn.util.traits.TraitJsonHandlerBase import CcpNmrJsonClassHandlerABC
[docs]class DataStoreTrait(Instance): """Specific trait for a Datastore instance encoding the path and dataFormat of the (binary) spectrum data. None indicates no spectrum data file path has been defined """ klass = DataStore def __init__(self, **kwds): Instance.__init__(self, klass=self.klass, allow_none=True, **kwds)
[docs] class jsonHandler(CcpNmrJsonClassHandlerABC): # klass = klass pass