python类PatternMatchingEventHandler()的实例源码

def __init__(self, runner):
        self.runner = runner

        self.event_handler = PatternMatchingEventHandler(patterns=["*.fast5"],
                ignore_patterns=[],
                ignore_directories=True)
        self.event_handler.on_created = self.on_created
        self.event_handler.on_moved = self.on_moved

        self.observer = Observer()

        self.observedPaths = []
        for path in self.runner.input:
            if os.path.isdir(path):
                self.observer.schedule(self.event_handler, path, recursive=True)
                self.observedPaths.append(path)
        log("Monitoring {} in real time. Press Ctrl+C to exit.".format(", ".join(self.observedPaths)))

def __init__(self, container, host_dir, container_dir):
        """
        Initialize a new instance of ContainerNotifier

        Args:
            container: Container
            host_dir (str): Host directory
            container_dir (str): Container directory
        """
        self.container = container
        self.host_dir = host_dir
        self.container_dir = container_dir

        event_handler = PatternMatchingEventHandler(ignore_directories=False)
        handler = self.__change_handler
        event_handler.on_created = handler
        event_handler.on_moved = handler
        event_handler.on_modified = handler

        self.observer = Observer()
        self.observer.schedule(event_handler, host_dir, recursive=True)
        self.observer.start()

def __init__(self, loop, *args, **kwargs):
        self.loop = loop

        # awaitable future to race on
        self.changed = asyncio.Future(loop=loop)

        # Continue init for EventHandler
        return super(Handler, self).__init__(*args, **kwargs)

def watch(mode, target, only_page="", pdf_file=DEFAULT_PDF_FILE,
          es_upload=NO_ES_UP):
    """Look for changed files and re-run the build whenever there's an update.
       Runs until interrupted."""
    target = get_target(target)

    class UpdaterHandler(PatternMatchingEventHandler):
        """Updates to pattern-matched files means rendering."""
        def on_any_event(self, event):
            logger.info("got event!")
            # bypass_errors=True because Watch shouldn't
            #  just die if a file is temporarily not found
            if mode == "pdf":
                make_pdf(pdf_file, target=target, bypass_errors=True,
                    only_page=only_page, es_upload=es_upload)
            else:
                render_pages(target, mode=mode, bypass_errors=True,
                            only_page=only_page, es_upload=es_upload)
            logger.info("done rendering")

    patterns = ["*template-*.html",
                "*.md",
                "*code_samples/*"]

    event_handler = UpdaterHandler(patterns=patterns)
    observer = Observer()
    observer.schedule(event_handler, config["template_path"], recursive=True)
    observer.schedule(event_handler, config["content_path"], recursive=True)
    observer.start()
    # The above starts an observing thread,
    #   so the main thread can just wait
    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        observer.stop()
    observer.join()