code_gallery

Web News Blog Press Pricing

Learn

Training

Profiles

Executive

Templates

Animation VFX

Code

Code Gallery Designer

v5.20

Nuke

EXR to Review

Jorge Hernandez Ibañez 04/12/2025

Tags

EXR to Review

Jorge Herna…

Action v8

04/12/2025

Nuke

Pseudo-Code

Parameters

Talk with TD Support

Import to my Pipeline

Code

import nuke
import pipelinepro as pipeline
import os
import ast
import json
import requests

TEMPLATE_FILENAME = "ExrToReview_Template.nk"

def get_template_file_from_storage(filename):
    """
    Query the database to get the template file data and download if needed.
    
    Args:
        filename (str): The name of the template file to retrieve.
        
    Returns:
        str: The local path to the template file, or None if not found.
    """
    query = """
    query {
      file(fileName: "{filename}", token:"{token}"){
        fileName
        key
        fileLocal
        pathToFile
        link
        lastUpdate
        completed
        deleted
        fileSize
      }
    }
    """
    query = query.replace("{filename}", filename)
    response = exec_query(query)
    
    if not response:
        print(f"Error: Could not query template file '{filename}' from database.")
        return None
    
    file_data = response.get("file")
    if not file_data:
        print(f"Error: Template file '{filename}' not found in database.")
        return None
    
    # Check if file is valid
    if file_data.get("deleted") or not file_data.get("completed"):
        print(f"Error: Template file '{filename}' is deleted or not completed.")
        return None
    
    # Get the local path and download link
    file_path = file_data.get("pathToFile") or file_data.get("fileLocal")
    file_link = file_data.get("link")
    
    if not file_path:
        print(f"Error: No local path found for template file '{filename}'.")
        return None
    
    # Normalize the path
    file_path = os.path.normpath(file_path)
    
    # Check if file exists locally, if not download it
    if not os.path.exists(file_path):
        if not file_link:
            print(f"Error: Template file '{filename}' not found locally and no download link available.")
            return None
        
        print(f"Template file not found locally. Downloading from storage...")
        try:
            # Create directory if it doesn't exist
            folder_path = os.path.dirname(file_path)
            if not os.path.exists(folder_path):
                os.makedirs(folder_path)
            
            # Download the file
            response = requests.get(file_link)
            response.raise_for_status()
            
            with open(file_path, "wb") as f:
                f.write(response.content)
            
            print(f"Template file downloaded successfully to: {file_path}")
        except Exception as e:
            print(f"Error downloading template file: {e}")
            return None
    
    return file_path

def get_node_frame_range(node):
    """Return the first and last frame for a Nuke node."""
    import nuke
    if node.Class() == "Read":
        # Custom node frame range
        start = int(node.knob("first").value())
        end = int(node.knob("last").value())
    elif node.knob("node") and node["node"].value() == "ReadAOV":
        with node:
            reads = nuke.allNodes("Read")
            start = int(reads[0].knob("first").value())
            end = int(reads[0].knob("last").value())
    elif node.knob("use_limit") and node["use_limit"].value():
        # Custom node frame range
        start = int(node.knob("first").value())
        end = int(node.knob("last").value())
    else:
        start = int(nuke.root().knob("first_frame").value())
        end = int(nuke.root().knob("last_frame").value())
    return (start, end)

def exec_query(query):
    if query.find("token:") < 0:
        print("Warning: The token key is not included in"
               "the query, it will probably fail.")
    try:
        response = pipeline.query(query)
    except Exception as e:
        print("Error on DB query: ", e)
        return
    return response

def get_render_data(exr_filename):
    query = """
    query {
      file(fileName: "{filename}", token:"{token}"){
        fileName
        key
        pathToFile
        deleted
        folder{
          pathToFolder
          key
          name
          code
          folderLocal
        }
      }
    }
    """
    query = query.replace("{filename}", exr_filename)
    response = exec_query(query)
    if not response:
        return
    return response["file"]

def get_cdl(filename): # TODO get the cdl with a query
    query = """
    query {
      file(fileName: "{filename}", token:"{token}"){
        folder{
          relTaskFolder{
            relShotTask{
              folderLocal
              data
            }
          }
        }
      }
    }
    """
    query = query.replace("{filename}", filename)
    response = exec_query(query)

if not response:
        return
    file_data = response["file"]
    if not file_data:
        return
    shot = file_data["folder"]["relTaskFolder"][0]["relShotTask"][0]
    shot_data = shot["data"]
    if not shot_data:
        return
    if type(shot_data) == str:
        cdl_data = ast.literal_eval(shot_data).get("cdl")
    else:
        cdl_data = shot_data.get("cdl")

if not cdl_data:
        return
    cdl_version = max(cdl_data, key=lambda x: x["version"])
    cdl_file = cdl_version.get("file")
    if not cdl_file:
        return
    cdl_path = os.path.normpath(
        f"{shot.get('folderLocal')}/editorial/color/{cdl_file}").replace("\\","/")
    return cdl_path

def get_lut(filename):
    query = """
    query {
      file(fileName: "{filename}", token:"{token}"){
        shot{
          code
          data
          relSequenceShot{
            number
            data
          }
          relEpisodeShots{
            code
            data
          }
        }
        project{
          lut: data(field:"lut")
          folderLocal
        }
      }
    }
    """
    query = query.replace("{filename}", filename)
    response = exec_query(query)
    if not response:
        return
    file_data = response["file"]
    shot = file_data["shot"]
    project = file_data["project"]
    lut_folder_path = os.path.normpath(os.path.join(project["folderLocal"], "storage/library/LUTs/"))
    shot_data = shot["data"]
    if shot_data:
        if type(shot_data) == str:
            lut_data = ast.literal_eval(shot_data).get("lut")  # TODO fix
        else:
            lut_data = shot_data.get("lut")
        if lut_data:
            return os.path.normpath(
                 f"{lut_folder_path}/{lut_data}").replace("\\","/")
    sec_data = shot["relSequenceShot"][0]["data"]
    if sec_data:
        if type(sec_data) == str:
            lut_data = ast.literal_eval(sec_data).get("lut")  # TODO fix
        else:
            lut_data = sec_data.get("lut")
        if lut_data:
            return os.path.normpath(
                 f"{lut_folder_path}/{lut_data}").replace("\\","/")
    ep_data = shot["relEpisodeShots"][0]["data"]
    if ep_data:
        if type(ep_data) == str:
            lut_data = ast.literal_eval(ep_data).get("lut")
        else:
            lut_data = ep_data.get("lut")
        if lut_data:
            return os.path.normpath(
                 f"{lut_folder_path}/{lut_data}").replace("\\","/")
    lut_proj = project.get("lut")
    if lut_proj:
        return os.path.normpath(
            f"{lut_folder_path}/{lut_proj}").replace("\\","/")
    return

def set_luts(nodes, filename):
    lut_file = get_lut(filename)
    if lut_file:
        for node in nodes:
            # Get overrides for lut node in the nuke template from project
            node["file"].setValue(lut_file)
            #n["reload"].execute()  # Reload the lut from the node.
            node["disable"].setValue(False)
    else:
        for node in nodes:
            node["disable"].setValue(True)

def set_cdls(nodes, filename):
    cdl_file = get_cdl(filename)
    if cdl_file:
        for n in nodes:
            n["read_from_file"].setValue(True)
            n["file"].setValue(filepath)
            n["reload"].execute()  # Reload the cdl from the node.
            m["disable"].setValue(False)
    else:
        for n in nodes:
            n["disable"].setValue(True)
            
def load_template_file(filepath):
    nodes = nuke.nodePaste(filepath)

return nodes

def set_write_oputput_from_folder(write, folder, filename):
    write_name = write.name()
    out_folder = os.path.normpath(os.path.join(folder['folderLocal'], write_name))
    out_path = os.path.normpath(os.path.join(out_folder, filename)).replace("\\", "/")
    write["file"].setValue(out_path)
    return out_path

def load_mov(mov_path, fr_rg):
    readNode = nuke.nodes.Read(file=mov_path)
    readNode["file"].fromUserText(mov_path)
    readNode["frame_mode"].setValue("start at")
    readNode["frame"].setValue(str(fr_rg[0]))
    # TODO set colorspace
    return readNode

def create_file_render(folder_key, filename, pathToFile):
    folder_render_query = """
    mutation{
      newFile(token:"{token}", fileName:"{fileName}", folderKey:"{folderKey}", pathToFile:"{pathToFile}"){
        file{
          fileName
          key
          fileLocal
          uploadLink
          pathToFile
        }
      }
    }
    """
    folder_render_query = (
        folder_render_query.replace("{folderKey}", folder_key)
        .replace("{fileName}", filename)
        .replace("{pathToFile}", pathToFile)
    )
    return exec_query(folder_render_query)

def complete_file_render(file_key):
    file_complete_query = """
    mutation{
    completedFile(token:"{token}", fileKey:"{fileKey}", fileSize: 0){
        file{
            fileName
            }
        }

}
    """
    file_complete_query = file_complete_query.replace("{fileKey}", file_key)
    return exec_query(file_complete_query)

def data_write(model, key, field, value):
    mutation = """
    mutation{
      dataWrite(token:"{token}", model:"{model}", key:"{key}", value:{value}){
        data
      }
    }
    """
    mutation = (
        mutation.replace("{model}", model)
        .replace("{key}", key)
        .replace("{value}", json.dumps(json.dumps({field: value})))
    )
    return exec_query(mutation)

def publish_mov(folder_key, filename, pathToFile, fr, node_name, 
                colorspace=None, override_file=None, codec=None):
    if override_file is None:  # Create File
        response = create_file_render(folder_key, filename, pathToFile)
        file_key = response["newFile"]["file"]["key"]
    else:
        response = create_file_render(folder_key, filename, pathToFile)
        file_key = response["newFile"]["file"]["key"] #override_file["key"]
    complete_response = complete_file_render(file_key)

file_data = {
        "frame_range": fr,
        "path": pathToFile,
        "is_sequence": False,
        "format": "mov",
        "node": node_name,
        "colorspace": colorspace,
        "codec": codec
    }

response_file_data_write = data_write("File", file_key, "render", file_data)
    
    # TODO update folder data
    
    return response["newFile"]["file"]

def get_readaov_file(read_node):
    load_folder_key = read_node["load_folder_key"].value()
    if not load_folder_key: 
        nuke.message("The selected ReadAOV do not have a render loaded.")
        return
    response = pipeline.api(
        call_type="query",
        operation="folder",
        fields=("files{ fileName } dependence{ name }"),
        folderKey=load_folder_key
    )
    files = response.get("files", [])
    if not files:
        nuke.message("The selected ReadAOV do not have a render loaded.")
        return
    render_name = f"{response['dependence']['name']}_{read_node['load_layer'].value()}"
    file_name = f"{render_name}_Beauty.%04d.exr"
    for f in files:
        if f["fileName"] == file_name:
            return f["fileName"], render_name

def get_read_file(read_node):
    """Get the file path from a Read node."""
    if read_node.Class() == "Read":
        file_path = read_node["file"].value()
        render_name = file_path.split(".")[0]
        filename = os.path.basename(file_path)
    elif read_node.knob("node") and read_node["node"].value() == "ReadAOV":
        filename, render_name = get_readaov_file(read_node)
    else:
        raise ValueError("Node not supported.")
    return filename, render_name

def main():
    exr_to_review = nuke.selectedNode()
    is_readaov = True if exr_to_review.knob("node") and exr_to_review["node"].value() == "ReadAOV" else False
    if exr_to_review.Class() != "Read" and not is_readaov:
        nuke.message("To use the Ext to Review tool you need to have selected "
                     "1 Read node with a Nuke EXR render loaded.")
        return
    filename, render_name = get_read_file(exr_to_review)
    
    fr = get_node_frame_range(exr_to_review)
    # Check selected exr and get data from DB
    exr_data = get_render_data(filename)
    if not exr_data:
        nuke.message(f"The file {filename} was not found in the DB. Make sure you use a published render.")
        return
    # TODO Force user review of exr
    
    # Show a dialog with exr data to confirm
    if not nuke.ask(f"Do you want to sumit the exr {filename} to "
                    f"review with frame range {fr[0]}-{fr[1]}?"):
        return
    mov_filename = f"{render_name}.mov"
    mov_data = get_render_data(mov_filename) or {}
    print(mov_data.get("deleted"))
    if (mov_data and mov_data.get("deleted") is None) and not nuke.ask("This exr render already have a mov published. Are you sure you want to override it?"):
        return
    
    # Get template path from storage
    template_path = get_template_file_from_storage(TEMPLATE_FILENAME)
    if not template_path:
        nuke.message(f"Could not load template file '{TEMPLATE_FILENAME}'. Make sure it exists in the storage.")
        return
    
    # Load Template
    load_nodes = load_template_file(template_path)
    # Get selected nodes -> is a better filter to avoid nested nodes
    load_nodes = nuke.selectedNodes()
    
    # Get the node to connect the three
    connect_node = None
    cdl_nodes = []
    lut_nodes = []
    write_node = None
    for node in load_nodes:
        if node.name() == "connectExr":
            connect_node = node
        elif node.name().startswith("CDL"):
            cdl_nodes.append(node)
        elif node.name().startswith("LUT"):
            lut_nodes.append(node)
        elif node.Class() == "Write":
            write_node = node
            
    # Connect input
    if not connect_node or not write_node:
        return  # Abort
    connect_node.setInput(0, exr_to_review)
    # Move Read to top of connect node
    y_pos = connect_node.ypos() - 100
    x_pos = connect_node.xpos() + connect_node.screenWidth()//2 - exr_to_review.screenWidth()//2
    exr_to_review.setXYpos(x_pos, y_pos)
    # Fill CDL
    set_cdls(cdl_nodes, filename)
    # Fill LUT
    set_luts(lut_nodes, filename)
    # Set Write output from the folder of the exr
    out_mov_path = set_write_oputput_from_folder(write_node, exr_data["folder"], mov_filename)
    out_colorspace = write_node["colorspace"].value()    
    out_codec = {"data_range": write_node["dataRange"].value()}
    # Execute the render of the mov
    print(f"Rendering write '{write_node.name()}' with frame range {fr[0]}-{fr[1]}")
    nuke.execute(write_node, fr[0], fr[1])
    mov_read = load_mov(out_mov_path, fr)
    mov_read.setXYpos(exr_to_review.xpos(), exr_to_review.ypos() + 100)
    # Publish the mov
    if mov_data:
        mov_file = publish_mov(exr_data["folder"]["key"], mov_filename, out_mov_path, fr, write_node.name(),
                    colorspace=out_colorspace, override_file=mov_data, codec=out_codec)
    else:
        mov_file = publish_mov(exr_data["folder"]["key"], mov_filename, out_mov_path, fr, 
                    write_node.name(), colorspace=out_colorspace)

# Remove loaded nodes
    for n in load_nodes:
        nuke.delete(n)
    # Render a mp4 for preview
    query_preview = '''
    mutation{
      executeLogic(token: "{token}" , logicName:"Create_Review_From_File", data: {data}){
        success
        message
      }
    }
    '''
    data = {}
    data["file_list"] = [mov_file]
    query_preview = query_preview.replace("{data}", json.dumps(json.dumps(data)))
    response = pipeline.query(query_preview)
    # Set Task For approval int status
try:
    main()
except Exception as e:
    print(f"Error on rendering mov: {e}")

Import Action

1. Copy Key

Copy

2. Go to your Action pannel

Go