Source code for pyworkflow.project.config

#!/usr/bin/env python
# -*- coding: utf-8 -*-
# **************************************************************************
# *
# * Authors:     J.M. De la Rosa Trevin ( [1]
# *
# * [1] SciLifeLab, Stockholm University
# *
# * This program is free software: you can redistribute it and/or modify
# * it under the terms of the GNU General Public License as published by
# * the Free Software Foundation, either version 3 of the License, or
# * (at your option) any later version.
# *
# * This program is distributed in the hope that it will be useful,
# * but WITHOUT ANY WARRANTY; without even the implied warranty of
# * GNU General Public License for more details.
# *
# * You should have received a copy of the GNU General Public License
# * along with this program.  If not, see <>.
# *
# *  All comments concerning this program package may be sent to the
# *  e-mail address ''
# *
# **************************************************************************

import logging
logger = logging.getLogger(__name__)
import json
import datetime as dt

import pyworkflow.object as pwobj
from pyworkflow.mapper import SqliteMapper

[docs]class ProjectSettings(pwobj.Object): """ Store settings related to a project. """ COLOR_MODE_STATUS = 0 COLOR_MODE_LABELS = 1 COLOR_MODE_AGE = 2 COLOR_MODE_SIZE = 3 COLOR_MODES = (COLOR_MODE_STATUS, COLOR_MODE_LABELS, COLOR_MODE_AGE, COLOR_MODE_SIZE) def __init__(self, confs={}, **kwargs): super().__init__(**kwargs) self.config = ProjectConfig() # Store the current view selected by the user self.currentProtocolsView = pwobj.String() # Store the color mode: 0= Status, 1=Labels, ... self.colorMode = pwobj.Integer(ProjectSettings.COLOR_MODE_STATUS) # Store graph nodes positions and other info self.nodeList = NodeConfigList() self.labelsList = LabelsList() # Label list self.mapper = None # This should be set when load, or write self.runsView = pwobj.Integer(1) # by default the graph view self.readOnly = pwobj.Boolean(False) self.runSelection = pwobj.CsvList(int) # Store selected runs self.dataSelection = pwobj.CsvList(int) # Store selected runs # Some extra settings stored, now mainly used # from the webtools # Time when the project was created self.creationTime = pwobj.String( # Number of days that this project is active # if None, the project will not expire # This is used in webtools where a limited time # is allowed for each project self.lifeTime = pwobj.Integer() # Set a disk quota for the project (in Gb) # if None, quota is unlimited self.diskQuota = pwobj.Integer()
[docs] def commit(self): """ Commit changes made. """ self.mapper.commit()
[docs] def getRunsView(self): return self.runsView.get()
[docs] def setRunsView(self, value): self.runsView.set(value)
[docs] def getReadOnly(self): return self.readOnly.get()
[docs] def setReadOnly(self, value): self.readOnly.set(value)
[docs] def getCreationTime(self): return self.creationTime.datetime()
[docs] def setCreationTime(self, value): self.creationTime.set(value)
[docs] def getLifeTime(self): return self.lifeTime.get()
[docs] def setLifeTime(self, value): self.lifeTime.set(value)
[docs] def getConfig(self): return self.config
[docs] def getProtocolView(self): return self.currentProtocolsView.get()
[docs] def setProtocolView(self, protocolView): """ Set the new protocol Menu given its index. The new ProtocolMenu will be returned. """ self.currentProtocolsView.set(protocolView)
[docs] def getColorMode(self): return self.colorMode.get()
[docs] def setColorMode(self, colorMode): """ Set the color mode to use when drawing the graph. """ self.colorMode.set(colorMode)
[docs] def statusColorMode(self): return self.getColorMode() == self.COLOR_MODE_STATUS
[docs] def labelsColorMode(self): return self.getColorMode() == self.COLOR_MODE_LABELS
[docs] def ageColorMode(self): return self.getColorMode() == self.COLOR_MODE_AGE
[docs] def sizeColorMode(self): return self.getColorMode() == self.COLOR_MODE_SIZE
[docs] def write(self, dbPath=None): self.setName('ProjectSettings') if dbPath is not None: self.mapper = SqliteMapper(dbPath, globals()) else: if self.mapper is None: raise Exception("Can't write ProjectSettings without " "mapper or dbPath") self.mapper.deleteAll() self.mapper.insert(self) self.mapper.commit()
[docs] def getNodes(self): return self.nodeList
[docs] def cleanUpNodes(self, runsIds, toRemove=True): """ This will clean up all the nodes that do not have a matching run. This is because until now, the nodes here weren't removes when protocols were removed. :param runsIds: iterable with protocol's objId to be removed. :param toRemove: Passed is are to be removed. Otherwise, are the ones to keep """ try:"Cleaning up unused graphical nodes.") nodesToDelete = [] for node in self.getNodes(): nodeId = str(node.getId()) # if it is not the root node if nodeId != '0': if (nodeId in runsIds) == toRemove: nodesToDelete.append(node.getId())"Following graphical nodes %s unmatched. Deleting them" % nodesToDelete) for key in nodesToDelete: self.getNodes().removeNode(key) except Exception as e: logger.error("Couldn't clean up graphical nodes.", exc_info=e)
[docs] def getNodeById(self, nodeId): return self.nodeList.getNode(nodeId)
[docs] def addNode(self, nodeId, **kwargs): return self.nodeList.addNode(nodeId, **kwargs)
[docs] def removeNode(self, nodeId): """ Removes a graphical node based on its id""" self.nodeList.removeNode(nodeId)
[docs] def getLabels(self): return self.labelsList
[docs] @classmethod def load(cls, dbPath): """ Load a ProjectSettings from dbPath. """ classDict = dict(globals()) classDict.update(pwobj.__dict__) mapper = SqliteMapper(dbPath, classDict) settingList = mapper.selectByClass('ProjectSettings') n = len(settingList) if n == 0: raise Exception("Can't load ProjectSettings from %s" % dbPath) elif n > 1: raise Exception("Only one ProjectSettings is expected in db, " "found %d in %s" % (n, dbPath)) settings = settingList[0] settings.mapper = mapper return settings
[docs]class ProjectConfig(pwobj.Object): """A simple base class to store ordered parameters""" def __init__(self, **args): super().__init__(**args) self.logo = pwobj.String('scipion_logo_small.png') # Do not store this object, unless we implement some kind of # icon customization self._objDoStore = False
[docs]class NodeConfig(pwobj.Scalar): """ Store Graph node information such as x, y. """ def __init__(self, nodeId=0, x=None, y=None, selected=False, expanded=True, visible=True): pwobj.Scalar.__init__(self) # Special node id 0 for project node self._values = {'id': nodeId, 'x': pwobj.Integer(x).get(0), 'y': pwobj.Integer(y).get(0), 'selected': selected, 'expanded': expanded, 'visible': pwobj.Boolean(visible).get(0), 'labels': []} def _convertValue(self, value): """Value should be a str with comma separated values or a list. """ self._values = json.loads(value)
[docs] def getObjValue(self): self._objValue = json.dumps(self._values) return self._objValue
[docs] def get(self): return self.getObjValue()
[docs] def getId(self): return self._values['id']
[docs] def setX(self, x): self._values['x'] = x
[docs] def getX(self): return self._values['x']
[docs] def setY(self, y): self._values['y'] = y
[docs] def getY(self): return self._values['y']
[docs] def setPosition(self, x, y): self.setX(x) self.setY(y)
[docs] def getPosition(self): return self.getX(), self.getY()
[docs] def setSelected(self, selected): self._values['selected'] = selected
[docs] def isSelected(self): return self._values['selected']
[docs] def setExpanded(self, expanded): self._values['expanded'] = expanded
[docs] def isExpanded(self): return self._values['expanded']
[docs] def setVisible(self, visible): self._values['visible'] = visible
[docs] def isVisible(self): if self._values.get('visible') is None: self._values['visible'] = True return self._values['visible']
[docs] def setLabels(self, labels): self._values['labels'] = labels
[docs] def getLabels(self): return self._values.get('labels', None)
def __str__(self): return 'NodeConfig: %s' % self._values
[docs]class NodeConfigList(pwobj.List): """ Store all nodes information items and also store a dictionary for quick access to nodes query. """ def __init__(self): self._nodesDict = {} pwobj.List.__init__(self)
[docs] def getNode(self, nodeId): return self._nodesDict.get(nodeId, None)
[docs] def addNode(self, nodeId, **kwargs): node = NodeConfig(nodeId, **kwargs) self._nodesDict[node.getId()] = node self.append(node) return node
[docs] def removeNode(self, nodeId): """ Removes a node with the id = nodeId""" nodeToRemove = self._nodesDict[nodeId] self._nodesDict.pop(nodeId) self.remove(nodeToRemove)
[docs] def updateDict(self): self._nodesDict.clear() for node in self: self._nodesDict[node.getId()] = node
[docs] def clear(self): pwobj.List.clear(self) self._nodesDict.clear()
[docs]class Label(pwobj.Scalar): """ Store Label information """ EMPTY_OLD_NAME = None def __init__(self, name='', color=None): pwobj.Scalar.__init__(self) # Special node id 0 for project node self._values = {'name': name, 'color': color} self._oldName = self.EMPTY_OLD_NAME def _convertValue(self, value): """Value should be a str with comma separated values or a list. """ self._values = json.loads(value) # Clean unused "id" field if "id" in self._values: self._values.pop("id")
[docs] def getObjValue(self): self._objValue = json.dumps(self._values) return self._objValue
[docs] def get(self): return self.getObjValue()
[docs] def getName(self)->str: return self._values['name']
[docs] def setName(self, newName): # For recurrent edit, # we keep the old name only the first time if not self.hasOldName(): self._oldName = self._values['name'] self._values['name'] = newName
[docs] def hasOldName(self)->bool: return self._oldName != self.EMPTY_OLD_NAME
[docs] def clearOldName(self): self._oldName = self.EMPTY_OLD_NAME
[docs] def getOldName(self)->str: return self._oldName
[docs] def setColor(self, color): self._values['color'] = color
[docs] def getColor(self)->str: return self._values.get('color', None)
def __str__(self): return 'Label: %s' % self._values def __eq__(self, other): return self.getName() == other.getName()
[docs]class LabelsList(pwobj.List): """ Store all labels information""" def __init__(self): self._labelsDict = {} pwobj.List.__init__(self)
[docs] def getLabel(self, name): return self._labelsDict.get(name, None)
[docs] def addLabel(self, label): self._labelsDict[label.getName()] = label self.append(label) return label
[docs] def updateDict(self): self._labelsDict.clear() for label in self: self._labelsDict[label.getName()] = label
[docs] def deleteLabel(self, label): self._labelsDict.pop(label.getName()) self.remove(label)
[docs] def clear(self): pwobj.List.clear(self) self._labelDict.clear()