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

+#!/usr/bin/env python
+
+import os
+import json
+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, experimentName, dataDirectory, daqInfo={}):
+        url = '%s/experiments/startDaq' % (self.getContextRoot())
+        if experimentName is None or not len(experimentName):
+            raise InvalidRequest('Experiment name must be provided.')
+        if dataDirectory is None or not len(dataDirectory):
+            raise InvalidRequest('Experiment data directory must be provided.')
+        daqInfo['experimentName'] = experimentName
+        daqInfo['dataDirectory'] = dataDirectory
+        url += '?daqInfo=%s' % (Encoder.encode(json.dumps(daqInfo)))
+        responseDict = self.sendSessionRequest(url=url, method='POST')
+        return Experiment(responseDict)
+
+    @DaqRestApi.execute
+    def stopDaq(self, experimentName):
+        url = '%s/experiments/stopDaq' % (self.getContextRoot())
+        if experimentName is None or not len(experimentName):
+            raise InvalidRequest('Experiment name must be provided.')
+        url += '?experimentName=%s' % Encoder.encode(experimentName)
+        responseDict = self.sendSessionRequest(url=url, method='POST')
+        return Experiment(responseDict)
+
+    @DaqRestApi.execute
+    def upload(self, experimentName, dataDirectory, daqInfo={}):
+        url = '%s/experiments/upload' % (self.getContextRoot())
+        if experimentName is None or not len(experimentName):
+            raise InvalidRequest('Experiment name must be provided.')
+        if dataDirectory is None or not len(dataDirectory):
+            raise InvalidRequest('Experiment data directory must be provided.')
+        daqInfo['experimentName'] = experimentName
+        daqInfo['dataDirectory'] = dataDirectory
+        url += '?daqInfo=%s' % (Encoder.encode(json.dumps(daqInfo)))
+        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, validArgCount=self.ANY_NUMBER_OF_POSITIONAL_ARGS)
+        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
+        [key1:value1, key2:value2, ...]
+
+Description:
+    Starts DAQ for a given experiment. Specified data directory will be 
+    monitored for data files. All provided key/value pairs will be passed to 
+    file processing plugins. 
+        """)
+        self.checkArgs()
+        api = ExperimentRestApi(self.getLoginUsername(), self.getLoginPassword(), self.getServiceHost(), self.getServicePort(), self.getServiceProtocol())
+        daqInfo = self.splitArgsIntoDict()
+        experiment = api.startDaq(self.getExperimentName(), self.getDataDirectory(), daqInfo=daqInfo)
+        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/uploadCli.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 UploadCli(DaqWebServiceSessionCli):
+    def __init__(self):
+        DaqWebServiceSessionCli.__init__(self, validArgCount=self.ANY_NUMBER_OF_POSITIONAL_ARGS)
+        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 --experiment=EXPERIMENTNAME --data-directory=DATADIRECTORY
+        [key1:value1, key2:value2, ...]
+
+Description:
+    Schedules data upload for a given experiment. All existing files in the
+    specified directory will be uploaded to storage. Relative directory 
+    structure will be preserved. All provided key/value pairs will be passed 
+    to file processing plugins.
+        """)
+        self.checkArgs()
+        api = ExperimentRestApi(self.getLoginUsername(), self.getLoginPassword(), self.getServiceHost(), self.getServicePort(), self.getServiceProtocol())
+        experiment = api.upload(self.getExperimentName(), self.getDataDirectory())
+        print experiment.getDisplayString(self.getDisplayKeys(), self.getDisplayFormat())
+
+#######################################################################
+# Run command.
+if __name__ == '__main__':
+    cli = UploadCli()
+    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.common.processing.fileProcessingManager import FileProcessingManager
+from dm.daq_web_service.service.impl.fileSystemObserver import FileSystemObserver 
+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']
+            },
+
+            # Upload experiment data 
+            {
+                'name' : 'upload',
+                'path' : '%s/experiments/upload' % contextRoot,
+                'controller' : experimentSessionController,
+                'action' : 'upload',
+                'method' : ['POST']
+            },
+
+        ]
+       
+        return routes
+
+

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

+#!/usr/bin/env python
+
+import cherrypy
+import json
+
+from dm.common.service.dmSessionController import DmSessionController
+from dm.common.exceptions.invalidRequest import InvalidRequest
+from dm.common.utility.encoder import Encoder
+
+from dm.daq_web_service.service.impl.experimentSessionControllerImpl import ExperimentSessionControllerImpl
+
+
+class ExperimentSessionController(DmSessionController):
+
+    def __init__(self):
+        DmSessionController.__init__(self)
+        self.experimentSessionControllerImpl = ExperimentSessionControllerImpl()
+
+    @cherrypy.expose
+    @DmSessionController.require(DmSessionController.isAdministrator())
+    @DmSessionController.execute
+    def startDaq(self, **kwargs):
+        encodedDaqInfo = kwargs.get('daqInfo')
+        if not encodedDaqInfo:
+            raise InvalidRequest('Invalid DAQ info provided.')
+        daqInfo = json.loads(Encoder.decode(encodedDaqInfo))
+
+        if not daqInfo.has_key('experimentName'):
+            raise InvalidRequest('Experiment name is missing.')
+        if not daqInfo.has_key('dataDirectory'):
+            raise InvalidRequest('Data directory is missing.')
+        response = self.experimentSessionControllerImpl.startDaq(daqInfo).getFullJsonRep()
+        self.logger.debug('Returning: %s' % response)
+        return response
+
+    @cherrypy.expose
+    @DmSessionController.require(DmSessionController.isAdministrator())
+    @DmSessionController.execute
+    def stopDaq(self, **kwargs):
+        experimentName = kwargs.get('experimentName')
+        if not experimentName:
+            raise InvalidRequest('Missing experiment name.')
+        experimentName = Encoder.decode(experimentName)
+        response = self.experimentSessionControllerImpl.stopDaq(experimentName).getFullJsonRep()
+        self.logger.debug('Returning: %s' % response)
+        return response
+
+    @cherrypy.expose
+    @DmSessionController.require(DmSessionController.isAdministrator())
+    @DmSessionController.execute
+    def upload(self, **kwargs):
+        encodedDaqInfo = kwargs.get('daqInfo')
+        if not encodedDaqInfo:
+            raise InvalidRequest('Invalid DAQ info provided.')
+        daqInfo = json.loads(Encoder.decode(encodedDaqInfo))
+
+        if not daqInfo.has_key('experimentName'):
+            raise InvalidRequest('Experiment name is missing.')
+        if not daqInfo.has_key('dataDirectory'):
+            raise InvalidRequest('Data directory is missing.')
+        response = self.experimentSessionControllerImpl.upload(daqInfo).getFullJsonRep()
+        self.logger.debug('Returning: %s' % response)
+        return response
+

src/python/dm/daq_web_service/service/impl/dmFileSystemEventHandler.py

+#!/usr/bin/env python
+
+import os
+
+from watchdog.events import FileSystemEventHandler
+
+from dm.common.utility.loggingManager import LoggingManager
+
+class DmFileSystemEventHandler(FileSystemEventHandler):
+
+    def __init__(self, fileSystemObserver, dataDirectory, experiment):
+        FileSystemEventHandler.__init__(self)
+        self.logger = LoggingManager.getInstance().getLogger(self.__class__.__name__)
+        self.fileSystemObserver = fileSystemObserver
+        self.dataDirectory = dataDirectory
+        self.experiment = experiment
+
+    def dispatch(self, event):
+        FileSystemEventHandler.dispatch(self, event)
+
+    def on_any_event(self, event):
+        FileSystemEventHandler.on_any_event(self, event)
+
+    def on_created(self, event):
+        FileSystemEventHandler.on_created(self, event)
+
+    def on_deleted(self, event):
+        FileSystemEventHandler.on_deleted(self, event)
+
+    def on_modified(self, event):
+        FileSystemEventHandler.on_modified(self, event)
+        filePath = event.src_path
+        self.logger.debug('File system modified event: %s' % (event.__dict__))
+        if not event.is_directory:
+            self.fileSystemObserver.fileUpdated(filePath, self.dataDirectory, self.experiment)
+        
+    def on_moved(self, event):
+        FileSystemEventHandler.on_moved(self, event)
+
+####################################################################
+# Testing
+
+if __name__ == '__main__':
+    import sys
+    import time
+    import logging
+    from watchdog.observers import Observer
+    from watchdog.observers.polling import PollingObserver
+    from watchdog.observers.api import ObservedWatch
+    from watchdog.observers.api import EventQueue
+    from watchdog.observers.api import EventEmitter
+    from watchdog.events import LoggingEventHandler
+    logging.basicConfig(level=logging.INFO,
+                        format='%(asctime)s - %(message)s',
+                        datefmt='%Y-%m-%d %H:%M:%S')
+    path = sys.argv[1] if len(sys.argv) > 1 else '.'
+    eventHandler = DmFileSystemEventHandler()
+
+    observer = PollingObserver()
+    observedWatch = observer.schedule(eventHandler, path, recursive=True)
+    print 'OBSERVED WATCH: ', observedWatch
+
+    #observer.add_handler_for_watch(eventHandler2, observedWatch)
+    #observer._clear_emitters()
+    print observer.emitters
+
+    observer.start()
+    try:
+        while True:
+            time.sleep(1)
+            print time.time()
+    except KeyboardInterrupt:
+        observer.stop()
+    observer.join()

src/python/dm/daq_web_service/service/impl/dsProcessFileNotificationPlugin.py

+#!/usr/bin/env python
+
+import os
+from dm.common.utility.loggingManager import LoggingManager
+from dm.common.processing.plugins.fileProcessor import FileProcessor
+from dm.ds_web_service.api.dsRestApiFactory import DsRestApiFactory
+
+class DsProcessFileNotificationPlugin(FileProcessor):
+
+    def __init__(self):
+        FileProcessor.__init__(self)
+        self.dsFileApi = DsRestApiFactory.getFileRestApi()
+        self.logger = LoggingManager.getInstance().getLogger(self.__class__.__name__)
+
+    def processFile(self, fileInfo):
+        experimentFilePath = fileInfo.get('experimentFilePath')
+        experiment = fileInfo.get('experiment')
+        experimentName = experiment.get('name')
+        self.logger.debug('Processing file %s for experiment %s' % (experimentFilePath, experimentName))
+
+        # Prepare dictionary for processing. Only send needed data.
+        fileInfo2 = {}
+        fileInfo2['experimentFilePath'] = experimentFilePath
+        fileInfo2['experimentName'] = experimentName
+        fileInfo2['daqInfo'] = experiment.get('daqInfo')
+        self.dsFileApi.processFile(experimentFilePath, experimentName, fileInfo2)
+
+#######################################################################
+# Testing.
+if __name__ == '__main__':
+    pass

src/python/dm/daq_web_service/service/impl/experimentSessionControllerImpl.py

+#!/usr/bin/env python
+
+#
+# Implementation for experiment session controller.
+#
+
+import time
+
+from dm.common.objects.experiment import Experiment
+from dm.common.objects.dmObjectManager import DmObjectManager
+from dm.common.exceptions.invalidRequest import InvalidRequest
+from dm.common.utility.osUtility import OsUtility
+from dm.common.objects.observedFile import ObservedFile
+from dm.common.processing.fileProcessingManager import FileProcessingManager
+from dm.ds_web_service.api.dsRestApiFactory import DsRestApiFactory
+from experimentTracker import ExperimentTracker
+from fileSystemObserver import FileSystemObserver
+
+class ExperimentSessionControllerImpl(DmObjectManager):
+    """ Experiment session controller implementation class. """
+
+    def __init__(self):
+        DmObjectManager.__init__(self)
+        self.dsExperimentApi = DsRestApiFactory.getExperimentRestApi()
+
+    def startDaq(self, daqInfo):
+        experimentName = daqInfo.get('experimentName')
+        experiment = ExperimentTracker.getInstance().get(experimentName)
+        if experiment is not None:
+            oldDaqInfo = experiment.get('daqInfo')
+            if oldDaqInfo.get('daqEndTime') is None:
+                raise InvalidRequest('DAQ for experiment %s is already active in directory %s.' % (experimentName,oldDaqInfo.get('dataDirectory')))
+
+        dataDirectory = daqInfo.get('dataDirectory')
+        experiment = self.dsExperimentApi.getExperimentByName(experimentName)
+        storageDirectory = experiment.get('storageDirectory')
+        if storageDirectory is None:
+            raise InvalidRequest('Experiment %s has not been started.' % experimentName)
+        startTime = time.time()
+        daqInfo['daqStartTime'] = startTime
+        experiment['daqInfo'] = daqInfo
+        self.logger.debug('Starting DAQ %s' % daqInfo)
+        FileSystemObserver.getInstance().startObservingPath(dataDirectory, experiment)
+        ExperimentTracker.getInstance().put(experimentName, experiment)
+        return experiment
+
+    def stopDaq(self, experimentName):
+        experiment = ExperimentTracker.getInstance().get(experimentName)
+        if experiment is not None:
+            daqInfo = experiment.get('daqInfo')
+        if experiment is None or daqInfo.get('daqEndTime') is not None:
+            raise InvalidRequest('Experiment %s is not active.' % experimentName)
+        dataDirectory = daqInfo.get('dataDirectory')
+        daqInfo['daqEndTime'] = time.time()
+        FileSystemObserver.getInstance().stopObservingPath(dataDirectory, experiment)
+        return experiment
+
+    def upload(self, daqInfo):
+        experimentName = daqInfo.get('experimentName')
+        experiment = self.dsExperimentApi.getExperimentByName(experimentName)
+        experiment['daqInfo'] = daqInfo
+        storageDirectory = experiment.get('storageDirectory')
+        if storageDirectory is None:
+            raise InvalidRequest('Experiment %s has not been started.' % experimentName)
+        dataDirectory = daqInfo.get('dataDirectory')
+        filePaths = OsUtility.findFiles(dataDirectory)
+        fileProcessingManager = FileProcessingManager.getInstance()
+        for filePath in filePaths:
+            fileInfo = ObservedFile(filePath=filePath, dataDirectory=dataDirectory, experiment=experiment)
+            fileProcessingManager.processFile(fileInfo)
+        ExperimentTracker.getInstance().put(experimentName, experiment)
+        return experiment
+

src/python/dm/daq_web_service/service/impl/experimentTracker.py

+#!/usr/bin/env python
+
+from dm.common.objects.experiment import Experiment
+from dm.common.utility.objectTracker import ObjectTracker
+
+class ExperimentTracker(ObjectTracker):
+
+    # Cache configuration
+    objectClass = Experiment
+
+####################################################################
+# Testing
+
+if __name__ == '__main__':
+    et = ExperimentTracker.getInstance()
+    print et
+    et2 = ExperimentTracker.getInstance()
+    print et2
+
+
+

src/python/dm/daq_web_service/service/impl/fileSystemObserver.py

+#!/usr/bin/env python
+
+import threading
+import time
+
+from watchdog.observers.polling import PollingObserver
+
+from dm.common.utility.loggingManager import LoggingManager
+from dm.common.utility.configurationManager import ConfigurationManager
+from dm.common.objects.observedFile import ObservedFile
+from dm.common.utility.singleton import Singleton
+from dm.common.utility.threadingUtility import ThreadingUtility
+from dm.common.processing.fileProcessingManager import FileProcessingManager
+
+from dmFileSystemEventHandler import DmFileSystemEventHandler
+
+class FileSystemObserver(threading.Thread,Singleton):
+
+    CONFIG_SECTION_NAME = 'FileSystemObserver'
+    MIN_FILE_PROCESSING_DELAY_IN_SECONDS_KEY = 'minfileprocessingdelayinseconds'
+    FILE_SYSTEM_EVENT_TIMEOUT_IN_SECONDS_KEY = 'filesystemeventtimeoutinseconds'
+
+    # Singleton.
+    __instanceLock = threading.RLock()
+    __instance = None
+
+    def __init__(self):
+        FileSystemObserver.__instanceLock.acquire()
+        try:
+            if FileSystemObserver.__instance:
+                return
+            FileSystemObserver.__instance = self
+            threading.Thread.__init__(self)
+            self.setName('FileSystemObserverThread')
+            self.logger = LoggingManager.getInstance().getLogger(self.__class__.__name__)
+
+            self.logger.debug('Initializing')
+            self.lock = threading.RLock()
+            self.eventFlag = threading.Event()
+            self.exitFlag = False
+
+            self.observedFileMap = {}
+            self.observer = PollingObserver()
+            self.observedWatchDict = {}
+            self.__configure()
+            self.fileProcessingManager = FileProcessingManager.getInstance()
+            self.logger.debug('Initialization complete')
+        finally:
+            FileSystemObserver.__instanceLock.release()
+
+    def __configure(self):
+        cm = ConfigurationManager.getInstance()
+        configItems = cm.getConfigItems(FileSystemObserver.CONFIG_SECTION_NAME)
+        self.logger.debug('Got config items: %s' % configItems)
+        self.minFileProcessingDelayInSeconds = int(cm.getConfigOption(FileSystemObserver.CONFIG_SECTION_NAME, FileSystemObserver.MIN_FILE_PROCESSING_DELAY_IN_SECONDS_KEY))
+        self.logger.debug('Minimum file processing delay: %s seconds' % self.minFileProcessingDelayInSeconds)
+        self.fileSystemEventTimeoutInSeconds = int(cm.getConfigOption(FileSystemObserver.CONFIG_SECTION_NAME, FileSystemObserver.FILE_SYSTEM_EVENT_TIMEOUT_IN_SECONDS_KEY))
+        self.logger.debug('File system event timeout: %s seconds' % self.fileSystemEventTimeoutInSeconds)
+
+    @ThreadingUtility.synchronize
+    def startObservingPath(self, dataDirectory, experiment):
+        self.logger.debug('Starting observer for %s' % dataDirectory)
+        eventHandler = DmFileSystemEventHandler(self, dataDirectory, experiment)
+        observedWatch = self.observer.schedule(eventHandler, dataDirectory, recursive=True)
+        self.observedWatchDict[dataDirectory] = observedWatch
+
+    @ThreadingUtility.synchronize
+    def stopObservingPath(self, dataDirectory, experiment):
+        observedWatch = self.observedWatchDict.get(dataDirectory)
+        if observedWatch:
+            self.logger.debug('Stopping observer for %s' % dataDirectory)
+            self.observer.unschedule(observedWatch)
+            del self.observedWatchDict[dataDirectory] 
+        else:
+            self.logger.debug('Observer for %s is not active' % dataDirectory)
+        
+    @ThreadingUtility.synchronize
+    def fileUpdated(self, filePath, dataDirectory, experiment):
+        observedFile = self.observedFileMap.get(filePath, ObservedFile(filePath=filePath, dataDirectory=dataDirectory, experiment=experiment))
+        observedFile.setLastUpdatedTimestampToNow()
+        self.observedFileMap[filePath] = observedFile
+        self.logger.debug('Observed file updated: %s', observedFile)
+       
+    @ThreadingUtility.synchronize
+    def checkObservedFilesForProcessing(self):
+        now = time.time()
+        filePathsForProcessing = []
+        for (filePath,observedFile) in self.observedFileMap.items():
+            timestamp = observedFile.get('lastUpdateTimestamp')
+            deltaT = now - timestamp
+            if deltaT > self.minFileProcessingDelayInSeconds:
+                self.logger.debug('File %s was last modified %s seconds ago, will process it.' % (filePath, deltaT))
+                filePathsForProcessing.append(filePath)
+        return filePathsForProcessing
+    
+    @ThreadingUtility.synchronize
+    def processFile(self, filePath):
+        self.logger.debug('Processing file %s' % filePath)
+        observedFile = self.observedFileMap.get(filePath)
+        if observedFile is not None:
+            del self.observedFileMap[filePath]
+            self.fileProcessingManager.processFile(observedFile)
+     
+    @ThreadingUtility.synchronize
+    def start(self):
+        self.logger.debug('Starting file observer thread')
+        threading.Thread.start(self)
+
+        self.logger.debug('Starting watchdog observer')
+        self.observer.start()
+
+    def run(self):
+        self.logger.debug('Starting thread: %s' % self.getName())
+        while True:
+            if self.exitFlag:
+                self.logger.debug('Exit flag set, %s done' % self.getName())
+                break
+            try:
+                self.logger.debug('Checking observed files')
+
+                filePathsForProcessing = self.checkObservedFilesForProcessing()
+                for filePath in filePathsForProcessing:
+                    self.processFile(filePath)
+            except Exception, ex:
+                self.logger.exception(ex)
+            self.eventFlag.wait(timeout=self.fileSystemEventTimeoutInSeconds)
+
+    @ThreadingUtility.synchronize
+    def stop(self):
+        self.logger.debug('Stopping watchdog observer')
+        self.observer.stop()
+        self.observer.join()
+
+        self.logger.debug('Stopping file observer thread')
+        self.exitFlag = True
+        self.eventFlag.set()
+        self.logger.debug('Event is set, joining thread')
+        threading.Thread.join(self)
+        self.logger.debug('Module stopped')
+
+    @ThreadingUtility.synchronize
+    def setEvent(self):
+        self.eventFlag.set()
+
+    @ThreadingUtility.synchronize
+    def clearEvent(self):
+        self.eventFlag.clear()
+
+####################################################################
+# Testing
+
+if __name__ == '__main__':
+    fp = FileSystemObserver.getInstance()
+    fp.start()
+    time.sleep(30)
+    fp.stop()
+
+

src/python/dm/daq_web_service/service/impl/userInfoControllerImpl.py

+#!/usr/bin/env python
+
+#
+# Implementation for user info controller.
+#
+
+#######################################################################
+
+from dm.common.objects.dmObject import DmObject
+from dm.common.objects.dmObjectManager import DmObjectManager
+from dm.common.db.api.userDbApi import UserDbApi
+
+#######################################################################
+
+class UserInfoControllerImpl(DmObjectManager):
+    """ User info controller implementation class. """
+
+    def __init__(self):
+        DmObjectManager.__init__(self)
+        self.userDbApi = UserDbApi()
+
+    def getUsers(self):
+        return self.userDbApi.getUsers()
+
+    def getUserById(self, id):
+        return self.userDbApi.getUserById(id)
+
+    def getUserByUsername(self, username):
+        return self.userDbApi.getUserByUsername(username)
+