baseimage-tutorial-framework/lib/tfw/components/inotify/inotify.py
2019-06-27 17:42:41 +02:00

177 lines
4.7 KiB
Python

# 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