fileSystemObserver.py

#!/usr/bin/env python

import threading
import time
import os

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.valueUtility import ValueUtility
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
from daqTracker import DaqTracker

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 createDirectory(self, dataDirectory):
        self.fileSystemObserverAgent.createDirectory(dataDirectory)

    @ThreadingUtility.synchronize
    def getFiles(self, dataDirectory):
        self.logger.debug('Agent is retrieving files for %s' % dataDirectory)
        return self.fileSystemObserverAgent.getFiles(dataDirectory)

    @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):
        daqInfo = DaqTracker.getInstance().getDaqInfoByExperimentAndDataDirectory(experiment, dataDirectory)
        experimentName = experiment.get('name')

        # No daq info, ignore
        if not daqInfo:
            self.logger.debug('No daq for data directory %s and experiment %s, file path %s will not be processed' % (dataDirectoy, experimentName, experimentfilePath))
            return

        # Do not process hidden files unless requested
        if not ValueUtility.toBoolean(daqInfo.get('processHiddenFiles')):
            fileName = os.path.basename(filePath)
            if fileName.startswith('.'):
                self.logger.debug('File path %s is hidden file, will not process it' % filePath)
                return

        observedFile = self.observedFileMap.get(filePath)
        if not observedFile:
            daqInfo.fileAdded(filePath)
            observedFile = ObservedFile(filePath=filePath, dataDirectory=dataDirectory, experiment=experiment)
            observedFile['daqInfo'] = daqInfo.toDictWithOriginalKeys()
            observedFile['statusMonitor'] = daqInfo
            self.observedFileMap[filePath] = observedFile
            self.logger.debug('New observed file: %s', filePath)
        else:
            self.logger.debug('Observed file updated: %s', filePath)
        observedFile.setLastUpdateTimeToNow()
       
    @ThreadingUtility.synchronize
    def checkObservedFilesForProcessing(self):
        now = time.time()
        filePathsForProcessing = []
        for (filePath,observedFile) in self.observedFileMap.items():
            timestamp = observedFile.get('lastUpdateTime')
            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:
                filePathsForProcessing = self.checkObservedFilesForProcessing()
                if len(filePathsForProcessing):
                    self.logger.debug('Checking observed files')
                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()