import logging
logger = logging.getLogger(__file__)
import ast
import importlib
import inspect
import json
import os
import shutil
import sys
import types
from .constants import *
HOME = os.path.abspath(os.path.dirname(__file__))
PYTHON = os.environ.get(SCIPION_PYTHON, SCIPION_PYTHON_DEFAULT)
[docs]def join(*paths):
""" join paths from HOME . """
return os.path.join(HOME, *paths)
__resourcesPath = [join('resources')]
[docs]def findResource(filename):
from .utils.path import findFile
return findFile(filename, *__resourcesPath)
[docs]def genNotesHeading():
return SCIPION_NOTES_HEADING_MSG
[docs]def getAppsPath(*paths):
return join(APPS, *paths)
[docs]def getSyncDataScript():
return getAppsPath(PW_SYNC_DATA)
[docs]def getScheduleScript():
return getAppsPath(PW_SCHEDULE_RUN)
[docs]def getPwProtMpiRunScript():
return getAppsPath(PW_PROTOCOL_MPIRUN)
[docs]def getTestsScript():
return getAppsPath(PW_RUN_TESTS)
[docs]def getViewerScript():
return getAppsPath(PW_VIEWER)
[docs]def getPyworkflowPath():
""" Returns the path where pyworkflow is"""
return os.path.dirname(__file__)
[docs]def getModuleFolder(moduleName):
""" Returns the path of a module without importing it"""
spec = importlib.util.find_spec(moduleName)
return os.path.dirname(spec.origin)
[docs]class Config:
""" Main Config for pyworkflow. It contains the main Scipion configuration variables
providing default values or, if present, taking them from the environment.
Necessary value is SCIPION_HOME and has to be present in the environment"""
@staticmethod
def __get(key, default):
value = os.environ.get(key, default)
# Expand user and variables if string value
if isinstance(value, str):
value = os.path.expandvars(os.path.expanduser(value))
return value
[docs] class Root:
""" Simple helper to return path from a root. """
def __init__(self, root):
self._root = root
[docs] def join(self, *path):
# We need to consider variable in the config with ~
expanded = os.path.expanduser(os.path.join(*path))
# join will not join if expanded is absolute
return os.path.join(self._root, expanded)
# Home for scipion
_get = __get.__func__
SCIPION_HOME = os.path.abspath(_get(SCIPION_HOME_VAR, ''))
"Path where Scipion is installed. Other paths are based on this like SCIPION_SOFTWARE, SCIPION_TESTS,... unless specified"
_root = Root(SCIPION_HOME)
_join = _root.join
# SCIPION PATHS
SCIPION_SOFTWARE = _join(_get('SCIPION_SOFTWARE', 'software'))
"Path where Scipion will install the software. Defaults to SCIPION_HOME/software."
SCIPION_TESTS = _join(_get('SCIPION_TESTS', os.path.join('data', 'tests')))
"Path where to find/download test data. Defaults to SCIPION_HOME/data/tests."
# User dependent paths
SCIPION_USER_DATA = _get('SCIPION_USER_DATA', '~/ScipionUserData')
"Path where Scipion projects are or will be created. Defaults to ~/ScipionUserData"
SCIPION_TMP = _get('SCIPION_TMP', _join(SCIPION_USER_DATA, 'tmp'))
"General purpose scipion tmp folder. Defaults to SCIPION_USER_DATA/tmp"
# LOGGING variables
SCIPION_LOGS = _get('SCIPION_LOGS', _join(SCIPION_USER_DATA, 'logs'))
"Path for Scipion logs folder used by the GUI. Defaults to SCIPION_USER_DATA/logs."
SCIPION_LOG_CONFIG = _get('SCIPION_LOG_CONFIG', None)
"Optional. Path to a python logging configuration file fine tune the logging."
SCIPION_LOG = _join(SCIPION_LOGS, 'scipion.log')
"Path to the file where scipion will write GUI logging messages. Defaults to SCIPION_LOGS/scipion.log"
SCIPION_LOG_FORMAT = _get('SCIPION_LOG_FORMAT', "%(message)s")
"Format for all the log lines, defaults to %(message)s. To compose the format see https://docs.python.org/3/library/logging.html#logrecord-attributes"
SCIPION_LOG_LEVEL = _get(SCIPION_LOG_LEVEL, 'INFO')
"Default logging level. String among CRITICAL, ERROR, WARNING, INFO, DEBUG, NOTSET. Default value is INFO."
NO_COLOR = _get('NO_COLOR', '')
"Comply with https://no-color.org/ initiative. Set it to something different than '' to deactivate colors in the output."
SCIPION_SCRATCH = _get(SCIPION_SCRATCH, None)
"Optional. Path to a location mounted in a scratch drive (SSD,...)"
SCIPION_TESTS_OUTPUT = _get('SCIPION_TESTS_OUTPUT', _join(SCIPION_USER_DATA, 'Tests'))
"Path to a folder Where the output of the tests will be written. Defaults to SCIPION_USER_DATA/Tests."
SCIPION_TEST_NOSYNC = _get('SCIPION_TEST_NOSYNC', "False")
"Set it to 1, True, Yes or y to cancel test dataset synchronization. Needed when updating files in a dataset."
SCIPION_SUPPORT_EMAIL = _get('SCIPION_SUPPORT_EMAIL', 'scipion@cnb.csic.es')
SCIPION_LOGO = _get('SCIPION_LOGO', 'scipion_logo.gif')
# Config variables
SCIPION_CONFIG = _get('SCIPION_CONFIG', 'scipion.conf')
"Path to the scipion configuration file where all this variables could be defined."
SCIPION_LOCAL_CONFIG = _get('SCIPION_LOCAL_CONFIG', SCIPION_CONFIG)
"Path to an optional/extra/user configuration file meant to overwrite default variables."
SCIPION_HOSTS = _get('SCIPION_HOSTS', 'hosts.conf')
"Path to the host.cof file to allow scipion to use queue engines and run in HPC environments."
SCIPION_PROTOCOLS = _get('SCIPION_PROTOCOLS', 'protocols.conf')
""
SCIPION_PLUGIN_JSON = _get('SCIPION_PLUGIN_JSON', None)
"Optional. Path to get the json file with all the plugins available for Scipion."
SCIPION_PLUGIN_REPO_URL = _get('SCIPION_PLUGIN_REPO_URL',
'http://scipion.i2pc.es/getplugins/')
"Url from where to get the list of plugins."
# REMOTE Section
SCIPION_URL = _get('SCIPION_URL', 'http://scipion.cnb.csic.es/downloads/scipion')
SCIPION_URL_SOFTWARE = _get('SCIPION_URL_SOFTWARE', SCIPION_URL + '/software')
SCIPION_URL_TESTDATA = _get('SCIPION_URL_TESTDATA', SCIPION_URL + '/data/tests')
# Scipion Notes
SCIPION_NOTES_FILE = _get(SCIPION_NOTES_FILE, 'notes.txt')
"Name of the file where to write per project notes."
SCIPION_NOTES_PROGRAM = _get(SCIPION_NOTES_PROGRAM, None)
"Command or program to use to open the notes file. Otherwise system will extension association will take place."
SCIPION_NOTES_ARGS = _get(SCIPION_NOTES_ARGS, None)
# Aspect
SCIPION_FONT_NAME = _get('SCIPION_FONT_NAME', "Helvetica")
"Name of the font to use in Scipion GUI. Defaults to Helvetica."
SCIPION_FONT_SIZE = int(_get('SCIPION_FONT_SIZE', SCIPION_DEFAULT_FONT_SIZE))
"Size of the 'normal' font to be used in Scipion GUI. Defaults to 10."
WIZARD_MASK_COLOR = _get('WIZARD_MASK_COLOR', '[0.125, 0.909, 0.972]')
"Color to use in some wizards."
# Notification
SCIPION_NOTIFY = _get('SCIPION_NOTIFY', 'True')
"If set to False, Scipion developers will know almost nothing about Scipion usage and will have less information to improve it."
SCIPION_CWD = _get('SCIPION_CWD', os.path.abspath(os.getcwd()))
"Directory when scipion was launched"
SCIPION_GUI_REFRESH_IN_THREAD = _get('SCIPION_GUI_REFRESH_IN_THREAD', 'False')
"True to refresh the runs graph with a thread. Unstable."
SCIPION_GUI_REFRESH_INITIAL_WAIT = int(_get("SCIPION_GUI_REFRESH_INITIAL_WAIT", 5))
"Seconds to wait after a manual refresh"
SCIPION_GUI_CANCEL_AUTO_REFRESH = _get("SCIPION_GUI_CANCEL_AUTO_REFRESH","False")
"Set it to True to cancel automatic refresh of the runs."
# Cancel shutil fast copy. In GPFS, shutil.copy does fail when trying a fastcopy and does not fallback on the slow copy.
SCIPION_CANCEL_FASTCOPY = _get('SCIPION_CANCEL_FASTCOPY', None)
"Cancel fast copy done by shutil (copying files) when it fails. Has happened in GPFS environments."
# Priority package list: This variable is used in the view protocols in
# order to load first the plugins that contains the main protocols.conf
# sections, so other plugins can define only their sections avoiding
# duplicating all the sections in all plugins
SCIPION_PRIORITY_PACKAGE_LIST = _get('SCIPION_PRIORITY_PACKAGE_LIST', None)
SCIPION_STEPS_CHECK_SEC = int(_get('SCIPION_STEPS_CHECK_SEC', 5))
"Number of seconds to wait before checking if new input is available in streamified protocols."
SCIPION_UPDATE_SET_ATTEMPTS = int(_get('SCIPION_UPDATE_SET_ATTEMPTS', 3))
"Number of attempts to modify the protocol output before failing. The default value is 3"
SCIPION_UPDATE_SET_ATTEMPT_WAIT = int(_get('SCIPION_UPDATE_SET_ATTEMPT_WAIT', 2))
"Time in seconds to wait until the next attempt when checking new outputs. The default value is 2 seconds"
try:
VIEWERS = ast.literal_eval(_get('VIEWERS', "{}"))
except Exception as e:
VIEWERS = {}
print("ERROR loading preferred viewers, VIEWERS variable will be ignored")
print(e)
SCIPION_DOMAIN = _get(SCIPION_DOMAIN, None)
SCIPION_TESTS_CMD = _get(SCIPION_TESTS_CMD, getTestsScript())
# ---- Getters ---- #
# Getters are alternatives to offer a variable, but preventing it to be stored in the config
[docs] @classmethod
def getLibFolder(cls):
"""
:return: Folder where libraries must be placed in case a binding needs them
"""
lib = cls._join(cls.SCIPION_SOFTWARE, 'lib')
os.makedirs(lib, exist_ok=True)
return lib
[docs] @classmethod
def getBindingsFolder(cls):
"""
Folder where bindings must be placed. This folder is added to sys.path at launching time.
If the binding depends on a dynamic libraries, those must be placed at cls.getLibFolder()
:return: The bindings folder
"""
bindings = cls._join(cls.SCIPION_SOFTWARE, 'bindings')
os.makedirs(bindings, exist_ok=True)
return bindings
[docs] @classmethod
def getLogsFolder(cls):
"""
Folder where scipion logs must be placed. The folder is created
"""
logsFolder = cls.SCIPION_LOGS
os.makedirs(logsFolder, exist_ok=True)
return logsFolder
[docs] @classmethod
def getVars(cls):
""" Return a dictionary with all variables defined
in this Config.
"""
configVars = dict()
# For each variable, also in base classes
for baseCls in inspect.getmro(cls):
for name, value in vars(baseCls).items():
# Skip methods and internal attributes starting with __
# (e.g __doc__, __module__, etc)
if (isinstance(value, str) or isinstance(value, int)) and not name.startswith('__'):
configVars[name] = str(value)
return configVars
[docs] @classmethod
def printVars(cls):
""" Print the variables dict, mostly for debugging. """
from .utils import prettyDict
prettyDict(cls.getVars())
[docs] @classmethod
def getDomain(cls):
""" Import domain module from path or name defined in SCIPION_DOMAIN.
"""
value = cls.SCIPION_DOMAIN
if not value:
return None
if os.path.isdir(value):
dirname, value = os.path.split(value)
sys.path.append(dirname)
return importlib.import_module(value).Domain
[docs] @classmethod
def setDomain(cls, moduleOrNameOrPath):
if isinstance(moduleOrNameOrPath, types.ModuleType):
value = os.path.abspath(moduleOrNameOrPath.__path__[0])
else:
value = moduleOrNameOrPath
cls.SCIPION_DOMAIN = value
os.environ[SCIPION_DOMAIN] = value
[docs] @staticmethod
def getPythonLibFolder():
from sysconfig import get_paths
return os.path.join(get_paths()['data'], "lib")
[docs] @staticmethod
def debugOn():
""" Returns a True if debug mode (SCIPION_DEBUG variable) is active """
from .utils import envVarOn
return bool(envVarOn(SCIPION_DEBUG))
[docs] @staticmethod
def toggleDebug():
debugOn = not Config.debugOn()
os.environ[SCIPION_DEBUG] = str(debugOn)
os.environ[SCIPION_DEBUG_NOCLEAN] = str(debugOn)
os.environ[SCIPION_LOG_LEVEL] = "INFO" if not debugOn else "DEBUG"
[docs] @staticmethod
def debugSQLOn():
from .utils import envVarOn
return bool(envVarOn(SCIPION_DEBUG_SQLITE))
[docs] @staticmethod
def toggleDebugSQL():
newValue = not Config.debugSQLOn()
os.environ[SCIPION_DEBUG_SQLITE] = str(newValue)
[docs] @classmethod
def refreshInThreads(cls):
from .utils import strToBoolean
return strToBoolean(cls.SCIPION_GUI_REFRESH_IN_THREAD)
[docs] @classmethod
def getExternalJsonTemplates(cls):
return os.path.dirname(cls.SCIPION_CONFIG)
[docs] @classmethod
def getWizardMaskColor(cls):
return json.loads(cls.WIZARD_MASK_COLOR)
[docs] @classmethod
def getPriorityPackageList(cls):
if cls.SCIPION_PRIORITY_PACKAGE_LIST is not None:
return cls.SCIPION_PRIORITY_PACKAGE_LIST.split(" ")
else:
return []
[docs] @classmethod
def getStepsCheckSeconds(cls):
return cls.SCIPION_STEPS_CHECK_SEC
[docs] @classmethod
def getUpdateSetAttempts(cls):
return cls.SCIPION_UPDATE_SET_ATTEMPTS
[docs] @classmethod
def getUpdateSetAttemptsWait(cls):
return cls.SCIPION_UPDATE_SET_ATTEMPT_WAIT
[docs] @classmethod
def colorsInTerminal(cls):
""" Returns true if colors are allowed. Based on NO_COLOR variable. Undefined or '' colors are enabled"""
return cls.NO_COLOR == ''
# Add bindings folder to sys.path
sys.path.append(Config.getBindingsFolder())
# Cancel fast copy
if Config.SCIPION_CANCEL_FASTCOPY:
shutil._USE_CP_SENDFILE = False