Lucent Watcher

lucent_watcher is an extension module for lucent that monitors filesystem events relevant to a Codex.

A watcher monitors directories and triggers callbacks when files matching specific Conventions are created, deleted, or moved.

Requirements

• Python >= 3.9

Note: The new type annotation syntax was introduced in Python 3.9 (PEP 585), and while there is no plan for removal at the moment, using modern annotations is more future-proof.

Installation

Use your preferred package installer:

pip install lucent-watcher

uv add lucent-watcher

poetry add lucent-watcher

To try it quickly, lucent-watcher provides an example configuration for testing purposes:

from lucent_watcher.lucent_watcher_example import main

main()

Step-by-Step Explanation

1. Define directories to monitor

Start by creating a list of directories that the Watcher should monitor.

dirs = [Path("D:/monitor_me")]

You can also dynamically build paths using your Conventions:

project_root = Path(
    codex.convs.project_root.format(
        {"project": "myAwesomeProject"}
    )
)
dirs.append(project_root)

Recursive Monitoring

Directory monitoring is recursive. Therefore, adding subfolders explicitly is unnecessary and duplicate child directories will be ignored

# This path will be ignored if `"D:/monitor_me"` is already monitored.
dirs.append(Path("D:/monitor_me/subfolder"))

2. Create the Watcher

Once directories are defined, create a Watcher. The Codex passed as argument determines which file patterns specifically need to be monitored. Files that do not match known conventions are ignored.

watcher = Watcher(
    codex=codex,
    watch_dirs=dirs
)

3. Create a WatcherProtocol

A WatcherProtocol defines what event to watch and what to execute when an event occurs. This example will run a custom callback when a Maya file is created.

def created_asset_maya_file(
    path: Path,
    convention: Convention,
    fields: dict[str, str]
):
    print("Asset Maya file created")
    print(f"- Path : {path}")
    print(f"- Convention : {convention}")
    print(f"- Fields : {fields}")

protocol = WatcherProtocol(
    codex.convs.asset_maya_file,
    on_created=created_asset_maya_file
)
watcher.register(protocol)

Supported Events

A WatcherProtocol can handle three types of events:

• Creation
• Deletion
• Move / Rename

Note that the "on_moved" callback has more arguments, because it needs to handle both the source and destination.

def deleted_asset_maya_file(
    path: Path,
    convention: Convention,
    fields: dict[str, str]
):
    print(f"The file {path.name} was deleted")

def moved_asset_maya_file(
    src: Path,
    dst: Path,
    src_convention: Convention,
    dst_convention: Convention,
    src_fields: dict[str, str],
    dst_fields: dict[str, str],
):
    print(f"The file {src.name} was moved to {dst}")

protocol = WatcherProtocol(
    codex.convs.asset_maya_file,
    on_created=created_asset_maya_file,
    on_deleted=deleted_asset_maya_file,
    on_moved=moved_asset_maya_file,
)
watcher.register(protocol)

Note: Moving/renaming files

Conceptually, moving a file is equivalent to deleting it and creating it elsewhere.

If the "on_moved" callback is not defined, the "on_created" and "on_deleted" callbacks will be triggered instead. Define on_moved when:

• You need precise control over rename/move operations
• You want to avoid duplicate logic from create/delete events

6. Start the Watcher

We are all set! Finally, start the watcher to begin monitoring directories.

watcher.start()

This begins the file monitoring loop, the watcher will continue running until the program exits. You may also set blocking=False if your watcher runs in a gui or something.

Full code

"""
This script demonstrates how to set up a Lucent Watcher service.
"""

import logging
from pathlib import Path

from lucent.lucent_example_config import Convention, codex

from lucent_watcher import Watcher, WatcherProtocol


def main():
    # Directories to monitor
    dirs = [Path("D:/monitor_me")]
    project_root = Path(codex.convs.project_root.format({"project": "myAwesomeProject"}))
    dirs.append(project_root)

    # Watcher
    watcher = Watcher(codex=codex, watch_dirs=dirs)

    # Callbacks
    def created_asset_maya_file(path: Path, convention: Convention, fields: dict[str, str]):
        print("Asset Maya file created")
        print(f"- Path : {path}")
        print(f"- Convention : {convention}")
        print(f"- Fields : {fields}")

    def deleted_asset_maya_file(path: Path, convention: Convention, fields: dict[str, str]):
        print(f"The file {path.name} was deleted")

    def moved_asset_maya_file(
        src: Path,
        dst: Path,
        src_convention: Convention,
        dst_convention: Convention,
        src_fields: dict[str, str],
        dst_fields: dict[str, str],
    ):
        print(f"The file {src.name} was moved to {dst}")
        print(f"{src_convention.name} -> {dst_convention.name}")
        print(src_fields)
        print(dst_fields)

    # WatcherProtocols
    protocol = WatcherProtocol(
        codex.convs.asset_maya_file,
        on_created=created_asset_maya_file,
        on_deleted=deleted_asset_maya_file,
        on_moved=moved_asset_maya_file,
    )
    watcher.register(protocol)

    # Start Watcher
    watcher.start()


def _main():
    logging.basicConfig(
        level=logging.INFO,
        format="%(asctime)s %(levelname)-4s  %(message)s",
        datefmt="%Y-%m-%dT%H:%M:%S",
    )
    main()


if __name__ == "__main__":
    _main()

Acknowledgements

• Big Company: while being a personal project, Lucent was tested and improved there, so shoutout to the team!
• Lucidity team: The years I've spent working with Lucidity were a great inspiration for Lucent.
• The VFX/Animation community