src/python/dm/common/mongodb/impl/datasetCollection.py

+#!/usr/bin/env python
+
+from dm.common.utility.loggingManager import LoggingManager
+from dm.common.exceptions.invalidArgument import InvalidArgument
+from dm.common.exceptions.objectAlreadyExists import ObjectAlreadyExists
+from dm.common.exceptions.objectNotFound import ObjectNotFound
+from dm.common.exceptions.dbError import DbError
+from dmMongoCollection import DmMongoCollection
+
+class DatasetCollection(DmMongoCollection):
+    """Class responsible for updating dataset collection in mongo db."""
+
+    UNIQUE_KEYS_LIST = [ '_id', 'datasetName', 'experimentName' ]
+    ITEM_TYPE = 'dataset'
+
+    def __init__(self, dbClient):
+        DmMongoCollection.__init__(self, 'datasets', dbClient)
+
+#######################################################################
+# Testing
+if __name__ == '__main__':
+    from dmMongoClient import DmMongoClient
+    mongo = DmMongoClient('dm')
+    datasetCollection = DatasetCollection(mongo)
+    datasetInfo = {'name' : 'ds-001', 
+        'owner' : 'sv',
+        'experimentName' : 'exp-001',
+        'voltage' : { 'gt' : 400},
+        'current' : { 'lt' : 100},
+    }
+    #print datasetCollection.add(datasetInfo)
+    print datasetCollection.updateByName(datasetInfo)
+    print datasetCollection.findByName('ds-001')
+    print datasetCollection.findByQueryDict({'dataset' : 'ds-001'}, {'owner' : 1})
+
+
+

src/python/dm/common/mongodb/impl/dmMongoClient.py

+#!/usr/bin/env python
+
+from pymongo import MongoClient
+
+from dm.common.utility.loggingManager import LoggingManager
+from dm.common.exceptions.dbError import DbError
+
+class DmMongoClient(object):
+    """ DM MongoDB client."""
+
+    def __init__(self, dbName, dbUri='mongodb://localhost:27017/'):
+        self.logger = LoggingManager.getInstance().getLogger(self.__class__.__name__)
+        try:
+            self.client = MongoClient(dbUri)
+            self.db = self.client[dbName]
+            self.dbName = dbName
+        except Exception, ex:
+            self.logger.error('Cannot connect to Mongo DB: %s' % ex)
+            raise DbError(exception=ex)
+
+    def drop(self, collection):
+        return self.db[collection].drop()
+
+    def insert(self, collection, dict, **kwargs):
+        return self.db[collection].insert(dict, **kwargs)
+
+    def update(self, collection, query, update, **kwargs):
+        key = '.'.join((collection, str(update['$set'].keys()[0])))
+        return self.db[collection].update(query, update, **kwargs)
+
+    def findOne(self, collection, criteria={}):
+        return self.db[collection].find_one(criteria)
+
+    def find(self, collection, criteria={}, projections={}):
+        return self.db[collection].find(criteria, projections)
+
+    def findAsList(self, collection, criteria={}, projections={}):
+        return list(self.db[collection].find(criteria, projections))
+
+    def getCollectionNames(self):
+        return self.db.collection_names()
+
+#######################################################################
+# Testing
+if __name__ == '__main__':
+    mongo = DmMongoClient('dm')
+    mongo.drop('students')
+    id = mongo.insert('students', { '_id' : 1, 'semester' : 1, 'grades' : [ 70, 87, 90 ]   })
+    print 'Student #1 id: ', id
+    id = mongo.insert('students', { '_id' : 2, 'semester' : 1, 'grades' : [ 90, 88, 92 ] } )
+    print 'Student #2 id: ', id
+    print mongo.findAsList('students', criteria={ 'semester' : 1, 'grades': { '$gte' : 85 } }, projections={ 'grades.$' : 1 } )
+    result = mongo.update('files', {'_id' : 2}, {'$set' : {'experimentId' : 2}}, upsert=True)
+    print result
+    doc = mongo.findOne('files', {'experimentId' : 1})
+    print doc
+    docs = mongo.find('files', projections={'experimentId' : 1})
+    print docs.count()
+    docs = mongo.findAsList('files', projections={'experimentId' : 1})
+    print docs
+    print 'Collection names: ', mongo.getCollectionNames()
+    mongo.drop('students')
+    print 'Collection names (after drop): ', mongo.getCollectionNames()

src/python/dm/common/mongodb/impl/dmMongoCollection.py

+#!/usr/bin/env python
+
+from bson.objectid import ObjectId
+from dm.common.utility.loggingManager import LoggingManager
+from dm.common.exceptions.invalidArgument import InvalidArgument
+from dm.common.exceptions.objectAlreadyExists import ObjectAlreadyExists
+from dm.common.exceptions.objectNotFound import ObjectNotFound
+from dm.common.exceptions.dbError import DbError
+
+class DmMongoCollection(object):
+    """Base collection in mongo db."""
+
+    ALL_FIELDS_DICT = {'__return_only_id__' : False}
+    UNIQUE_KEYS_LIST = []
+    ITEM_TYPE = 'item'
+
+    def __init__(self, collectionName, dbClient):
+        self.logger = LoggingManager.getInstance().getLogger(self.__class__.__name__)
+        self.dbClient = dbClient
+
+        # Collection consists of items: [item]
+        self.collectionName = collectionName
+
+        # items => item
+        self.itemName = self.ITEM_TYPE
+
+        # item => Item
+        self.capitalizedItemName = self.itemName.capitalize() 
+
+    @classmethod
+    def getUniqueKeys(cls):
+        return cls.UNIQUE_KEYS_LIST
+
+    def findByKey(self, key, value):
+        dbObjectDict = self.dbClient.findOne(self.collectionName, {key : value})
+        if dbObjectDict is None:
+            raise ObjectNotFound('%s with %s=%s not found.' % (self.capitalizedItemName, key, value))
+        return dbObjectDict
+
+    def findByKeys(self, keyList, queryDict):
+        for key in keyList:
+            if not queryDict.has_key(key):
+                raise InvalidArgument('%s query dictionary does not specify key %s.' % (self.capitalizedItemName, key))
+
+        dbObjectDict = self.dbClient.findOne(self.collectionName, queryDict)
+        if dbObjectDict is None:
+            raise ObjectNotFound('%s with properties %s not found.' % (self.capitalizedItemName, queryDict))
+        return dbObjectDict
+
+    def findByUniqueKeys(self, queryDict):
+        return self.findByKeys(self.UNIQUE_KEYS_LIST, queryDict)
+
+    def findById(self, id):
+        return self.findByKey('_id', ObjectId(id))
+
+    def findByName(self, name):
+        return self.findByKey(self.NAME_KEY, name)
+
+    def findByQueryDict(self, queryDict, returnFieldDict=ALL_FIELDS_DICT):
+        return self.dbClient.findAsList(self.collectionName, queryDict, returnFieldDict)
+
+    def listByKey(self, key):
+        return self.dbClient.findAsList(self.collectionName, {}, {key : True})
+
+    def listById(self):
+        return self.dbClient.findAsList(self.collectionName, {}, {})
+
+    def listByName(self):
+        return self.findByName('_name')
+
+    def listByKeys(self, keyList):
+        returnFieldDict = {}
+        for key in keyList:
+            returnFieldDict[key] = True
+        return self.dbClient.findAsList(self.collectionName, {}, returnFieldDict)
+    def listAll(self):
+        return self.dbClient.findAsList(self.collectionName, {}, DmMongoCollection.ALL_FIELDS_DICT)
+
+    def __addDbObject(self, objectDict):
+        try:
+            self.dbClient.insert(self.collectionName, objectDict)
+        except Exception, ex:
+            self.logger.error('Cannot add %s with %s set to %s: %s' % (self.itemName, key, value, ex))
+            raise DbError(exception=ex)
+        return objectDict
+
+    def addByKey(self, key, objectDict):
+        value = objectDict.get(key)
+        if value is None:
+            raise InvalidArgument('%s info dictionary does not specify key %s.' % (self.capitalizedItemName, key))
+
+        dbObjectDict = self.dbClient.findOne(self.collectionName, {key : value})
+        if dbObjectDict is not None:
+            raise ObjectAlreadyExists('%s with %s set to %s already exists.' % (self.capitalizedItemName, key, value))
+
+        return self.__addDbObject(objectDict)
+
+    def addByName(self, objectDict):
+        return self.addByKey('_name', objectDict)
+
+    def addByKeys(self, keyList, objectDict):
+        queryDict = {}
+        for key in keyList:
+            value = objectDict.get(key)
+            if value is None:
+                raise InvalidArgument('%s info dictionary does not specify key %s.' % (self.capitalizedItemName, key))
+            queryDict[key] = value
+
+        dbObjectDict = self.dbClient.findOne(self.collectionName, queryDict)
+        if dbObjectDict is not None:
+            raise ObjectAlreadyExists('%s with properties %s already exists.' % (self.capitalizedItemName, queryDict))
+
+        return self.__addDbObject(objectDict)
+
+    def addByUniqueKeys(self, objectDict):
+        return self.addByKeys(self.UNIQUE_KEYS_LIST, objectDict)
+
+    def __updateDbObject(self, dbObjectDict, objectDict):
+        try:
+            id = dbObjectDict.get('_id')
+            dbObjectDict.update(objectDict)
+            self.dbClient.update(self.collectionName,
+                {'_id' : id},
+                {'$set' : dbObjectDict})
+        except Exception, ex:
+            self.logger.error('Cannot update %s %s with %s: %s' % (self.itemName, dbObjectDict, objectDict, ex))
+            raise DbError(exception=ex)
+        return dbObjectDict
+
+    def updateByKey(self, key, objectDict):
+        value = objectDict.get(key)
+        if value is None:
+            raise InvalidArgument('%s info dictionary does not specify key %s.' % (self.capitalizedItemName, key))
+
+        dbObjectDict = self.dbClient.findOne(self.collectionName, {key : value})
+        if dbObjectDict is None:
+            raise ObjectNotFound('%s with %s set to %s not found.' % (self.capitalizedItemName, key, value))
+
+        return self.__updateDbObject(dbObjectDict, objectDict)
+
+    def updateByKeys(self, keyList, objectDict):
+        queryDict = {}
+        for key in keyList:
+            value = objectDict.get(key)
+            if value is None:
+                raise InvalidArgument('%s info dictionary does not specify key %s.' % (self.capitalizedItemName, key))
+            queryDict[key] = value
+
+        dbObjectDict = self.dbClient.findOne(self.collectionName, queryDict)
+        if dbObjectDict is None:
+            raise ObjectNotFound('%s with properties %s not found.' % (self.capitalizedItemName, queryDict))
+
+        return self.__updateDbObject(dbObjectDict, objectDict)
+
+    def updateByUniqueKeys(self, objectDict):
+        return self.updateByKeys(self.UNIQUE_KEYS_LIST, objectDict)
+
+    def updateById(self, objectDict):
+        # Convert 'id' to '_id' key if needed, and wrap it with ObjectId()
+        if not objectDict.has_key('_id') and objectDict.has_key('id'):
+            objectDict['_id'] = ObjectId(objectDict['id'])
+            del objectDict['id']
+        return self.updateByKey('_id', objectDict)
+
+    def updateByName(self, objectDict):
+        return self.updateByKey('_name', objectDict)
+
+    def updateOrAddByKey(self, key, objectDict):
+        value = objectDict.get(key)
+        if value is None:
+            raise InvalidArgument('%s info dictionary does not specify key %s.' % (self.capitalizedItemName, key))
+
+        dbObjectDict = self.dbClient.findOne(self.collectionName, {key : value})
+        if dbObjectDict is None:
+            return self.__addDbObject(objectDict)
+        else:
+            return self.__updateDbObject(dbObjectDict, objectDict)
+
+    def updateOrAddByKeys(self, keyList, objectDict):
+        queryDict = {}
+        for key in keyList:
+            value = objectDict.get(key)
+            if value is None:
+                raise InvalidArgument('%s info dictionary does not specify key %s.' % (self.capitalizedItemName, key))
+            queryDict[key] = value
+
+        dbObjectDict = self.dbClient.findOne(self.collectionName, queryDict)
+        if dbObjectDict is None:
+            return self.__addDbObject(objectDict)
+        else:
+            return self.__updateDbObject(dbObjectDict, objectDict)
+
+    def updateOrAddByUniqueKeys(self, objectDict):
+        return self.updateOrAddByKeys(self.UNIQUE_KEYS_LIST, objectDict)
+
+    def updateOrAddByName(self, objectDict):
+        return self.updateOrAddByKey('_name', objectDict)
+
+#######################################################################
+# Testing
+if __name__ == '__main__':
+    from dmMongoClient import DmMongoClient
+    from bson.objectid import ObjectId
+    mongo = DmMongoClient('dm')
+    fileCollectionImpl = DmMongoCollection('files', mongo)
+    #objectDict = {'name' : 'xyz-001', 'experiment' : 'myexp-D', 'update' : 'sv2', 'locationList' : '[/opt/xyz, /data/xyz]'}
+    #print fileCollectionImpl.updateOrAddByKey('name', objectDict)
+    #f = fileCollectionImpl.findByKey('name', 'xyz-001')
+    #print f
+    #print type(f['_id'])
+    #print fileCollectionImpl.findById('556de0059e058b0ef4c4413b')
+    #print fileCollectionImpl.findByQueryDict({'experiment' : 'exp-001'}, {'locationList' : 1})
+
+    #print 'LIST BY ID\n', fileCollectionImpl.listById()
+    #print 'LIST BY NAME\n', fileCollectionImpl.listByKey('name')
+    print 'LIST ALL\n'
+    for f in fileCollectionImpl.listAll():
+        print 'FILE: %s\n' % f
+
+
+    #print fileCollectionImpl.updateByKeys(['name', 'experiment'], objectDict)
+

src/python/dm/common/mongodb/impl/experimentCollection.py

+#!/usr/bin/env python
+
+from dm.common.utility.loggingManager import LoggingManager
+from dm.common.exceptions.invalidArgument import InvalidArgument
+from dm.common.exceptions.objectAlreadyExists import ObjectAlreadyExists
+from dm.common.exceptions.objectNotFound import ObjectNotFound
+from dm.common.exceptions.dbError import DbError
+from dmMongoCollection import DmMongoCollection
+
+class ExperimentCollection(DmMongoCollection):
+    """Class responsible for updating experiment collection in mongo db."""
+
+    UNIQUE_KEYS_LIST = [ '_id', 'name' ]
+
+    def __init__(self, dbClient):
+        DmMongoCollection.__init__(self, 'experiments', dbClient)
+
+#######################################################################
+# Testing
+if __name__ == '__main__':
+    from dmMongoClient import DmMongoClient
+    mongo = DmMongoClient('dm')
+    experimentCollection = ExperimentCollection(mongo)
+    experimentInfo = {'name' : 'exp-001', 'owner' : 'sv'}
+    #print experimentCollection.add(experimentInfo)
+    print experimentCollection.updateByName(experimentInfo)
+    print experimentCollection.findByName('exp-001')
+    print experimentCollection.findByQueryDict({'experiment' : 'exp-001'}, {'owner' : 1})
+
+
+

src/python/dm/common/mongodb/impl/fileCollection.py

+#!/usr/bin/env python
+
+from dm.common.utility.loggingManager import LoggingManager
+from dm.common.exceptions.invalidArgument import InvalidArgument
+from dm.common.exceptions.objectAlreadyExists import ObjectAlreadyExists
+from dm.common.exceptions.objectNotFound import ObjectNotFound
+from dm.common.exceptions.dbError import DbError
+from dmMongoCollection import DmMongoCollection
+
+class FileCollection(DmMongoCollection):
+    """Class responsible for updating file collection in mongo db."""
+
+    UNIQUE_KEYS_LIST = [ 'experimentFilePath', 'experimentName' ]
+    NAME_KEY = 'experimentFilePath'
+    ITEM_TYPE = 'file'
+
+    def __init__(self, dbClient, experimentName=None):
+        collectionName = 'files'
+        if experimentName:
+            collectionName = '%s-files' % experimentName
+        DmMongoCollection.__init__(self, collectionName, dbClient)
+
+#######################################################################
+# Testing
+if __name__ == '__main__':
+    from dmMongoClient import DmMongoClient
+    mongo = DmMongoClient('dm')
+    fileCollection = FileCollection(mongo, 'exp01')
+    fileInfo = {'fileName' : 'xyz-001', 'experimentName' : 'exp01', 'update' : 'sv2', 'locationList' : '[/opt/xyz, /data/xyz]', 'experimentFilePath' : 'd1/xyz-001'}
+    print fileCollection.addByUniqueKeys(fileInfo)
+    #print type(fileCollection.findByName('xyz-001'))
+    print fileCollection.findByQueryDict({'experimentName' : 'exp01'}, {'locationList' : 1})
+    fileInfo['owner'] = 'ddm1'
+    print fileCollection.updateByUniqueKeys(fileInfo)
+
+
+

src/python/dm/common/mongodb/impl/mongoDbManager.py

+#!/usr/bin/env python
+
+import threading
+import os.path
+import sqlalchemy 
+from sqlalchemy.orm import sessionmaker
+from sqlalchemy.orm import mapper
+from sqlalchemy.orm import relationship
+
+from dm.common.exceptions.commandFailed import CommandFailed
+from dm.common.exceptions.configurationError import ConfigurationError
+from dm.common.utility.loggingManager import LoggingManager
+from dm.common.utility.configurationManager import ConfigurationManager
+from dmMongoClient import DmMongoClient
+
+class MongoDbManager:
+    """ Singleton class for mongo db management. """
+
+    CONFIG_SECTION_NAME = 'MongoDbManager'
+    MONGO_DB_NAME_KEY = 'mongoDbName'
+    MONGO_DB_USER_KEY = 'mongoDbUser'
+    MONGO_DB_URI_KEY = 'mongoDbUri' 
+    MONGO_DB_PASSWORD_FILE_KEY = 'mongoDbPasswordFile' 
+    CONFIG_OPTION_NAME_LIST = [ 'dbName', 'dbUser', 'dbPasswordFile' ]
+
+    # Singleton.
+    __lock = threading.RLock()
+    __instance = None
+
+    @classmethod
+    def getInstance(cls):
+        from dm.common.mongodb.impl.mongoDbManager import MongoDbManager
+        try:
+            mgr = MongoDbManager()
+        except MongoDbManager, ex:
+            mgr = ex
+        return mgr 
+
+    def __init__(self):
+        MongoDbManager.__lock.acquire()
+        try:
+            if MongoDbManager.__instance is not None:
+                raise MongoDbManager.__instance
+            MongoDbManager.__instance = self
+            self.lock = threading.RLock()
+            self.logger = LoggingManager.getInstance().getLogger(self.__class__.__name__)
+            cm = ConfigurationManager.getInstance()
+            self.dbName = cm.getConfigOption(MongoDbManager.CONFIG_SECTION_NAME, MongoDbManager.MONGO_DB_NAME_KEY)
+            self.logger.debug('Mongo DB name: %s' % self.dbName)
+            self.dbUri = cm.getConfigOption(MongoDbManager.CONFIG_SECTION_NAME, MongoDbManager.MONGO_DB_URI_KEY)
+            self.logger.debug('Mongo DB URI: %s' % self.dbUri)
+            self.dbUser = cm.getConfigOption(MongoDbManager.CONFIG_SECTION_NAME, MongoDbManager.MONGO_DB_USER_KEY)
+            self.logger.debug('Mongo DB user: %s' % self.dbUser)
+            self.dbPasswordFile = cm.getConfigOption(MongoDbManager.CONFIG_SECTION_NAME, MongoDbManager.MONGO_DB_PASSWORD_FILE_KEY)
+            self.logger.debug('Mongo DB password file: %s' % self.dbPasswordFile)
+            #dbPassword = open(dbPasswordFile, 'r').readline().strip()
+
+        finally:
+            MongoDbManager.__lock.release()
+
+    def getLogger(self):
+        return self.logger
+
+    def getDbClient(self):
+        return DmMongoClient(self.dbName, self.dbUri)
+
+
+#######################################################################
+# Testing.
+if __name__ == '__main__':
+    ConfigurationManager.getInstance().setConsoleLogLevel('debug')
+    mgr = MongoDbManager.getInstance()
+
+
+

src/python/dm/common/mongodb/impl/processingJobCollection.py

+#!/usr/bin/env python
+
+from dm.common.utility.loggingManager import LoggingManager
+from dm.common.exceptions.invalidArgument import InvalidArgument
+from dm.common.exceptions.objectAlreadyExists import ObjectAlreadyExists
+from dm.common.exceptions.objectNotFound import ObjectNotFound
+from dm.common.exceptions.dbError import DbError
+from dmMongoCollection import DmMongoCollection
+
+class ProcessingJobCollection(DmMongoCollection):
+    """Class responsible for updating processing job collection in mongo db."""
+
+    UNIQUE_KEYS_LIST = [ 'workflowId', 'timeStamp' ]
+    NAME_KEY = 'name'
+    ITEM_TYPE = 'processingJob'
+
+    def __init__(self, dbClient, prefix=None):
+        collectionName = 'processing-jobs'
+        if prefix:
+            collectionName = '%s-processing-jobs' % prefix
+        DmMongoCollection.__init__(self, collectionName, dbClient)
+
+#######################################################################
+# Testing
+if __name__ == '__main__':
+    from dmMongoClient import DmMongoClient
+    mongo = DmMongoClient('dm')
+    processingJobCollection = ProcessingJobCollection(mongo)
+    workflowInfo = {'name' : 'workflow-01', 'owner' : 'sveseli', 'stages' : 
+        [ 
+            { 'id' : '1', 'executable' : '/bin/date', 'args' : [ ] }, 
+            { 'id' : '2', 'executable' : '/bin/ls', 'args' : [ '-l', 'INPUT_FILE' ], 'parallelizeExecution' : False},
+            { 'id' : '3', 'executable' : '/bin/cp', 'args' : [ 'INPUT_FILE', 'OUTPUT_FILE=INPUT_FILE.processed' ], 'parallelizeExecution' : False} 
+        ]
+    }
+    print workflowCollection.addByUniqueKeys(workflowInfo)
+    print workflowCollection.findByQueryDict({'name' : 'workflow-01', 'owner' : 'sveseli'}, {'steps' : 1})
+    workflowInfo['description'] = 'My first workflow'
+    print workflowCollection.updateByUniqueKeys(workflowInfo)
+
+
+

src/python/dm/common/mongodb/impl/workflowCollection.py

+#!/usr/bin/env python
+
+from dm.common.utility.loggingManager import LoggingManager
+from dm.common.exceptions.invalidArgument import InvalidArgument
+from dm.common.exceptions.objectAlreadyExists import ObjectAlreadyExists
+from dm.common.exceptions.objectNotFound import ObjectNotFound
+from dm.common.exceptions.dbError import DbError
+from dmMongoCollection import DmMongoCollection
+
+class WorkflowCollection(DmMongoCollection):
+    """Class responsible for updating workflow collection in mongo db."""
+
+    UNIQUE_KEYS_LIST = [ 'name', 'owner' ]
+    NAME_KEY = 'name'
+    ITEM_TYPE = 'workflow'
+
+    def __init__(self, dbClient, prefix=None):
+        collectionName = 'workflows'
+        if prefix:
+            collectionName = '%s-workflows' % prefix
+        DmMongoCollection.__init__(self, collectionName, dbClient)
+
+#######################################################################
+# Testing
+if __name__ == '__main__':
+    from dmMongoClient import DmMongoClient
+    mongo = DmMongoClient('dm')
+    workflowCollection = WorkflowCollection(mongo)
+    workflowInfo = {'name' : 'workflow-01', 'owner' : 'sveseli', 'stages' : 
+        [ 
+            { 'id' : '1', 'executable' : '/bin/date', 'args' : [ ] }, 
+            { 'id' : '2', 'executable' : '/bin/ls', 'args' : [ '-l', 'INPUT_FILE' ], 'parallelizeExecution' : False},
+            { 'id' : '3', 'executable' : '/bin/cp', 'args' : [ 'INPUT_FILE', 'OUTPUT_FILE=INPUT_FILE.processed' ], 'parallelizeExecution' : False} 
+        ]
+    }
+    print workflowCollection.addByUniqueKeys(workflowInfo)
+    print workflowCollection.findByQueryDict({'name' : 'workflow-01', 'owner' : 'sveseli'}, {'steps' : 1})
+    workflowInfo['description'] = 'My first workflow'
+    print workflowCollection.updateByUniqueKeys(workflowInfo)
+
+
+

src/python/dm/common/objects/allowedExperimentStationExperimentType.py

+#!/usr/bin/env python
+
+from dmObject import DmObject
+
+class AllowedExperimentStationExperimentType(DmObject):
+
+    DEFAULT_KEY_LIST = [ 'experimentStationId', 'experimentTypeId' ]
+
+    def __init__(self, dict):
+        DmObject.__init__(self, dict)
+

src/python/dm/common/objects/apsUserInfo.py

+#!/usr/bin/env python
+
+from dmObject import DmObject
+
+class ApsUserInfo(DmObject):
+
+    DEFAULT_KEY_LIST = [ 'id', 'badgeNumber', 'firstName', 'middleName', 'lastName', 'email', 'passwordHashValue' ]
+
+    def __init__(self, dict):
+        DmObject.__init__(self, dict)
+

src/python/dm/common/objects/authorizationPrincipal.py

+#!/usr/bin/env python
+
+from dmObject import DmObject
+
+class AuthorizationPrincipal(DmObject):
+
+    def __init__(self, dict={}, name=None, token=None, userInfo={}):
+        DmObject.__init__(self, dict)
+        if name is not None:
+            self['name'] = name
+        if token is not None:
+            self['token'] = token
+        if userInfo is not None and len(userInfo):
+            self['userInfo'] = userInfo
+
+    def getName(self):
+        return self.get('name')
+
+    def getAuthenticationToken(self):
+        return self.get('token')
+
+    def getToken(self):
+        return self.get('token')
+
+    def setSessionRole(self, role):
+        self['sessionRole'] = role
+
+    def getSessionRole(self):
+        return self.get('sessionRole')
+
+    def setUserInfo(self, userInfo):
+        self['userInfo'] = userInfo
+
+    def getUserInfo(self):
+        return self.get('userInfo')
+
+    def setUserSystemRoleDict(self, userSystemRoleDict):
+        self['userSystemRoleDict'] = userSystemRoleDict
+
+    def getUserSystemRoleDict(self):
+        return self.get('userSystemRoleDict')
+
+    def setUserExperimentRoleDict(self, userExperimentRoleDict):
+        self['userExperimentRoleDict'] = userExperimentRoleDict
+
+    def getUserExperimentRoleDict(self):
+        return self.get('userExperimentRoleDict')
+

src/python/dm/common/objects/beamlineInfo.py

+#!/usr/bin/env python
+
+from dmObject import DmObject
+
+class BeamlineInfo(DmObject):
+
+    DEFAULT_KEY_LIST = [ 'id', 'name' ]
+
+    def __init__(self, dict):
+        DmObject.__init__(self, dict)
+

src/python/dm/common/objects/childProcess.py

+#!/usr/bin/env python
+
+from dmObject import DmObject
+
+class ChildProcess(DmObject):
+
+    DEFAULT_KEY_LIST = [ 'command', 'exitStatus', 'stdErr', 'stdOut', 'workingDir', 'childProcessNumber', 'stageId', 'status' ]
+
+    def __init__(self, dict):
+        DmObject.__init__(self, dict)
+

src/python/dm/common/objects/daqInfo.py

+#!/usr/bin/env python
+
+import copy
+import time
+import threading
+
+from dmObject import DmObject
+from dm.common.constants import dmProcessingStatus
+from dm.common.utility.dictUtility import DictUtility
+from dm.common.utility.timeUtility import TimeUtility
+
+class DaqInfo(DmObject):
+
+    DEFAULT_KEY_LIST = [ 'id', 'experimentName', 'dataDirectory', 'status', 'nProcessedFiles', 'nProcessingErrors', 'nCompletedFiles', 'nWaitingFiles', 'nFiles', 'startTime', 'endTime', 'runTime', 'startTimestamp', 'endTimestamp', 'errorMessage' ]
+    
+
+    def __init__(self, dict={}):
+        DmObject.__init__(self, dict)
+        self.lock = threading.RLock()
+        self.originalKeys = dict.keys()
+
+    def fileAdded(self, filePath):
+        self.lock.acquire()
+        try:
+            self['nFiles'] = self.get('nFiles', 0) + 1
+        finally:
+            self.lock.release()
+
+    def fileProcessed(self, filePath, processingEndTime):
+        self.lock.acquire()
+        try:
+            self['nProcessedFiles'] = self.get('nProcessedFiles', 0) + 1
+            lastFileProcessedTime = self.get('lastFileProcessedTime', 0)
+            if processingEndTime is not None and processingEndTime > lastFileProcessedTime:
+                self['lastFileProcessed'] = filePath
+                self['lastFileProcessedTime'] = processingEndTime
+        finally:
+            self.lock.release()
+
+    def fileProcessingError(self, filePath, processingError, processingEndTime):
+        self.lock.acquire()
+        try:
+            # file can be processed multiple times, keep only the last error
+            self['nProcessingErrors'] = self.get('nProcessingErrors', 0) + 1
+            processingErrors = self.get('processingErrors', {})
+            processingErrors[filePath] = processingError 
+            self['processingErrors'] = processingErrors
+            lastFileProcessingErrorTime = self.get('lastFileProcessingErrorTime', 0)
+            if processingEndTime is not None and processingEndTime > lastFileProcessingErrorTime:
+                self['lastFileProcessingErrorTime'] = processingEndTime
+        finally:
+            self.lock.release()
+
+    def fileProcessingSkipped(self, processorName, filePath, processingError, processingEndTime):
+        self.lock.acquire()
+        try:
+            pluginStatsDict = self.get('pluginStats', {})
+            self['pluginStats'] = pluginStatsDict
+
+            pluginStats = pluginStatsDict.get(processorName, {})
+            pluginStatsDict[processorName] = pluginStats
+
+            pluginStats['nSkippedFiles'] = pluginStats.get('nSkippedFiles', 0) + 1
+            if processingError:
+                processingErrors = pluginStats.get('processingErrors', {})
+                processingErrors[filePath] = processingError
+                pluginStats['processingErrors'] = processingErrors
+            lastFileSkippedTime = pluginStats.get('lastFileSkippedTime', 0)
+            if processingEndTime is not None and processingEndTime > lastFileSkippedTime:
+                pluginStats['lastFileSkippedTime'] = processingEndTime
+        finally:
+            self.lock.release()
+
+    def updateStatus(self):
+        now = time.time()
+        daqStatus = self.get('status', dmProcessingStatus.DM_PROCESSING_STATUS_RUNNING)
+        if daqStatus in dmProcessingStatus.DM_INACTIVE_PROCESSING_STATUS_LIST:
+            return 
+        nFiles = self.get('nFiles', 0)
+        nProcessedFiles = self.get('nProcessedFiles', 0)
+        nProcessingErrors = self.get('nProcessingErrors', 0)
+        processingErrors = self.get('processingErrors', {})
+        nCompletedFiles = nProcessedFiles+nProcessingErrors
+        self['nCompletedFiles'] = nCompletedFiles
+        nWaitingFiles = nFiles-nCompletedFiles
+        self['nWaitingFiles'] = nWaitingFiles
+
+        startTime = self.get('startTime', now)
+        runTime = now - startTime
+        endTime = None
+
+        percentageComplete = 100.0
+        percentageProcessed = 100.0
+        percentageProcessingErrors = 0.0
+        if nFiles > 0:
+             percentageComplete = float(nCompletedFiles)/float(nFiles)*100.0
+             percentageProcessed = float(nProcessedFiles)/float(nFiles)*100.0
+             percentageProcessingErrors = float(nProcessingErrors)/float(nFiles)*100.0
+        self['percentageComplete'] = '%.2f' % percentageComplete
+        self['percentageProcessed'] = '%.2f' % percentageProcessed
+        self['percentageProcessingErrors'] = '%.2f' % percentageProcessingErrors
+
+        if self.get('endTime'):
+            daqStatus = dmProcessingStatus.DM_PROCESSING_STATUS_FINALIZING
+            if nCompletedFiles >= nFiles:
+                daqStatus = dmProcessingStatus.DM_PROCESSING_STATUS_DONE
+                if nProcessingErrors:
+                    daqStatus = dmProcessingStatus.DM_PROCESSING_STATUS_FAILED
+            lastFileProcessingErrorTime = self.get('lastFileProcessingErrorTime')
+            lastFileProcessedTime = self.get('lastFileProcessedTime')
+            endTime = lastFileProcessedTime
+            if not endTime or lastFileProcessingErrorTime > endTime:
+                endTime = lastFileProcessingErrorTime
+            if self.get('endTime') > endTime:
+                endTime = self.get('endTime')
+
+        if endTime:
+            runTime = endTime - startTime
+            self['endTime'] = endTime
+            self['endTimestamp'] = TimeUtility.formatLocalTimestamp(endTime)
+        self['runTime'] = runTime
+        self['status'] = daqStatus
+
+    def toDictWithOriginalKeys(self):
+        dict = {}
+        for key in self.originalKeys:
+            if self.has_key(key):
+                dict[key] = self.get(key)
+        return dict
+

src/python/dm/common/objects/dataFolder.py

+#!/usr/bin/env python
+
+import time
+from dmObject import DmObject
+
+class DataFolder(DmObject):
+
+    DEFAULT_KEY_LIST = [ 'id', 'name', 'description', 'storageId', 'experimentId', 'dataPath' ]
+
+    def __init__(self, dict={}):
+        DmObject.__init__(self, dict)
+
+####################################################################
+# Testing
+if __name__ == '__main__':
+    pass
+

src/python/dm/common/objects/datasetMetadata.py

+#!/usr/bin/env python
+
+from dmObject import DmObject
+
+class DatasetMetadata(DmObject):
+
+    DEFAULT_KEY_LIST = [ 'id', 'datasetName', 'experimentName' ]
+
+    def __init__(self, dict):
+        DmObject.__init__(self, dict)
+

src/python/dm/common/objects/directoryMetadata.py

+#!/usr/bin/env python
+
+from dmObject import DmObject
+
+class DirectoryMetadata(DmObject):
+
+    DEFAULT_KEY_LIST = [ 'id', 'directoryPath', 'experimentDirectoryPath' ]
+
+    def __init__(self, dict):
+        DmObject.__init__(self, dict)
+

src/python/dm/common/objects/directoryUploadInfo.py

+#!/usr/bin/env python
+
+import time
+import threading
+
+from dmObject import DmObject
+from dm.common.constants import dmProcessingStatus
+from dm.common.utility.dictUtility import DictUtility
+from dm.common.utility.timeUtility import TimeUtility
+
+class DirectoryUploadInfo(DmObject):
+
+    DEFAULT_KEY_LIST = [ 'id', 'experimentName', 'dataDirectory', 'status', 'nFiles', 'startTime', 'endTime', 'runTime', 'startTimestamp', 'endTimestamp', 'errorMessage' ]
+
+    def __init__(self, dict={}):
+        DmObject.__init__(self, dict)
+        self.lock = threading.RLock()
+
+    def updateStatus(self):
+        now = time.time()
+        uploadStatus = self.get('status', dmProcessingStatus.DM_PROCESSING_STATUS_RUNNING)
+        if uploadStatus in dmProcessingStatus.DM_INACTIVE_PROCESSING_STATUS_LIST:
+            return 
+
+        startTime = self.get('startTime', now)
+        runTime = now - startTime
+
+        processingInfo = self.get('processingInfo')
+        endTime = 0
+        uploadStatus = dmProcessingStatus.DM_PROCESSING_STATUS_DONE
+        for processorName in processingInfo.keys():
+            processingEndTime = processingInfo[processorName].get('processingEndTime')
+            status = processingInfo[processorName].get('status')
+            if status in [dmProcessingStatus.DM_PROCESSING_STATUS_ABORTED, dmProcessingStatus.DM_PROCESSING_STATUS_FAILED]:
+                uploadStatus = status
+
+            if not processingEndTime and status != dmProcessingStatus.DM_PROCESSING_STATUS_SKIPPED:
+                endTime = None
+                break
+
+            if processingEndTime > endTime:
+                endTime = processingEndTime
+        
+        if endTime:
+            runTime = endTime - startTime
+            self['endTime'] = endTime
+            self['endTimestamp'] = TimeUtility.formatLocalTimestamp(endTime)
+            self['status'] = uploadStatus
+        self['runTime'] = runTime
+
+
+

src/python/dm/common/objects/dmObject.py

+#!/usr/bin/env python
+
+#
+# DM Object class.
+#
+
+#######################################################################
+
+import UserDict
+import UserList
+import types
+import json
+import datetime
+
+from dm.common.exceptions.invalidArgument import InvalidArgument
+from dm.common.exceptions.objectNotFound import ObjectNotFound
+from dm.common.utility import loggingManager
+
+class DmObject(UserDict.UserDict):
+    """ Base dm object class. """
+    ALL_KEYS = 'ALL'
+    DEFAULT_KEYS = 'DEFAULT'
+
+    DICT_DISPLAY_FORMAT = 'dict'
+    TEXT_DISPLAY_FORMAT = 'text'
+    JSON_DISPLAY_FORMAT = 'json'
+
+    DEFAULT_KEY_LIST = [ 'id', 'name' ]
+
+    def __init__(self, dict={}):
+        if isinstance(dict, types.DictType): 
+            UserDict.UserDict.__init__(self, dict)
+        elif isinstance(dict, UserDict.UserDict):
+            UserDict.UserDict.__init__(self, dict.data)
+        else:
+            raise InvalidArgument('DmObject instance must be initialized using dictionary.')
+        self.logger = None
+
+    def getLogger(self):
+        if not self.logger:
+            self.logger = loggingManager.getLogger(self._class__.__name__)
+        return self.logger
+
+    def getRequiredKeyValue(self, key):
+        value = self.get(key)
+        if value is None:
+            errorMsg = 'Required dictionary key %s is missing.' % key
+            raise ObjectNotFound(errorMsg)
+
+    @classmethod
+    def getFromDict(cls, dict):
+        inst = cls()
+        for key in dict.keys():
+            inst[key] = dict[key]
+        return inst
+
+    def getRepKeyList(self, keyList):
+        if keyList is None:
+            return self.DEFAULT_KEY_LIST
+        elif type(keyList) == types.ListType:
+            if not len(keyList):
+                return self.DEFAULT_KEY_LIST
+            else:
+                return keyList
+        elif type(keyList) == types.StringType:
+            if keyList == DmObject.ALL_KEYS:
+                return self.data.keys()
+            elif keyList == DmObject.DEFAULT_KEYS:
+                return self.DEFAULT_KEY_LIST
+            else:
+                # Assume keys are separated by comma
+                return keyList.split(',')
+        else: 
+            # Unknown key list parameter.
+            raise InvalidArgument('Key list parameter must be one of: None, string "%s", string "%s", string containing comma-separated keys, or list of strings.' (DmObject.ALL_KEYS, DmObject.DEFAULT_KEYS))
+            
+
+    def getDictRep(self, keyList=None):
+        # Dict representation is dict
+        dictRep = {}
+        displayKeyList = self.getRepKeyList(keyList)
+        for key in displayKeyList:
+            value = self.get(key)
+            if isinstance(value, DmObject):
+                dictRep[key] = value.getDictRep('ALL')
+            elif type(value) == types.ListType:
+                itemList = []
+                for item in value:
+                    if isinstance(item, DmObject):
+                        itemList.append(item.getDictRep('ALL'))
+                    else:
+                        itemList.append(item)
+                dictRep[key] = itemList
+            else:
+                if value is not None:
+                    if isinstance(value, datetime.datetime):
+                        dictRep[key] = str(value)
+                    else:
+                        dictRep[key] = value
+        return dictRep
+
+    def getTextRep(self, keyList=None):
+        display = ''
+        displayKeyList = self.getRepKeyList(keyList)
+        for key in displayKeyList:
+            value = self.get(key)
+            if isinstance(value, DmObject):
+                display = display + '%s={ %s} ' % (key, value.getTextRep())
+            elif isinstance(value, types.ListType):
+                display = display + '%s=[ ' % key
+                for item in value:
+                    if isinstance(item, DmObject):
+                        display = display + '{ %s}, ' % (item)
+                    else:
+                        display = display + ' %s, ' % (item)
+                display = display + '] '
+            else:
+                if value is not None:
+                    display = display + '%s=%s ' % (key, value)
+        return display
+
+    def getJsonRep(self, keyList=None):
+        dictRep = self.getDictRep(keyList)
+        return json.dumps(dictRep)
+
+    def getFullJsonRep(self):
+        dictRep = self.getDictRep(DmObject.ALL_KEYS)
+        return json.dumps(dictRep)
+
+    @classmethod 
+    def fromJsonString(cls, jsonString):
+        return cls.getFromDict(json.loads(jsonString))
+
+    def getDisplayString(self, displayKeyList=[], displayFormat=TEXT_DISPLAY_FORMAT):
+        """ Get display string. """
+        if displayFormat == DmObject.DICT_DISPLAY_FORMAT:
+            return self.getDictRep(displayKeyList)
+        elif displayFormat == DmObject.TEXT_DISPLAY_FORMAT:
+            return self.getTextRep(displayKeyList)
+        elif displayFormat == DmObject.JSON_DISPLAY_FORMAT:
+            return self.getJsonRep(displayKeyList)
+        raise InvalidArgument('Unrecognized display displayFormat: %s.' (displayFormat))
+
+
+#######################################################################
+# Testing.
+
+if __name__ == '__main__':
+    x = {'name' : 'XYZ', 'one':1, 'two':2 }
+    o = DmObject(x)
+    print 'DM Object: ', o
+    print 'Type of DM object: ', type(o)
+    print 'JSON Rep:  ', o.getJsonRep()
+    print 'Type of JSON rep: ', type(o.getJsonRep())
+    j = '{"name" : "XYZ", "one":1, "two":2 }'
+    print 'String: ', j
+    x2 = DmObject.fromJsonString(j)
+    print 'DM Object 2: ', x2
+    print 'Type of DM object 2: ', type(x2)
+    print x2.getDisplayString(displayKeyList='ALL')
+
+