pinto/VFXProxy
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Initial Commit

Browse Source
This commit is contained in:
twotalesanimation
2026-07-04 15:54:19 +02:00
commit 1a2ece6f12
17 changed files with 1844 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
PROMPT.md
+162
View File
@@ -0,0 +1,162 @@
Create a Python desktop application using PySide6 called "VFX Pull Processor".
Purpose:
This application scans one or more selected folders recursively for EXR image sequences, then uses Nuke in terminal mode to render proxy movies using a template Nuke script.
Architecture:
- PySide6 GUI
- Separate modules
- app.py
- gui.py
- scanner.py
- parser.py
- jobs.py
- nuke_runner.py
- process.py (runs inside Nuke)
- config.json
The GUI should contain:
------------------------------------------------
VFX Pull Processor
Pull Folder
[_________________________] [Browse]
Output Root
[ V:\_FOOTAGE ] [Browse]
Nuke Executable
[ C:\Program Files\Nuke16.0v1\Nuke16.0.exe ] [Browse]
Template Script
[ proxy_template.nk ] [Browse]
------------------------------------------------
Sequences Found
(QTableWidget)
Columns:
Checkbox
Shot
Frames
Resolution
Status
------------------------------------------------
[ Scan ]
[ Render Selected ]
Progress Bar
Log Window
------------------------------------------------
Requirements
When Scan is pressed:
- recursively search every selected folder
- identify EXR sequences
- group frames into sequences
- detect missing frames
- determine first and last frame
- populate the table
Filename format is always:
SHOW_EPISODE_SEQUENCE_SHOT_ELEMENT_TASK_VERSION.FRAME.exr
Example:
UNG_101_012_010_BG01_TT_v001.01001.exr
Parse into:
show = UNG
episode = 101
sequence = 012
shot = 010
element = BG01
task = TT
version = v001
Create a Shot object containing:
show
episode
sequence
shot
element
task
version
basename
firstFrame
lastFrame
inputSequence
outputFolder
outputMovie
basename should equal:
UNG_101_012_010_BG01_TT_v001
Output folder should be:
V:/_FOOTAGE/
{show}/
{show}_{episode}/
{basename}/
Output files should be:
{basename}.exr
{basename}.mov
When Render Selected is pressed:
Launch Nuke ONCE using:
Nuke.exe -t process.py jobs.json
Do NOT launch Nuke once per shot.
Instead:
- create jobs.json
- pass it to process.py
process.py should:
- load proxy_template.nk
- locate nodes named:
Read_Input
Write_Output
- for each job:
set Read_Input.file
set Write_Output.file
execute render
- clear caches between jobs if needed
- print progress to stdout
The GUI should capture stdout from Nuke and display it in the log window.
Design the code cleanly using dataclasses where appropriate.
Keep parsing logic, scanning logic, GUI, and Nuke execution completely separate.
The code should be easy to extend later for:
- JPG proxies
- PNG proxies
- Contact sheets
- Thumbnail generation
- Validation reports
Do not put any pipeline logic inside the Nuke script. The Nuke script is only a visual template.
__pycache__/gui.cpython-312.pyc
BIN
View File
Binary file not shown.
__pycache__/jobs.cpython-312.pyc
BIN
View File
Binary file not shown.
__pycache__/nuke_runner.cpython-312.pyc
BIN
View File
Binary file not shown.
__pycache__/parser.cpython-312.pyc
BIN
View File
Binary file not shown.
__pycache__/scanner.cpython-312.pyc
BIN
View File
Binary file not shown.
app.py
+25
View File
@@ -0,0 +1,25 @@
"""
app.py — Entry point for VFX Pull Processor.
"""
from __future__ import annotations
import sys
from PySide6.QtWidgets import QApplication
from gui import MainWindow
def main() -> None:
app = QApplication(sys.argv)
app.setApplicationName("VFX Pull Processor")
app.setOrganizationName("VFX")
window = MainWindow()
window.show()
sys.exit(app.exec())
if __name__ == "__main__":
main()
config.json
+6
View File
@@ -0,0 +1,6 @@
{
"pull_folder": "V:/_FOOTAGE/UNG/_PULLS/UNG_111_029_SERIES_EL01_TT_GATE_ADDITIONAL_v001",
"output_root": "V:\\_FOOTAGE",
"nuke_executable": "C:/Program Files/Nuke17.0v1/Nuke17.0.exe",
"template_script": "proxy_template.nkind"
}
gui.py
+944
View File
@@ -0,0 +1,944 @@
"""
gui.py — PySide6 main window for VFX Pull Processor.
Dark, Nuke-adjacent theme with a single amber accent.
All pipeline behaviour (scan, jobs file, Nuke launch) is unchanged from
the original implementation — this module only changes presentation:
* top bar: pull folder + Scan, with rarely-used paths in a
collapsible settings panel behind the gear button
* sequences table with colour-coded status pills and a
missing-frames warning in the Frames column
* primary Render button with a live selection count
* progress bar with a "n of m · shot" label
* collapsible, timestamped, colour-coded log that auto-opens
on the first error
* drag-and-drop a folder anywhere on the window to set Pull Folder
"""
from __future__ import annotations
import html
import json
import os
import tempfile
from datetime import datetime
from pathlib import Path
from PySide6.QtCore import Qt, Slot
from PySide6.QtGui import QColor, QFont
from PySide6.QtWidgets import (
QApplication,
QCheckBox,
QFileDialog,
QFrame,
QHBoxLayout,
QHeaderView,
QLabel,
QLineEdit,
QMainWindow,
QMessageBox,
QProgressBar,
QPushButton,
QSizePolicy,
QStackedWidget,
QTableWidget,
QTableWidgetItem,
QTextEdit,
QToolButton,
QVBoxLayout,
QWidget,
)
from jobs import build_jobs_payload, write_jobs_file
from nuke_runner import NukeRunner
from parser import Shot
from scanner import ScanWorker
# ──────────────────────────────────────────────────────────────────────────────
# Palette / stylesheet
# ──────────────────────────────────────────────────────────────────────────────
ACCENT = "#f5a623"
ACCENT_HOVER = "#ffb840"
COL_TEXT = "#e4e6e8"
COL_DIM = "#9a9ea3"
COL_FAINT = "#6b6f74"
COL_GREEN = "#4caf7d"
COL_BLUE = "#4a9eda"
COL_AMBER = "#e0a83c"
COL_RED = "#e05c5c"
_QSS = f"""
QMainWindow {{
background-color: #232527;
}}
QWidget {{
color: {COL_TEXT};
font-size: 13px;
}}
QLabel {{
background: transparent;
}}
QLabel#sectionLabel {{
color: {COL_FAINT};
font-size: 11px;
font-weight: 600;
letter-spacing: 1px;
}}
QLabel#fieldLabel {{
color: {COL_DIM};
font-size: 12px;
}}
QLabel#hintLabel {{
color: {COL_FAINT};
font-size: 12px;
}}
QLineEdit {{
background-color: #1c1e20;
border: 1px solid #3a3d40;
border-radius: 6px;
padding: 7px 10px;
color: {COL_TEXT};
selection-background-color: {ACCENT};
selection-color: #1a1a1a;
}}
QLineEdit:focus {{
border-color: {ACCENT};
}}
QPushButton {{
background-color: #2b2d30;
border: 1px solid #3a3d40;
border-radius: 6px;
padding: 8px 18px;
color: {COL_TEXT};
}}
QPushButton:hover {{
background-color: #333639;
border-color: #46494d;
}}
QPushButton:pressed {{
background-color: #1c1e20;
}}
QPushButton:disabled {{
color: {COL_FAINT};
background-color: #26282a;
border-color: #313437;
}}
QPushButton#primaryButton {{
background-color: {ACCENT};
border: 1px solid {ACCENT};
color: #1a1a1a;
font-weight: 600;
padding: 9px 22px;
}}
QPushButton#primaryButton:hover {{
background-color: {ACCENT_HOVER};
border-color: {ACCENT_HOVER};
}}
QPushButton#primaryButton:disabled {{
background-color: #4d4a42;
border-color: #4d4a42;
color: #8a8a8a;
}}
QToolButton {{
background-color: #2b2d30;
border: 1px solid #3a3d40;
border-radius: 6px;
padding: 7px 10px;
color: {COL_DIM};
font-size: 14px;
}}
QToolButton:hover {{
background-color: #333639;
}}
QToolButton:checked {{
background-color: #333639;
border-color: {ACCENT};
color: {ACCENT};
}}
QToolButton#logToggle {{
background: transparent;
border: none;
color: {COL_DIM};
font-size: 12px;
font-weight: 600;
padding: 4px 6px;
}}
QToolButton#logToggle:hover {{
color: {COL_TEXT};
}}
QToolButton#logToggle:checked {{
background: transparent;
border: none;
color: {COL_TEXT};
}}
QFrame#card {{
background-color: #2b2d30;
border: 1px solid #313437;
border-radius: 8px;
}}
QFrame#settingsCard {{
background-color: #26282b;
border: 1px solid #313437;
border-radius: 8px;
}}
QTableWidget {{
background-color: #2b2d30;
alternate-background-color: #292b2e;
border: none;
gridline-color: transparent;
outline: none;
selection-background-color: #3a3d42;
selection-color: {COL_TEXT};
}}
QTableWidget::item {{
padding: 4px 10px;
border: none;
}}
QTableWidget::item:selected {{
background-color: #3a3d42;
}}
QTableWidget::indicator {{
width: 16px;
height: 16px;
border: 1px solid #4a4d51;
border-radius: 4px;
background-color: #1c1e20;
}}
QTableWidget::indicator:checked {{
background-color: {ACCENT};
border-color: {ACCENT};
}}
QHeaderView::section {{
background-color: #2b2d30;
color: {COL_FAINT};
border: none;
border-bottom: 1px solid #3a3d40;
padding: 8px 10px;
font-size: 11px;
font-weight: 600;
}}
QCheckBox {{
color: {COL_DIM};
font-size: 12px;
spacing: 7px;
}}
QCheckBox::indicator {{
width: 16px;
height: 16px;
border: 1px solid #4a4d51;
border-radius: 4px;
background-color: #1c1e20;
}}
QCheckBox::indicator:checked {{
background-color: {ACCENT};
border-color: {ACCENT};
}}
QProgressBar {{
background-color: #1c1e20;
border: none;
border-radius: 3px;
max-height: 6px;
text-align: center;
}}
QProgressBar::chunk {{
background-color: {ACCENT};
border-radius: 3px;
}}
QTextEdit#logView {{
background-color: #151617;
border: none;
border-top: 1px solid #313437;
color: {COL_DIM};
padding: 6px;
}}
QScrollBar:vertical {{
background: transparent;
width: 10px;
margin: 0;
}}
QScrollBar::handle:vertical {{
background: #46494d;
border-radius: 5px;
min-height: 24px;
}}
QScrollBar::handle:vertical:hover {{
background: #55585c;
}}
QScrollBar::add-line:vertical, QScrollBar::sub-line:vertical {{
height: 0;
}}
QScrollBar::add-page:vertical, QScrollBar::sub-page:vertical {{
background: transparent;
}}
QMessageBox {{
background-color: #2b2d30;
}}
QToolTip {{
background-color: #1c1e20;
color: {COL_TEXT};
border: 1px solid #3a3d40;
padding: 4px 8px;
}}
"""
# Status → (text colour, pill background)
_BADGE_STYLES: dict[str, tuple[str, str]] = {
"Ready": (COL_DIM, "rgba(154, 158, 163, 30)"),
"Queued": (COL_AMBER, "rgba(224, 168, 60, 33)"),
"Rendering": (COL_BLUE, "rgba(74, 158, 218, 33)"),
"Done": (COL_GREEN, "rgba(76, 175, 125, 33)"),
"Failed": (COL_RED, "rgba(224, 92, 92, 33)"),
}
class StatusBadge(QLabel):
"""Small rounded pill showing a job status in its own colour."""
def __init__(self, status: str = "Ready", parent: QWidget | None = None) -> None:
super().__init__(parent)
self.setAlignment(Qt.AlignmentFlag.AlignCenter)
self._status = status
self.set_status(status)
@property
def status(self) -> str:
return self._status
def set_status(self, status: str) -> None:
self._status = status
colour, bg = _BADGE_STYLES.get(status, _BADGE_STYLES["Ready"])
self.setText(("● " if status == "Rendering" else "") + status)
self.setStyleSheet(
f"color: {colour}; background-color: {bg};"
"border-radius: 9px; padding: 2px 10px;"
"font-size: 11px; font-weight: 600;"
)
def _badge_cell(status: str) -> tuple[QWidget, StatusBadge]:
"""Wrap a StatusBadge in a container so it sits left-aligned in a cell."""
container = QWidget()
layout = QHBoxLayout(container)
layout.setContentsMargins(8, 3, 8, 3)
layout.setAlignment(Qt.AlignmentFlag.AlignLeft | Qt.AlignmentFlag.AlignVCenter)
badge = StatusBadge(status)
layout.addWidget(badge)
return container, badge
# ──────────────────────────────────────────────────────────────────────────────
# Config helpers
# ──────────────────────────────────────────────────────────────────────────────
_CONFIG_PATH = Path(__file__).parent / "config.json"
_DEFAULTS: dict = {
"pull_folder": "",
"output_root": r"V:\_FOOTAGE",
"nuke_executable": r"C:\Program Files\Nuke16.0v1\Nuke16.0.exe",
"template_script": str(Path(__file__).parent / "proxy_template.nk"),
}
def _load_config() -> dict:
if _CONFIG_PATH.exists():
try:
with _CONFIG_PATH.open(encoding="utf-8") as fh:
return {**_DEFAULTS, **json.load(fh)}
except Exception:
pass
return dict(_DEFAULTS)
def _save_config(cfg: dict) -> None:
try:
with _CONFIG_PATH.open("w", encoding="utf-8") as fh:
json.dump(cfg, fh, indent=2)
except Exception:
pass
# ──────────────────────────────────────────────────────────────────────────────
# Main window
# ──────────────────────────────────────────────────────────────────────────────
class MainWindow(QMainWindow):
def __init__(self) -> None:
super().__init__()
self.setWindowTitle("VFX Pull Processor")
self.resize(1000, 800)
app = QApplication.instance()
if app is not None:
app.setStyle("Fusion")
self.setStyleSheet(_QSS)
self.setAcceptDrops(True)
self._shots: list[Shot] = []
self._basename_row: dict[str, int] = {}
self._badges: dict[str, StatusBadge] = {}
self._runner: NukeRunner | None = None
self._scan_worker: ScanWorker | None = None
self._config = _load_config()
self._updating_checks = False
self._error_count = 0
self._build_ui()
self._apply_config()
# ── UI construction ───────────────────────────────────────────────
def _build_ui(self) -> None:
central = QWidget()
self.setCentralWidget(central)
root = QVBoxLayout(central)
root.setSpacing(12)
root.setContentsMargins(16, 16, 16, 16)
root.addLayout(self._build_top_bar())
root.addWidget(self._build_settings_card())
root.addWidget(self._build_table_card(), stretch=1)
root.addLayout(self._build_action_bar())
root.addWidget(self._build_log_card())
def _build_top_bar(self) -> QHBoxLayout:
bar = QHBoxLayout()
bar.setSpacing(8)
lbl = QLabel("Pull Folder")
lbl.setObjectName("fieldLabel")
bar.addWidget(lbl)
self._pull_field = QLineEdit()
self._pull_field.setPlaceholderText(
"Folder to scan for EXR sequences — or drop a folder here…"
)
bar.addWidget(self._pull_field, stretch=1)
browse = QPushButton("Browse")
browse.clicked.connect(
lambda: self._browse(self._pull_field, is_dir=True, file_filter="")
)
bar.addWidget(browse)
self._scan_btn = QPushButton("Scan")
self._scan_btn.clicked.connect(self._on_scan)
bar.addWidget(self._scan_btn)
self._settings_btn = QToolButton()
self._settings_btn.setText("⚙") # gear
self._settings_btn.setToolTip("Settings")
self._settings_btn.setCheckable(True)
self._settings_btn.toggled.connect(
lambda on: self._settings_card.setVisible(on)
)
bar.addWidget(self._settings_btn)
return bar
def _build_settings_card(self) -> QFrame:
self._settings_card = QFrame()
self._settings_card.setObjectName("settingsCard")
self._settings_card.setVisible(False)
layout = QVBoxLayout(self._settings_card)
layout.setSpacing(8)
layout.setContentsMargins(14, 12, 14, 12)
self._output_field = self._add_path_row(
layout, "Output Root", is_dir=True
)
self._nuke_field = self._add_path_row(
layout,
"Nuke Executable",
is_dir=False,
file_filter="Executables (*.exe);;All Files (*)",
)
self._tmpl_field = self._add_path_row(
layout,
"Template Script",
is_dir=False,
file_filter="Nuke Scripts (*.nk);;All Files (*)",
)
return self._settings_card
def _build_table_card(self) -> QFrame:
card = QFrame()
card.setObjectName("card")
layout = QVBoxLayout(card)
layout.setSpacing(0)
layout.setContentsMargins(0, 0, 0, 0)
# Header bar
head = QHBoxLayout()
head.setContentsMargins(14, 10, 14, 10)
title = QLabel("SEQUENCES FOUND")
title.setObjectName("sectionLabel")
head.addWidget(title)
head.addStretch(1)
self._select_all = QCheckBox("Select all")
self._select_all.clicked.connect(self._on_select_all)
self._select_all.setVisible(False)
head.addWidget(self._select_all)
self._count_label = QLabel("")
self._count_label.setObjectName("hintLabel")
head.addWidget(self._count_label)
layout.addLayout(head)
# Stacked: empty state ↔ table
self._stack = QStackedWidget()
empty = QLabel(
"Choose a pull folder and hit Scan to find EXR sequences."
)
empty.setObjectName("hintLabel")
empty.setAlignment(Qt.AlignmentFlag.AlignCenter)
self._stack.addWidget(empty)
self._table = QTableWidget(0, 5)
self._table.setHorizontalHeaderLabels(
["", "SHOT", "FRAMES", "RESOLUTION", "STATUS"]
)
hh = self._table.horizontalHeader()
hh.setSectionResizeMode(0, QHeaderView.ResizeMode.Fixed)
hh.setSectionResizeMode(1, QHeaderView.ResizeMode.Stretch)
hh.setSectionResizeMode(2, QHeaderView.ResizeMode.ResizeToContents)
hh.setSectionResizeMode(3, QHeaderView.ResizeMode.ResizeToContents)
hh.setSectionResizeMode(4, QHeaderView.ResizeMode.ResizeToContents)
hh.setHighlightSections(False)
self._table.setColumnWidth(0, 34)
self._table.setAlternatingRowColors(True)
self._table.setShowGrid(False)
self._table.setEditTriggers(QTableWidget.EditTrigger.NoEditTriggers)
self._table.setSelectionBehavior(QTableWidget.SelectionBehavior.SelectRows)
self._table.verticalHeader().setVisible(False)
self._table.verticalHeader().setDefaultSectionSize(34)
self._table.setFocusPolicy(Qt.FocusPolicy.NoFocus)
self._table.itemChanged.connect(self._on_item_changed)
self._stack.addWidget(self._table)
self._stack.setCurrentIndex(0)
layout.addWidget(self._stack, stretch=1)
return card
def _build_action_bar(self) -> QHBoxLayout:
bar = QHBoxLayout()
bar.setSpacing(12)
self._render_btn = QPushButton("Render Selected")
self._render_btn.setObjectName("primaryButton")
self._render_btn.setEnabled(False)
self._render_btn.clicked.connect(self._on_render)
bar.addWidget(self._render_btn)
self._progress = QProgressBar()
self._progress.setTextVisible(False)
self._progress.setValue(0)
self._progress.setSizePolicy(
QSizePolicy.Policy.Expanding, QSizePolicy.Policy.Fixed
)
bar.addWidget(self._progress, stretch=1)
self._progress_label = QLabel("")
self._progress_label.setObjectName("hintLabel")
self._progress_label.setMinimumWidth(220)
self._progress_label.setAlignment(
Qt.AlignmentFlag.AlignRight | Qt.AlignmentFlag.AlignVCenter
)
bar.addWidget(self._progress_label)
return bar
def _build_log_card(self) -> QFrame:
card = QFrame()
card.setObjectName("card")
layout = QVBoxLayout(card)
layout.setSpacing(0)
layout.setContentsMargins(0, 0, 0, 0)
head = QHBoxLayout()
head.setContentsMargins(8, 4, 14, 4)
self._log_toggle = QToolButton()
self._log_toggle.setObjectName("logToggle")
self._log_toggle.setCheckable(True)
self._log_toggle.setChecked(False)
self._log_toggle.setText("▸ Log")
self._log_toggle.toggled.connect(self._on_log_toggled)
head.addWidget(self._log_toggle)
head.addStretch(1)
self._log_hint = QLabel("")
self._log_hint.setObjectName("hintLabel")
head.addWidget(self._log_hint)
layout.addLayout(head)
self._log = QTextEdit()
self._log.setObjectName("logView")
self._log.setReadOnly(True)
mono = QFont("Cascadia Code")
mono.setStyleHint(QFont.StyleHint.Monospace)
mono.setPointSize(9)
self._log.setFont(mono)
self._log.setFixedHeight(170)
self._log.setVisible(False)
layout.addWidget(self._log)
return card
def _add_path_row(
self,
layout: QVBoxLayout,
label: str,
is_dir: bool,
file_filter: str = "",
) -> QLineEdit:
"""Add a labelled path field + Browse button row to *layout*."""
row = QHBoxLayout()
row.setSpacing(8)
lbl = QLabel(label)
lbl.setObjectName("fieldLabel")
lbl.setFixedWidth(130)
row.addWidget(lbl)
edit = QLineEdit()
row.addWidget(edit)
btn = QPushButton("Browse")
# Use default-argument binding to capture loop variables correctly
btn.clicked.connect(
lambda _checked=False, e=edit, d=is_dir, f=file_filter:
self._browse(e, d, f)
)
row.addWidget(btn)
layout.addLayout(row)
return edit
# ── Drag & drop ───────────────────────────────────────────────────
def dragEnterEvent(self, event) -> None: # noqa: N802 (Qt naming)
if event.mimeData().hasUrls():
for url in event.mimeData().urls():
if os.path.isdir(url.toLocalFile()):
event.acceptProposedAction()
return
event.ignore()
def dropEvent(self, event) -> None: # noqa: N802 (Qt naming)
for url in event.mimeData().urls():
path = url.toLocalFile()
if os.path.isdir(path):
self._pull_field.setText(path)
self._save_config()
break
# ── Config / browsing ─────────────────────────────────────────────
def _browse(self, edit: QLineEdit, is_dir: bool, file_filter: str) -> None:
if is_dir:
path = QFileDialog.getExistingDirectory(
self, "Select Folder", edit.text()
)
else:
path, _ = QFileDialog.getOpenFileName(
self, "Select File", edit.text(), file_filter
)
if path:
edit.setText(path)
self._save_config()
def _apply_config(self) -> None:
self._pull_field.setText(self._config.get("pull_folder", ""))
self._output_field.setText(self._config.get("output_root", ""))
self._nuke_field.setText(self._config.get("nuke_executable", ""))
self._tmpl_field.setText(self._config.get("template_script", ""))
def _save_config(self) -> None:
self._config.update(
{
"pull_folder": self._pull_field.text(),
"output_root": self._output_field.text(),
"nuke_executable": self._nuke_field.text(),
"template_script": self._tmpl_field.text(),
}
)
_save_config(self._config)
# ── Scan ──────────────────────────────────────────────────────────
@Slot()
def _on_scan(self) -> None:
folder = self._pull_field.text().strip()
if not folder or not os.path.isdir(folder):
QMessageBox.warning(self, "Invalid Folder", "Please set a valid Pull Folder.")
return
output_root = self._output_field.text().strip() or r"V:\_FOOTAGE"
# Reset state
self._shots.clear()
self._basename_row.clear()
self._badges.clear()
self._table.setRowCount(0)
self._render_btn.setEnabled(False)
self._log.clear()
self._error_count = 0
self._log_hint.setText("")
self._scan_btn.setEnabled(False)
self._save_config()
# Indeterminate bar while scanning
self._progress.setRange(0, 0)
self._progress_label.setText("Scanning…")
self._scan_worker = ScanWorker([folder], output_root)
self._scan_worker.log_message.connect(self._append_log)
self._scan_worker.scan_complete.connect(self._on_scan_complete)
self._scan_worker.start()
@Slot(list)
def _on_scan_complete(self, shots: list) -> None:
self._shots = shots
self._updating_checks = True
self._table.setRowCount(len(shots))
for row, shot in enumerate(shots):
self._basename_row[shot.basename] = row
# Column 0 — checkbox
chk = QTableWidgetItem()
chk.setFlags(
Qt.ItemFlag.ItemIsUserCheckable | Qt.ItemFlag.ItemIsEnabled
)
chk.setCheckState(Qt.CheckState.Checked)
self._table.setItem(row, 0, chk)
# Column 1 — shot basename
self._table.setItem(row, 1, QTableWidgetItem(shot.basename))
# Column 2 — frame range (amber when frames are missing)
frames_item = QTableWidgetItem(shot.display_frames)
if shot.missing_frames:
frames_item.setForeground(QColor(COL_AMBER))
frames_item.setToolTip(
f"Missing frames: {_summarise_frames(shot.missing_frames)}"
)
self._table.setItem(row, 2, frames_item)
# Column 3 — resolution
self._table.setItem(row, 3, QTableWidgetItem(shot.display_resolution))
# Column 4 — status badge
cell, badge = _badge_cell(shot.status)
self._badges[shot.basename] = badge
self._table.setCellWidget(row, 4, cell)
self._updating_checks = False
self._scan_btn.setEnabled(True)
self._progress.setRange(0, 1)
self._progress.setValue(0)
self._progress_label.setText(
f"{len(shots)} sequence(s) found" if shots else "No sequences found"
)
self._stack.setCurrentIndex(1 if shots else 0)
self._select_all.setVisible(bool(shots))
self._update_selection_ui()
# ── Selection bookkeeping ─────────────────────────────────────────
def _checked_rows(self) -> list[int]:
return [
row
for row in range(self._table.rowCount())
if self._table.item(row, 0)
and self._table.item(row, 0).checkState() == Qt.CheckState.Checked
]
@Slot(QTableWidgetItem)
def _on_item_changed(self, item: QTableWidgetItem) -> None:
if item.column() == 0 and not self._updating_checks:
self._update_selection_ui()
@Slot(bool)
def _on_select_all(self, checked: bool) -> None:
self._updating_checks = True
state = Qt.CheckState.Checked if checked else Qt.CheckState.Unchecked
for row in range(self._table.rowCount()):
item = self._table.item(row, 0)
if item:
item.setCheckState(state)
self._updating_checks = False
self._update_selection_ui()
def _update_selection_ui(self) -> None:
total = self._table.rowCount()
selected = len(self._checked_rows())
self._count_label.setText(
f"{total} sequence(s) · {selected} selected" if total else ""
)
self._select_all.setChecked(bool(total) and selected == total)
rendering = self._runner is not None and self._runner.isRunning()
self._render_btn.setText(
f"Render Selected ({selected})" if selected else "Render Selected"
)
self._render_btn.setEnabled(selected > 0 and not rendering)
# ── Render ────────────────────────────────────────────────────────
@Slot()
def _on_render(self) -> None:
selected = [self._shots[row] for row in self._checked_rows()]
if not selected:
QMessageBox.warning(self, "Nothing Selected", "No sequences are checked.")
return
nuke_exe = self._nuke_field.text().strip()
template = self._tmpl_field.text().strip()
if not os.path.isfile(nuke_exe):
QMessageBox.warning(
self,
"Nuke Not Found",
f"Nuke executable not found:\n{nuke_exe}",
)
return
if not os.path.isfile(template):
QMessageBox.warning(
self,
"Template Not Found",
f"Template script not found:\n{template}",
)
return
process_script = str(Path(__file__).parent / "process.py")
tmp_dir = tempfile.mkdtemp(prefix="vfxproxy_")
jobs_file = os.path.join(tmp_dir, "jobs.json")
payload = build_jobs_payload(selected, template)
write_jobs_file(payload, jobs_file)
self._append_log(f"Jobs file: {jobs_file} ({len(selected)} job(s))")
self._progress.setRange(0, len(selected))
self._progress.setValue(0)
self._progress_label.setText(f"0 of {len(selected)}")
for shot in selected:
self._set_row_status(shot.basename, "Queued")
self._runner = NukeRunner(
nuke_exe=nuke_exe,
process_script=process_script,
jobs_file=jobs_file,
total=len(selected),
)
self._runner.log_message.connect(self._append_log)
self._runner.progress_update.connect(self._on_progress)
self._runner.job_started.connect(self._on_job_started)
self._runner.job_done.connect(self._on_job_done)
self._runner.finished_signal.connect(self._on_render_finished)
self._scan_btn.setEnabled(False)
self._render_btn.setEnabled(False)
self._runner.start()
@Slot(int, int)
def _on_progress(self, current: int, total: int) -> None:
self._progress.setMaximum(total)
self._progress.setValue(current)
self._progress_label.setText(f"{current} of {total}")
@Slot(str)
def _on_job_started(self, basename: str) -> None:
self._set_row_status(basename, "Rendering")
label = basename if len(basename) <= 32 else basename[:31] + "…"
self._progress_label.setText(
f"{self._progress.value()} of {self._progress.maximum()} · {label}"
)
@Slot(str)
def _on_job_done(self, basename: str) -> None:
self._set_row_status(basename, "Done")
@Slot(bool)
def _on_render_finished(self, success: bool) -> None:
self._scan_btn.setEnabled(True)
if success:
self._progress.setValue(self._progress.maximum())
self._progress_label.setText(
f"Complete · {self._progress.maximum()} job(s)"
)
self._append_log("── Render complete ──")
else:
# Anything still queued or mid-render did not make it
for basename, badge in self._badges.items():
if badge.status in ("Queued", "Rendering"):
self._set_row_status(basename, "Failed")
self._progress_label.setText("Finished with errors")
self._append_log("ERROR: Render finished with errors")
self._update_selection_ui()
# ── Helpers ───────────────────────────────────────────────────────
def _set_row_status(self, basename: str, status: str) -> None:
badge = self._badges.get(basename)
if badge is not None:
badge.set_status(status)
@Slot(bool)
def _on_log_toggled(self, open_: bool) -> None:
self._log.setVisible(open_)
self._log_toggle.setText(("▾" if open_ else "▸") + " Log")
def _append_log(self, msg: str) -> None:
stamp = datetime.now().strftime("%H:%M:%S")
lower = msg.lower()
colour = COL_DIM
if "error" in lower or "traceback" in lower or "failed" in lower:
colour = COL_RED
self._error_count += 1
self._log_hint.setText(f"{self._error_count} error(s)")
self._log_hint.setStyleSheet(f"color: {COL_RED}; font-size: 12px;")
if not self._log_toggle.isChecked():
self._log_toggle.setChecked(True) # auto-open on first error
elif "done:" in lower or "complete" in lower:
colour = COL_GREEN
elif "missing" in lower or "skipping" in lower or "warning" in lower:
colour = COL_AMBER
self._log.append(
f'<span style="color:{COL_FAINT}">{stamp}</span>&nbsp;&nbsp;'
f'<span style="color:{colour}">{html.escape(msg)}</span>'
)
def _summarise_frames(frames: list[int], limit: int = 12) -> str:
"""Compact string for a sorted frame list, e.g. '1104–1106, 1210'."""
if not frames:
return ""
runs: list[tuple[int, int]] = []
start = prev = frames[0]
for f in frames[1:]:
if f == prev + 1:
prev = f
continue
runs.append((start, prev))
start = prev = f
runs.append((start, prev))
parts = [f"{a}" if a == b else f"{a}–{b}" for a, b in runs[:limit]]
if len(runs) > limit:
parts.append("…")
return ", ".join(parts)
jobs.py
+61
View File
@@ -0,0 +1,61 @@
"""
jobs.py — Build and serialise the jobs.json payload passed to process.py.
The jobs file is the sole interface between the GUI/host Python environment
and the Nuke Python environment. Keeping it as plain JSON means process.py
has no dependency on this project's other modules.
"""
from __future__ import annotations
import json
from pathlib import Path
from parser import Shot
# ──────────────────────────────────────────────────────────────────────────────
# Public API
# ──────────────────────────────────────────────────────────────────────────────
def build_jobs_payload(shots: list[Shot], template_script: str) -> dict:
"""Return the dict that will be serialised to jobs.json.
Parameters
----------
shots:
Shots selected for rendering.
template_script:
Absolute path to the Nuke template (.nk) file.
"""
return {
"template_script": Path(template_script).as_posix(),
"jobs": [_shot_to_job(s) for s in shots],
}
def write_jobs_file(payload: dict, path: str) -> None:
"""Write *payload* to *path* as indented JSON."""
with open(path, "w", encoding="utf-8") as fh:
json.dump(payload, fh, indent=2)
def read_jobs_file(path: str) -> dict:
"""Read and return the jobs payload from *path*."""
with open(path, encoding="utf-8") as fh:
return json.load(fh)
# ──────────────────────────────────────────────────────────────────────────────
# Internal helpers
# ──────────────────────────────────────────────────────────────────────────────
def _shot_to_job(shot: Shot) -> dict:
return {
"basename": shot.basename,
"input_sequence": shot.input_sequence,
"first_frame": shot.first_frame,
"last_frame": shot.last_frame,
"output_folder": shot.output_folder,
"output_movie": shot.output_movie,
"output_exr": shot.output_exr,
}
nuke_runner.py
+111
View File
@@ -0,0 +1,111 @@
"""
nuke_runner.py — Launch Nuke in terminal mode and stream its output.
Nuke is started ONCE with:
Nuke.exe -t process.py jobs.json
stdout/stderr are merged and forwarded line-by-line via Qt signals so the
GUI log window updates in real time. Progress is inferred from the
structured output lines that process.py prints.
"""
from __future__ import annotations
import re
import subprocess
from PySide6.QtCore import QThread, Signal
# Lines emitted by process.py that carry progress information:
# [3/10] Rendering: UNG_101_012_010_BG01_TT_v001
# [3/10] Done: UNG_101_012_010_BG01_TT_v001
_RENDERING_RE = re.compile(r"^\[(\d+)/(\d+)\]\s+Rendering:\s+(.+)$")
_DONE_RE = re.compile(r"^\[(\d+)/(\d+)\]\s+Done:\s+(.+)$")
class NukeRunner(QThread):
"""Runs a single Nuke process and emits its output as Qt signals.
Signals
-------
log_message(str)
Each non-empty output line from Nuke.
progress_update(int, int)
(jobs_done, jobs_total) whenever a job completes.
job_started(str)
basename of the job Nuke has just started rendering.
job_done(str)
basename of the job Nuke has just finished.
finished_signal(bool)
True if Nuke exited with code 0, False otherwise.
"""
log_message: Signal = Signal(str)
progress_update: Signal = Signal(int, int)
job_started: Signal = Signal(str)
job_done: Signal = Signal(str)
finished_signal: Signal = Signal(bool)
def __init__(
self,
nuke_exe: str,
process_script: str,
jobs_file: str,
total: int = 0,
parent=None,
) -> None:
super().__init__(parent)
self.nuke_exe = nuke_exe
self.process_script = process_script
self.jobs_file = jobs_file
self.total = total
def set_total(self, total: int) -> None:
self.total = total
# ── Thread entry point ────────────────────────────────────────────
def run(self) -> None:
cmd = [self.nuke_exe, "--indie", "-t", self.process_script, self.jobs_file]
self.log_message.emit("Launching: " + " ".join(cmd))
try:
proc = subprocess.Popen(
cmd,
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
text=True,
bufsize=1,
)
except FileNotFoundError:
self.log_message.emit(
f"ERROR: Nuke executable not found: {self.nuke_exe}"
)
self.finished_signal.emit(False)
return
except OSError as exc:
self.log_message.emit(f"ERROR: Could not launch Nuke: {exc}")
self.finished_signal.emit(False)
return
for raw_line in proc.stdout:
line = raw_line.rstrip()
if not line:
continue
self.log_message.emit(line)
m = _DONE_RE.match(line)
if m:
current, total, basename = int(m.group(1)), int(m.group(2)), m.group(3)
self.total = total
self.progress_update.emit(current, total)
self.job_done.emit(basename.strip())
continue
m = _RENDERING_RE.match(line)
if m:
basename = m.group(3)
self.job_started.emit(basename.strip())
proc.wait()
self.finished_signal.emit(proc.returncode == 0)
parser.py
+177
View File
@@ -0,0 +1,177 @@
"""
parser.py — Filename parsing and Shot data model.
Filename format:
SHOW_EPISODE_SEQUENCE_SHOT_ELEMENT_TASK_VERSION.FRAME.exr
Example:
UNG_101_012_010_BG01_TT_v001.01001.exr
"""
from __future__ import annotations
import re
import struct
from dataclasses import dataclass, field
from pathlib import Path
# ──────────────────────────────────────────────────────────────────────────────
# Regex patterns
# ──────────────────────────────────────────────────────────────────────────────
# Matches the basename (no frame, no extension):
# SHOW_EPISODE_SEQUENCE_SHOT_ELEMENT_TASK_VERSION
BASENAME_RE = re.compile(
r"^(?P<show>[^_]+)_"
r"(?P<episode>[^_]+)_"
r"(?P<sequence>[^_]+)_"
r"(?P<shot>[^_]+)_"
r"(?P<element>[^_]+)_"
r"(?P<task>[^_]+)_"
r"(?P<version>v\d+)$",
re.IGNORECASE,
)
# Matches a full EXR sequence filename: {basename}.{4-7 digit frame}.exr
SEQ_FILE_RE = re.compile(r"^(.+)\.(\d{4,7})\.exr$", re.IGNORECASE)
# ──────────────────────────────────────────────────────────────────────────────
# Data model
# ──────────────────────────────────────────────────────────────────────────────
@dataclass
class Shot:
show: str
episode: str
sequence: str
shot: str
element: str
task: str
version: str
basename: str
first_frame: int
last_frame: int
frame_count: int
missing_frames: list[int] = field(default_factory=list)
width: int = 0
height: int = 0
input_sequence: str = "" # Nuke-style path, e.g. /path/basename.%05d.exr
output_folder: str = ""
output_movie: str = ""
output_exr: str = ""
status: str = "Ready"
@property
def display_frames(self) -> str:
s = f"{self.first_frame}\u2013{self.last_frame} ({self.frame_count}f)"
if self.missing_frames:
s += f" [{len(self.missing_frames)} missing]"
return s
@property
def display_resolution(self) -> str:
return f"{self.width}\u00d7{self.height}" if self.width and self.height else "\u2014"
# ──────────────────────────────────────────────────────────────────────────────
# Parsing helpers
# ──────────────────────────────────────────────────────────────────────────────
def parse_basename(basename: str) -> dict | None:
"""Return a dict of named fields for *basename*, or None if it does not match."""
m = BASENAME_RE.match(basename)
return m.groupdict() if m else None
def build_shot(
basename: str,
source_dir: str,
frames: list[int],
output_root: str,
parsed: dict,
frame_padding: int = 4,
) -> Shot:
"""Construct a :class:`Shot` from parsed filename parts and frame list."""
frames_sorted = sorted(frames)
first = frames_sorted[0]
last = frames_sorted[-1]
missing = sorted(set(range(first, last + 1)) - set(frames_sorted))
show = parsed["show"]
episode = parsed["episode"]
# Nuke requires forward slashes even on Windows
input_dir = Path(source_dir).as_posix()
input_seq = f"{input_dir}/{basename}.%0{frame_padding}d.exr"
out_folder = Path(output_root) / show / f"{show}_{episode}" / basename
out_folder_str = out_folder.as_posix()
# Attempt to read resolution from the first frame of the sequence
first_exr = Path(source_dir) / f"{basename}.{str(first).zfill(frame_padding)}.exr"
width, height = read_exr_resolution(str(first_exr))
return Shot(
show=show,
episode=episode,
sequence=parsed["sequence"],
shot=parsed["shot"],
element=parsed["element"],
task=parsed["task"],
version=parsed["version"],
basename=basename,
first_frame=first,
last_frame=last,
frame_count=len(frames_sorted),
missing_frames=missing,
width=width,
height=height,
input_sequence=input_seq,
output_folder=out_folder_str,
output_movie=f"{out_folder_str}/{basename}.mov",
output_exr=f"{out_folder_str}/{basename}.%0{frame_padding}d.exr",
)
# ──────────────────────────────────────────────────────────────────────────────
# EXR header reader (no external dependencies)
# ──────────────────────────────────────────────────────────────────────────────
_EXR_MAGIC = b"\x76\x2f\x31\x01"
def read_exr_resolution(filepath: str) -> tuple[int, int]:
"""Read (width, height) from an OpenEXR file header without external libraries.
Returns (0, 0) if the file cannot be read or is not a valid EXR.
"""
try:
with open(filepath, "rb") as fh:
if fh.read(4) != _EXR_MAGIC:
return 0, 0
fh.read(4) # version + feature flags
while True:
name = _read_cstring(fh)
if not name:
break # end of header
type_str = _read_cstring(fh)
size = struct.unpack("<I", fh.read(4))[0]
data = fh.read(size)
if name == "dataWindow" and type_str == "box2i":
xmin, ymin, xmax, ymax = struct.unpack("<iiii", data)
return xmax - xmin + 1, ymax - ymin + 1
except Exception:
pass
return 0, 0
def _read_cstring(fh) -> str:
"""Read a null-terminated ASCII string from a binary file handle."""
buf = bytearray()
while True:
c = fh.read(1)
if not c or c == b"\x00":
break
buf.extend(c)
return buf.decode("latin-1")
process.py
+172
View File
@@ -0,0 +1,172 @@
"""
process.py — Nuke terminal-mode batch renderer.
This script runs INSIDE Nuke and must not import anything outside the
Python standard library or Nuke's own modules.
Usage (called by nuke_runner.py):
Nuke.exe -t process.py <jobs.json>
jobs.json schema
----------------
{
"template_script": "/abs/path/to/proxy_template.nk",
"jobs": [
{
"basename": "SHOW_EP_SEQ_SHOT_ELEM_TASK_v001",
"input_sequence": "/path/SHOW_..._%05d.exr",
"first_frame": 1001,
"last_frame": 1100,
"output_folder": "/output/SHOW/SHOW_EP/basename",
"output_movie": "/output/.../basename.mov",
"output_exr": "/output/.../basename.%05d.exr"
},
...
]
}
Expected template nodes
-----------------------
Read_Input — source EXR sequence
Overlay — text burn-in; knob 'bottomleft' is set to the shot basename
Write_4k — EXR sequence output (file = output_exr)
Write_h264 — MOV / H.264 output (file = output_movie)
Both Write nodes are executed in a single nuke.execute() call so the
graph is only evaluated once per frame.
Progress lines printed to stdout (parsed by nuke_runner.py):
[1/10] Rendering: <basename>
[1/10] Done: <basename>
[1/10] ERROR: <basename>: <message>
"""
from __future__ import annotations
import json
import os
import sys
def _log(msg: str) -> None:
"""Print *msg* and flush immediately so the parent process receives it."""
print(msg)
sys.stdout.flush()
def main() -> None:
# ── Locate jobs file ──────────────────────────────────────────────
# When launched as: Nuke.exe -t process.py jobs.json
# sys.argv is typically: ['process.py', 'jobs.json']
if len(sys.argv) < 2:
_log("ERROR: Usage: Nuke.exe -t process.py <jobs.json>")
sys.exit(1)
jobs_path = sys.argv[1]
if not os.path.isfile(jobs_path):
_log(f"ERROR: Jobs file not found: {jobs_path}")
sys.exit(1)
try:
with open(jobs_path, encoding="utf-8") as fh:
payload = json.load(fh)
except Exception as exc:
_log(f"ERROR: Cannot parse jobs file: {exc}")
sys.exit(1)
template_script = payload.get("template_script", "")
jobs = payload.get("jobs", [])
total = len(jobs)
if not jobs:
_log("No jobs to process.")
return
# ── Import Nuke ───────────────────────────────────────────────────
try:
import nuke # noqa: PLC0415
except ImportError:
_log("ERROR: 'nuke' module not available. This script must run inside Nuke.")
sys.exit(1)
# ── Open template ─────────────────────────────────────────────────
if not os.path.isfile(template_script):
_log(f"ERROR: Template script not found: {template_script}")
sys.exit(1)
_log(f"Loading template: {template_script}")
nuke.scriptOpen(template_script)
read_node = nuke.toNode("Read_Input")
overlay_node = nuke.toNode("Overlay")
write_4k = nuke.toNode("Write_4k")
write_h264 = nuke.toNode("Write_h264")
if read_node is None:
_log("ERROR: Node 'Read_Input' not found in template.")
sys.exit(1)
if write_4k is None:
_log("ERROR: Node 'Write_4k' not found in template.")
sys.exit(1)
if write_h264 is None:
_log("ERROR: Node 'Write_h264' not found in template.")
sys.exit(1)
if overlay_node is None:
_log("WARNING: Node 'Overlay' not found in template — bottomleft will not be set.")
# ── Process jobs ──────────────────────────────────────────────────
for idx, job in enumerate(jobs, start=1):
basename = job.get("basename", f"job_{idx}")
_log(f"[{idx}/{total}] Rendering: {basename}")
try:
first = int(job["first_frame"])
last = int(job["last_frame"])
input_seq = job["input_sequence"]
output_mov = job["output_movie"]
output_exr = job["output_exr"]
output_folder = job.get("output_folder", "")
# Ensure output directory exists
target_dir = output_folder or os.path.dirname(output_mov)
if target_dir:
os.makedirs(target_dir, exist_ok=True)
# Configure Read node
read_node["file"].setValue(input_seq)
read_node["first"].setValue(first)
read_node["last"].setValue(last)
read_node["origfirst"].setValue(first)
read_node["origlast"].setValue(last)
# Set burn-in text on the Overlay node
if overlay_node is not None:
overlay_node["bottomleft"].setValue(basename)
# Configure both Write nodes
write_4k["file"].setValue(output_exr)
write_h264["file"].setValue(output_mov)
# Render both Write nodes (sequential calls; list form not supported in all versions)
nuke.execute(write_4k, first, last, 1)
nuke.execute(write_h264, first, last, 1)
_log(f"[{idx}/{total}] Done: {basename}")
except Exception as exc: # noqa: BLE001
_log(f"[{idx}/{total}] ERROR: {basename}: {exc}")
# Free memory caches between jobs to avoid RAM build-up
for fn in ("clearRAMCache", "memory2", "memory"):
func = getattr(nuke, fn, None)
if func is not None:
try:
func("free")
except Exception:
pass
break
_log("All jobs complete.")
if __name__ == "__main__":
main()
proxy_template.nk
+54
View File
@@ -0,0 +1,54 @@
# proxy_template.nk
#
# VFX Pull Processor — Nuke proxy render template
#
# process.py sets the following knobs at render time:
# Read_Input.file, Read_Input.first, Read_Input.last,
# Read_Input.origfirst, Read_Input.origlast
# Write_Output.file
#
# You may insert any intermediate nodes (grades, reformats, LUT transforms,
# etc.) between Read_Input and Write_Output. process.py only touches the
# two named nodes, so your pipeline can be as complex as needed.
#
# Codec / format notes:
# The Write_Output node below is pre-set to Apple ProRes 422 HQ (apch)
# inside a QuickTime container. Change "codec" to suit your facility:
# apco = ProRes 422 Proxy
# apcs = ProRes 422 LT
# apcn = ProRes 422
# apch = ProRes 422 HQ
# ap4h = ProRes 4444
# hap1 = HAP
# jpeg = Photo-JPEG
# On Windows you may need mov64 instead of mov32 — swap the codec
# knob accordingly.
#
Root {
inputs 0
name proxy_template.nk
first_frame 1001
last_frame 1100
fps 24
lock_range false
colorManagement Nuke
format "2048 1152 0 0 2048 1152 1 2K_Super_35(full-ap)"
}
Read {
inputs 0
file ""
name Read_Input
first 1001
last 1100
origfirst 1001
origlast 1100
on_error nearest
}
Write {
file ""
file_type mov
codec apch
mov32_codec apch
name Write_Output
create_directories true
}
proxy_template.nkind
+1
View File
File diff suppressed because one or more lines are too long
proxy_template.nkind~
+1
View File
File diff suppressed because one or more lines are too long
scanner.py
+130
View File
@@ -0,0 +1,130 @@
"""
scanner.py — Recursive EXR sequence discovery.
Uses a background QThread so the GUI stays responsive during scanning.
The scan logic itself (_scan_folder) is kept free of Qt types so it can
be tested independently.
"""
from __future__ import annotations
import os
from collections import defaultdict
from pathlib import Path
from typing import Callable, Optional
from PySide6.QtCore import QThread, Signal
from parser import SEQ_FILE_RE, Shot, build_shot, parse_basename
class ScanWorker(QThread):
"""Scans one or more folders for EXR sequences in a background thread.
Signals
-------
log_message(str)
Informational text suitable for display in a log window.
scan_complete(list[Shot])
Emitted when scanning finishes (always, even on error).
"""
log_message: Signal = Signal(str)
scan_complete: Signal = Signal(list)
def __init__(
self,
folders: list[str],
output_root: str,
parent=None,
) -> None:
super().__init__(parent)
self.folders = folders
self.output_root = output_root
def run(self) -> None:
shots: list[Shot] = []
try:
for folder in self.folders:
self.log_message.emit(f"Scanning: {folder}")
found = scan_folder(folder, self.output_root, self.log_message.emit)
shots.extend(found)
except Exception as exc:
self.log_message.emit(f"ERROR during scan: {exc}")
finally:
self.log_message.emit(f"Scan complete — {len(shots)} sequence(s) found.")
self.scan_complete.emit(shots)
# ──────────────────────────────────────────────────────────────────────────────
# Core scan logic (Qt-free, testable standalone)
# ──────────────────────────────────────────────────────────────────────────────
def scan_folder(
root: str,
output_root: str,
log: Optional[Callable[[str], None]] = None,
) -> list[Shot]:
"""Recursively scan *root* for EXR sequences.
Parameters
----------
root:
Top-level directory to walk.
output_root:
Base path used to derive output folder/movie/exr paths on each Shot.
log:
Optional callable that receives progress strings (e.g. ``print``).
Returns
-------
list[Shot]
One Shot per discovered sequence, sorted by basename.
"""
def _log(msg: str) -> None:
if log:
log(msg)
# Collect files: dir → basename → [(frame_int, frame_str)]
dir_seqs: dict[str, dict[str, list[tuple[int, str]]]] = defaultdict(
lambda: defaultdict(list)
)
for dirpath, _dirs, files in os.walk(root):
for fname in files:
m = SEQ_FILE_RE.match(fname)
if not m:
continue
basename, frame_str = m.group(1), m.group(2)
dir_seqs[dirpath][basename].append((int(frame_str), frame_str))
shots: list[Shot] = []
for dirpath in sorted(dir_seqs):
for basename in sorted(dir_seqs[dirpath]):
parsed = parse_basename(basename)
if parsed is None:
_log(f" Skipping (unrecognised name): {basename}")
continue
frame_list = dir_seqs[dirpath][basename]
frames = [f for f, _ in frame_list]
# Detect padding length from the first filename encountered
frame_padding = len(frame_list[0][1])
_log(
f" Found: {basename} "
f"({len(frames)} frames, padding={frame_padding})"
)
shot = build_shot(
basename=basename,
source_dir=dirpath,
frames=frames,
output_root=output_root,
parsed=parsed,
frame_padding=frame_padding,
)
shots.append(shot)
return shots