justinas/path-juggler
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Initial development by Claude

Browse Source
This commit is contained in:
Justinas Petravičius
2025-12-09 22:40:12 +02:00
parent 652e09ccdf
commit 6479b24195
6 changed files with 1280 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files

155
README.md
View File

@@ -1,2 +1,155 @@
# path-juggler
# Path Juggler
Automates file organization by watching folders and routing files to structured destinations based on matching source material.
## Current Workflow: DIT Editorial Proxies
The default configuration watches for Resolve render outputs and organizes them into editorial proxy folder structures:
1. Watches `~/Movies/AM` and `~/Movies/PM` for MXF files from Resolve
2. Parses filenames like `A080C002_251209R2.mxf` to extract the reel prefix (`A080`)
3. Searches mounted `HONEY 1`, `HONEY 2`, etc. volumes for matching bin folders
4. Creates the editorial proxy folder structure in `~/Movies/Honey Dailies Transfer`
5. Moves the rendered files to the correct location
### Example
- Resolve outputs: `~/Movies/AM/A080C002_251209R2.mxf`
- Path Juggler finds on HONEY 1: `Honey/20251209_Day 22/02 FOOTAGE/CAM_A/A080CY75/`
- Path Juggler creates: `~/Movies/Honey Dailies Transfer/20251209_Day 22 AM/03 EDITORIAL PROXIES/CAM_A/A080CY75/`
- Path Juggler moves the MXF file there
## Installation
### Option 1: Standalone App (Recommended)
Build a self-contained macOS application - no Python dependencies needed after building:
```bash
# Make the build script executable
chmod +x build_app.sh
# Build the app
./build_app.sh
```
This creates `dist/Path Juggler.app` which you can drag to your Applications folder.
**First launch:** Right-click the app and select "Open" to bypass Gatekeeper (since it's self-signed).
#### Optional: Add a custom icon
```bash
# Requires Pillow
pip3 install Pillow
python3 create_icon.py
# Then edit build_app.sh and change:
# icon=None,
# to:
# icon='AppIcon.icns',
# Rebuild
./build_app.sh
```
### Option 2: Run from Source
If you prefer to run the Python scripts directly:
```bash
pip3 install watchdog
python3 path_juggler_gui.py
```
## Usage
### GUI Application
Launch the app (double-click or from terminal):
```bash
open "Path Juggler.app"
# or
python3 path_juggler_gui.py
```
The GUI provides:
- Start/Stop button to control watching
- Live status of mounted HONEY volumes
- Real-time activity log with color-coded messages
- Clean macOS-style interface
### Command Line Interface
**Watch mode (default)** - Continuously watches for new files:
```bash
python3 path_juggler.py
```
**Dry run** - See what would happen without moving anything:
```bash
python3 path_juggler.py --dry-run
```
**Process once and exit** - Process existing files without watching:
```bash
python3 path_juggler.py --once
```
**Verbose mode** - See detailed logging:
```bash
python3 path_juggler.py -v
```
## Files
- `path_juggler_gui.py` - GUI application
- `path_juggler.py` - Command line interface
- `path_juggler_core.py` - Core logic (shared by both interfaces)
- `build_app.sh` - Build script for standalone .app
- `create_icon.py` - Optional icon generator
## Customization
Edit the configuration section at the top of `path_juggler_core.py`:
```python
WATCH_FOLDERS = {
"AM": Path.home() / "Movies" / "AM",
"PM": Path.home() / "Movies" / "PM",
}
DESTINATION_BASE = Path.home() / "Movies" / "Honey Dailies Transfer"
VOLUME_PATTERN = re.compile(r"^HONEY \d+$")
PROJECT_FOLDER = "Honey"
FOOTAGE_FOLDER = "02 FOOTAGE"
PROXIES_FOLDER = "03 EDITORIAL PROXIES"
```
After changing configuration, rebuild the app with `./build_app.sh`
## Troubleshooting
### "No HONEY volumes found"
- Make sure your backup drives are mounted
- Check they're named exactly `HONEY 1`, `HONEY 2`, etc. (with space before number)
### "Could not find bin folder for A080"
- Verify the bin exists in SilverStack's folder structure
- Check the path follows: `HONEY X/Honey/YYYYMMDD_Day XX/02 FOOTAGE/CAM_X/A080XXXX/`
### Files not being detected
- Ensure Resolve is outputting to `~/Movies/AM` or `~/Movies/PM`
- Check file extension is `.mxf` (case insensitive)
### App won't open ("unidentified developer")
- Right-click the app and select "Open" instead of double-clicking
- Or: System Preferences → Security & Privacy → "Open Anyway"
### Build fails
- Ensure you have Xcode Command Line Tools: `xcode-select --install`
- Check Python 3.8+ is installed: `python3 --version`

169
build_app.sh Normal file
View File

@@ -0,0 +1,169 @@
#!/bin/bash
#
# Build script for Path Juggler
# Creates a standalone macOS .app bundle with ad-hoc code signing
#
set -e
SCRIPT_DIR="$(cd "$(dirname "$0")" && pwd)"
APP_NAME="Path Juggler"
BUNDLE_ID="com.pathjuggler.app"
VENV_DIR="$SCRIPT_DIR/.build_venv"
DIST_DIR="$SCRIPT_DIR/dist"
echo "=== Path Juggler - Build Script ==="
echo ""
# Check for Python 3
if ! command -v python3 &> /dev/null; then
echo "Error: python3 is required but not found"
exit 1
fi
PYTHON_VERSION=$(python3 -c 'import sys; print(f"{sys.version_info.major}.{sys.version_info.minor}")')
echo "Using Python $PYTHON_VERSION"
# Create virtual environment
echo ""
echo "=== Creating virtual environment ==="
if [ -d "$VENV_DIR" ]; then
echo "Removing existing build environment..."
rm -rf "$VENV_DIR"
fi
python3 -m venv "$VENV_DIR"
source "$VENV_DIR/bin/activate"
# Upgrade pip
echo ""
echo "=== Installing build dependencies ==="
pip install --upgrade pip --quiet
pip install pyinstaller watchdog --quiet
echo "Installed:"
pip list | grep -E "(pyinstaller|watchdog)"
# Create PyInstaller spec file for more control
echo ""
echo "=== Creating build configuration ==="
cat > "$SCRIPT_DIR/path_juggler.spec" << 'SPEC'
# -*- mode: python ; coding: utf-8 -*-
import sys
from pathlib import Path
block_cipher = None
a = Analysis(
['path_juggler_gui.py'],
pathex=[],
binaries=[],
datas=[],
hiddenimports=[
'watchdog.observers',
'watchdog.observers.fsevents',
'watchdog.events',
],
hookspath=[],
hooksconfig={},
runtime_hooks=[],
excludes=[],
win_no_prefer_redirects=False,
win_private_assemblies=False,
cipher=block_cipher,
noarchive=False,
)
pyz = PYZ(a.pure, a.zipped_data, cipher=block_cipher)
exe = EXE(
pyz,
a.scripts,
[],
exclude_binaries=True,
name='Path Juggler',
debug=False,
bootloader_ignore_signals=False,
strip=False,
upx=True,
console=False, # No terminal window
disable_windowed_traceback=False,
argv_emulation=True, # Important for macOS
target_arch=None,
codesign_identity=None,
entitlements_file=None,
)
coll = COLLECT(
exe,
a.binaries,
a.zipfiles,
a.datas,
strip=False,
upx=True,
upx_exclude=[],
name='Path Juggler',
)
app = BUNDLE(
coll,
name='Path Juggler.app',
icon=None,
bundle_identifier='com.pathjuggler.app',
info_plist={
'CFBundleName': 'Path Juggler',
'CFBundleDisplayName': 'Path Juggler',
'CFBundleVersion': '1.0.0',
'CFBundleShortVersionString': '1.0.0',
'NSHighResolutionCapable': True,
'LSMinimumSystemVersion': '10.13.0',
'NSAppleEventsUsageDescription': 'Path Juggler needs to access files.',
},
)
SPEC
# Build the app
echo ""
echo "=== Building application ==="
pyinstaller --clean --noconfirm path_juggler.spec
# Sign the app with ad-hoc signature
echo ""
echo "=== Signing application (ad-hoc) ==="
APP_PATH="$DIST_DIR/Path Juggler.app"
if [ -d "$APP_PATH" ]; then
# Remove any existing signatures
codesign --remove-signature "$APP_PATH" 2>/dev/null || true
# Sign with ad-hoc signature (dash means ad-hoc)
codesign --force --deep --sign - "$APP_PATH"
echo "Signature applied"
codesign -dv "$APP_PATH" 2>&1 | head -5
else
echo "Error: App bundle not found at $APP_PATH"
exit 1
fi
# Clean up
echo ""
echo "=== Cleaning up ==="
rm -rf "$SCRIPT_DIR/build"
rm -f "$SCRIPT_DIR/path_juggler.spec"
deactivate
rm -rf "$VENV_DIR"
echo ""
echo "=== Build complete ==="
echo ""
echo "Application created at:"
echo " $APP_PATH"
echo ""
echo "To install, drag the app to your Applications folder:"
echo " open \"$DIST_DIR\""
echo ""
echo "Note: On first launch, you may need to right-click and select 'Open'"
echo "to bypass Gatekeeper since this is a self-signed app."

134
create_icon.py Normal file
View File

@@ -0,0 +1,134 @@
#!/usr/bin/env python3
"""
Generate a simple icon for Path Juggler.
Creates an .icns file for use with the macOS app bundle.
Requires: pip install Pillow
"""
import subprocess
import tempfile
from pathlib import Path
try:
from PIL import Image, ImageDraw
except ImportError:
print("Pillow not installed. Install with: pip3 install Pillow")
print("Or skip icon generation - the app will use a default icon.")
exit(1)
def create_icon_image(size: int) -> Image.Image:
"""Create a single icon image at the given size."""
img = Image.new('RGBA', (size, size), (0, 0, 0, 0))
draw = ImageDraw.Draw(img)
# Background - purple/blue gradient-like color (juggler theme)
margin = int(size * 0.08)
radius = int(size * 0.18)
# Draw rounded rectangle background
draw.rounded_rectangle(
[margin, margin, size - margin, size - margin],
radius=radius,
fill=(88, 86, 214, 255) # Purple-blue
)
center_x = size // 2
center_y = size // 2
# Draw three "juggling balls" in an arc
ball_radius = int(size * 0.09)
arc_radius = int(size * 0.22)
# Ball positions (arc above center)
balls = [
(center_x - arc_radius, center_y - int(size * 0.08)), # Left
(center_x, center_y - int(size * 0.18)), # Top center
(center_x + arc_radius, center_y - int(size * 0.08)), # Right
]
ball_colors = [
(255, 107, 107, 255), # Red/coral
(78, 205, 196, 255), # Teal
(255, 230, 109, 255), # Yellow
]
for (bx, by), color in zip(balls, ball_colors):
draw.ellipse(
[bx - ball_radius, by - ball_radius, bx + ball_radius, by + ball_radius],
fill=color
)
# Draw path arrows (two crossing paths)
arrow_color = (255, 255, 255, 200)
line_width = max(2, int(size * 0.025))
# Left-to-right arrow
arrow_y = center_y + int(size * 0.15)
arrow_start = margin + int(size * 0.1)
arrow_end = size - margin - int(size * 0.1)
draw.line(
[(arrow_start, arrow_y), (arrow_end - int(size * 0.08), arrow_y)],
fill=arrow_color,
width=line_width
)
# Arrow head
head_size = int(size * 0.06)
draw.polygon([
(arrow_end, arrow_y),
(arrow_end - head_size, arrow_y - head_size // 2),
(arrow_end - head_size, arrow_y + head_size // 2),
], fill=arrow_color)
return img
def create_icns(output_path: Path):
"""Create an .icns file with all required sizes."""
sizes = [16, 32, 64, 128, 256, 512, 1024]
with tempfile.TemporaryDirectory() as tmpdir:
iconset_path = Path(tmpdir) / "icon.iconset"
iconset_path.mkdir()
for size in sizes:
img = create_icon_image(size)
img.save(iconset_path / f"icon_{size}x{size}.png")
if size <= 512:
img_2x = create_icon_image(size * 2)
img_2x.save(iconset_path / f"icon_{size}x{size}@2x.png")
result = subprocess.run(
["iconutil", "-c", "icns", str(iconset_path), "-o", str(output_path)],
capture_output=True,
text=True
)
if result.returncode != 0:
print(f"iconutil failed: {result.stderr}")
return False
return True
def main():
script_dir = Path(__file__).parent
output_path = script_dir / "AppIcon.icns"
print("Generating Path Juggler icon...")
if create_icns(output_path):
print(f"Icon created: {output_path}")
print("\nTo use this icon in the build:")
print("1. Edit build_app.sh")
print("2. Find the line: icon=None,")
print("3. Change it to: icon='AppIcon.icns',")
else:
print("Failed to create icon")
if __name__ == "__main__":
main()

99
path_juggler.py Normal file
View File

@@ -0,0 +1,99 @@
#!/usr/bin/env python3
"""
Path Juggler - Command Line Interface
Watches folders for rendered files and automatically organizes them
into structured folder hierarchies based on source material locations.
Usage:
python3 path_juggler.py [--dry-run] [--once] [-v]
For the GUI version, run:
python3 path_juggler_gui.py
"""
import argparse
import logging
import time
from path_juggler_core import PathJuggler, get_honey_volumes
# Setup logging for CLI
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s - %(levelname)s - %(message)s",
datefmt="%H:%M:%S"
)
logger = logging.getLogger(__name__)
def main():
parser = argparse.ArgumentParser(
description="Watch for rendered files and organize into structured folders"
)
parser.add_argument(
"--dry-run",
action="store_true",
help="Show what would be done without actually moving files"
)
parser.add_argument(
"--once",
action="store_true",
help="Process existing files and exit (don't watch)"
)
parser.add_argument(
"--verbose", "-v",
action="store_true",
help="Enable verbose logging"
)
parser.add_argument(
"--gui",
action="store_true",
help="Launch the GUI version"
)
args = parser.parse_args()
# Launch GUI if requested
if args.gui:
from path_juggler_gui import main as gui_main
gui_main()
return
if args.verbose:
logging.getLogger().setLevel(logging.DEBUG)
if args.dry_run:
logger.info("DRY RUN MODE - no files will be moved")
# Create juggler instance
juggler = PathJuggler(dry_run=args.dry_run)
if args.once:
# Just process existing files and exit
juggler.setup_folders()
volumes = get_honey_volumes()
if volumes:
logger.info(f"Found HONEY volumes: {[v.name for v in volumes]}")
else:
logger.warning("No HONEY volumes currently mounted")
logger.info("Processing existing files...")
juggler.process_existing()
logger.info("Done")
return
# Start watching
juggler.start()
try:
while juggler.is_running():
time.sleep(1)
except KeyboardInterrupt:
logger.info("Interrupted")
finally:
juggler.stop()
if __name__ == "__main__":
main()

398
path_juggler_core.py Normal file
View File

@@ -0,0 +1,398 @@
"""
Path Juggler - Core Logic
This module contains the file watching and processing logic,
separated from any UI concerns.
"""
import os
import re
import shutil
import subprocess
import time
import logging
from pathlib import Path
from threading import Thread, Event
from typing import Callable
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler
# Configuration
WATCH_FOLDERS = {
"AM": Path.home() / "Movies" / "AM",
"PM": Path.home() / "Movies" / "PM",
}
DESTINATION_BASE = Path.home() / "Movies" / "Honey Dailies Transfer"
VOLUMES_PATH = Path("/Volumes")
VOLUME_PATTERN = re.compile(r"^HONEY \d+$")
PROJECT_FOLDER = "Honey"
FOOTAGE_FOLDER = "02 FOOTAGE"
PROXIES_FOLDER = "03 EDITORIAL PROXIES"
# Regex to parse filenames like A080C002_251209R2.mxf
FILENAME_PATTERN = re.compile(
r"^([A-Z])(\d{3})C(\d{3})_(\d{6})([A-Z0-9]*)\.mxf$",
re.IGNORECASE
)
logger = logging.getLogger(__name__)
def is_file_open(file_path: Path) -> bool:
"""Check if a file is currently open by any process using lsof."""
try:
result = subprocess.run(
["lsof", str(file_path)],
capture_output=True,
text=True,
timeout=10
)
return result.returncode == 0
except subprocess.TimeoutExpired:
logger.warning(f"lsof timed out for {file_path}")
return True
except Exception as e:
logger.warning(f"lsof failed: {e}")
return True
def get_file_owner_process(file_path: Path) -> str | None:
"""Get the name of the process that has the file open."""
try:
result = subprocess.run(
["lsof", "-t", str(file_path)],
capture_output=True,
text=True,
timeout=10
)
if result.returncode == 0 and result.stdout.strip():
pid = result.stdout.strip().split('\n')[0]
ps_result = subprocess.run(
["ps", "-p", pid, "-o", "comm="],
capture_output=True,
text=True,
timeout=5
)
if ps_result.returncode == 0:
return ps_result.stdout.strip()
except Exception:
pass
return None
def wait_for_file_ready(
file_path: Path,
stop_event: Event | None = None,
check_interval: float = 2.0,
stable_duration: float = 5.0,
timeout: float = 3600.0
) -> bool:
"""Wait until a file is no longer being written to."""
start_time = time.time()
last_size = -1
size_stable_since = None
logged_process = False
logger.info(f"Waiting for file to be ready: {file_path.name}")
while time.time() - start_time < timeout:
# Check if we should stop
if stop_event and stop_event.is_set():
logger.info(f"Stop requested, abandoning: {file_path.name}")
return False
if not file_path.exists():
logger.warning(f"File disappeared: {file_path.name}")
return False
if not is_file_open(file_path):
logger.info(f"File closed by all processes: {file_path.name}")
return True
if not logged_process:
owner = get_file_owner_process(file_path)
if owner:
logger.info(f"File held by: {owner}")
logged_process = True
current_size = file_path.stat().st_size
if current_size == last_size:
if size_stable_since is None:
size_stable_since = time.time()
elif time.time() - size_stable_since >= stable_duration:
time.sleep(1)
if not is_file_open(file_path):
logger.info(f"File ready (closed): {file_path.name}")
return True
else:
size_stable_since = None
size_mb = current_size / (1024 * 1024)
logger.debug(f"File growing: {file_path.name} ({size_mb:.1f} MB)")
last_size = current_size
time.sleep(check_interval)
logger.error(f"Timeout waiting for file: {file_path.name}")
return False
def get_honey_volumes() -> list[Path]:
"""Find all mounted HONEY volumes."""
if not VOLUMES_PATH.exists():
return []
volumes = []
for item in VOLUMES_PATH.iterdir():
if item.is_dir() and VOLUME_PATTERN.match(item.name):
volumes.append(item)
return sorted(volumes)
def parse_filename(filename: str) -> dict | None:
"""Parse an MXF filename to extract camera and reel info."""
match = FILENAME_PATTERN.match(filename)
if not match:
return None
camera, reel, clip, date, suffix = match.groups()
return {
"camera": camera.upper(),
"reel": reel,
"clip": clip,
"date": date,
"suffix": suffix,
"reel_prefix": f"{camera.upper()}{reel}",
"original_filename": filename,
}
def find_bin_folder(reel_prefix: str, volumes: list[Path]) -> dict | None:
"""Search HONEY volumes for a bin folder matching the reel prefix."""
for volume in volumes:
project_path = volume / PROJECT_FOLDER
if not project_path.exists():
continue
for day_folder in project_path.iterdir():
if not day_folder.is_dir():
continue
footage_path = day_folder / FOOTAGE_FOLDER
if not footage_path.exists():
continue
for cam_folder in footage_path.iterdir():
if not cam_folder.is_dir():
continue
for bin_folder in cam_folder.iterdir():
if not bin_folder.is_dir():
continue
if bin_folder.name.upper().startswith(reel_prefix.upper()):
return {
"volume": volume,
"day_folder": day_folder.name,
"camera_folder": cam_folder.name,
"bin_folder": bin_folder.name,
"full_path": bin_folder,
}
return None
def build_destination_path(bin_info: dict, am_pm: str, destination_base: Path) -> Path:
"""Build the destination path for the proxy file."""
day_with_ampm = f"{bin_info['day_folder']} {am_pm}"
return (
destination_base
/ day_with_ampm
/ PROXIES_FOLDER
/ bin_info["camera_folder"]
/ bin_info["bin_folder"]
)
def process_file(file_path: Path, am_pm: str, dry_run: bool = False) -> bool:
"""Process a single MXF file: find matching bin and move to destination."""
filename = file_path.name
file_info = parse_filename(filename)
if not file_info:
logger.warning(f"Could not parse filename: {filename}")
return False
logger.info(f"Processing: {filename} (Reel: {file_info['reel_prefix']}, {am_pm})")
volumes = get_honey_volumes()
if not volumes:
logger.error("No HONEY volumes found!")
return False
bin_info = find_bin_folder(file_info["reel_prefix"], volumes)
if not bin_info:
logger.error(f"Could not find bin folder for {file_info['reel_prefix']}")
return False
logger.info(f"Found bin: {bin_info['bin_folder']} in {bin_info['day_folder']}")
dest_folder = build_destination_path(bin_info, am_pm, DESTINATION_BASE)
dest_file = dest_folder / filename
logger.info(f"Destination: {dest_file}")
if dry_run:
logger.info("[DRY RUN] Would create folder and move file")
return True
dest_folder.mkdir(parents=True, exist_ok=True)
try:
shutil.move(str(file_path), str(dest_file))
logger.info(f"✓ Moved: {filename}")
return True
except Exception as e:
logger.error(f"Failed to move file: {e}")
return False
class MXFHandler(FileSystemEventHandler):
"""Watchdog handler for new MXF files."""
def __init__(self, am_pm: str, stop_event: Event, dry_run: bool = False):
self.am_pm = am_pm
self.stop_event = stop_event
self.dry_run = dry_run
self.processing = set()
def on_created(self, event):
if event.is_directory:
return
file_path = Path(event.src_path)
if file_path.suffix.lower() != ".mxf":
return
if str(file_path) in self.processing:
return
self.processing.add(str(file_path))
try:
if wait_for_file_ready(file_path, self.stop_event):
process_file(file_path, self.am_pm, self.dry_run)
finally:
self.processing.discard(str(file_path))
def on_moved(self, event):
if event.is_directory:
return
dest_path = Path(event.dest_path)
if dest_path.suffix.lower() == ".mxf":
if str(dest_path) not in self.processing:
self.processing.add(str(dest_path))
try:
if wait_for_file_ready(dest_path, self.stop_event):
process_file(dest_path, self.am_pm, self.dry_run)
finally:
self.processing.discard(str(dest_path))
class PathJuggler:
"""
Main controller class for Path Juggler.
Can be used by CLI or GUI.
"""
def __init__(self, dry_run: bool = False):
self.dry_run = dry_run
self.observer: Observer | None = None
self.stop_event = Event()
self.running = False
self._thread: Thread | None = None
def setup_folders(self):
"""Ensure watch and destination folders exist."""
for name, folder in WATCH_FOLDERS.items():
folder.mkdir(parents=True, exist_ok=True)
logger.info(f"Watch folder ({name}): {folder}")
DESTINATION_BASE.mkdir(parents=True, exist_ok=True)
logger.info(f"Destination: {DESTINATION_BASE}")
def process_existing(self):
"""Process any MXF files already in watch folders."""
for am_pm, folder in WATCH_FOLDERS.items():
if not folder.exists():
continue
for file_path in folder.glob("*.mxf"):
if self.stop_event.is_set():
return
if is_file_open(file_path):
logger.info(f"Skipping (still being written): {file_path.name}")
continue
process_file(file_path, am_pm, self.dry_run)
def start(self):
"""Start watching for files."""
if self.running:
logger.warning("Already running")
return
self.stop_event.clear()
self.running = True
self.setup_folders()
# Log volumes
volumes = get_honey_volumes()
if volumes:
logger.info(f"Found volumes: {[v.name for v in volumes]}")
else:
logger.warning("No HONEY volumes currently mounted")
# Process existing files
logger.info("Checking for existing files...")
self.process_existing()
# Start observer
self.observer = Observer()
for am_pm, folder in WATCH_FOLDERS.items():
handler = MXFHandler(am_pm, self.stop_event, self.dry_run)
self.observer.schedule(handler, str(folder), recursive=False)
logger.info(f"Watching: {folder}")
self.observer.start()
logger.info("Watcher started - waiting for new files...")
def stop(self):
"""Stop watching for files."""
if not self.running:
return
logger.info("Stopping watcher...")
self.stop_event.set()
if self.observer:
self.observer.stop()
self.observer.join(timeout=5)
self.observer = None
self.running = False
logger.info("Watcher stopped")
def start_async(self):
"""Start watching in a background thread."""
self._thread = Thread(target=self.start, daemon=True)
self._thread.start()
def is_running(self) -> bool:
return self.running

326
path_juggler_gui.py Normal file
View File

@@ -0,0 +1,326 @@
#!/usr/bin/env python3
"""
Path Juggler - GUI Application
A simple macOS GUI for watching Resolve render outputs and
organizing them into editorial proxy folder structures.
"""
import tkinter as tk
from tkinter import ttk, scrolledtext
import logging
import queue
from datetime import datetime
from pathlib import Path
from path_juggler_core import (
PathJuggler,
get_honey_volumes,
WATCH_FOLDERS,
DESTINATION_BASE,
)
class QueueHandler(logging.Handler):
"""Logging handler that puts log records into a queue for GUI consumption."""
def __init__(self, log_queue: queue.Queue):
super().__init__()
self.log_queue = log_queue
def emit(self, record):
self.log_queue.put(record)
class PathJugglerApp:
"""Main GUI Application."""
def __init__(self):
self.root = tk.Tk()
self.root.title("Path Juggler")
self.root.geometry("700x500")
self.root.minsize(500, 300)
# Set macOS-like appearance
self.root.configure(bg="#f0f0f0")
# Core juggler instance
self.juggler = PathJuggler(dry_run=False)
# Logging queue for thread-safe log updates
self.log_queue: queue.Queue = queue.Queue()
# Setup logging to our queue
self._setup_logging()
# Build UI
self._create_widgets()
# Start polling the log queue
self._poll_log_queue()
# Update status periodically
self._update_status()
# Handle window close
self.root.protocol("WM_DELETE_WINDOW", self._on_close)
def _setup_logging(self):
"""Configure logging to use our queue handler."""
# Create queue handler
queue_handler = QueueHandler(self.log_queue)
queue_handler.setLevel(logging.DEBUG)
# Format
formatter = logging.Formatter("%(asctime)s - %(levelname)s - %(message)s", "%H:%M:%S")
queue_handler.setFormatter(formatter)
# Add to root logger
root_logger = logging.getLogger()
root_logger.setLevel(logging.INFO)
root_logger.addHandler(queue_handler)
# Also add to our module logger
core_logger = logging.getLogger("path_juggler_core")
core_logger.setLevel(logging.INFO)
def _create_widgets(self):
"""Create all UI widgets."""
# Main container with padding
main_frame = ttk.Frame(self.root, padding="10")
main_frame.pack(fill=tk.BOTH, expand=True)
# === Header / Status Section ===
header_frame = ttk.Frame(main_frame)
header_frame.pack(fill=tk.X, pady=(0, 10))
# Title
title_label = ttk.Label(
header_frame,
text="Path Juggler",
font=("SF Pro Display", 18, "bold")
)
title_label.pack(side=tk.LEFT)
# Status indicator
self.status_frame = ttk.Frame(header_frame)
self.status_frame.pack(side=tk.RIGHT)
self.status_dot = tk.Canvas(
self.status_frame,
width=12,
height=12,
highlightthickness=0,
bg="#f0f0f0"
)
self.status_dot.pack(side=tk.LEFT, padx=(0, 5))
self._draw_status_dot("stopped")
self.status_label = ttk.Label(
self.status_frame,
text="Stopped",
font=("SF Pro Text", 12)
)
self.status_label.pack(side=tk.LEFT)
# === Info Section ===
info_frame = ttk.LabelFrame(main_frame, text="Status", padding="10")
info_frame.pack(fill=tk.X, pady=(0, 10))
# Volumes status
volumes_frame = ttk.Frame(info_frame)
volumes_frame.pack(fill=tk.X, pady=(0, 5))
ttk.Label(volumes_frame, text="Volumes:", font=("SF Pro Text", 11, "bold")).pack(side=tk.LEFT)
self.volumes_label = ttk.Label(volumes_frame, text="Checking...", font=("SF Pro Text", 11))
self.volumes_label.pack(side=tk.LEFT, padx=(5, 0))
# Watch folders
watch_frame = ttk.Frame(info_frame)
watch_frame.pack(fill=tk.X, pady=(0, 5))
ttk.Label(watch_frame, text="Watching:", font=("SF Pro Text", 11, "bold")).pack(side=tk.LEFT)
watch_paths = ", ".join([f"~/{p.relative_to(Path.home())}" for p in WATCH_FOLDERS.values()])
ttk.Label(watch_frame, text=watch_paths, font=("SF Pro Text", 11)).pack(side=tk.LEFT, padx=(5, 0))
# Destination
dest_frame = ttk.Frame(info_frame)
dest_frame.pack(fill=tk.X)
ttk.Label(dest_frame, text="Output:", font=("SF Pro Text", 11, "bold")).pack(side=tk.LEFT)
dest_path = f"~/{DESTINATION_BASE.relative_to(Path.home())}"
ttk.Label(dest_frame, text=dest_path, font=("SF Pro Text", 11)).pack(side=tk.LEFT, padx=(5, 0))
# === Control Buttons ===
button_frame = ttk.Frame(main_frame)
button_frame.pack(fill=tk.X, pady=(0, 10))
# Style for buttons
style = ttk.Style()
style.configure("Start.TButton", font=("SF Pro Text", 12))
style.configure("Stop.TButton", font=("SF Pro Text", 12))
self.start_button = ttk.Button(
button_frame,
text="▶ Start Juggling",
command=self._start,
style="Start.TButton",
width=20
)
self.start_button.pack(side=tk.LEFT, padx=(0, 10))
self.stop_button = ttk.Button(
button_frame,
text="◼ Stop",
command=self._stop,
style="Stop.TButton",
state=tk.DISABLED,
width=15
)
self.stop_button.pack(side=tk.LEFT)
# Clear log button
clear_button = ttk.Button(
button_frame,
text="Clear Log",
command=self._clear_log,
width=10
)
clear_button.pack(side=tk.RIGHT)
# === Log Section ===
log_frame = ttk.LabelFrame(main_frame, text="Activity Log", padding="5")
log_frame.pack(fill=tk.BOTH, expand=True)
# Scrolled text widget for logs
self.log_text = scrolledtext.ScrolledText(
log_frame,
wrap=tk.WORD,
font=("SF Mono", 11),
bg="#1e1e1e",
fg="#d4d4d4",
insertbackground="#d4d4d4",
selectbackground="#264f78",
state=tk.DISABLED,
height=15
)
self.log_text.pack(fill=tk.BOTH, expand=True)
# Configure log text tags for different log levels
self.log_text.tag_configure("INFO", foreground="#d4d4d4")
self.log_text.tag_configure("WARNING", foreground="#dcdcaa")
self.log_text.tag_configure("ERROR", foreground="#f14c4c")
self.log_text.tag_configure("DEBUG", foreground="#808080")
self.log_text.tag_configure("timestamp", foreground="#6a9955")
def _draw_status_dot(self, status: str):
"""Draw the status indicator dot."""
self.status_dot.delete("all")
colors = {
"running": "#34c759", # Green
"stopped": "#8e8e93", # Gray
"error": "#ff3b30", # Red
}
color = colors.get(status, colors["stopped"])
self.status_dot.create_oval(2, 2, 10, 10, fill=color, outline=color)
def _start(self):
"""Start the file watcher."""
self.start_button.config(state=tk.DISABLED)
self.stop_button.config(state=tk.NORMAL)
self._draw_status_dot("running")
self.status_label.config(text="Juggling")
# Start juggler in background thread
self.juggler.start_async()
def _stop(self):
"""Stop the file watcher."""
self.stop_button.config(state=tk.DISABLED)
self.juggler.stop()
self.start_button.config(state=tk.NORMAL)
self._draw_status_dot("stopped")
self.status_label.config(text="Stopped")
def _clear_log(self):
"""Clear the log text area."""
self.log_text.config(state=tk.NORMAL)
self.log_text.delete(1.0, tk.END)
self.log_text.config(state=tk.DISABLED)
def _poll_log_queue(self):
"""Poll the log queue and update the text widget."""
while True:
try:
record = self.log_queue.get_nowait()
self._append_log(record)
except queue.Empty:
break
# Schedule next poll
self.root.after(100, self._poll_log_queue)
def _append_log(self, record: logging.LogRecord):
"""Append a log record to the text widget."""
self.log_text.config(state=tk.NORMAL)
# Format: [HH:MM:SS] LEVEL - Message
timestamp = datetime.fromtimestamp(record.created).strftime("%H:%M:%S")
# Insert timestamp
self.log_text.insert(tk.END, f"[{timestamp}] ", "timestamp")
# Insert message with level-based coloring
level = record.levelname
message = f"{record.getMessage()}\n"
self.log_text.insert(tk.END, message, level)
# Auto-scroll to bottom
self.log_text.see(tk.END)
self.log_text.config(state=tk.DISABLED)
def _update_status(self):
"""Update the volumes status periodically."""
volumes = get_honey_volumes()
if volumes:
names = ", ".join([v.name for v in volumes])
self.volumes_label.config(text=names, foreground="")
else:
self.volumes_label.config(text="No HONEY volumes mounted", foreground="#ff3b30")
# Check if juggler stopped unexpectedly
if not self.juggler.is_running() and self.stop_button["state"] == tk.NORMAL:
self._stop()
# Schedule next update
self.root.after(5000, self._update_status)
def _on_close(self):
"""Handle window close."""
if self.juggler.is_running():
self.juggler.stop()
self.root.destroy()
def run(self):
"""Start the application main loop."""
# Initial log message
logging.info("Path Juggler ready")
logging.info("Click 'Start Juggling' to begin")
self.root.mainloop()
def main():
app = PathJugglerApp()
app.run()
if __name__ == "__main__":
main()