""" VFXReview Connector for Nuke Dockable PySide2 panel — Phase 1 + Phase 2 Install: Copy this file to: ~/.nuke/VFXReviewConnector.py Add to ~/.nuke/menu.py: import VFXReviewConnector VFXReviewConnector.add_menu() """ from __future__ import annotations import json import os import re import threading import urllib.error import urllib.request from typing import Any try: import nuke import nukescripts _IN_NUKE = True except ImportError: _IN_NUKE = False try: from PySide2 import QtCore, QtGui, QtWidgets from PySide2.QtCore import Qt, Signal except ImportError: from PySide6 import QtCore, QtGui, QtWidgets from PySide6.QtCore import Qt, Signal # ── Configuration ───────────────────────────────────────────────────────────── # Edit these values for your facility / show. TOKEN = "am3O0PWUtqMJkAqsZ+bO7lho4cxQItxgukF6FHteAx4=" BASE_URL = "https://review.twotalesvfx.com" PROJECT_CODE = "UNG_S1" PROJECT_ROOT = "X:/shared_projects_2026/UNGO_VFX" FOOTAGE_ROOT = PROJECT_ROOT + "/production/plates" EXPORT_ROOT = PROJECT_ROOT + "/production/renders" PICLOCK_ROOT = PROJECT_ROOT + "/production/piclocks" NK_ROOT = PROJECT_ROOT + "/production/nuke" OCIO_CONFIG = PROJECT_ROOT + "/config/aces/config.ocio" # ── API client ──────────────────────────────────────────────────────────────── class APIError(Exception): pass class VFXReviewAPI: """Thin wrapper around the VFXReview HTTP API.""" def __init__(self, base_url: str = BASE_URL, token: str = TOKEN, project_code: str = PROJECT_CODE): self.base_url = base_url.rstrip("/") self.token = token self.project_code = project_code # ── Low-level ───────────────────────────────────────────────────────────── def _get(self, path: str, params: dict[str, str] | None = None) -> Any: """Make a GET request and return parsed JSON.""" url = self.base_url + path if params: query = "&".join( f"{urllib.parse.quote(k)}={urllib.parse.quote(str(v))}" for k, v in params.items() if v is not None ) url = url + "?" + query req = urllib.request.Request( url, headers={ "Authorization": f"Bearer {self.token}", "Accept": "application/json", }, ) try: with urllib.request.urlopen(req, timeout=15) as resp: return json.loads(resp.read().decode("utf-8")) except urllib.error.HTTPError as exc: body = exc.read().decode("utf-8", errors="replace") raise APIError(f"HTTP {exc.code} — {body}") from exc except Exception as exc: raise APIError(str(exc)) from exc # ── Endpoints ───────────────────────────────────────────────────────────── def get_episodes(self) -> list[dict]: """Return list of episode dicts for the project.""" data = self._get(f"/api/ext/projects/{self.project_code}/episodes") if isinstance(data, list): return data return data.get("episodes", []) def get_shots(self, episode: str | None = None, sequence: str | None = None) -> list[dict]: """Return shots for the project, optionally filtered by episode / sequence.""" params: dict[str, str] = {"limit": "500"} if episode: params["episode"] = episode if sequence: params["sequence"] = sequence data = self._get(f"/api/ext/projects/{self.project_code}/shots", params) return data.get("shots", []) def lookup_shot(self, shot_code: str) -> dict | None: """Return full shot metadata dict, or None if not found.""" try: data = self._get("/api/ext/shots/lookup", { "shotCode": shot_code, "projectCode": self.project_code, }) return data.get("shot") or data except APIError: return None def test_connection(self) -> tuple[bool, str]: """Return (ok, message) — used by the panel health-check.""" try: self.get_episodes() return True, "Connected" except APIError as exc: return False, str(exc) # ── urllib.parse shim — included so the module works without extra imports ──── import urllib.parse # noqa: E402 (already in stdlib, just ensuring it's imported) # ── Nuke helpers ───────────────────────────────────────────────────────────── def _safe(name: str) -> str: """Return a Nuke-safe node name from an arbitrary string.""" return re.sub(r"[^A-Za-z0-9_]", "_", name) def _find_node(name: str) -> "nuke.Node | None": """Return the first node whose name starts with *name*, or None.""" for node in nuke.allNodes(): if node.name().startswith(name): return node return None def _create_backdrop(label: str, nodes: list, color: int = 0x2A2A3AFF, margin: int = 40) -> "nuke.Node": """Create a backdrop that encloses *nodes*.""" if not nodes: return None bd = nuke.nodes.BackdropNode() bd["label"].setValue(f"{label}") bd["note_font_size"].setValue(18) bd["tile_color"].setValue(color) x_positions = [n.xpos() for n in nodes] y_positions = [n.ypos() for n in nodes] x = min(x_positions) - margin y = min(y_positions) - margin r = max(n.xpos() + n.screenWidth() for n in nodes) + margin b = max(n.ypos() + n.screenHeight() for n in nodes) + margin bd.setXYpos(x, y) bd["bdwidth"].setValue(r - x) bd["bdheight"].setValue(b - y) return bd def _format_exists(name: str) -> bool: for fmt in nuke.formats(): if fmt.name() == name: return True return False # ── Shot builder ────────────────────────────────────────────────────────────── def _find_plates(shot_code: str) -> list[str]: """ Scan FOOTAGE_ROOT for EXR sequences matching *shot_code*. Returns a list of first-frame file paths. """ episode = shot_code[:7] # e.g. UNG_106 shot_dir = os.path.join(FOOTAGE_ROOT, episode, shot_code) results = [] if not os.path.isdir(shot_dir): # Fall back: look for any subfolder under the episode root ep_dir = os.path.join(FOOTAGE_ROOT, episode) if os.path.isdir(ep_dir): for sub in sorted(os.listdir(ep_dir)): if sub.startswith(shot_code): results.extend(_scan_for_exr(os.path.join(ep_dir, sub))) return results results.extend(_scan_for_exr(shot_dir)) return results def _scan_for_exr(directory: str) -> list[str]: """Return the first frame path of every EXR sequence found in *directory*.""" results = [] try: files = sorted(f for f in os.listdir(directory) if f.lower().endswith(".exr")) except OSError: return results seen_stems: set[str] = set() for f in files: stem = re.sub(r"[\._]\d+\.exr$", "", f, flags=re.IGNORECASE) if stem not in seen_stems: seen_stems.add(stem) results.append(os.path.join(directory, f)) return results def _nuke_sequence_path(first_frame_path: str) -> str: """Convert a concrete first-frame path to a Nuke ####.exr pattern.""" return re.sub(r"(\d+)(\.exr)$", lambda m: "#" * len(m.group(1)) + m.group(2), first_frame_path, flags=re.IGNORECASE) def build_shot(shot_code: str, api: VFXReviewAPI) -> bool: """ Build a complete Nuke shot script for *shot_code*. - Fetches metadata from VFXReview - Locates plates on disk - Creates Read → OCIOColorSpace nodes - Creates Write_EXR and Write_Review nodes - Saves the script Returns True on success. """ if not _IN_NUKE: print(f"[VFXReview] Would build shot: {shot_code}") return False # ── Fetch metadata ──────────────────────────────────────────────────────── shot = api.lookup_shot(shot_code) if not shot: nuke.message(f"Shot not found in VFXReview:\n{shot_code}") return False episode = shot.get("episode", "") frame_start = int(shot.get("frameStart") or 1001) frame_end = int(shot.get("frameEnd") or 1100) fps = float(shot.get("fps") or 24) description = shot.get("description", "") exr_output = shot.get("exrOutput") or f"{shot_code}_comp_TT_v001" # ── Check / create save path ────────────────────────────────────────────── ep_code = f"UNG_{str(episode).zfill(3)}" if str(episode).isdigit() else str(episode) script_dir = os.path.join(NK_ROOT, ep_code, shot_code) script_path = os.path.join(script_dir, f"{shot_code}_comp_v001.nk") os.makedirs(script_dir, exist_ok=True) # ── Project settings ────────────────────────────────────────────────────── root = nuke.root() root["fps"].setValue(fps) root["first_frame"].setValue(frame_start) root["last_frame"].setValue(frame_end) if os.path.isfile(OCIO_CONFIG): root["colorManagement"].setValue("OCIO") root["OCIO_config"].setValue("custom") root["customOCIOConfigPath"].setValue(OCIO_CONFIG) # ── Locate plates ───────────────────────────────────────────────────────── plate_paths = _find_plates(shot_code) read_nodes: list[nuke.Node] = [] x_cursor = 0 for plate_path in plate_paths: seq_path = _nuke_sequence_path(plate_path) read = nuke.createNode("Read", inpanel=False) read["file"].setValue(seq_path) read["first"].setValue(frame_start) read["last"].setValue(frame_end) read["colorspace"].setValue("ACES2065-1") read.setName(_safe(f"Read_{os.path.basename(os.path.dirname(plate_path))}")) read.setXYpos(x_cursor, 0) color = nuke.createNode("OCIOColorSpace", inpanel=False) color["in_colorspace"].setValue("ACES2065-1") color["out_colorspace"].setValue("ACEScg") color.setInput(0, read) color.setXYpos(x_cursor, 100) color.setName(_safe(f"CS_{read.name()}")) read_nodes.append(read) x_cursor += 200 # ── Merge / dot into comp area ──────────────────────────────────────────── # If we have plates, connect the first into a "COMP" dot comp_dot = nuke.createNode("Dot", inpanel=False) comp_dot.setXYpos(0, 250) comp_dot.setName("Dot_COMP") if read_nodes: # Wire the OCIOColorSpace node below the first read into the dot cs_nodes = [n for n in nuke.allNodes("OCIOColorSpace")] if cs_nodes: comp_dot.setInput(0, cs_nodes[0]) # ── Write EXR ───────────────────────────────────────────────────────────── exr_dir = os.path.join(EXPORT_ROOT, shot_code) os.makedirs(exr_dir, exist_ok=True) exr_path = os.path.join(exr_dir, f"{exr_output}.####.exr").replace("\\", "/") write_exr = nuke.createNode("Write", inpanel=False) write_exr.setName("Write_EXR") write_exr["file"].setValue(exr_path) write_exr["file_type"].setValue("exr") write_exr["datatype"].setValue("16 bit half") write_exr["compression"].setValue("ZIP (1 scanline)") write_exr["colorspace"].setValue("ACEScg") write_exr["create_directories"].setValue(True) write_exr.setInput(0, comp_dot) write_exr.setXYpos(0, 400) # ── Write Review MOV ────────────────────────────────────────────────────── mov_path = os.path.join(EXPORT_ROOT, f"{shot_code}_cmp_TT_v001.mov").replace("\\", "/") write_mov = nuke.createNode("Write", inpanel=False) write_mov.setName("Write_Review") write_mov["file"].setValue(mov_path) write_mov["file_type"].setValue("mov") write_mov["colorspace"].setValue("Output - Rec.709") write_mov["create_directories"].setValue(True) write_mov.setInput(0, comp_dot) write_mov.setXYpos(200, 400) # ── Backdrops ───────────────────────────────────────────────────────────── if read_nodes: all_input_nodes = nuke.allNodes("Read") + nuke.allNodes("OCIOColorSpace") _create_backdrop("PLATES", all_input_nodes, color=0x1F3A4AFF) _create_backdrop("COMP", [comp_dot], color=0x2A3A2AFF) _create_backdrop("WRITE", [write_exr, write_mov], color=0x3A2A2AFF) # ── Viewer ──────────────────────────────────────────────────────────────── viewer = nuke.createNode("Viewer", inpanel=False) viewer.setInput(0, comp_dot) viewer.setXYpos(400, 250) # ── Save script ─────────────────────────────────────────────────────────── nuke.scriptSaveAs(script_path, overwrite=False) return True # ── Panel ───────────────────────────────────────────────────────────────────── class StatusBar(QtWidgets.QWidget): """Three-line status bar shown at the bottom of the panel.""" def __init__(self, parent=None): super().__init__(parent) layout = QtWidgets.QVBoxLayout(self) layout.setContentsMargins(6, 4, 6, 4) layout.setSpacing(2) self._shot_label = QtWidgets.QLabel("No shot loaded") self._action_label = QtWidgets.QLabel("Ready") self._warn_label = QtWidgets.QLabel("") for lbl in (self._shot_label, self._action_label, self._warn_label): lbl.setWordWrap(True) layout.addWidget(lbl) self._warn_label.setStyleSheet("color: #E8A020;") frame = QtWidgets.QFrame() frame.setFrameShape(QtWidgets.QFrame.StyledPanel) frame_layout = QtWidgets.QVBoxLayout(frame) frame_layout.setContentsMargins(0, 0, 0, 0) frame_layout.addWidget(self) self._frame = frame def set_shot(self, text: str): self._shot_label.setText(f"Shot: {text}") def set_action(self, text: str): self._action_label.setText(text) def set_warning(self, text: str): self._warn_label.setText(text) self._warn_label.setVisible(bool(text)) def clear_warning(self): self.set_warning("") class ConnectionWidget(QtWidgets.QGroupBox): """Shows live API connection status with a test button.""" def __init__(self, api: VFXReviewAPI, parent=None): super().__init__("Connection", parent) self._api = api layout = QtWidgets.QHBoxLayout(self) layout.setContentsMargins(8, 6, 8, 6) self._indicator = QtWidgets.QLabel("●") self._indicator.setFixedWidth(16) self._status = QtWidgets.QLabel("Not tested") self._test_btn = QtWidgets.QPushButton("Test") self._test_btn.setFixedWidth(50) self._test_btn.clicked.connect(self._test) layout.addWidget(self._indicator) layout.addWidget(self._status, 1) layout.addWidget(self._test_btn) self._set_state(None) def _set_state(self, ok: bool | None, message: str = ""): if ok is None: color, text = "#888888", "Not tested" elif ok: color, text = "#44CC44", f"OK — {message}" else: color, text = "#CC4444", message or "Failed" self._indicator.setStyleSheet(f"color: {color}; font-size: 14px;") self._status.setText(text) def _test(self): self._status.setText("Testing…") QtWidgets.QApplication.processEvents() ok, msg = self._api.test_connection() self._set_state(ok, msg) class ShotBuilderWidget(QtWidgets.QGroupBox): """Episode → Shot dropdowns + Build Shot button.""" shot_built = Signal(str) # emitted with shot_code on success status_msg = Signal(str) # status line updates warning_msg = Signal(str) # warning line updates def __init__(self, api: VFXReviewAPI, parent=None): super().__init__("Shot Builder", parent) self._api = api layout = QtWidgets.QVBoxLayout(self) layout.setContentsMargins(8, 8, 8, 8) layout.setSpacing(6) # ── Row 1: episode + shot dropdowns ────────────────────────────────── row1 = QtWidgets.QHBoxLayout() self._episode_combo = QtWidgets.QComboBox() self._episode_combo.setMinimumWidth(90) self._episode_combo.setToolTip("Episode") self._shot_combo = QtWidgets.QComboBox() self._shot_combo.setMinimumWidth(160) self._shot_combo.setToolTip("Shot code") row1.addWidget(self._episode_combo) row1.addWidget(self._shot_combo, 1) layout.addLayout(row1) # ── Row 2: action buttons ───────────────────────────────────────────── row2 = QtWidgets.QHBoxLayout() self._refresh_btn = QtWidgets.QPushButton("Refresh") self._build_btn = QtWidgets.QPushButton("Build Shot") self._build_btn.setDefault(True) row2.addWidget(self._refresh_btn) row2.addWidget(self._build_btn) layout.addLayout(row2) # ── Row 3: shot info label ──────────────────────────────────────────── self._info_label = QtWidgets.QLabel("") self._info_label.setWordWrap(True) self._info_label.setStyleSheet("color: #AAAAAA; font-size: 11px;") layout.addWidget(self._info_label) # ── Wire signals ────────────────────────────────────────────────────── self._episode_combo.currentIndexChanged.connect(self._on_episode_changed) self._shot_combo.currentIndexChanged.connect(self._on_shot_changed) self._refresh_btn.clicked.connect(self._load_episodes) self._build_btn.clicked.connect(self._build) # ── Public ──────────────────────────────────────────────────────────────── def load(self): """Populate the episode dropdown (call once after the panel is shown).""" self._load_episodes() # ── Private ─────────────────────────────────────────────────────────────── def _load_episodes(self): self._episode_combo.blockSignals(True) self._episode_combo.clear() self.status_msg.emit("Loading episodes…") QtWidgets.QApplication.processEvents() try: episodes = self._api.get_episodes() except APIError as exc: self.warning_msg.emit(f"API error: {exc}") self.status_msg.emit("Failed to load episodes") self._episode_combo.blockSignals(False) return for ep in episodes: ep_val = str(ep.get("episode") or ep.get("code") or ep) ep_label = ep_val if len(ep_val) > 3 else f"UNG_{ep_val.zfill(3)}" self._episode_combo.addItem(ep_label, userData=ep_val) self._episode_combo.blockSignals(False) if self._episode_combo.count(): self._episode_combo.setCurrentIndex(0) self._on_episode_changed(0) else: self.status_msg.emit("No episodes found") def _on_episode_changed(self, _index: int): ep_val = self._episode_combo.currentData() if ep_val is None: return self._shot_combo.clear() self.status_msg.emit(f"Loading shots for episode {ep_val}…") QtWidgets.QApplication.processEvents() try: shots = self._api.get_shots(episode=str(ep_val)) except APIError as exc: self.warning_msg.emit(f"API error: {exc}") self.status_msg.emit("Failed to load shots") return for shot in shots: code = shot.get("shotCode", "") if code: self._shot_combo.addItem(code, userData=shot) count = self._shot_combo.count() self.status_msg.emit(f"{count} shot(s) for episode {ep_val}") self.warning_msg.emit("") if count: self._shot_combo.setCurrentIndex(0) self._on_shot_changed(0) def _on_shot_changed(self, _index: int): shot = self._shot_combo.currentData() if not shot: self._info_label.setText("") return parts = [] if shot.get("description"): parts.append(shot["description"]) fr = shot.get("frameStart") to = shot.get("frameEnd") if fr and to: parts.append(f"Frames {fr}–{to}") status = shot.get("status", "") if status: parts.append(status.replace("_", " ").title()) self._info_label.setText(" | ".join(parts)) def _build(self): shot_data = self._shot_combo.currentData() if not shot_data: self.warning_msg.emit("Select a shot first") return shot_code = shot_data.get("shotCode", "") if not shot_code: self.warning_msg.emit("Invalid shot selection") return self.status_msg.emit(f"Building {shot_code}…") self._build_btn.setEnabled(False) QtWidgets.QApplication.processEvents() try: ok = build_shot(shot_code, self._api) except Exception as exc: self.warning_msg.emit(str(exc)) self.status_msg.emit("Build failed") self._build_btn.setEnabled(True) return self._build_btn.setEnabled(True) if ok: self.shot_built.emit(shot_code) self.status_msg.emit(f"Built: {shot_code}") self.warning_msg.emit("") else: self.status_msg.emit("Build cancelled or failed") class VFXReviewPanel(QtWidgets.QWidget): """Main dockable panel widget.""" TITLE = "VFXReview Connector" OBJECT_NAME = "VFXReviewConnectorPanel" def __init__(self, parent=None): super().__init__(parent) self.setObjectName(self.OBJECT_NAME) self.setWindowTitle(self.TITLE) self.setMinimumWidth(300) self._api = VFXReviewAPI() layout = QtWidgets.QVBoxLayout(self) layout.setContentsMargins(8, 8, 8, 8) layout.setSpacing(8) # ── Connection ──────────────────────────────────────────────────────── self._connection = ConnectionWidget(self._api) layout.addWidget(self._connection) # ── Shot Builder ────────────────────────────────────────────────────── self._shot_builder = ShotBuilderWidget(self._api) layout.addWidget(self._shot_builder) # ── Spacer ──────────────────────────────────────────────────────────── layout.addStretch(1) # ── Status bar ──────────────────────────────────────────────────────── self._status = StatusBar() frame = QtWidgets.QFrame() frame.setFrameShape(QtWidgets.QFrame.StyledPanel) frame_layout = QtWidgets.QVBoxLayout(frame) frame_layout.setContentsMargins(0, 0, 0, 0) frame_layout.addWidget(self._status) layout.addWidget(frame) # ── Wire inter-widget signals ───────────────────────────────────────── self._shot_builder.shot_built.connect(self._status.set_shot) self._shot_builder.status_msg.connect(self._status.set_action) self._shot_builder.warning_msg.connect(self._status.set_warning) # Load episodes asynchronously after the panel is shown QtCore.QTimer.singleShot(200, self._shot_builder.load) def sizeHint(self): return QtCore.QSize(320, 480) # ── Nuke integration ────────────────────────────────────────────────────────── _panel_instance: VFXReviewPanel | None = None def show_panel(): """Create (or raise) the floating panel.""" global _panel_instance if _panel_instance is None or not _panel_instance.isVisible(): _panel_instance = VFXReviewPanel() _panel_instance.setWindowFlags(Qt.Window | Qt.WindowStaysOnTopHint) _panel_instance.show() else: _panel_instance.raise_() _panel_instance.activateWindow() class _NukePanel(nukescripts.PythonPanel if _IN_NUKE else object): """Wrapper so Nuke can dock the panel in its workspace.""" def __init__(self): if not _IN_NUKE: return nukescripts.PythonPanel.__init__( self, VFXReviewPanel.TITLE, VFXReviewPanel.OBJECT_NAME ) self._widget = VFXReviewPanel() self.customKnob = nuke.PyCustom_Knob( VFXReviewPanel.OBJECT_NAME, "", "_NukePanel._get_widget()" ) self.addKnob(self.customKnob) @staticmethod def _get_widget(): global _panel_instance if _panel_instance is None: _panel_instance = VFXReviewPanel() return _panel_instance def add_menu(): """Called from menu.py to register the panel in the Nuke menus.""" if not _IN_NUKE: return menu = nuke.menu("Nuke") vfx_menu = menu.addMenu("VFXReview") vfx_menu.addCommand( "Open Connector", "import VFXReviewConnector; VFXReviewConnector.show_panel()", "F8", ) # Also register as a dockable pane nukescripts.registerPanel( VFXReviewPanel.OBJECT_NAME, "import VFXReviewConnector; return VFXReviewConnector._NukePanel()", ) # ── Standalone test (run outside Nuke with: python VFXReviewConnector.py) ───── if __name__ == "__main__": import sys app = QtWidgets.QApplication.instance() or QtWidgets.QApplication(sys.argv) win = VFXReviewPanel() win.setWindowFlags(Qt.Window) win.show() sys.exit(app.exec_())