"""This module provide file watcher toolkit.
Whole module is built on a python package named `watchdog <https://github.com/gorakhargosh/watchdog>`_.
"""
from PySide6.QtCore import QObject, Signal
from watchdog.observers import Observer
from watchdog.events import FileSystemEvent, PatternMatchingEventHandler, RegexMatchingEventHandler, LoggingEventHandler
_observer = Observer()
[docs]class PatternWatcher(QObject):
"""Watch files match given patterns.
"""
[docs] sigCreated = Signal(FileSystemEvent)
"""PySide6.QtCore.Signal(FileSystemEvent): Signal when a file or directory is created."""
[docs] sigDeleted = Signal(FileSystemEvent)
"""PySide6.QtCore.Signal(FileSystemEvent): Signal when a file or directory is deleted."""
[docs] sigModified = Signal(FileSystemEvent)
"""PySide6.QtCore.Signal(FileSystemEvent): Signal when a file or directory is modified."""
[docs] sigMoved = Signal(FileSystemEvent)
"""PySide6.QtCore.Signal(FileSystemEvent): Signal when a file or directory is moved or renamed."""
def __init__(self,
patterns = None, ignorePatterns = None,
ignoreDirectories = True, caseSensitive = True,
onCreated = None, onDeleted = None, onModified = None, onMoved = None):
"""
Args:
list[str] patterns: Patterns to allow matching event paths.
list[str] ignorePatterns: Patterns to ignore matching event paths.
bool ignoreDirectories: ``True`` if directories should be ignored; ``False`` otherwise.
bool caseSensitive: ``True`` if path names should be matched sensitive to case; ``False`` otherwise.
function onCreated: Called when a file or directory is created.
function onDeleted: Called when a file or directory is deleted.
function onModified: Called when a file or directory is modified.
function onMoved: Called when a file or directory is moved or renamed.
"""
super().__init__()
[docs] self.handler = PatternMatchingEventHandler(
patterns = patterns,
ignore_patterns = ignorePatterns,
ignore_directories = ignoreDirectories,
case_sensitive = caseSensitive
)
"""PatternMatchingEventHandler: Internal ``PatternMatchingEventHandler`` instance."""
if onCreated:
self.sigCreated.connect(onCreated)
self.handler.on_created = self.sigCreated.emit
if onDeleted:
self.sigDeleted.connect(onDeleted)
self.handler.on_deleted = self.sigDeleted.emit
if onModified:
self.sigModified.connect(onModified)
self.handler.on_modified = self.sigModified.emit
if onMoved:
self.sigMoved.connect(onMoved)
self.handler.on_moved = self.sigMoved.emit
[docs] def start(self, path, recursive = True):
"""Schedules watching a path and calls appropriate methods specified in the given event handler in response to file system events.
Args:
str path: Directory path that will be monitored.
bool recursive: ``True`` if events will be emitted for sub-directories traversed recursively; ``False`` otherwise.
"""
_observer.schedule(self.handler, path, recursive)
_observer.start()
[docs] def stop(self):
"""Unschedules this watch.
"""
_observer.unschedule(self.handler)
[docs]class RegexWatcher(QObject):
"""Watch files match given regexes.
"""
[docs] sigCreated = Signal(FileSystemEvent)
"""PySide6.QtCore.Signal(FileSystemEvent): Signal when a file or directory is created."""
[docs] sigDeleted = Signal(FileSystemEvent)
"""PySide6.QtCore.Signal(FileSystemEvent): Signal when a file or directory is deleted."""
[docs] sigModified = Signal(FileSystemEvent)
"""PySide6.QtCore.Signal(FileSystemEvent): Signal when a file or directory is modified."""
[docs] sigMoved = Signal(FileSystemEvent)
"""PySide6.QtCore.Signal(FileSystemEvent): Signal when a file or directory is moved or renamed."""
def __init__(self,
regexes = ['.*'], ignoreRegexes = [],
ignoreDirectories = True, caseSensitive = True,
onCreated = None, onDeleted = None, onModified = None, onMoved = None):
"""
Args:
list[str] regexes: Regexes to allow matching event paths.
list[str] ignoreRegexes: Regexes to ignore matching event paths.
bool ignoreDirectories: ``True`` if directories should be ignored; ``False`` otherwise.
bool caseSensitive: ``True`` if path names should be matched sensitive to case; ``False`` otherwise.
function onCreated: Called when a file or directory is created.
function onDeleted: Called when a file or directory is deleted.
function onModified: Called when a file or directory is modified.
function onMoved: Called when a file or directory is moved or renamed.
"""
super().__init__()
[docs] self.handler = RegexMatchingEventHandler(
regexes = regexes,
ignore_regexes = ignoreRegexes,
ignore_directories = ignoreDirectories,
case_sensitive = caseSensitive
)
"""RegexMatchingEventHandler: Internal ``RegexMatchingEventHandler`` instance."""
self.handler.on_modified = lambda e: print(e)
if onCreated:
self.sigCreated.connect(onCreated)
self.handler.on_created = self.sigCreated.emit
if onDeleted:
self.sigDeleted.connect(onDeleted)
self.handler.on_deleted = self.sigDeleted.emit
if onModified:
self.sigModified.connect(onModified)
self.handler.on_modified = self.sigModified.emit
if onMoved:
self.sigMoved.connect(onMoved)
self.handler.on_moved = self.sigMoved.emit
[docs] def start(self, path, recursive = True):
"""Schedules watching a path and calls appropriate methods specified in the given event handler in response to file system events.
Args:
str path: Directory path that will be monitored.
bool recursive: ``True`` if events will be emitted for sub-directories traversed recursively; ``False`` otherwise.
"""
self.watch = _observer.schedule(self.handler, path, recursive)
_observer.start()
[docs] def stop(self):
"""Unschedules this watch.
"""
_observer.unschedule(self.watch)
self.watch = None
[docs]def stopAllWatches():
"""Unschedules all watches and detaches all associated event handlers.
"""
_observer.unschedule_all()
_observer.stop()