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

+#!/usr/bin/env python
+
+import os
+import glob
+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)
+        self.logger.debug('File system any_event event: %s' % (event.__dict__))
+
+    def on_created(self, event):
+        FileSystemEventHandler.on_created(self, event)
+        self.logger.debug('File system created event: %s' % (event.__dict__))
+        self.processEvent(event)
+
+    def on_moved(self, event):
+        FileSystemEventHandler.on_moved(self, event)
+        self.logger.debug('File system moved event: %s' % (event.__dict__))
+
+    def on_deleted(self, event):
+        FileSystemEventHandler.on_deleted(self, event)
+        self.logger.debug('File system deleted event: %s' % (event.__dict__))
+
+    def on_modified(self, event):
+        FileSystemEventHandler.on_modified(self, event)
+        self.logger.debug('File system directory modified event: %s' % (event.__dict__))
+        self.processEvent(event)
+
+    def processEvent(self, event):
+        if event.is_directory:
+            self.logger.debug('Processing directory event: %s , src path: %s ,  latest files: %s' % (event.__dict__, event.src_path, files))
+            try:
+                files = glob.glob(os.path.join(event.src_path,'*.*'))
+                if len(files) > 0:
+                    filePath = max(files, key=os.path.getctime)
+                    self.logger.debug('Latest file: %s' % (filePath))
+                    self.fileSystemObserver.fileUpdated(filePath, self.dataDirectory, self.experiment)
+            except Exception, ex:
+                self.logger.error('Exception occured when searching for file in directory %s: %s' % (event.__dict__, ex))
+        else:
+            filePath = event.src_path
+            self.logger.debug('Processing file event: %s' % (event.__dict__))
+            self.fileSystemObserver.fileUpdated(filePath, self.dataDirectory, self.experiment)
+        
+####################################################################
+# 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 os
+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. """
+
+    DAQ_PERMISSIONS_MODE = 0777
+
+    def __init__(self):
+        DmObjectManager.__init__(self)
+        self.dsExperimentApi = DsRestApiFactory.getExperimentRestApi()
+
+    def startDaq(self, experimentName, dataDirectory, daqInfo):
+        OsUtility.createDir(dataDirectory, mode=self.DAQ_PERMISSIONS_MODE)
+        if daqInfo is None:
+            daqInfo={}
+	    daqInfo['experimentName'] = experimentName
+	    daqInfo['dataDirectory'] = dataDirectory
+        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')))
+
+        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.objectUtility import ObjectUtility
+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'
+    FILE_SYSTEM_OBSERVER_AGENT_KEY = 'filesystemobserveragent'
+
+    # 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.__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)
+
+        agentClass = cm.getConfigOption(FileSystemObserver.CONFIG_SECTION_NAME, FileSystemObserver.FILE_SYSTEM_OBSERVER_AGENT_KEY)
+        (moduleName,className,constructor) = cm.getModuleClassConstructorTuple(agentClass)
+        self.logger.debug('Creating file system observer agent instance of class %s' % className)
+        self.fileSystemObserverAgent = ObjectUtility.createObjectInstance(moduleName, className, constructor)
+        self.fileSystemObserverAgent.setFileSystemObserver(self)
+
+    @ThreadingUtility.synchronize
+    def startObservingPath(self, dataDirectory, experiment):
+        self.logger.debug('Agent is starting observer for %s' % dataDirectory)
+        self.fileSystemObserverAgent.startObservingPath(dataDirectory, experiment)
+
+    @ThreadingUtility.synchronize
+    def stopObservingPath(self, dataDirectory, experiment):
+        self.logger.debug('Agent is stopping observer for %s' % dataDirectory)
+        self.fileSystemObserverAgent.stopObservingPath(dataDirectory, experiment)
+        
+    @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 file observer agent')
+        self.fileSystemObserverAgent.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 file observer agent')
+        self.fileSystemObserverAgent.stop()
+
+        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/fileSystemObserverAgent.py

+#!/usr/bin/env python
+
+from dm.common.utility.loggingManager import LoggingManager
+
+class FileSystemObserverAgent:
+
+    def __init__(self):
+        self.logger = LoggingManager.getInstance().getLogger(self.__class__.__name__)
+        self.fileSystemObserver = None
+
+    def setFileSystemObserver(self, fileSystemObserver):
+        self.fileSystemObserver = fileSystemObserver
+
+    def startObservingPath(self, dataDirectory, experiment):
+        pass
+
+    def stopObservingPath(self, dataDirectory, experiment):
+        pass
+
+    def start(self):
+        pass
+
+    def stop(self):
+        pass
+

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

+#!/usr/bin/env python
+
+from threading import Timer
+from fileSystemObserverAgent import FileSystemObserverAgent
+from dm.common.utility.ftpUtility import FtpUtility
+
+class FtpFileSystemObserverAgent(FileSystemObserverAgent):
+
+    DEFAULT_POLLING_PERIOD_IN_SECONDS = 5
+
+    def __init__(self, host, port, username=None, password=None, pollingPeriod=DEFAULT_POLLING_PERIOD_IN_SECONDS):
+        FileSystemObserverAgent.__init__(self)
+        self.host = host
+        self.port = port
+        self.username = username
+        self.password = password
+        self.pollingPeriod = pollingPeriod
+        self.observedDirDict = {}
+        self.isDone = False
+
+    def getFiles(self, dataDirectory):
+        ftpUtility = FtpUtility(self.host, self.port, self.username, self.password)
+        return ftpUtility.getFiles(dataDirectory, {})
+
+    def updateFile(self, filePath, dataDirectory, experiment):
+        if self.fileSystemObserver:
+            self.logger.debug('Processing file path: %s' % filePath)
+            self.fileSystemObserver.fileUpdated(filePath, dataDirectory, experiment)
+
+    def processFiles(self, fileDict, oldFileDict, dataDirectory, experiment):
+        for filePath in fileDict.keys():
+            if not oldFileDict.has_key(filePath):
+                # new file, must be updated
+                self.logger.debug('New file path detected: %s' % filePath)
+                self.updateFile(filePath, dataDirectory, experiment)
+            else:
+                # old file, check timestamp
+                oldFileInfo = oldFileDict.get(filePath)
+                oldModifyTime = oldFileInfo.get('Modify', '')
+                fileInfo = fileDict.get(filePath)
+                modifyTime = fileInfo.get('Modify')
+                if modifyTime != oldModifyTime:
+                    # file has been modified, need to process it
+                    self.logger.debug('Modified file path detected: %s' % filePath)
+                    self.updateFile(filePath, dataDirectory, experiment)
+
+
+    def pollFileSystem(self, dataDirectory, experiment):
+        try:
+            fileDict = self.getFiles(dataDirectory)
+            observedDirInfo = self.observedDirDict.get(dataDirectory)
+            if not observedDirInfo:
+                self.logger.debug('Polling cancelled for directory: %s' % dataDirectory)
+                return 
+            oldFileDict = observedDirInfo.get('files')
+            observedDirInfo['files'] = fileDict
+            self.processFiles(fileDict, oldFileDict, dataDirectory, experiment)
+            self.startPollingTimer(observedDirInfo, dataDirectory, experiment)
+        except Exception, ex:
+            self.logger.error('Could not poll directory %s: %s' % (dataDirectory,ex))
+
+    def startPollingTimer(self, observedDirInfo, dataDirectory, experiment):
+        if self.isDone:
+            return
+
+        t = Timer(self.pollingPeriod, self.pollFileSystem, [dataDirectory, experiment])
+        observedDirInfo['pollTimer'] = t
+        t.start()
+
+    def startObservingPath(self, dataDirectory, experiment):
+        observedDirInfo = self.observedDirDict.get(dataDirectory)
+        if observedDirInfo:
+            self.logger.debug('Observer for %s is already active' % dataDirectory)
+            return
+        self.logger.debug('Starting observer for %s' % dataDirectory)
+        fileDict = self.getFiles(dataDirectory)
+        observedDirInfo = self.observedDirDict.get(dataDirectory, {})
+        observedDirInfo['files'] = fileDict
+        observedDirInfo['experiment'] = experiment
+        self.observedDirDict[dataDirectory] = observedDirInfo 
+        self.startPollingTimer(observedDirInfo, dataDirectory, experiment)
+        
+    def stopObservingPath(self, dataDirectory, experiment):
+        observedDirInfo = self.observedDirDict.get(dataDirectory)
+        if not observedDirInfo:
+            self.logger.debug('Observer for %s is not active' % dataDirectory)
+            return
+
+        self.logger.debug('Stopping observer for %s' % dataDirectory)
+        t = observedDirInfo.get('pollTimer')
+        if t:
+            t.cancel()
+        del self.observedDirDict[dataDirectory]
+
+    def start(self):
+        self.logger.debug('Starting ftp observer agent')
+
+    def stop(self):
+        self.logger.debug('Stopping ftp observer agent')
+        self.isDone = True
+        for (dataDirectory,observedDirInfo)  in self.observedDirDict.items():
+            experiment = observedDirInfo.get('experiment')
+            self.stopObservingPath(dataDirectory, experiment)
+
+####################################################################
+# Testing
+
+if __name__ == '__main__':
+    import time
+    agent = FtpFileSystemObserverAgent('zagreb', 2811)
+    print 'ORIGINAL FILES: ', agent.getFiles('/tmp/test')
+    agent.startObservingPath('/tmp/test', 'e1')
+    time.sleep(100)
+    agent.stopObservingPath('/tmp/test', 'e1')
+

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

+#!/usr/bin/env python
+
+import os
+from dm.common.utility.loggingManager import LoggingManager
+from dm.common.utility.osUtility import OsUtility
+from dm.common.utility.fileUtility import FileUtility
+from dm.common.exceptions.fileProcessingError import FileProcessingError
+from dm.common.processing.plugins.fileTransferPlugin import FileTransferPlugin
+from dm.ds_web_service.api.dsRestApiFactory import DsRestApiFactory
+
+class RsyncWithChecksumAndDeleteFileTransferPlugin(FileTransferPlugin):
+
+    COMMAND = 'rsync -arvlPR'
+
+    def __init__(self, src=None, dest=None):
+        FileTransferPlugin.__init__(self, self.COMMAND, src, dest)
+        self.dsFileApi = DsRestApiFactory.getFileRestApi()
+        self.logger = LoggingManager.getInstance().getLogger(self.__class__.__name__)
+
+    def processFile(self, fileInfo):
+        filePath = fileInfo.get('filePath')
+        dataDirectory = fileInfo.get('dataDirectory')
+        experiment = fileInfo.get('experiment')
+        experimentName = experiment.get('name')
+        experimentFilePath = fileInfo.get('experimentFilePath')
+        storageHost = experiment.get('storageHost')
+        storageDirectory = experiment.get('storageDirectory')
+        destUrl = self.getDestUrl(storageHost, storageDirectory)
+        # Use relative path with respect to data directory as a source
+        os.chdir(dataDirectory)
+        srcUrl = self.getSrcUrl(filePath, dataDirectory)
+
+        # Calculate checksum
+        FileUtility.statFile(filePath, fileInfo)
+        FileUtility.getMd5Sum(filePath, fileInfo)
+        self.logger.debug('File info before transfer: %s' % fileInfo)
+
+        # Transfer file
+        self.start(srcUrl, destUrl)
+
+        # Get remote checksum
+        fileInfo2 = {}
+        fileInfo2['experimentFilePath'] = experimentFilePath
+        fileInfo2['experimentName'] = experimentName
+        fileMetadata = self.dsFileApi.statFile(experimentFilePath, experimentName, fileInfo2)
+        self.logger.debug('File stat after transfer: %s' % fileMetadata)
+
+        # Verify checksum
+        if fileMetadata.get('md5Sum') != fileInfo.get('md5Sum'):
+            self.logger.error('Checksum mismatch for file: %s' % filePath)
+            raise FileProcessingError('Checksum mismatch detected for file: %s' % filePath)
+
+        # Remove file
+        self.logger.debug('Checksum test passed, deleting file %s' % filePath)
+        OsUtility.removeFile(srcUrl)
+
+#######################################################################
+# Testing.
+if __name__ == '__main__':
+    ft = RsyncWithChecksumAndDeleteFileTransferPlugin('/tmp/xyz', '/tmp/xyz2')
+    ft.start()
+    print 'StdOut: ', ft.getStdOut()
+    print 'StdErr: ', ft.getStdErr()
+    print 'Exit Status: ', ft.getExitStatus()

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)
+

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

+#!/usr/bin/env python
+
+from watchdog.observers.polling import PollingObserver
+from fileSystemObserverAgent import FileSystemObserverAgent
+from dmFileSystemEventHandler import DmFileSystemEventHandler
+
+class WatchdogFileSystemObserverAgent(FileSystemObserverAgent):
+
+    def __init__(self):
+        FileSystemObserverAgent.__init__(self)
+        self.observer = PollingObserver()
+        self.observedWatchDict = {}
+
+    def startObservingPath(self, dataDirectory, experiment):
+        self.logger.debug('Starting observer for %s' % dataDirectory)
+        eventHandler = DmFileSystemEventHandler(self.fileSystemObserver, dataDirectory, experiment)
+        observedWatch = self.observer.schedule(eventHandler, dataDirectory, recursive=True)
+        self.observedWatchDict[dataDirectory] = observedWatch
+
+    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)
+
+    def start(self):
+        self.logger.debug('Starting watchdog observer agent')
+        self.observer.start()
+
+    def stop(self):
+        self.logger.debug('Stopping watchdog observer agent')
+        self.observer.stop()
+        self.observer.join()
+

src/python/dm/ds_web_service/api/authRestApi.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.exceptions.invalidRequest import InvalidRequest
+from dm.common.objects.authorizationPrincipal import AuthorizationPrincipal
+from dm.common.objects.dmSession import DmSession
+from dsRestApi import DsRestApi
+
+class AuthRestApi(DsRestApi):
+   
+    def __init__(self, username=None, password=None, host=None, port=None, protocol=None):
+        DsRestApi.__init__(self, username, password, host, port, protocol)
+
+    @DsRestApi.execute
+    def getAuthorizationPrincipal(self, username):
+        if username is None:
+            raise InvalidRequest('Username must be provided.')
+        url = '%s/authorizationPrincipals/%s' % (self.getContextRoot(), username)
+        self.logger.debug('Retrieving principal for user %s' % (username))
+        responseData = self.sendSessionRequest(url=url, method='GET')
+        return AuthorizationPrincipal(responseData)
+
+    @DsRestApi.execute
+    def addSession(self, sessionId, sessionInfo):
+        if sessionId is None:
+            raise InvalidRequest('Session id must be provided.')
+        if sessionInfo is None:
+            raise InvalidRequest('Session info must be provided.')
+        url = '%s/sessions/%s' % (self.getContextRoot(), sessionId)
+        url += '?sessionInfo=%s' % Encoder.encode(json.dumps(sessionInfo))
+        responseData = self.sendSessionRequest(url=url, method='POST')
+        return DmSession(responseData)
+
+    @DsRestApi.execute
+    def checkSession(self, sessionId):
+        if sessionId is None:
+            raise InvalidRequest('Session id must be provided.')
+        url = '%s/sessions/%s' % (self.getContextRoot(), sessionId)
+        responseData = self.sendSessionRequest(url=url, method='PUT')
+        return DmSession(responseData)
+
+#######################################################################
+# Testing.
+
+if __name__ == '__main__':
+    api = AuthRestApi('dm', 'dm', 'zagreb.svdev.net', 22236, 'http')
+    print api.authenticateUser('sveseli', 'sv')
+        
+
+
+

src/python/dm/ds_web_service/api/dsRestApi.py

+#!/usr/bin/env python
+
+from dm.common.api.dmRestApi import DmRestApi
+from dm.common.utility.configurationManager import ConfigurationManager
+
+class DsRestApi(DmRestApi):
+    """ Base DS DM REST api class. """
+
+    def __init__(self, username=None, password=None, host=None, port=None, protocol=None):
+        if host == None:
+            host = ConfigurationManager.getInstance().getDsWebServiceHost()
+        if port == None:
+            port = ConfigurationManager.getInstance().getDsWebServicePort()
+        DmRestApi.__init__(self, username, password, host, port, protocol)
+
+#######################################################################
+# Testing.
+
+if __name__ == '__main__':
+    pass
+
+

src/python/dm/ds_web_service/api/dsRestApiFactory.py

+#!/usr/bin/env python
+
+from dm.common.utility.loggingManager import LoggingManager
+from dm.common.utility.configurationManager import ConfigurationManager
+
+class DsRestApiFactory:
+
+    CONFIG_SECTION_NAME = 'DsRestApiFactory'
+    USERNAME_KEY = 'username'
+    PASSWORD_FILE_KEY = 'passwordfile'
+    HOST_KEY = 'host'
+    PORT_KEY = 'port'
+    PROTOCOL_KEY = 'protocol'
+
+    __logger = None
+    __username = None
+    __password = None
+    __host = None
+    __port = None
+    __protocol = None
+
+    @classmethod
+    def getLogger(cls):
+        if cls.__logger is None:
+            cls.__logger = LoggingManager.getInstance().getLogger(cls.__name__)
+        return cls.__logger
+
+    @classmethod
+    def __getConfiguration(cls):
+        if cls.__username is None:
+            cls.__username = ConfigurationManager.getInstance().getConfigOption(cls.CONFIG_SECTION_NAME, cls.USERNAME_KEY)
+            cls.__password = open(ConfigurationManager.getInstance().getConfigOption(cls.CONFIG_SECTION_NAME, cls.PASSWORD_FILE_KEY)).read().strip()
+            cls.__host = ConfigurationManager.getInstance().getConfigOption(cls.CONFIG_SECTION_NAME, cls.HOST_KEY)
+            cls.__port = ConfigurationManager.getInstance().getConfigOption(cls.CONFIG_SECTION_NAME, cls.PORT_KEY)
+            cls.__protocol = ConfigurationManager.getInstance().getConfigOption(cls.CONFIG_SECTION_NAME, cls.PROTOCOL_KEY)
+        return (cls.__username, cls.__password, cls.__host, cls.__port, cls.__protocol)
+
+    @classmethod
+    def getUserRestApi(cls):
+        from userRestApi import UserRestApi
+        (username, password, host, port, protocol) = cls.__getConfiguration()
+        api = UserRestApi(username, password, host, port, protocol)
+        return api
+
+    @classmethod
+    def getAuthRestApi(cls):
+        from authRestApi import AuthRestApi
+        (username, password, host, port, protocol) = cls.__getConfiguration()
+        api = AuthRestApi(username, password, host, port, protocol)
+        return api
+
+    @classmethod
+    def getExperimentRestApi(cls):
+        from experimentRestApi import ExperimentRestApi
+        (username, password, host, port, protocol) = cls.__getConfiguration()
+        api = ExperimentRestApi(username, password, host, port, protocol)
+        return api
+
+    @classmethod
+    def getFileRestApi(cls):
+        from fileRestApi import FileRestApi
+        (username, password, host, port, protocol) = cls.__getConfiguration()
+        api = FileRestApi(username, password, host, port, protocol)
+        return api
+####################################################################
+# Testing
+
+if __name__ == '__main__':
+    pass
+

src/python/dm/ds_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 dm.common.objects.experimentType import ExperimentType
+from dsRestApi import DsRestApi
+
+class ExperimentRestApi(DsRestApi):
+    
+    def __init__(self, username=None, password=None, host=None, port=None, protocol=None):
+        DsRestApi.__init__(self, username, password, host, port, protocol)
+
+    @DsRestApi.execute
+    def getExperimentTypes(self):
+        url = '%s/experimentTypes' % (self.getContextRoot())
+        responseData = self.sendSessionRequest(url=url, method='GET')
+        return self.toDmObjectList(responseData, ExperimentType)
+
+    @DsRestApi.execute
+    def getExperiments(self):
+        url = '%s/experiments' % (self.getContextRoot())
+        responseData = self.sendSessionRequest(url=url, method='GET')
+        return self.toDmObjectList(responseData, Experiment)
+
+    @DsRestApi.execute
+    def getExperimentByName(self, name):
+        url = '%s/experimentsByName/%s' % (self.getContextRoot(), name)
+        if name is None or not len(name):
+            raise InvalidRequest('Experiment name must be provided.')
+        responseDict = self.sendSessionRequest(url=url, method='GET')
+        return Experiment(responseDict)
+
+    @DsRestApi.execute
+    def getExperimentById(self, id):
+        url = '%s/experiments/%s' % (self.getContextRoot(), id)
+        if id is None:
+            raise InvalidRequest('Experiment id must be provided.')
+        responseDict = self.sendSessionRequest(url=url, method='GET')
+        return Experiment(responseDict)
+
+    @DsRestApi.execute
+    def startExperiment(self, name):
+        url = '%s/experiments/start' % (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='PUT')
+        return Experiment(responseDict)
+
+    @DsRestApi.execute
+    def updateExperiment(self, name):
+        url = '%s/experiments/update' % (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='PUT')
+        return Experiment(responseDict)
+
+    @DsRestApi.execute
+    def stopExperiment(self, name):
+        url = '%s/experiments/stop' % (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='PUT')
+        return Experiment(responseDict)
+
+    @DsRestApi.execute
+    def addExperiment(self, name, experimentTypeId, description, startDate, endDate):
+        url = '%s/experiments' % (self.getContextRoot())
+        if name is None or not len(name):
+            raise InvalidRequest('Experiment name must be provided.')
+        url += '?name=%s' % Encoder.encode(name)
+        if experimentTypeId is None:
+            raise InvalidRequest('Experiment type id must be provided.')
+        url += '&experimentTypeId=%s' % experimentTypeId
+        if description is not None:
+            url += '&description=%s' % Encoder.encode(description)
+        if startDate is not None:
+            url += '&startDate=%s' % Encoder.encode(startDate)
+        if endDate is not None:
+            url += '&endDate=%s' % Encoder.encode(endDate)
+        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.startExperiment('experiment1')
+

src/python/dm/ds_web_service/api/fileRestApi.py

+#!/usr/bin/env python
+
+import os
+import urllib
+import json
+
+from dm.common.utility.encoder import Encoder
+from dm.common.exceptions.dmException import DmException
+from dm.common.objects.fileMetadata import FileMetadata
+from dsRestApi import DsRestApi
+
+class FileRestApi(DsRestApi):
+    
+    def __init__(self, username=None, password=None, host=None, port=None, protocol=None):
+        DsRestApi.__init__(self, username, password, host, port, protocol)
+
+    @DsRestApi.execute
+    def processFile(self, experimentFilePath, experimentName, fileInfo={}):
+        url = '%s/files/processFile' % (self.getContextRoot())
+        if not experimentFilePath:
+            raise InvalidRequest('Experiment file path must be provided.')
+        if not experimentName:
+            raise InvalidRequest('Experiment name must be provided.')
+        fileInfo['experimentFilePath'] = experimentFilePath
+        fileInfo['experimentName'] = experimentName
+        url += '?fileInfo=%s' % (Encoder.encode(json.dumps(fileInfo)))
+        responseDict = self.sendSessionRequest(url=url, method='POST')
+        return FileMetadata(responseDict)
+
+    @DsRestApi.execute
+    def statFile(self, experimentFilePath, experimentName, fileInfo={}):
+        url = '%s/files/statFile' % (self.getContextRoot())
+        if not experimentFilePath:
+            raise InvalidRequest('Experiment file path must be provided.')
+        if not experimentName:
+            raise InvalidRequest('Experiment name must be provided.')
+        fileInfo['experimentFilePath'] = experimentFilePath
+        fileInfo['experimentName'] = experimentName
+        url += '?fileInfo=%s' % (Encoder.encode(json.dumps(fileInfo)))
+        responseDict = self.sendSessionRequest(url=url, method='POST')
+        return FileMetadata(responseDict)
+
+#######################################################################
+# Testing.
+
+if __name__ == '__main__':
+    api = FileRestApi('sveseli', 'sveseli', 'zagreb.svdev.net', 22236, 'http')
+    print api.processFile('file1', '/ESAF/exp1', 'exp1')
+

src/python/dm/ds_web_service/api/userRestApi.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.exceptions.invalidRequest import InvalidRequest
+from dm.common.objects.userInfo import UserInfo
+from dm.common.objects.userExperimentRole import UserExperimentRole
+from dsRestApi import DsRestApi
+
+class UserRestApi(DsRestApi):
+    
+    def __init__(self, username=None, password=None, host=None, port=None, protocol=None):
+        DsRestApi.__init__(self, username, password, host, port, protocol)
+
+    @DsRestApi.execute
+    def getUsers(self):
+        url = '%s/users' % (self.getContextRoot())
+        responseData = self.sendSessionRequest(url=url, method='GET')
+        return self.toDmObjectList(responseData, UserInfo)
+
+    @DsRestApi.execute
+    def getUserById(self, id):
+        if id is None:
+            raise InvalidRequest('User id must be provided.')
+        url = '%s/users/%s' % (self.getContextRoot(), id)
+        responseData = self.sendSessionRequest(url=url, method='GET')
+        return UserInfo(responseData)
+
+    @DsRestApi.execute
+    def getUserByUsername(self, username):
+        if username is None:
+            raise InvalidRequest('Username must be provided.')
+        url = '%s/usersByUsername/%s' % (self.getContextRoot(), username)
+        responseData = self.sendSessionRequest(url=url, method='GET')
+        return UserInfo(responseData)
+
+    @DsRestApi.execute
+    def addUserExperimentRole(self, username, experimentName, roleName):
+        if not username:
+            raise InvalidRequest('Username must be provided.')
+        if not experimentName:
+            raise InvalidRequest('Experiment name must be provided.')
+        if not roleName:
+            raise InvalidRequest('Role name must be provided.')
+        url = '%s/usersByExperiment/%s/%s/%s' % (self.getContextRoot(), username, experimentName, roleName)
+        responseData = self.sendSessionRequest(url=url, method='POST')
+        return UserExperimentRole(responseData)
+
+    @DsRestApi.execute
+    def deleteUserExperimentRole(self, username, experimentName, roleName):
+        if not username:
+            raise InvalidRequest('Username must be provided.')
+        if not experimentName:
+            raise InvalidRequest('Experiment name must be provided.')
+        if not roleName:
+            raise InvalidRequest('Role name must be provided.')
+        url = '%s/usersByExperiment/%s/%s/%s' % (self.getContextRoot(), username, experimentName, roleName)
+        responseData = self.sendSessionRequest(url=url, method='DELETE')
+        return UserExperimentRole(responseData)
+
+#######################################################################
+# Testing.
+
+if __name__ == '__main__':
+    pass
+
+
+

src/python/dm/ds_web_service/cli/addExperimentCli.py

+#!/usr/bin/env python
+
+from dm.ds_web_service.api.experimentRestApi import ExperimentRestApi
+from dm.common.exceptions.invalidRequest import InvalidRequest
+from dsWebServiceSessionCli import DsWebServiceSessionCli
+
+class AddExperimentCli(DsWebServiceSessionCli):
+    def __init__(self):
+        DsWebServiceSessionCli.__init__(self)
+        self.addOption('', '--experiment', dest='experimentName', help='Experiment name.')
+        self.addOption('', '--type-id', dest='typeId', help='Experiment type id.')
+        self.addOption('', '--description', dest='description', help='Experiment description.')
+        self.addOption('', '--start-date', dest='startDate', help='Experiment start date in format 31-AUG-15.')
+        self.addOption('', '--end-date', dest='endDate', help='Experiment end date in format 31-AUG-15.')
+
+    def checkArgs(self):
+        if self.options.experimentName is None:
+            raise InvalidRequest('Experiment name must be provided.')
+        if self.options.typeId is None:
+            raise InvalidRequest('Experiment type id must be provided.')
+
+
+    def getExperimentName(self):
+        return self.options.experimentName
+
+    def getTypeId(self):
+        return self.options.typeId
+
+    def getDescription(self):
+        return self.options.description
+
+    def getStartDate(self):
+        return self.options.startDate
+
+    def getEndDate(self):
+        return self.options.endDate
+
+    def runCommand(self):
+        self.parseArgs(usage="""
+    dm-add-experiment --experiment=EXPERIMENTNAME --type-id=TYPEID
+        [--description=DESCRIPTION] 
+        [--start-date=STARTDATE] 
+        [--end-date=ENDDATE]
+
+Description:
+    Add new experiment to the DM database. 
+        """)
+        self.checkArgs()
+        api = ExperimentRestApi(self.getLoginUsername(), self.getLoginPassword(), self.getServiceHost(), self.getServicePort(), self.getServiceProtocol())
+        experiment = api.addExperiment(self.getExperimentName(), self.getTypeId(), self.getDescription(), self.getStartDate(), self.getEndDate())
+        print experiment.getDisplayString(self.getDisplayKeys(), self.getDisplayFormat())
+
+#######################################################################
+# Run command.
+if __name__ == '__main__':
+    cli = AddExperimentCli()
+    cli.run()
+