baseimage-tutorial-framework/lib/tfw/components/inotify/inotify.py

# pylint: disable=too-few-public-methods

from time import time

from watchdog.observers import Observer
from watchdog.events import FileSystemMovedEvent, PatternMatchingEventHandler
from watchdog.events import (
    FileCreatedEvent, FileModifiedEvent, FileMovedEvent, FileDeletedEvent,
    DirCreatedEvent, DirModifiedEvent, DirMovedEvent, DirDeletedEvent
)


class InotifyEvent:
    def __init__(self, src_path):
        self.date = time()
        self.src_path = src_path

    def __str__(self):
        return self.__repr__()

    def __repr__(self):
        return f'{self.date}\t{self.__class__.__name__}:\t{self.src_path}'


class InotifyMovedEvent(InotifyEvent):
    def __init__(self, src_path, dest_path):
        self.dest_path = dest_path
        super().__init__(src_path)

    def __repr__(self):
        return super().__repr__()+f'\t-> {self.dest_path}'


class InotifyFileCreatedEvent(InotifyEvent):
    pass


class InotifyFileModifiedEvent(InotifyEvent):
    pass


class InotifyFileMovedEvent(InotifyMovedEvent):
    pass


class InotifyFileDeletedEvent(InotifyEvent):
    pass


class InotifyDirCreatedEvent(InotifyEvent):
    pass


class InotifyDirModifiedEvent(InotifyEvent):
    pass


class InotifyDirMovedEvent(InotifyMovedEvent):
    pass


class InotifyDirDeletedEvent(InotifyEvent):
    pass


class InotifyObserver:
    def __init__(self, path, patterns=None, exclude=None):
        self._path = path
        self._patterns = patterns
        self._exclude = exclude
        self.observer = Observer()
        self.reset(path, patterns, exclude)

    def reset(self, path, patterns, exclude):
        dispatch_event = self.dispatch_event
        class TransformerEventHandler(PatternMatchingEventHandler):
            def on_any_event(self, event):
                dispatch_event(event)
        self.handler = TransformerEventHandler(patterns, exclude)
        self.observer.unschedule_all()
        self.observer.schedule(self.handler, path, recursive=True)

    @property
    def path(self):
        return self._path

    @path.setter
    def path(self, path):
        self._path = path
        self.reset(path, self._patterns, self._exclude)

    @property
    def patterns(self):
        return self._patterns

    @patterns.setter
    def patterns(self, patterns):
        self._patterns = patterns
        self.reset(self._path, patterns, self._exclude)

    @property
    def exclude(self):
        return self._exclude

    @exclude.setter
    def exclude(self, exclude):
        self._exclude = exclude
        self.reset(self._path, self._patterns, exclude)

    def start(self):
        self.observer.start()

    def stop(self):
        self.observer.stop()
        self.observer.join()

    def dispatch_event(self, event):
        event_to_action = {
            InotifyFileCreatedEvent  : self.on_created,
            InotifyFileModifiedEvent : self.on_modified,
            InotifyFileMovedEvent    : self.on_moved,
            InotifyFileDeletedEvent  : self.on_deleted,
            InotifyDirCreatedEvent   : self.on_created,
            InotifyDirModifiedEvent  : self.on_modified,
            InotifyDirMovedEvent     : self.on_moved,
            InotifyDirDeletedEvent   : self.on_deleted
        }

        event = self.transform_event(event)
        self.on_any_event(event)
        event_to_action[type(event)](event)

    @staticmethod
    def transform_event(event):
        watchdog_to_inotify = {
            FileCreatedEvent  : InotifyFileCreatedEvent,
            FileModifiedEvent : InotifyFileModifiedEvent,
            FileMovedEvent    : InotifyFileMovedEvent,
            FileDeletedEvent  : InotifyFileDeletedEvent,
            DirCreatedEvent   : InotifyDirCreatedEvent,
            DirModifiedEvent  : InotifyDirModifiedEvent,
            DirMovedEvent     : InotifyDirMovedEvent,
            DirDeletedEvent   : InotifyDirDeletedEvent
        }

        try:
            cls = watchdog_to_inotify[type(event)]
        except KeyError:
            raise NameError('Watchdog API returned an unknown event.')

        if isinstance(event, FileSystemMovedEvent):
            return cls(event.src_path, event.dest_path)
        return cls(event.src_path)

    def on_any_event(self, event):
        pass

    def on_created(self, event):
        pass

    def on_modified(self, event):
        pass

    def on_moved(self, event):
        pass

    def on_deleted(self, event):
        pass