"""
A class vor managing Structure calculation Run's
IN_PROGRESS
"""
#=========================================================================================
# Licence, Reference and Credits
#=========================================================================================
__copyright__ = "Copyright (C) CCPN project (https://www.ccpn.ac.uk) 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 http://www.ccpn.ac.uk/v3-software/downloads/license",
)
__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, http://doi.org/10.1007/s10858-016-0060-y"
)
#=========================================================================================
# Last code modification
#=========================================================================================
__modifiedBy__ = "$modifiedBy: Geerten Vuister $"
__dateModified__ = "$dateModified: 2022-03-11 15:07:06 +0000 (Fri, March 11, 2022) $"
__version__ = "$Revision: 3.1.0 $"
#=========================================================================================
# Created
#=========================================================================================
__author__ = "$Author: geertenv $"
__date__ = "$Date: 2020-02-10 10:28:41 +0000 (Thu, February 10, 2022) $"
#=========================================================================================
# Start of code
#=========================================================================================
import sys
from ccpn.util.traits.CcpNmrJson import Constants, update, CcpNmrJson
from ccpn.util.traits.CcpNmrTraits import \
Unicode, Dict, List, V3ObjectList, V3Object, Bool, CPath, Int
from ccpn.util.Logging import getLogger
from ccpn.util.Path import aPath, Path
from ccpn.util.Time import timeStamp
from ccpn.framework.Preferences import getPreferences, \
XPLOR_NIH_PATH, TALOS_PATH, CYANA_PATH, ARIA_PATH
[docs]class RunManagerABC(CcpNmrJson):
"""
Class that maintains structure calculation and save/restore functionality
"""
_RUN_TYPE = None
_JSON_FILE = 'runManager.json'
_INPUT_NEF_PATH = 'input.nef'
_OUTPUT_NEF_PATH = 'out.nef'
_SCRIPT_PATH = 'script.sh'
classVersion = 3.1
saveAllTraitsToJson = True
WARNING = Unicode('This is a runManager file; it will be automatically overwritten')
runType = Unicode().tag(
info='The type of structure calculation run'
)
runName = Unicode(allow_none=False, default_value='run').tag(
info='The name of the structure calculation run'
)
runId = Int(allow_none=False, default_value=1).tag(
info='The id of the structure calculation run'
)
useTimeStamp = Bool(default_value=True).tag(info='flag to indicate if a timestamp should be used in generating the run directory'
)
_timeStamp = Unicode(allow_none=True, default_value=None).tag(
info='The timestamp of the structure calculation run'
)
# data-related attributes
_project = V3Object(allow_none=True).tag(
info='The Project instance; for reference'
)
peakLists = V3ObjectList(default_value=[]).tag(
info='The list of peakList objects used in the structure calculation run'
)
chain = V3Object().tag(
info='The chain object used in the structure calculation run'
)
chemicalShiftList = V3Object().tag(
info='The chemicalShiftList used in the structure calculation run'
)
# Executables, not saved to json; settings taken from preferences
# "preferences-get", e.g. XPLOR_NIH_PATH; should be subclassed
_EXECUTABLE1 = None
_executable1 = CPath(allow_none=True, default_value=None).tag(
info='The path of executable-1; set from preferences if defined',
saveToJson=False
)
# "preferences-get", e.g. XPLOR_NIH_PATH; should be subclassed
_EXECUTABLE2 = None
_executable2 = CPath(allow_none=True, default_value=None).tag(
info='The path of executable-2; set from preferences if defined',
saveToJson=False
)
# Can be subclassed
_ALLOW_PARALLEL = True
# setting up the script, paths
useParallel = Bool(default_value=False).tag(info='flag to indicate if parallel mode should be used'
)
numberOfCores = Int(default_value=4).tag(info='number of cores for the calculations if useParallel is True'
)
scriptPath = CPath(allow_none=False, default_value=_SCRIPT_PATH).tag(
info='The (relative) path of the script, used in the structure calculation run'
)
nefInputPath = CPath(allow_none=True, default_value=_INPUT_NEF_PATH).tag(
info='The (relative) path of the input file in Nef format, used in the structure calculation run'
)
nefOutputPath = CPath(allow_none=True, default_value=None).tag(
info='The (relative) path of the output file in Nef format, generated by the structure calculation run'
)
wwPdbPath = CPath(allow_none=True, default_value=None).tag(
info='The (relative) path of the wwPDB XML file, generated by the wwPDB validation server'
)
# progress monitoring
setupDone = Bool(default_value=False).tag(
info='flag to indicate if setup has been done'
)
calculationDone = Bool(default_value=False).tag(
info='flag to indicate if the calculation has been done'
)
processDone = Bool(default_value=False).tag(
info='flag to indicate if the post-calculation processing has been done'
)
def __init__(self, project, runName='run', runId=1):
"""
:param project: the project instance
"""
super().__init__()
self.runType = self._RUN_TYPE
self.runName = runName
self.runId = runId
self._project = project
# assure directory for RUN_TYPE
self._project.dataPath.fetchDir(self.runType)
self._runPath = None
# Find any executables (if defined)
_preferences = project.application.preferences
if self._EXECUTABLE1 is not None:
if (_path := _preferences.get(self._EXECUTABLE1)) is not None and \
len(_path) > 0:
self._executable1 = aPath(_path)
if not self._executable1.exists():
getLogger().warning(f'Executable-1 not found at "{self._executable1}"')
if self._EXECUTABLE2 is not None:
if (_path := _preferences.get(self._EXECUTABLE2)) is not None and \
len(_path) > 0:
self._executable2 = aPath(_path)
if not self._executable2.exists():
getLogger().warning(f'Executable-2 not found at "{self._executable2}"')
@property
def project(self):
""":return the Project instance"""
return self._project
def _getDirName(self) -> str:
"""Create a directory name from the current settings of runName, runId and
timestamp"""
_dirName = '%s-%s' % (self.runName, self.runId)
if self.useTimeStamp and self._timeStamp is not None:
_dirName += '-' + self._timeStamp
return _dirName
@property
def runPath(self) -> Path:
"""Absolute path to the structure calculation run data
Either _runPath as currently defined, or
auto constructed from runType, runName, runId and timestamp.
"""
if self._runPath is not None:
return aPath(self._runPath)
else:
return self.project.application.dataPath / self.runType / self._getDirName()
[docs] def fetchDirectory(self) -> Path:
"""Fetch (i.e. get or create if needed) the directory defined by current settings,
optionally setting the timestamp depending on self.useTimeStamp
save directory as the _runPath attribute for later
:return a Path instance to the directory
"""
if self.useTimeStamp:
self._timeStamp = timeStamp()
# This assures the data/runType directory exists; run will be generated there
self.project.application.dataPath.fetchDir(self.runType)
self._runPath = None # This "resets" the runPath, and a new one will be generated
runPath = self.runPath
if not runPath.exists():
runPath.mkdir()
else:
raise RuntimeError(f'Directory {runPath} already exists')
self._runPath = runPath
return runPath
[docs] def saveState(self) -> Path:
"""Save the state of self to a json file in directory self.runPath
:return the path of the json file as a Path instance
"""
path = self.runPath / self._JSON_FILE
self.save(path.asString())
return path
[docs] def restoreState(self, runPath=None):
"""Restore the settings from json-file in directory runPath
(defaults to the directory defined by current settings).
:param runPath: the path to the directory
"""
if runPath is None:
runPath = self.runPath
runPath = aPath(runPath)
if not runPath.exists():
raise RuntimeError('runPath "%s" does not exist' % runPath)
if not runPath.is_dir():
raise RuntimeError('invalid runPath "%s"; not a directory' % runPath)
path = runPath / self._JSON_FILE
if not path.exists():
raise FileNotFoundError('Json file "%s" does not exist' % path)
_project = self._project
self.restore(path.asString()) # This restores the data from path
if self._project is None: # This occurs when importing from another project
getLogger().warning("Project appears to have changed!")
self._project = _project
self._runPath = runPath
def __str__(self):
return '<%s: %r>' % (self.__class__.__name__, self._getDirName())
#-------------------------------------------------------------------------------
[docs] def setupCalculation(self, useTimeStamp) -> Path:
"""This sets up the calculation;
Needs Sub-classing
:param useTimeStamp: add timestamp to the run path
:return The absolute path to the run directory
"""
raise NotImplemented('requires subclassing')