src/python/dm/common/utility/osUtility.py

+#!/usr/bin/env python
+
+import os
+
+class OsUtility:
+
+    @classmethod
+    def createDir(cls, path, mode=None):
+        """ Create directory if it does not exist already. """
+        if not os.path.isdir(path):
+            os.makedirs(path)
+            if mode is not None:
+                os.chmod(path, mode)
+
+    @classmethod
+    def removeLink(cls, path):
+        """ Remove link on a given path. """
+        if not os.path.islink(path):
+            return
+        os.remove(path)
+
+    @classmethod
+    def removeFile(cls, path):
+        """ Remove file on a given path. """
+        if not os.path.isfile(path):
+            return
+        os.remove(path)
+
+    @classmethod
+    def removeAndIgnoreErrors(cls, path):
+        """ Remove file on a given path and ignore any errors. """
+        try:
+            os.remove(path)
+        except Exception, ex:
+            pass
+
+    @classmethod
+    def removeDir(cls, path):
+        """ Remove dir on a given path, even if it is not empty. """
+        if not os.path.isdir(path):
+            return
+        files=os.listdir(path)
+        for f in files:
+            fullpath=os.path.join(path, f)
+            if os.path.islink(fullpath) or not os.path.isdir(fullpath):
+                os.remove(fullpath)
+            else:
+                removeDir(fullpath)
+        os.rmdir(path)
+        
+    @classmethod
+    def chownPath(cls, path, uid, gid):
+        """ Change owner on a given path recursively.  """
+        if os.path.isfile(path):
+            os.chown(path, uid, gid)
+            return 
+        elif os.path.islink(path):
+            os.lchown(path, uid, gid)
+            return 
+        elif os.path.isdir(path):
+            files=os.listdir(path)
+            for f in files:
+                fullpath=os.path.join(path, f)
+                chownPath(fullpath, uid, gid)
+            os.chown(path, uid, gid)
+
+    @classmethod
+    def chownPathByUserName(cls, path, userName):
+        """ Change owner on a given path recursively.  """
+        import pwd
+        user = pwd.getpwnam(userName)
+        chownPath(path, user.pw_uid, user.pw_gid)
+
+    @classmethod
+    def findFiles(cls, dirPath, fileList=None):
+        """ List files in a given directory. Return list of absolute paths.
+            Do not follow symbolic links.
+        """
+        fList = fileList
+        if not fList:
+            fList = []
+        if os.path.isdir(dirPath):
+            files = os.listdir(dirPath)
+            for f in files:
+                fullPath = os.path.join(dirPath, f)
+                if os.path.isfile(fullPath):
+                    fList.append(fullPath)
+                elif os.path.isdir(fullPath):
+                    fList = findFiles(fullPath, fList)
+        return fList
+
+    @classmethod
+    def importNameFromFile(cls, name, filePath):
+        """ Import specified name from file. """
+        import sys
+        import os.path
+        dirName = os.path.dirname(filePath)
+        moduleName = os.path.basename(filePath).replace('.py', '')
+        sys.path = [dirName] + sys.path
+        cmd = 'from %s import %s as tmpObject' % (moduleName, name)
+        exec cmd
+        del sys.path[0]
+        return tmpObject
+
+    @classmethod
+    def getUserHomeDir(cls):
+        """ Get current user home directory. """
+        from os.path import expanduser
+        home = expanduser('~')
+        return home
+
+#######################################################################
+# Testing.
+
+if __name__ == '__main__':
+    pass

src/python/dm/common/utility/singleton.py

+#!/usr/bin/env python
+
+class Singleton(object):
+
+    __instance = None
+
+    def __new__(cls, *args, **kwargs):
+        # Allow subclasses to create their own instances.
+        if cls.__instance is None or cls != type(cls.__instance):
+            instance = object.__new__(cls, *args, **kwargs)
+            cls.__instance = instance
+            instance.__init__(*args, **kwargs)
+        return cls.__instance
+
+    @classmethod
+    def getInstance(cls, *args, **kwargs):
+        return cls.__new__(cls, *args, **kwargs)
+
+    def __init__(self, *args, **kwargs):
+        # Only initialize once.
+        if Singleton.__instance is not None:
+            return
+
+####################################################################
+# Testing
+if __name__ == '__main__':
+    s1 = Singleton.getInstance()
+    s2 = Singleton()
+    s3 = Singleton.getInstance()
+    s4 = Singleton()
+    print 'S1: ', s1
+    print 'S2: ', s2
+    print 'S3: ', s3
+    print 'S4: ', s4
+    class A(Singleton):
+        def __init__(self, x):
+            self.x = x
+    class B(Singleton):
+        def __init__(self, x):
+            self.x = x
+    class C(Singleton):
+        def __init__(self):
+            self.x = 14
+
+    a1 = A(3)
+    a2 = A(4)
+    print a1
+    print a2
+    print a2.x, a1.x
+
+    b1 = B(6)
+    b2 = B(5)
+    print b1
+    print b2
+    print b2.x, b1.x
+
+    c1 = C()
+    c2 = C()
+    print c1
+    print c2
+    print c2.x, c1.x

src/python/dm/common/utility/sslUtility.py

+#!/usr/bin/env python
+
+import ssl
+
+class SslUtility:
+
+    DEFAULT_SSL_CONTEXT = ssl._create_default_https_context
+
+    @classmethod
+    def useUnverifiedSslContext(cls, func):
+
+        def wrapper(*args, **kwargs):
+            # Disable SSL checking
+            ssl._create_default_https_context = ssl._create_unverified_context
+
+            # Perform function call 
+            result = func(*args, **kwargs)
+
+            # Revert back to original SSL settings
+            ssl._create_default_https_context = SslUtility.DEFAULT_SSL_CONTEXT 
+            return result
+
+        return wrapper
+

src/python/dm/common/utility/threadSafeQueue.py

+#!/usr/bin/env python
+
+import threading
+
+class ThreadSafeQueue:
+
+    def __init__(self):
+        self.lock = threading.RLock()
+        self.queue = []
+
+    def push(self, item):
+        self.lock.acquire()
+        try:
+            self.queue.insert(0,item)
+        finally:
+            self.lock.release()
+
+    def pop(self):
+        # Return None if work queue is empty.
+        self.lock.acquire()
+        try:
+            item = None
+            if len(self.queue):
+                item = self.queue.pop()
+            return item
+        finally:
+            self.lock.release()
+
+    def getLength(self):
+        return len(self.queue)
+
+    def isEmpty(self):
+        return len(self.queue) == 0
+
+####################################################################
+# Testing
+
+if __name__ == '__main__':
+    q = ThreadSafeQueue()
+    for i in range(0,10):
+        q.push(i)
+        print 'Added: ', i
+
+    while not q.isEmpty():
+        i = q.pop() 
+        print 'Got: ', i
+
+
+

src/python/dm/common/utility/timeUtility.py

+#!/usr/bin/env python
+
+import time
+class TimeUtility:
+
+    @classmethod
+    def getCurrentGMTimeStamp(cls):
+        """ Formats GMT timestamp. """
+        return time.strftime('%Y-%m-%d %H:%M:%S', time.gmtime(time.time()))
+
+    @classmethod
+    def formatGMTimeStamp(cls, t):
+        """ Format GMT timestamp. """
+        return time.strftime('%Y-%m-%d %H:%M:%S', time.gmtime(t))
+
+    @classmethod
+    def getCurrentLocalTimeStamp(cls):
+        """ Formats local timestamp. """
+        return time.strftime('%Y/%m/%d %H:%M:%S %Z', time.localtime(time.time()))
+
+    @classmethod
+    def formatLocalTimeStamp(cls, t):
+        """ Formats local timestamp. """
+        return time.strftime('%Y/%m/%d %H:%M:%S %Z', time.localtime(t))
+
+

src/python/dm/common/utility/timedRotatingFileLoggingHandler.py

+#!/usr/bin/env python
+
+#######################################################################
+
+import socket
+import pwd
+import os
+from logging.handlers import TimedRotatingFileHandler
+
+#######################################################################
+
+class TimedRotatingFileLoggingHandler(TimedRotatingFileHandler):
+    """ Class that enables logging into files. """
+    def __init__(self, filename, when='D', interval=1, backupCount=0, encoding=None):
+        TimedRotatingFileHandler.__init__(self, filename, when, interval, backupCount, encoding)
+        self.user = pwd.getpwuid(os.getuid())[0]
+        self.host = socket.gethostname()
+
+    def emit(self, record):
+        record.__dict__['user'] = self.user
+        record.__dict__['host'] = self.host
+        return TimedRotatingFileHandler.emit(self, record)
+                                                                                
+#######################################################################
+# Testing.
+
+if __name__ == '__main__':
+    pass

src/python/dm/daq_web_service/api/daqRestApi.py

+#!/usr/bin/env python
+
+from dm.common.api.dmRestApi import DmRestApi
+
+class DaqRestApi(DmRestApi):
+    """ Base DAQ DM REST api class. """
+
+    def __init__(self, username=None, password=None, host=None, port=None, protocol=None):
+        if host == None:
+            host = self.configurationManager.getDaqWebServiceHost()
+        if port == None:            
+            port = self.configurationManager.getDaqWebServicePort()
+        DmRestApi.__init__(self, username, password, host, port, protocol)
+
+#######################################################################
+# Testing.
+
+if __name__ == '__main__':
+    api = DaqRestApi('sveseli', 'sveseli')
+    #api.sendRequest('https://zagreb.svdev.net:10232/dm/directory/list', 'GET', data='')
+    import urllib
+    from dm.common.utility.configurationManager import ConfigurationManager
+    cm = ConfigurationManager.getInstance()
+    cm.setSessionCacheFile('/tmp/session')
+    #print 'Non-session request'
+    #print api.sendRequest('https://zagreb.svdev.net:10232/dm/directory/list?path=/tmp', 'GET')
+    print 'Session request'
+    data = { 'path' : '/tmp/xyz' }
+    #print api.sendSessionRequest('https://zagreb.svdev.net:10232/dm/file/write?path=/tmp/xyz&content=xyz', 'POST', contentType='application/x-www-form-urlencoded', data=urllib.urlencode(data))
+    #print api.sendSessionRequest('https://zagreb.svdev.net:10232/dm/file/write', 'POST', data=data)
+    postdata='path=/tmp/xyz'
+    postdata+='&content=%s' % urllib.quote_plus('Hey there')
+    print api.sendSessionRequest('https://zagreb.svdev.net:10232/dm/file/write', 'POST', contentType='application/x-www-form-urlencoded', data=postdata)
+
+
+

src/python/dm/daq_web_service/api/experimentRestApi.py

+#!/usr/bin/env python
+
+import os
+import urllib
+
+from dm.common.utility.encoder import Encoder
+from dm.common.exceptions.dmException import DmException
+from dm.common.objects.experiment import Experiment
+from daqRestApi import DaqRestApi
+
+class ExperimentRestApi(DaqRestApi):
+    
+    def __init__(self, username=None, password=None, host=None, port=None, protocol=None):
+        DaqRestApi.__init__(self, username, password, host, port, protocol)
+
+    @DaqRestApi.execute
+    def startDaq(self, name, dataDirectory):
+        url = '%s/experiments/startDaq' % (self.getContextRoot())
+        if name is None or not len(name):
+            raise InvalidRequest('Experiment name must be provided.')
+        url += '?name=%s' % Encoder.encode(name)
+        if dataDirectory is None or not len(dataDirectory):
+            raise InvalidRequest('Experiment data directory must be provided.')
+        url += '&dataDirectory=%s' % Encoder.encode(dataDirectory)
+        responseDict = self.sendSessionRequest(url=url, method='POST')
+        return Experiment(responseDict)
+
+    @DaqRestApi.execute
+    def stopDaq(self, name):
+        url = '%s/experiments/stopDaq' % (self.getContextRoot())
+        if name is None or not len(name):
+            raise InvalidRequest('Experiment name must be provided.')
+        url += '?name=%s' % Encoder.encode(name)
+        responseDict = self.sendSessionRequest(url=url, method='POST')
+        return Experiment(responseDict)
+
+#######################################################################
+# Testing.
+
+if __name__ == '__main__':
+    api = ExperimentRestApi('sveseli', 'sveseli', 'zagreb.svdev.net', 33336, 'http')
+    print api.startDaq('experiment1', '/tmp/data/experiment1')
+

src/python/dm/daq_web_service/cli/daqWebServiceCli.py

+#!/usr/bin/env python
+
+from dm.common.cli.dmRestCli import DmRestCli
+from dm.common.utility.configurationManager import ConfigurationManager
+
+class DaqWebServiceCli(DmRestCli):
+    """ DM DAQ web service cli class. """
+
+    def __init__(self, validArgCount=0):
+        DmRestCli.__init__(self, validArgCount)
+
+    def getDefaultServiceHost(self):
+        return ConfigurationManager.getInstance().getDaqWebServiceHost()
+
+    def getDefaultServicePort(self):
+        return ConfigurationManager.getInstance().getDaqWebServicePort()
+                        

src/python/dm/daq_web_service/cli/daqWebServiceSessionCli.py

+#!/usr/bin/env python
+
+from dm.common.cli.dmRestSessionCli import DmRestSessionCli
+from dm.common.utility.osUtility import OsUtility
+from dm.common.utility.configurationManager import ConfigurationManager
+
+class DaqWebServiceSessionCli(DmRestSessionCli):
+    """ DM DAQ web service session cli class. """
+
+    DEFAULT_SESSION_CACHE_FILE = OsUtility.getUserHomeDir() + '/.dm/.daq.session.cache'
+
+    def __init__(self, validArgCount=0):
+        DmRestSessionCli.__init__(self, validArgCount)
+        ConfigurationManager.getInstance().setSessionCacheFile(DaqWebServiceSessionCli.DEFAULT_SESSION_CACHE_FILE)
+
+    def getDefaultServiceHost(self):
+        return ConfigurationManager.getInstance().getDaqWebServiceHost()
+
+    def getDefaultServicePort(self):
+        return ConfigurationManager.getInstance().getDaqWebServicePort()
+

src/python/dm/daq_web_service/cli/startDaqCli.py

+#!/usr/bin/env python
+
+from dm.daq_web_service.api.experimentRestApi import ExperimentRestApi
+from dm.common.exceptions.invalidRequest import InvalidRequest
+from daqWebServiceSessionCli import DaqWebServiceSessionCli
+
+class StartDaqCli(DaqWebServiceSessionCli):
+    def __init__(self):
+        DaqWebServiceSessionCli.__init__(self)
+        self.addOption('', '--experiment', dest='experimentName', help='Experiment name.')
+        self.addOption('', '--data-directory', dest='dataDirectory', help='Experiment data directory.')
+
+    def checkArgs(self):
+        if self.options.experimentName is None:
+            raise InvalidRequest('Experiment name must be provided.')
+        if self.options.dataDirectory is None:
+            raise InvalidRequest('Experiment data directory must be provided.')
+
+    def getExperimentName(self):
+        return self.options.experimentName
+
+    def getDataDirectory(self):
+        return self.options.dataDirectory
+
+    def runCommand(self):
+        self.parseArgs(usage="""
+    dm-start-daq --experiment=EXPERIMENTNAME --data-directory=DATADIRECTORY
+
+Description:
+    Starts DAQ for a given experiment. Provided data directory will be monitored for data files.
+        """)
+        self.checkArgs()
+        api = ExperimentRestApi(self.getLoginUsername(), self.getLoginPassword(), self.getServiceHost(), self.getServicePort(), self.getServiceProtocol())
+        experiment = api.startDaq(self.getExperimentName(), self.getDataDirectory())
+        print experiment.getDisplayString(self.getDisplayKeys(), self.getDisplayFormat())
+
+#######################################################################
+# Run command.
+if __name__ == '__main__':
+    cli = StartDaqCli()
+    cli.run()
+

src/python/dm/daq_web_service/cli/stopDaqCli.py

+#!/usr/bin/env python
+
+from dm.daq_web_service.api.experimentRestApi import ExperimentRestApi
+from dm.common.exceptions.invalidRequest import InvalidRequest
+from daqWebServiceSessionCli import DaqWebServiceSessionCli
+
+class StopDaqCli(DaqWebServiceSessionCli):
+    def __init__(self):
+        DaqWebServiceSessionCli.__init__(self)
+        self.addOption('', '--experiment', dest='experimentName', help='Experiment name.')
+
+    def checkArgs(self):
+        if self.options.experimentName is None:
+            raise InvalidRequest('Experiment name must be provided.')
+
+    def getExperimentName(self):
+        return self.options.experimentName
+
+    def runCommand(self):
+        self.parseArgs(usage="""
+    dm-stop-daq --experiment=EXPERIMENTNAME
+
+Description:
+    Stop DAQ for a given experiment. 
+        """)
+        self.checkArgs()
+        api = ExperimentRestApi(self.getLoginUsername(), self.getLoginPassword(), self.getServiceHost(), self.getServicePort(), self.getServiceProtocol())
+        experiment = api.stopDaq(self.getExperimentName())
+        print experiment.getDisplayString(self.getDisplayKeys(), self.getDisplayFormat())
+
+#######################################################################
+# Run command.
+if __name__ == '__main__':
+    cli = StopDaqCli()
+    cli.run()
+

src/python/dm/daq_web_service/cli/uploadDataCli.py

+#!/usr/bin/env python
+
+from daqWebServiceSessionCli import DaqWebServiceSessionCli
+from dm.daq_web_service.api.experimentRestApi import ExperimentRestApi
+from dm.common.exceptions.invalidRequest import InvalidRequest
+
+class UploadDataCli(DaqWebServiceSessionCli):
+    def __init__(self):
+        DaqWebServiceCli.__init__(self)
+        self.addOption('', '--experiment', dest='experimentName', help='Experiment name.')
+        self.addOption('', '--data-directory', dest='dataDirectory', help='Experiment data directory.')
+
+    def checkArgs(self):
+        if self.options.experimentName is None:
+            raise InvalidRequest('Experiment name must be provided.')
+        if self.options.dataDirectory is None:
+            raise InvalidRequest('Experiment data directory must be provided.')
+
+    def getExperimentName(self):
+        return self.options.experimentName
+
+    def getDataDirectory(self):
+        return self.options.dataDirectory
+
+    def runCommand(self):
+        self.parseArgs(usage="""
+    dm-upload-data --name=EXPERIMENTNAME --data-directory=DATADIRECTORY
+
+Description:
+    Schedules data upload for a given experiment. 
+        """)
+        self.checkArgs()
+        api = ExperimentRestApi(self.getLoginUsername(), self.getLoginPassword(), self.getServiceHost(), self.getServicePort(), self.getServiceProtocol())
+        experiment = api.uploadData(self.getExperimentName(), self.getDataDirectory())
+        print experiment.getDisplayString(self.getDisplayKeys(), self.getDisplayFormat())
+
+#######################################################################
+# Run command.
+if __name__ == '__main__':
+    cli = UploadDataCli()
+    cli.run()
+

src/python/dm/daq_web_service/service/daqWebService.py

+#!/usr/bin/env python
+
+#
+# DM DAQ Web Service
+#
+
+from dm.common.service.dmRestWebServiceBase import DmRestWebServiceBase
+from dm.common.utility.dmModuleManager import DmModuleManager
+from dm.common.utility.configurationManager import ConfigurationManager
+from dm.daq_web_service.service.impl.fileSystemObserver import FileSystemObserver 
+from dm.daq_web_service.service.impl.fileProcessingManager import FileProcessingManager
+from daqWebServiceRouteMapper import DaqWebServiceRouteMapper
+
+class DaqWebService(DmRestWebServiceBase):
+ 
+    def __init__(self):
+        DmRestWebServiceBase.__init__(self, DaqWebServiceRouteMapper)
+
+    def initDmModules(self):
+        self.logger.debug('Initializing dm modules')
+
+        # Add modules that will be started.
+        moduleManager = DmModuleManager.getInstance()
+        moduleManager.addModule(FileSystemObserver.getInstance())
+        moduleManager.addModule(FileProcessingManager.getInstance())
+        self.logger.debug('Initialized dm modules')
+
+    def getDefaultServerHost(self):
+        return ConfigurationManager.getInstance().getServiceHost()
+
+    def getDefaultServerPort(self):
+        return ConfigurationManager.getInstance().getServicePort()
+
+####################################################################
+# Run service
+
+if __name__ == '__main__':
+    ConfigurationManager.getInstance().setServiceName('daq-web-service')
+    service = DaqWebService();
+    service.run()

src/python/dm/daq_web_service/service/daqWebServiceRouteMapper.py

+#!/usr/bin/env python
+
+#
+# Route mapper for DM DAQ web service.
+#
+
+import sys
+import os
+
+import cherrypy
+from dm.common.utility.loggingManager import LoggingManager
+from dm.common.utility.configurationManager import ConfigurationManager
+from dm.common.service.loginRouteDescriptor import LoginRouteDescriptor
+from experimentRouteDescriptor import ExperimentRouteDescriptor
+
+class DaqWebServiceRouteMapper:
+
+    @classmethod
+    def setupRoutes(cls):
+        """ Setup RESTFul routes. """
+        logger = LoggingManager.getInstance().getLogger(cls.__name__)
+        contextRoot = ConfigurationManager.getInstance().getContextRoot()
+        logger.debug('Using context root: %s' % contextRoot)
+
+        # Get routes.
+        routes = LoginRouteDescriptor.getRoutes() 
+        routes += ExperimentRouteDescriptor.getRoutes()
+
+        # Add routes to dispatcher. 
+        d = cherrypy.dispatch.RoutesDispatcher()
+        for route in routes:
+            logger.debug('Connecting route: %s' % route)
+            d.connect(route['name'], route['path'], action=route['action'], controller=route['controller'], conditions=dict(method=route['method']))
+        return d
+
+

src/python/dm/daq_web_service/service/experimentRouteDescriptor.py

+#!/usr/bin/env python
+
+#
+# User route descriptor.
+#
+
+from dm.common.utility.configurationManager import ConfigurationManager
+from experimentSessionController import ExperimentSessionController
+
+class ExperimentRouteDescriptor:
+
+    @classmethod
+    def getRoutes(cls):
+        contextRoot = ConfigurationManager.getInstance().getContextRoot()
+
+        # Static instances shared between different routes
+        experimentSessionController = ExperimentSessionController()
+
+        # Define routes.
+        routes = [
+
+            # Start experiment daq
+            {
+                'name' : 'startDaq',
+                'path' : '%s/experiments/startDaq' % contextRoot,
+                'controller' : experimentSessionController,
+                'action' : 'startDaq',
+                'method' : ['POST']
+            },
+
+            # Stop experiment daq
+            {
+                'name' : 'stopDaq',
+                'path' : '%s/experiments/stopDaq' % contextRoot,
+                'controller' : experimentSessionController,
+                'action' : 'stopDaq',
+                'method' : ['POST']
+            },
+
+        ]
+       
+        return routes
+
+