WatchFolder utility

[Rodney]

I've been messing about with a 'watchfolder' script (in python) that monitors a directory and when it finds a new sequence of PNG images it converts the sequence to MP4 video and moves the PNG sequence to a datetime stamped directory inside a 'processed' directory.

There are a number of things that are still rough about the program.

Firstly, the majority of people will not install python, set it up, run python programs etc. right?

Correct.

So, I compiled it into an executable .exe file.

That seems to work pretty well.

 

The python script has a watchfolder.ini file where users can quickly adjust settings.

If no .ini file is found default locations and values are used.

[settings]
watch_dir = F:/watch_folder
ffmpeg_path = ffmpeg
framerate = 24
timeout = 5
video_basename = video
max_runtime_minutes = 30
reset_timeout_on_video = true

Here we can see:

- The script uses a specific directory/folder so that's where the PNG sequence would need to be rendered to

- The script uses FFMPEG for the conversion and the path here suggests it is in the users environmental settings.  Perhaps better to specifically state where the FFMPEG executable files are located.  For example:  C:/ffmpeg/bin

- Framerate can be changed to allow more (or less) frames to be generated.

- The timeout is in seconds and suggests how long the utility waits to see if another frame is being generated.  If frames are expected to take longer than 5 seconds to render this value should be increased.

- The base name of the output video can be changed here (it's just named video by default and new videos get incremented with a number each time a new video is created (video1.mp4, video2.mp4, etc.)

- Max runtime (if set) limits how long the watchfolder program will monitor the folder.

- The timer for the max runtime can be set to refresh each time a new video is created so a new 30 minute timer starts.  Set to false if a reset of the timer is not desired.

 

The actual python script: 

import os
import time
import shutil
import keyboard
import subprocess
import re
from datetime import datetime
import configparser

# === DEFAULT CONFIG ===
DEFAULTS = {
    "watch_dir": "F:/watch_folder",
    "ffmpeg_path": "ffmpeg",
    "framerate": "24",
    "timeout": "5",
    "video_basename": "video",
    "max_runtime_minutes": "0",
    "reset_timeout_on_video": "false"
}

INI_FILE = "watchfolder.ini"

def load_config():
    config = configparser.ConfigParser()
    if not os.path.exists(INI_FILE):
        config["settings"] = DEFAULTS
        with open(INI_FILE, "w") as f:
            config.write(f)
        print(f"[i] Created default {INI_FILE}")
    else:
        config.read(INI_FILE)

    for key, val in DEFAULTS.items():
        if key not in config["settings"]:
            config["settings"][key] = val

    return config["settings"]

def get_png_files(watch_dir):
    return sorted([f for f in os.listdir(watch_dir) if f.lower().endswith('.png')])

def get_next_video_filename(watch_dir, basename):
    count = 1
    while True:
        candidate = f"{basename}_{count:04d}.mp4"
        if not os.path.exists(os.path.join(watch_dir, candidate)):
            return candidate
        count += 1

def guess_pattern(filename):
    match = re.search(r"([^.]+)\.(\d+)\.png$", filename)
    if match:
        prefix, digits = match.groups()
        return f"{prefix}.%0{len(digits)}d.png"
    return None

def convert_sequence_to_mp4(watch_dir, first_file, ffmpeg_path, framerate, video_basename):
    pattern = guess_pattern(first_file)
    if not pattern:
        print(f"[!] Could not determine pattern from {first_file}")
        return

    output_name = get_next_video_filename(watch_dir, video_basename)
    output_path = os.path.join(watch_dir, output_name)

    print(f"[+] Converting to MP4: {output_name}")
    try:
        subprocess.run([
            ffmpeg_path, "-y", "-framerate", str(framerate), "-i", pattern,
            "-c:v", "libx264", "-pix_fmt", "yuv420p", output_path
        ], cwd=watch_dir, check=True)
        print(f"[✓] Video saved as: {output_name}")
    except subprocess.CalledProcessError as e:
        print(f"[!] FFmpeg failed: {e}")

def move_sequence_to_archive(watch_dir, png_files):
    now = datetime.now().strftime("%Y%m%d_%H%M%S")
    archive_dir = os.path.join(watch_dir, "processed", now)
    os.makedirs(archive_dir, exist_ok=True)
    for f in png_files:
        shutil.move(os.path.join(watch_dir, f), os.path.join(archive_dir, f))
    print(f"[→] Moved PNGs to: {archive_dir}")

def monitor(settings):
    watch_dir = settings["watch_dir"]
    ffmpeg_path = settings["ffmpeg_path"]
    framerate = int(settings.get("framerate", 24))
    timeout = int(settings.get("timeout", 5))
    video_basename = settings["video_basename"]
    max_runtime = int(settings.get("max_runtime_minutes", "0").strip()) * 60
    reset_on_video = settings.get("reset_timeout_on_video", "false").lower() == "true"

    print(f"👁️  Monitoring folder: {watch_dir}")
    print(f"[i] FFmpeg: {ffmpeg_path}, timeout: {timeout}s, framerate: {framerate}fps")
    if max_runtime > 0:
        print(f"[i] Will auto-exit after {max_runtime // 60} minutes (unless reset)")

    start_time = time.time()
    previous_files = set(get_png_files(watch_dir))
    last_change_time = time.time()

    while True:
        # Check for Escape key press
        if keyboard.is_pressed("esc"):
            print("[✋] Escape key pressed. Exiting.")
            break

        time.sleep(1)

        # Auto-exit if timer exceeded
        if max_runtime > 0 and (time.time() - start_time > max_runtime):
            print("[!] Max runtime reached. Exiting.")
            break


        current_files = set(get_png_files(watch_dir))

        if current_files != previous_files:
            previous_files = current_files
            last_change_time = time.time()
            continue

        if current_files and (time.time() - last_change_time > timeout):
            png_files = sorted(current_files)
            print(f"[⏳] Sequence complete: {len(png_files)} files")

            convert_sequence_to_mp4(watch_dir, png_files[0], ffmpeg_path, framerate, video_basename)
            move_sequence_to_archive(watch_dir, png_files)

            previous_files = set()
            last_change_time = time.time()

            if reset_on_video:
                print("[i] Timer reset after video creation.")
                start_time = time.time()

    print("[✓] Monitoring stopped.")

if __name__ == "__main__":
    try:
        settings = load_config()
        monitor(settings)
    except KeyboardInterrupt:
        print("\n[✓] Monitoring stopped by user.")

My take is that this option for a hotwatch directory and execution of ffmpeg script would be best added to Animation:Master itself but if there is interest we can pursue this and more.

This script only converts PNG sequences to MP4 video but all manner of video formats is possible and even gif animation.

The utility currently does not have an interface/GUI but that would be a next step that allows the user to adjust settings in the interface and even opt for different outputs.

Here's the sequence I was testing with: