#!/usr/bin/env python
# -*- coding: utf-8 -*-
# **************************************************************************
# *
# * Authors: J.M. De la Rosa Trevin (delarosatrevin@scilifelab.se) [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
# * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# * 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 <https://www.gnu.org/licenses/>.
# *
# * All comments concerning this program package may be sent to the
# * e-mail address 'scipion@cnb.csic.es'
# *
# **************************************************************************
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)
# 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(dt.datetime.now())
# 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 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:
logger.info("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())
logger.info("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 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()