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

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

from time import time
from typing import Iterable

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.__class__.__name__}({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 f'{self.__class__.__name__}({self.src_path}, {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, paths, patterns=None, exclude=None, recursive=False):
        self._paths = paths
        self._patterns = patterns
        self._exclude = exclude
        self._recursive = recursive
        self._observer = Observer()
        self._reset(paths, patterns, exclude)

    def _reset(self, paths, 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()
        if isinstance(paths, str):
            self._observer.schedule(self.handler, paths, self._recursive)
        elif isinstance(paths, Iterable):
            for path in paths:
                self._observer.schedule(self.handler, path, self._recursive)
        else:
            raise ValueError('Expected one or more strings representing paths.')

    @property
    def paths(self):
        return self._paths

    @paths.setter
    def paths(self, paths):
        self._paths = paths
        self._reset(paths, self._patterns, self._exclude)

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

    @patterns.setter
    def patterns(self, patterns):
        self._patterns = patterns
        self._reset(self._paths, patterns, self._exclude)

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

    @exclude.setter
    def exclude(self, exclude):
        self._exclude = exclude
        self._reset(self._paths, 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