Module watch

Expand source code
# Copyright (c) 2016-2021 InSeven Limited
#
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:
#
# The above copyright notice and this permission notice shall be included in all
# copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
# SOFTWARE.

import atexit
import logging
import os
import signal
import subprocess
import sys
import threading
import time
import urllib
import webbrowser

import watchdog.events
import watchdog.observers

import incontext
import paths


class CallbackEventHandler(watchdog.events.FileSystemEventHandler):
    """Logs all the events captured."""

    def __init__(self, callback):
        self._callback = callback

    def on_moved(self, event):
        super(CallbackEventHandler, self).on_moved(event)
        self._callback()

    def on_created(self, event):
        self._callback()

    def on_deleted(self, event):
        self._callback()

    def on_modified(self, event):
        self._callback()


def watch_directory(paths, callback):
    observer = watchdog.observers.Observer()
    for path in paths:
        observer.schedule(CallbackEventHandler(callback=callback), path, recursive=True)
    observer.start()
    return observer


class Builder(threading.Thread):

    def __init__(self, incontext, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.incontext = incontext
        self.scheduled = False
        self.lock = threading.Lock()
        self.stopped = False

    def schedule(self):
        logging.info("Scheduling build...")
        with self.lock:
            self.scheduled = True

    def stop(self):
        logging.info("Stopping builder...")
        with self.lock:
            self.stopped = True

    def run(self):
        while True:
            time.sleep(1)
            scheduled = False
            with self.lock:
                if self.stopped:
                    return
                scheduled = self.scheduled
                self.scheduled = False
            if scheduled:
                try:
                    self.incontext.commands["build"].run()
                    logging.info("Done.")
                except Exception as e:
                    logging.error("Failed: %s", e)


def docker(command):
    prefix = []
    if sys.platform == "linux":
        prefix = ["sudo"]
    return subprocess.run(prefix + ["docker"] + command)


@incontext.command("watch", help="watch for changes and automatically build the website")
def command_watch(incontext, options):
    builder = Builder(incontext)
    builder.start()
    logging.info("Watching directory...")
    observer = watch_directory([incontext.configuration.site.paths.content,
                                incontext.configuration.site.paths.templates],
                               builder.schedule)
    logging.info("Performing initial build...")
    builder.schedule()
    try:
        while True:
            time.sleep(0.2)
    except KeyboardInterrupt:
        builder.stop()
        observer.stop()
    observer.join()
    builder.join()

Functions

def command_watch(incontext, options)
Expand source code
@incontext.command("watch", help="watch for changes and automatically build the website")
def command_watch(incontext, options):
    builder = Builder(incontext)
    builder.start()
    logging.info("Watching directory...")
    observer = watch_directory([incontext.configuration.site.paths.content,
                                incontext.configuration.site.paths.templates],
                               builder.schedule)
    logging.info("Performing initial build...")
    builder.schedule()
    try:
        while True:
            time.sleep(0.2)
    except KeyboardInterrupt:
        builder.stop()
        observer.stop()
    observer.join()
    builder.join()
def docker(command)
Expand source code
def docker(command):
    prefix = []
    if sys.platform == "linux":
        prefix = ["sudo"]
    return subprocess.run(prefix + ["docker"] + command)
def watch_directory(paths, callback)
Expand source code
def watch_directory(paths, callback):
    observer = watchdog.observers.Observer()
    for path in paths:
        observer.schedule(CallbackEventHandler(callback=callback), path, recursive=True)
    observer.start()
    return observer

Classes

class Builder (incontext, *args, **kwargs)

A class that represents a thread of control.

This class can be safely subclassed in a limited fashion. There are two ways to specify the activity: by passing a callable object to the constructor, or by overriding the run() method in a subclass.

This constructor should always be called with keyword arguments. Arguments are:

group should be None; reserved for future extension when a ThreadGroup class is implemented.

target is the callable object to be invoked by the run() method. Defaults to None, meaning nothing is called.

name is the thread name. By default, a unique name is constructed of the form "Thread-N" where N is a small decimal number.

args is the argument tuple for the target invocation. Defaults to ().

kwargs is a dictionary of keyword arguments for the target invocation. Defaults to {}.

If a subclass overrides the constructor, it must make sure to invoke the base class constructor (Thread.init()) before doing anything else to the thread.

Expand source code
class Builder(threading.Thread):

    def __init__(self, incontext, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.incontext = incontext
        self.scheduled = False
        self.lock = threading.Lock()
        self.stopped = False

    def schedule(self):
        logging.info("Scheduling build...")
        with self.lock:
            self.scheduled = True

    def stop(self):
        logging.info("Stopping builder...")
        with self.lock:
            self.stopped = True

    def run(self):
        while True:
            time.sleep(1)
            scheduled = False
            with self.lock:
                if self.stopped:
                    return
                scheduled = self.scheduled
                self.scheduled = False
            if scheduled:
                try:
                    self.incontext.commands["build"].run()
                    logging.info("Done.")
                except Exception as e:
                    logging.error("Failed: %s", e)

Ancestors

  • threading.Thread

Methods

def run(self)

Method representing the thread's activity.

You may override this method in a subclass. The standard run() method invokes the callable object passed to the object's constructor as the target argument, if any, with sequential and keyword arguments taken from the args and kwargs arguments, respectively.

Expand source code
def run(self):
    while True:
        time.sleep(1)
        scheduled = False
        with self.lock:
            if self.stopped:
                return
            scheduled = self.scheduled
            self.scheduled = False
        if scheduled:
            try:
                self.incontext.commands["build"].run()
                logging.info("Done.")
            except Exception as e:
                logging.error("Failed: %s", e)
def schedule(self)
Expand source code
def schedule(self):
    logging.info("Scheduling build...")
    with self.lock:
        self.scheduled = True
def stop(self)
Expand source code
def stop(self):
    logging.info("Stopping builder...")
    with self.lock:
        self.stopped = True
class CallbackEventHandler (callback)

Logs all the events captured.

Expand source code
class CallbackEventHandler(watchdog.events.FileSystemEventHandler):
    """Logs all the events captured."""

    def __init__(self, callback):
        self._callback = callback

    def on_moved(self, event):
        super(CallbackEventHandler, self).on_moved(event)
        self._callback()

    def on_created(self, event):
        self._callback()

    def on_deleted(self, event):
        self._callback()

    def on_modified(self, event):
        self._callback()

Ancestors

  • watchdog.events.FileSystemEventHandler

Methods

def on_created(self, event)

Called when a file or directory is created.

:param event: Event representing file/directory creation. :type event: :class:DirCreatedEvent or :class:FileCreatedEvent

Expand source code
def on_created(self, event):
    self._callback()
def on_deleted(self, event)

Called when a file or directory is deleted.

:param event: Event representing file/directory deletion. :type event: :class:DirDeletedEvent or :class:FileDeletedEvent

Expand source code
def on_deleted(self, event):
    self._callback()
def on_modified(self, event)

Called when a file or directory is modified.

:param event: Event representing file/directory modification. :type event: :class:DirModifiedEvent or :class:FileModifiedEvent

Expand source code
def on_modified(self, event):
    self._callback()
def on_moved(self, event)

Called when a file or a directory is moved or renamed.

:param event: Event representing file/directory movement. :type event: :class:DirMovedEvent or :class:FileMovedEvent

Expand source code
def on_moved(self, event):
    super(CallbackEventHandler, self).on_moved(event)
    self._callback()