mirror of
https://github.com/ynput/ayon-core.git
synced 2025-12-24 12:54:40 +01:00
1675 lines
53 KiB
Python
1675 lines
53 KiB
Python
import os
|
|
import re
|
|
import sys
|
|
import six
|
|
import platform
|
|
from collections import OrderedDict
|
|
|
|
|
|
from avalon import api, io, lib
|
|
from avalon.tools import workfiles
|
|
import avalon.nuke
|
|
from avalon.nuke import lib as anlib
|
|
from avalon.nuke import (
|
|
save_file, open_file
|
|
)
|
|
from openpype.api import (
|
|
Logger,
|
|
Anatomy,
|
|
get_version_from_path,
|
|
get_anatomy_settings,
|
|
get_hierarchy,
|
|
get_asset,
|
|
get_current_project_settings,
|
|
ApplicationManager
|
|
)
|
|
|
|
import nuke
|
|
|
|
from .utils import set_context_favorites
|
|
|
|
log = Logger().get_logger(__name__)
|
|
|
|
opnl = sys.modules[__name__]
|
|
opnl._project = None
|
|
opnl.project_name = os.getenv("AVALON_PROJECT")
|
|
opnl.workfiles_launched = False
|
|
opnl._node_tab_name = "{}".format(os.getenv("AVALON_LABEL") or "Avalon")
|
|
|
|
|
|
def get_created_node_imageio_setting(**kwarg):
|
|
''' Get preset data for dataflow (fileType, compression, bitDepth)
|
|
'''
|
|
log.debug(kwarg)
|
|
nodeclass = kwarg.get("nodeclass", None)
|
|
creator = kwarg.get("creator", None)
|
|
|
|
assert any([creator, nodeclass]), nuke.message(
|
|
"`{}`: Missing mandatory kwargs `host`, `cls`".format(__file__))
|
|
|
|
imageio = get_anatomy_settings(opnl.project_name)["imageio"]
|
|
imageio_nodes = imageio["nuke"]["nodes"]["requiredNodes"]
|
|
|
|
imageio_node = None
|
|
for node in imageio_nodes:
|
|
log.info(node)
|
|
if (node["nukeNodeClass"] != nodeclass) and (
|
|
creator not in node["plugins"]):
|
|
continue
|
|
|
|
imageio_node = node
|
|
|
|
log.info("ImageIO node: {}".format(imageio_node))
|
|
return imageio_node
|
|
|
|
|
|
def get_imageio_input_colorspace(filename):
|
|
''' Get input file colorspace based on regex in settings.
|
|
'''
|
|
imageio_regex_inputs = (
|
|
get_anatomy_settings(opnl.project_name)
|
|
["imageio"]["nuke"]["regexInputs"]["inputs"])
|
|
|
|
preset_clrsp = None
|
|
for regexInput in imageio_regex_inputs:
|
|
if bool(re.search(regexInput["regex"], filename)):
|
|
preset_clrsp = str(regexInput["colorspace"])
|
|
|
|
return preset_clrsp
|
|
|
|
|
|
def on_script_load():
|
|
''' Callback for ffmpeg support
|
|
'''
|
|
if nuke.env['LINUX']:
|
|
nuke.tcl('load ffmpegReader')
|
|
nuke.tcl('load ffmpegWriter')
|
|
else:
|
|
nuke.tcl('load movReader')
|
|
nuke.tcl('load movWriter')
|
|
|
|
|
|
def check_inventory_versions():
|
|
"""
|
|
Actual version idetifier of Loaded containers
|
|
|
|
Any time this function is run it will check all nodes and filter only
|
|
Loader nodes for its version. It will get all versions from database
|
|
and check if the node is having actual version. If not then it will color
|
|
it to red.
|
|
"""
|
|
# get all Loader nodes by avalon attribute metadata
|
|
for each in nuke.allNodes():
|
|
container = avalon.nuke.parse_container(each)
|
|
|
|
if container:
|
|
node = nuke.toNode(container["objectName"])
|
|
avalon_knob_data = avalon.nuke.read(
|
|
node)
|
|
|
|
# get representation from io
|
|
representation = io.find_one({
|
|
"type": "representation",
|
|
"_id": io.ObjectId(avalon_knob_data["representation"])
|
|
})
|
|
|
|
# Get start frame from version data
|
|
version = io.find_one({
|
|
"type": "version",
|
|
"_id": representation["parent"]
|
|
})
|
|
|
|
# get all versions in list
|
|
versions = io.find({
|
|
"type": "version",
|
|
"parent": version["parent"]
|
|
}).distinct('name')
|
|
|
|
max_version = max(versions)
|
|
|
|
# check the available version and do match
|
|
# change color of node if not max verion
|
|
if version.get("name") not in [max_version]:
|
|
node["tile_color"].setValue(int("0xd84f20ff", 16))
|
|
else:
|
|
node["tile_color"].setValue(int("0x4ecd25ff", 16))
|
|
|
|
|
|
def writes_version_sync():
|
|
''' Callback synchronizing version of publishable write nodes
|
|
'''
|
|
try:
|
|
rootVersion = get_version_from_path(nuke.root().name())
|
|
padding = len(rootVersion)
|
|
new_version = "v" + str("{" + ":0>{}".format(padding) + "}").format(
|
|
int(rootVersion)
|
|
)
|
|
log.debug("new_version: {}".format(new_version))
|
|
except Exception:
|
|
return
|
|
|
|
for each in nuke.allNodes(filter="Write"):
|
|
# check if the node is avalon tracked
|
|
if opnl._node_tab_name not in each.knobs():
|
|
continue
|
|
|
|
avalon_knob_data = avalon.nuke.read(
|
|
each)
|
|
|
|
try:
|
|
if avalon_knob_data['families'] not in ["render"]:
|
|
log.debug(avalon_knob_data['families'])
|
|
continue
|
|
|
|
node_file = each['file'].value()
|
|
|
|
node_version = "v" + get_version_from_path(node_file)
|
|
log.debug("node_version: {}".format(node_version))
|
|
|
|
node_new_file = node_file.replace(node_version, new_version)
|
|
each['file'].setValue(node_new_file)
|
|
if not os.path.isdir(os.path.dirname(node_new_file)):
|
|
log.warning("Path does not exist! I am creating it.")
|
|
os.makedirs(os.path.dirname(node_new_file))
|
|
except Exception as e:
|
|
log.warning(
|
|
"Write node: `{}` has no version in path: {}".format(
|
|
each.name(), e))
|
|
|
|
|
|
def version_up_script():
|
|
''' Raising working script's version
|
|
'''
|
|
import nukescripts
|
|
nukescripts.script_and_write_nodes_version_up()
|
|
|
|
|
|
def check_subsetname_exists(nodes, subset_name):
|
|
"""
|
|
Checking if node is not already created to secure there is no duplicity
|
|
|
|
Arguments:
|
|
nodes (list): list of nuke.Node objects
|
|
subset_name (str): name we try to find
|
|
|
|
Returns:
|
|
bool: True of False
|
|
"""
|
|
return next((True for n in nodes
|
|
if subset_name in avalon.nuke.read(n).get("subset", "")),
|
|
False)
|
|
|
|
|
|
def get_render_path(node):
|
|
''' Generate Render path from presets regarding avalon knob data
|
|
'''
|
|
data = {'avalon': avalon.nuke.read(node)}
|
|
data_preset = {
|
|
"nodeclass": data['avalon']['family'],
|
|
"families": [data['avalon']['families']],
|
|
"creator": data['avalon']['creator']
|
|
}
|
|
|
|
nuke_imageio_writes = get_created_node_imageio_setting(**data_preset)
|
|
|
|
application = lib.get_application(os.environ["AVALON_APP_NAME"])
|
|
data.update({
|
|
"application": application,
|
|
"nuke_imageio_writes": nuke_imageio_writes
|
|
})
|
|
|
|
anatomy_filled = format_anatomy(data)
|
|
return anatomy_filled["render"]["path"].replace("\\", "/")
|
|
|
|
|
|
def format_anatomy(data):
|
|
''' Helping function for formating of anatomy paths
|
|
|
|
Arguments:
|
|
data (dict): dictionary with attributes used for formating
|
|
|
|
Return:
|
|
path (str)
|
|
'''
|
|
# TODO: perhaps should be nonPublic
|
|
|
|
anatomy = Anatomy()
|
|
log.debug("__ anatomy.templates: {}".format(anatomy.templates))
|
|
|
|
try:
|
|
# TODO: bck compatibility with old anatomy template
|
|
padding = int(
|
|
anatomy.templates["render"].get(
|
|
"frame_padding",
|
|
anatomy.templates["render"].get("padding")
|
|
)
|
|
)
|
|
except KeyError as e:
|
|
msg = ("`padding` key is not in `render` "
|
|
"or `frame_padding` on is not available in "
|
|
"Anatomy template. Please, add it there and restart "
|
|
"the pipeline (padding: \"4\"): `{}`").format(e)
|
|
|
|
log.error(msg)
|
|
nuke.message(msg)
|
|
|
|
version = data.get("version", None)
|
|
if not version:
|
|
file = script_name()
|
|
data["version"] = get_version_from_path(file)
|
|
project_document = io.find_one({"type": "project"})
|
|
data.update({
|
|
"subset": data["avalon"]["subset"],
|
|
"asset": data["avalon"]["asset"],
|
|
"task": os.environ["AVALON_TASK"],
|
|
"family": data["avalon"]["family"],
|
|
"project": {"name": project_document["name"],
|
|
"code": project_document["data"].get("code", '')},
|
|
"hierarchy": get_hierarchy(),
|
|
"frame": "#" * padding,
|
|
})
|
|
return anatomy.format(data)
|
|
|
|
|
|
def script_name():
|
|
''' Returns nuke script path
|
|
'''
|
|
return nuke.root().knob('name').value()
|
|
|
|
|
|
def add_button_write_to_read(node):
|
|
name = "createReadNode"
|
|
label = "Cread Read From Rendered"
|
|
value = "import write_to_read;write_to_read.write_to_read(nuke.thisNode())"
|
|
knob = nuke.PyScript_Knob(name, label, value)
|
|
knob.clearFlag(nuke.STARTLINE)
|
|
node.addKnob(knob)
|
|
|
|
|
|
def create_write_node(name, data, input=None, prenodes=None, review=True):
|
|
''' Creating write node which is group node
|
|
|
|
Arguments:
|
|
name (str): name of node
|
|
data (dict): data to be imprinted
|
|
input (node): selected node to connect to
|
|
prenodes (list, optional): list of lists, definitions for nodes
|
|
to be created before write
|
|
review (bool): adding review knob
|
|
|
|
Example:
|
|
prenodes = [(
|
|
"NameNode", # string
|
|
"NodeClass", # string
|
|
( # OrderDict: knob and values pairs
|
|
("knobName", "knobValue"),
|
|
("knobName", "knobValue")
|
|
),
|
|
( # list outputs
|
|
"firstPostNodeName",
|
|
"secondPostNodeName"
|
|
)
|
|
)
|
|
]
|
|
|
|
Return:
|
|
node (obj): group node with avalon data as Knobs
|
|
'''
|
|
|
|
imageio_writes = get_created_node_imageio_setting(**data)
|
|
app_manager = ApplicationManager()
|
|
app_name = os.environ.get("AVALON_APP_NAME")
|
|
if app_name:
|
|
app = app_manager.applications.get(app_name)
|
|
|
|
for knob in imageio_writes["knobs"]:
|
|
if knob["name"] == "file_type":
|
|
representation = knob["value"]
|
|
|
|
try:
|
|
data.update({
|
|
"app": app.host_name,
|
|
"imageio_writes": imageio_writes,
|
|
"representation": representation,
|
|
})
|
|
anatomy_filled = format_anatomy(data)
|
|
|
|
except Exception as e:
|
|
msg = "problem with resolving anatomy template: {}".format(e)
|
|
log.error(msg)
|
|
nuke.message(msg)
|
|
|
|
# build file path to workfiles
|
|
fpath = str(anatomy_filled["work"]["folder"]).replace("\\", "/")
|
|
fpath = data["fpath_template"].format(
|
|
work=fpath, version=data["version"], subset=data["subset"],
|
|
frame=data["frame"],
|
|
ext=representation
|
|
)
|
|
|
|
# create directory
|
|
if not os.path.isdir(os.path.dirname(fpath)):
|
|
log.warning("Path does not exist! I am creating it.")
|
|
os.makedirs(os.path.dirname(fpath))
|
|
|
|
_data = OrderedDict({
|
|
"file": fpath
|
|
})
|
|
|
|
# adding dataflow template
|
|
log.debug("imageio_writes: `{}`".format(imageio_writes))
|
|
for knob in imageio_writes["knobs"]:
|
|
_data.update({knob["name"]: knob["value"]})
|
|
|
|
_data = anlib.fix_data_for_node_create(_data)
|
|
|
|
log.debug("_data: `{}`".format(_data))
|
|
|
|
if "frame_range" in data.keys():
|
|
_data["frame_range"] = data.get("frame_range", None)
|
|
log.debug("_data[frame_range]: `{}`".format(_data["frame_range"]))
|
|
|
|
GN = nuke.createNode("Group", "name {}".format(name))
|
|
|
|
prev_node = None
|
|
with GN:
|
|
connections = list()
|
|
if input:
|
|
# if connected input node was defined
|
|
connections.append({
|
|
"node": input,
|
|
"inputName": input.name()})
|
|
prev_node = nuke.createNode(
|
|
"Input", "name {}".format(input.name()))
|
|
prev_node.hideControlPanel()
|
|
else:
|
|
# generic input node connected to nothing
|
|
prev_node = nuke.createNode(
|
|
"Input", "name {}".format("rgba"))
|
|
prev_node.hideControlPanel()
|
|
|
|
# creating pre-write nodes `prenodes`
|
|
if prenodes:
|
|
for name, klass, properties, set_output_to in prenodes:
|
|
# create node
|
|
now_node = nuke.createNode(klass, "name {}".format(name))
|
|
now_node.hideControlPanel()
|
|
|
|
# add data to knob
|
|
for k, v in properties:
|
|
try:
|
|
now_node[k].value()
|
|
except NameError:
|
|
log.warning(
|
|
"knob `{}` does not exist on node `{}`".format(
|
|
k, now_node["name"].value()
|
|
))
|
|
continue
|
|
|
|
if k and v:
|
|
now_node[k].setValue(str(v))
|
|
|
|
# connect to previous node
|
|
if set_output_to:
|
|
if isinstance(set_output_to, (tuple or list)):
|
|
for i, node_name in enumerate(set_output_to):
|
|
input_node = nuke.createNode(
|
|
"Input", "name {}".format(node_name))
|
|
input_node.hideControlPanel()
|
|
connections.append({
|
|
"node": nuke.toNode(node_name),
|
|
"inputName": node_name})
|
|
now_node.setInput(1, input_node)
|
|
|
|
elif isinstance(set_output_to, str):
|
|
input_node = nuke.createNode(
|
|
"Input", "name {}".format(node_name))
|
|
input_node.hideControlPanel()
|
|
connections.append({
|
|
"node": nuke.toNode(set_output_to),
|
|
"inputName": set_output_to})
|
|
now_node.setInput(0, input_node)
|
|
|
|
else:
|
|
now_node.setInput(0, prev_node)
|
|
|
|
# swith actual node to previous
|
|
prev_node = now_node
|
|
|
|
# creating write node
|
|
write_node = now_node = anlib.add_write_node(
|
|
"inside_{}".format(name),
|
|
**_data
|
|
)
|
|
write_node.hideControlPanel()
|
|
# connect to previous node
|
|
now_node.setInput(0, prev_node)
|
|
|
|
# swith actual node to previous
|
|
prev_node = now_node
|
|
|
|
now_node = nuke.createNode("Output", "name Output1")
|
|
now_node.hideControlPanel()
|
|
|
|
# connect to previous node
|
|
now_node.setInput(0, prev_node)
|
|
|
|
# imprinting group node
|
|
anlib.set_avalon_knob_data(GN, data["avalon"])
|
|
anlib.add_publish_knob(GN)
|
|
add_rendering_knobs(GN)
|
|
|
|
if review:
|
|
add_review_knob(GN)
|
|
|
|
# add divider
|
|
GN.addKnob(nuke.Text_Knob('', 'Rendering'))
|
|
|
|
# Add linked knobs.
|
|
linked_knob_names = [
|
|
"_grp-start_",
|
|
"use_limit", "first", "last",
|
|
"_grp-end_",
|
|
"Render"
|
|
]
|
|
for name in linked_knob_names:
|
|
if "_grp-start_" in name:
|
|
knob = nuke.Tab_Knob(
|
|
"rnd_attr", "Rendering attributes", nuke.TABBEGINCLOSEDGROUP)
|
|
GN.addKnob(knob)
|
|
elif "_grp-end_" in name:
|
|
knob = nuke.Tab_Knob(
|
|
"rnd_attr_end", "Rendering attributes", nuke.TABENDGROUP)
|
|
GN.addKnob(knob)
|
|
else:
|
|
link = nuke.Link_Knob("")
|
|
link.makeLink(write_node.name(), name)
|
|
link.setName(name)
|
|
if "Render" in name:
|
|
link.setLabel("Render Local")
|
|
link.setFlag(0x1000)
|
|
GN.addKnob(link)
|
|
|
|
# adding write to read button
|
|
add_button_write_to_read(GN)
|
|
|
|
# Deadline tab.
|
|
add_deadline_tab(GN)
|
|
|
|
# open the our Tab as default
|
|
GN[opnl._node_tab_name].setFlag(0)
|
|
|
|
# set tile color
|
|
tile_color = _data.get("tile_color", "0xff0000ff")
|
|
GN["tile_color"].setValue(tile_color)
|
|
|
|
return GN
|
|
|
|
|
|
def add_rendering_knobs(node):
|
|
''' Adds additional rendering knobs to given node
|
|
|
|
Arguments:
|
|
node (obj): nuke node object to be fixed
|
|
|
|
Return:
|
|
node (obj): with added knobs
|
|
'''
|
|
if "render" not in node.knobs():
|
|
knob = nuke.Enumeration_Knob("render", "", [
|
|
"Use existing frames", "Local", "On farm"])
|
|
knob.clearFlag(nuke.STARTLINE)
|
|
node.addKnob(knob)
|
|
return node
|
|
|
|
|
|
def add_review_knob(node):
|
|
''' Adds additional review knob to given node
|
|
|
|
Arguments:
|
|
node (obj): nuke node object to be fixed
|
|
|
|
Return:
|
|
node (obj): with added knob
|
|
'''
|
|
if "review" not in node.knobs():
|
|
knob = nuke.Boolean_Knob("review", "Review")
|
|
knob.setValue(True)
|
|
node.addKnob(knob)
|
|
return node
|
|
|
|
|
|
def add_deadline_tab(node):
|
|
node.addKnob(nuke.Tab_Knob("Deadline"))
|
|
|
|
knob = nuke.Int_Knob("deadlineChunkSize", "Chunk Size")
|
|
knob.setValue(0)
|
|
node.addKnob(knob)
|
|
|
|
knob = nuke.Int_Knob("deadlinePriority", "Priority")
|
|
knob.setValue(50)
|
|
node.addKnob(knob)
|
|
|
|
|
|
def get_deadline_knob_names():
|
|
return ["Deadline", "deadlineChunkSize", "deadlinePriority"]
|
|
|
|
|
|
def create_backdrop(label="", color=None, layer=0,
|
|
nodes=None):
|
|
"""
|
|
Create Backdrop node
|
|
|
|
Arguments:
|
|
color (str): nuke compatible string with color code
|
|
layer (int): layer of node usually used (self.pos_layer - 1)
|
|
label (str): the message
|
|
nodes (list): list of nodes to be wrapped into backdrop
|
|
|
|
"""
|
|
assert isinstance(nodes, list), "`nodes` should be a list of nodes"
|
|
|
|
# Calculate bounds for the backdrop node.
|
|
bdX = min([node.xpos() for node in nodes])
|
|
bdY = min([node.ypos() for node in nodes])
|
|
bdW = max([node.xpos() + node.screenWidth() for node in nodes]) - bdX
|
|
bdH = max([node.ypos() + node.screenHeight() for node in nodes]) - bdY
|
|
|
|
# Expand the bounds to leave a little border. Elements are offsets
|
|
# for left, top, right and bottom edges respectively
|
|
left, top, right, bottom = (-20, -65, 20, 60)
|
|
bdX += left
|
|
bdY += top
|
|
bdW += (right - left)
|
|
bdH += (bottom - top)
|
|
|
|
bdn = nuke.createNode("BackdropNode")
|
|
bdn["z_order"].setValue(layer)
|
|
|
|
if color:
|
|
bdn["tile_color"].setValue(int(color, 16))
|
|
|
|
bdn["xpos"].setValue(bdX)
|
|
bdn["ypos"].setValue(bdY)
|
|
bdn["bdwidth"].setValue(bdW)
|
|
bdn["bdheight"].setValue(bdH)
|
|
|
|
if label:
|
|
bdn["label"].setValue(label)
|
|
|
|
bdn["note_font_size"].setValue(20)
|
|
return bdn
|
|
|
|
|
|
class WorkfileSettings(object):
|
|
"""
|
|
All settings for workfile will be set
|
|
|
|
This object is setting all possible root settings to the workfile.
|
|
Including Colorspace, Frame ranges, Resolution format. It can set it
|
|
to Root node or to any given node.
|
|
|
|
Arguments:
|
|
root (node): nuke's root node
|
|
nodes (list): list of nuke's nodes
|
|
nodes_filter (list): filtering classes for nodes
|
|
|
|
"""
|
|
|
|
def __init__(self,
|
|
root_node=None,
|
|
nodes=None,
|
|
**kwargs):
|
|
opnl._project = kwargs.get(
|
|
"project") or io.find_one({"type": "project"})
|
|
self._asset = kwargs.get("asset_name") or api.Session["AVALON_ASSET"]
|
|
self._asset_entity = get_asset(self._asset)
|
|
self._root_node = root_node or nuke.root()
|
|
self._nodes = self.get_nodes(nodes=nodes)
|
|
|
|
self.data = kwargs
|
|
|
|
def get_nodes(self, nodes=None, nodes_filter=None):
|
|
|
|
if not isinstance(nodes, list) and not isinstance(nodes_filter, list):
|
|
return [n for n in nuke.allNodes()]
|
|
elif not isinstance(nodes, list) and isinstance(nodes_filter, list):
|
|
nodes = list()
|
|
for filter in nodes_filter:
|
|
[nodes.append(n) for n in nuke.allNodes(filter=filter)]
|
|
return nodes
|
|
elif isinstance(nodes, list) and not isinstance(nodes_filter, list):
|
|
return [n for n in self._nodes]
|
|
elif isinstance(nodes, list) and isinstance(nodes_filter, list):
|
|
for filter in nodes_filter:
|
|
return [n for n in self._nodes if filter in n.Class()]
|
|
|
|
def set_viewers_colorspace(self, viewer_dict):
|
|
''' Adds correct colorspace to viewer
|
|
|
|
Arguments:
|
|
viewer_dict (dict): adjustments from presets
|
|
|
|
'''
|
|
if not isinstance(viewer_dict, dict):
|
|
msg = "set_viewers_colorspace(): argument should be dictionary"
|
|
log.error(msg)
|
|
nuke.message(msg)
|
|
return
|
|
|
|
filter_knobs = [
|
|
"viewerProcess",
|
|
"wipe_position"
|
|
]
|
|
|
|
erased_viewers = []
|
|
for v in nuke.allNodes(filter="Viewer"):
|
|
v['viewerProcess'].setValue(str(viewer_dict["viewerProcess"]))
|
|
if str(viewer_dict["viewerProcess"]) \
|
|
not in v['viewerProcess'].value():
|
|
copy_inputs = v.dependencies()
|
|
copy_knobs = {k: v[k].value() for k in v.knobs()
|
|
if k not in filter_knobs}
|
|
|
|
# delete viewer with wrong settings
|
|
erased_viewers.append(v['name'].value())
|
|
nuke.delete(v)
|
|
|
|
# create new viewer
|
|
nv = nuke.createNode("Viewer")
|
|
|
|
# connect to original inputs
|
|
for i, n in enumerate(copy_inputs):
|
|
nv.setInput(i, n)
|
|
|
|
# set coppied knobs
|
|
for k, v in copy_knobs.items():
|
|
print(k, v)
|
|
nv[k].setValue(v)
|
|
|
|
# set viewerProcess
|
|
nv['viewerProcess'].setValue(str(viewer_dict["viewerProcess"]))
|
|
|
|
if erased_viewers:
|
|
log.warning(
|
|
"Attention! Viewer nodes {} were erased."
|
|
"It had wrong color profile".format(erased_viewers))
|
|
|
|
def set_root_colorspace(self, root_dict):
|
|
''' Adds correct colorspace to root
|
|
|
|
Arguments:
|
|
root_dict (dict): adjustmensts from presets
|
|
|
|
'''
|
|
if not isinstance(root_dict, dict):
|
|
msg = "set_root_colorspace(): argument should be dictionary"
|
|
log.error(msg)
|
|
nuke.message(msg)
|
|
|
|
log.warning(">> root_dict: {}".format(root_dict))
|
|
|
|
# first set OCIO
|
|
if self._root_node["colorManagement"].value() \
|
|
not in str(root_dict["colorManagement"]):
|
|
self._root_node["colorManagement"].setValue(
|
|
str(root_dict["colorManagement"]))
|
|
log.debug("nuke.root()['{0}'] changed to: {1}".format(
|
|
"colorManagement", root_dict["colorManagement"]))
|
|
root_dict.pop("colorManagement")
|
|
|
|
# second set ocio version
|
|
if self._root_node["OCIO_config"].value() \
|
|
not in str(root_dict["OCIO_config"]):
|
|
self._root_node["OCIO_config"].setValue(
|
|
str(root_dict["OCIO_config"]))
|
|
log.debug("nuke.root()['{0}'] changed to: {1}".format(
|
|
"OCIO_config", root_dict["OCIO_config"]))
|
|
root_dict.pop("OCIO_config")
|
|
|
|
# third set ocio custom path
|
|
if root_dict.get("customOCIOConfigPath"):
|
|
unresolved_path = root_dict["customOCIOConfigPath"]
|
|
ocio_paths = unresolved_path[platform.system().lower()]
|
|
|
|
resolved_path = None
|
|
for ocio_p in ocio_paths:
|
|
resolved_path = str(ocio_p).format(**os.environ)
|
|
if not os.path.exists(resolved_path):
|
|
continue
|
|
|
|
if resolved_path:
|
|
self._root_node["customOCIOConfigPath"].setValue(
|
|
str(resolved_path).replace("\\", "/")
|
|
)
|
|
log.debug("nuke.root()['{}'] changed to: {}".format(
|
|
"customOCIOConfigPath", resolved_path))
|
|
root_dict.pop("customOCIOConfigPath")
|
|
|
|
# then set the rest
|
|
for knob, value in root_dict.items():
|
|
# skip unfilled ocio config path
|
|
# it will be dict in value
|
|
if isinstance(value, dict):
|
|
continue
|
|
if self._root_node[knob].value() not in value:
|
|
self._root_node[knob].setValue(str(value))
|
|
log.debug("nuke.root()['{}'] changed to: {}".format(
|
|
knob, value))
|
|
|
|
def set_writes_colorspace(self):
|
|
''' Adds correct colorspace to write node dict
|
|
|
|
'''
|
|
from avalon.nuke import read
|
|
|
|
for node in nuke.allNodes(filter="Group"):
|
|
|
|
# get data from avalon knob
|
|
avalon_knob_data = read(node)
|
|
|
|
if not avalon_knob_data:
|
|
continue
|
|
|
|
if avalon_knob_data["id"] != "pyblish.avalon.instance":
|
|
continue
|
|
|
|
# establish families
|
|
families = [avalon_knob_data["family"]]
|
|
if avalon_knob_data.get("families"):
|
|
families.append(avalon_knob_data.get("families"))
|
|
|
|
data_preset = {
|
|
"nodeclass": avalon_knob_data["family"],
|
|
"families": families,
|
|
"creator": avalon_knob_data['creator']
|
|
}
|
|
|
|
nuke_imageio_writes = get_created_node_imageio_setting(
|
|
**data_preset)
|
|
|
|
log.debug("nuke_imageio_writes: `{}`".format(nuke_imageio_writes))
|
|
|
|
if not nuke_imageio_writes:
|
|
return
|
|
|
|
write_node = None
|
|
|
|
# get into the group node
|
|
node.begin()
|
|
for x in nuke.allNodes():
|
|
if x.Class() == "Write":
|
|
write_node = x
|
|
node.end()
|
|
|
|
if not write_node:
|
|
return
|
|
|
|
# write all knobs to node
|
|
for knob in nuke_imageio_writes["knobs"]:
|
|
value = knob["value"]
|
|
if isinstance(value, six.text_type):
|
|
value = str(value)
|
|
if str(value).startswith("0x"):
|
|
value = int(value, 16)
|
|
|
|
write_node[knob["name"]].setValue(value)
|
|
|
|
|
|
def set_reads_colorspace(self, read_clrs_inputs):
|
|
""" Setting colorspace to Read nodes
|
|
|
|
Looping trought all read nodes and tries to set colorspace based
|
|
on regex rules in presets
|
|
"""
|
|
changes = {}
|
|
for n in nuke.allNodes():
|
|
file = nuke.filename(n)
|
|
if n.Class() != "Read":
|
|
continue
|
|
|
|
# check if any colorspace presets for read is mathing
|
|
preset_clrsp = None
|
|
|
|
for input in read_clrs_inputs:
|
|
if not bool(re.search(input["regex"], file)):
|
|
continue
|
|
preset_clrsp = input["colorspace"]
|
|
|
|
log.debug(preset_clrsp)
|
|
if preset_clrsp is not None:
|
|
current = n["colorspace"].value()
|
|
future = str(preset_clrsp)
|
|
if current != future:
|
|
changes.update({
|
|
n.name(): {
|
|
"from": current,
|
|
"to": future
|
|
}
|
|
})
|
|
log.debug(changes)
|
|
if changes:
|
|
msg = "Read nodes are not set to correct colospace:\n\n"
|
|
for nname, knobs in changes.items():
|
|
msg += str(
|
|
" - node: '{0}' is now '{1}' but should be '{2}'\n"
|
|
).format(nname, knobs["from"], knobs["to"])
|
|
|
|
msg += "\nWould you like to change it?"
|
|
|
|
if nuke.ask(msg):
|
|
for nname, knobs in changes.items():
|
|
n = nuke.toNode(nname)
|
|
n["colorspace"].setValue(knobs["to"])
|
|
log.info(
|
|
"Setting `{0}` to `{1}`".format(
|
|
nname,
|
|
knobs["to"]))
|
|
|
|
def set_colorspace(self):
|
|
''' Setting colorpace following presets
|
|
'''
|
|
# get imageio
|
|
imageio = get_anatomy_settings(opnl.project_name)["imageio"]
|
|
nuke_colorspace = imageio["nuke"]
|
|
|
|
try:
|
|
self.set_root_colorspace(nuke_colorspace["workfile"])
|
|
except AttributeError:
|
|
msg = "set_colorspace(): missing `workfile` settings in template"
|
|
nuke.message(msg)
|
|
|
|
try:
|
|
self.set_viewers_colorspace(nuke_colorspace["viewer"])
|
|
except AttributeError:
|
|
msg = "set_colorspace(): missing `viewer` settings in template"
|
|
nuke.message(msg)
|
|
log.error(msg)
|
|
|
|
try:
|
|
self.set_writes_colorspace()
|
|
except AttributeError as _error:
|
|
nuke.message(_error)
|
|
log.error(_error)
|
|
|
|
read_clrs_inputs = nuke_colorspace["regexInputs"].get("inputs", [])
|
|
if read_clrs_inputs:
|
|
self.set_reads_colorspace(read_clrs_inputs)
|
|
|
|
try:
|
|
for key in nuke_colorspace:
|
|
log.debug("Preset's colorspace key: {}".format(key))
|
|
except TypeError:
|
|
msg = "Nuke is not in templates! Contact your supervisor!"
|
|
nuke.message(msg)
|
|
log.error(msg)
|
|
|
|
def reset_frame_range_handles(self):
|
|
"""Set frame range to current asset"""
|
|
|
|
if "data" not in self._asset_entity:
|
|
msg = "Asset {} don't have set any 'data'".format(self._asset)
|
|
log.warning(msg)
|
|
nuke.message(msg)
|
|
return
|
|
data = self._asset_entity["data"]
|
|
|
|
log.debug("__ asset data: `{}`".format(data))
|
|
|
|
missing_cols = []
|
|
check_cols = ["fps", "frameStart", "frameEnd",
|
|
"handleStart", "handleEnd"]
|
|
|
|
for col in check_cols:
|
|
if col not in data:
|
|
missing_cols.append(col)
|
|
|
|
if len(missing_cols) > 0:
|
|
missing = ", ".join(missing_cols)
|
|
msg = "'{}' are not set for asset '{}'!".format(
|
|
missing, self._asset)
|
|
log.warning(msg)
|
|
nuke.message(msg)
|
|
return
|
|
|
|
# get handles values
|
|
handle_start = data["handleStart"]
|
|
handle_end = data["handleEnd"]
|
|
|
|
fps = float(data["fps"])
|
|
frame_start = int(data["frameStart"]) - handle_start
|
|
frame_end = int(data["frameEnd"]) + handle_end
|
|
|
|
self._root_node["lock_range"].setValue(False)
|
|
self._root_node["fps"].setValue(fps)
|
|
self._root_node["first_frame"].setValue(frame_start)
|
|
self._root_node["last_frame"].setValue(frame_end)
|
|
self._root_node["lock_range"].setValue(True)
|
|
|
|
# setting active viewers
|
|
try:
|
|
nuke.frame(int(data["frameStart"]))
|
|
except Exception as e:
|
|
log.warning("no viewer in scene: `{}`".format(e))
|
|
|
|
range = '{0}-{1}'.format(
|
|
int(data["frameStart"]),
|
|
int(data["frameEnd"]))
|
|
|
|
for node in nuke.allNodes(filter="Viewer"):
|
|
node['frame_range'].setValue(range)
|
|
node['frame_range_lock'].setValue(True)
|
|
node['frame_range'].setValue(range)
|
|
node['frame_range_lock'].setValue(True)
|
|
|
|
# adding handle_start/end to root avalon knob
|
|
if not anlib.set_avalon_knob_data(self._root_node, {
|
|
"handleStart": int(handle_start),
|
|
"handleEnd": int(handle_end)
|
|
}):
|
|
log.warning("Cannot set Avalon knob to Root node!")
|
|
|
|
def reset_resolution(self):
|
|
"""Set resolution to project resolution."""
|
|
log.info("Reseting resolution")
|
|
project = io.find_one({"type": "project"})
|
|
asset = api.Session["AVALON_ASSET"]
|
|
asset = io.find_one({"name": asset, "type": "asset"})
|
|
asset_data = asset.get('data', {})
|
|
|
|
data = {
|
|
"width": int(asset_data.get(
|
|
'resolutionWidth',
|
|
asset_data.get('resolution_width'))),
|
|
"height": int(asset_data.get(
|
|
'resolutionHeight',
|
|
asset_data.get('resolution_height'))),
|
|
"pixel_aspect": asset_data.get(
|
|
'pixelAspect',
|
|
asset_data.get('pixel_aspect', 1)),
|
|
"name": project["name"]
|
|
}
|
|
|
|
if any(x for x in data.values() if x is None):
|
|
msg = ("Missing set shot attributes in DB."
|
|
"\nContact your supervisor!."
|
|
"\n\nWidth: `{width}`"
|
|
"\nHeight: `{height}`"
|
|
"\nPixel Asspect: `{pixel_aspect}`").format(**data)
|
|
log.error(msg)
|
|
nuke.message(msg)
|
|
|
|
bbox = self._asset_entity.get('data', {}).get('crop')
|
|
|
|
if bbox:
|
|
try:
|
|
x, y, r, t = bbox.split(".")
|
|
data.update(
|
|
{
|
|
"x": int(x),
|
|
"y": int(y),
|
|
"r": int(r),
|
|
"t": int(t),
|
|
}
|
|
)
|
|
except Exception as e:
|
|
bbox = None
|
|
msg = ("{}:{} \nFormat:Crop need to be set with dots, "
|
|
"example: 0.0.1920.1080, "
|
|
"/nSetting to default").format(__name__, e)
|
|
log.error(msg)
|
|
nuke.message(msg)
|
|
|
|
existing_format = None
|
|
for format in nuke.formats():
|
|
if data["name"] == format.name():
|
|
existing_format = format
|
|
break
|
|
|
|
if existing_format:
|
|
# Enforce existing format to be correct.
|
|
existing_format.setWidth(data["width"])
|
|
existing_format.setHeight(data["height"])
|
|
existing_format.setPixelAspect(data["pixel_aspect"])
|
|
|
|
if bbox:
|
|
existing_format.setX(data["x"])
|
|
existing_format.setY(data["y"])
|
|
existing_format.setR(data["r"])
|
|
existing_format.setT(data["t"])
|
|
else:
|
|
format_string = self.make_format_string(**data)
|
|
log.info("Creating new format: {}".format(format_string))
|
|
nuke.addFormat(format_string)
|
|
|
|
nuke.root()["format"].setValue(data["name"])
|
|
log.info("Format is set.")
|
|
|
|
def make_format_string(self, **kwargs):
|
|
if kwargs.get("r"):
|
|
return (
|
|
"{width} "
|
|
"{height} "
|
|
"{x} "
|
|
"{y} "
|
|
"{r} "
|
|
"{t} "
|
|
"{pixel_aspect:.2f} "
|
|
"{name}".format(**kwargs)
|
|
)
|
|
else:
|
|
return (
|
|
"{width} "
|
|
"{height} "
|
|
"{pixel_aspect:.2f} "
|
|
"{name}".format(**kwargs)
|
|
)
|
|
|
|
def set_context_settings(self):
|
|
# replace reset resolution from avalon core to pype's
|
|
self.reset_resolution()
|
|
# replace reset resolution from avalon core to pype's
|
|
self.reset_frame_range_handles()
|
|
# add colorspace menu item
|
|
# self.set_colorspace()
|
|
|
|
def set_favorites(self):
|
|
work_dir = os.getenv("AVALON_WORKDIR")
|
|
asset = os.getenv("AVALON_ASSET")
|
|
favorite_items = OrderedDict()
|
|
|
|
# project
|
|
# get project's root and split to parts
|
|
projects_root = os.path.normpath(work_dir.split(
|
|
opnl.project_name)[0])
|
|
# add project name
|
|
project_dir = os.path.join(projects_root, opnl.project_name) + "/"
|
|
# add to favorites
|
|
favorite_items.update({"Project dir": project_dir.replace("\\", "/")})
|
|
|
|
# asset
|
|
asset_root = os.path.normpath(work_dir.split(
|
|
asset)[0])
|
|
# add asset name
|
|
asset_dir = os.path.join(asset_root, asset) + "/"
|
|
# add to favorites
|
|
favorite_items.update({"Shot dir": asset_dir.replace("\\", "/")})
|
|
|
|
# workdir
|
|
favorite_items.update({"Work dir": work_dir.replace("\\", "/")})
|
|
|
|
set_context_favorites(favorite_items)
|
|
|
|
|
|
def get_hierarchical_attr(entity, attr, default=None):
|
|
attr_parts = attr.split('.')
|
|
value = entity
|
|
for part in attr_parts:
|
|
value = value.get(part)
|
|
if not value:
|
|
break
|
|
|
|
if value or entity['type'].lower() == 'project':
|
|
return value
|
|
|
|
parent_id = entity['parent']
|
|
if (
|
|
entity['type'].lower() == 'asset'
|
|
and entity.get('data', {}).get('visualParent')
|
|
):
|
|
parent_id = entity['data']['visualParent']
|
|
|
|
parent = io.find_one({'_id': parent_id})
|
|
|
|
return get_hierarchical_attr(parent, attr)
|
|
|
|
|
|
def get_write_node_template_attr(node):
|
|
''' Gets all defined data from presets
|
|
|
|
'''
|
|
# get avalon data from node
|
|
data = dict()
|
|
data['avalon'] = avalon.nuke.read(
|
|
node)
|
|
data_preset = {
|
|
"nodeclass": data['avalon']['family'],
|
|
"families": [data['avalon']['families']],
|
|
"creator": data['avalon']['creator']
|
|
}
|
|
|
|
# get template data
|
|
nuke_imageio_writes = get_created_node_imageio_setting(**data_preset)
|
|
|
|
# collecting correct data
|
|
correct_data = OrderedDict({
|
|
"file": get_render_path(node)
|
|
})
|
|
|
|
# adding imageio template
|
|
{correct_data.update({k: v})
|
|
for k, v in nuke_imageio_writes.items()
|
|
if k not in ["_id", "_previous"]}
|
|
|
|
# fix badly encoded data
|
|
return anlib.fix_data_for_node_create(correct_data)
|
|
|
|
|
|
class ExporterReview:
|
|
"""
|
|
Base class object for generating review data from Nuke
|
|
|
|
Args:
|
|
klass (pyblish.plugin): pyblish plugin parent
|
|
instance (pyblish.instance): instance of pyblish context
|
|
|
|
"""
|
|
_temp_nodes = []
|
|
data = dict({
|
|
"representations": list()
|
|
})
|
|
|
|
def __init__(self,
|
|
klass,
|
|
instance
|
|
):
|
|
|
|
self.log = klass.log
|
|
self.instance = instance
|
|
self.path_in = self.instance.data.get("path", None)
|
|
self.staging_dir = self.instance.data["stagingDir"]
|
|
self.collection = self.instance.data.get("collection", None)
|
|
|
|
def get_file_info(self):
|
|
if self.collection:
|
|
self.log.debug("Collection: `{}`".format(self.collection))
|
|
# get path
|
|
self.fname = os.path.basename(self.collection.format(
|
|
"{head}{padding}{tail}"))
|
|
self.fhead = self.collection.format("{head}")
|
|
|
|
# get first and last frame
|
|
self.first_frame = min(self.collection.indexes)
|
|
self.last_frame = max(self.collection.indexes)
|
|
if "slate" in self.instance.data["families"]:
|
|
self.first_frame += 1
|
|
else:
|
|
self.fname = os.path.basename(self.path_in)
|
|
self.fhead = os.path.splitext(self.fname)[0] + "."
|
|
self.first_frame = self.instance.data.get("frameStartHandle", None)
|
|
self.last_frame = self.instance.data.get("frameEndHandle", None)
|
|
|
|
if "#" in self.fhead:
|
|
self.fhead = self.fhead.replace("#", "")[:-1]
|
|
|
|
def get_representation_data(self, tags=None, range=False):
|
|
add_tags = []
|
|
if tags:
|
|
add_tags = tags
|
|
|
|
repre = {
|
|
'name': self.name,
|
|
'ext': self.ext,
|
|
'files': self.file,
|
|
"stagingDir": self.staging_dir,
|
|
"tags": [self.name.replace("_", "-")] + add_tags
|
|
}
|
|
|
|
if range:
|
|
repre.update({
|
|
"frameStart": self.first_frame,
|
|
"frameEnd": self.last_frame,
|
|
})
|
|
|
|
self.data["representations"].append(repre)
|
|
|
|
def get_view_process_node(self):
|
|
"""
|
|
Will get any active view process.
|
|
|
|
Arguments:
|
|
self (class): in object definition
|
|
|
|
Returns:
|
|
nuke.Node: copy node of Input Process node
|
|
"""
|
|
anlib.reset_selection()
|
|
ipn_orig = None
|
|
for v in nuke.allNodes(filter="Viewer"):
|
|
ip = v['input_process'].getValue()
|
|
ipn = v['input_process_node'].getValue()
|
|
if "VIEWER_INPUT" not in ipn and ip:
|
|
ipn_orig = nuke.toNode(ipn)
|
|
ipn_orig.setSelected(True)
|
|
|
|
if ipn_orig:
|
|
# copy selected to clipboard
|
|
nuke.nodeCopy('%clipboard%')
|
|
# reset selection
|
|
anlib.reset_selection()
|
|
# paste node and selection is on it only
|
|
nuke.nodePaste('%clipboard%')
|
|
# assign to variable
|
|
ipn = nuke.selectedNode()
|
|
|
|
return ipn
|
|
|
|
def clean_nodes(self):
|
|
for node in self._temp_nodes:
|
|
nuke.delete(node)
|
|
self.log.info("Deleted nodes...")
|
|
|
|
|
|
class ExporterReviewLut(ExporterReview):
|
|
"""
|
|
Generator object for review lut from Nuke
|
|
|
|
Args:
|
|
klass (pyblish.plugin): pyblish plugin parent
|
|
instance (pyblish.instance): instance of pyblish context
|
|
|
|
|
|
"""
|
|
|
|
def __init__(self,
|
|
klass,
|
|
instance,
|
|
name=None,
|
|
ext=None,
|
|
cube_size=None,
|
|
lut_size=None,
|
|
lut_style=None):
|
|
# initialize parent class
|
|
ExporterReview.__init__(self, klass, instance)
|
|
|
|
# deal with now lut defined in viewer lut
|
|
if hasattr(klass, "viewer_lut_raw"):
|
|
self.viewer_lut_raw = klass.viewer_lut_raw
|
|
else:
|
|
self.viewer_lut_raw = False
|
|
|
|
self.name = name or "baked_lut"
|
|
self.ext = ext or "cube"
|
|
self.cube_size = cube_size or 32
|
|
self.lut_size = lut_size or 1024
|
|
self.lut_style = lut_style or "linear"
|
|
|
|
# set frame start / end and file name to self
|
|
self.get_file_info()
|
|
|
|
self.log.info("File info was set...")
|
|
|
|
self.file = self.fhead + self.name + ".{}".format(self.ext)
|
|
self.path = os.path.join(
|
|
self.staging_dir, self.file).replace("\\", "/")
|
|
|
|
def generate_lut(self):
|
|
# ---------- start nodes creation
|
|
|
|
# CMSTestPattern
|
|
cms_node = nuke.createNode("CMSTestPattern")
|
|
cms_node["cube_size"].setValue(self.cube_size)
|
|
# connect
|
|
self._temp_nodes.append(cms_node)
|
|
self.previous_node = cms_node
|
|
self.log.debug("CMSTestPattern... `{}`".format(self._temp_nodes))
|
|
|
|
# Node View Process
|
|
ipn = self.get_view_process_node()
|
|
if ipn is not None:
|
|
# connect
|
|
ipn.setInput(0, self.previous_node)
|
|
self._temp_nodes.append(ipn)
|
|
self.previous_node = ipn
|
|
self.log.debug("ViewProcess... `{}`".format(self._temp_nodes))
|
|
|
|
if not self.viewer_lut_raw:
|
|
# OCIODisplay
|
|
dag_node = nuke.createNode("OCIODisplay")
|
|
# connect
|
|
dag_node.setInput(0, self.previous_node)
|
|
self._temp_nodes.append(dag_node)
|
|
self.previous_node = dag_node
|
|
self.log.debug("OCIODisplay... `{}`".format(self._temp_nodes))
|
|
|
|
# GenerateLUT
|
|
gen_lut_node = nuke.createNode("GenerateLUT")
|
|
gen_lut_node["file"].setValue(self.path)
|
|
gen_lut_node["file_type"].setValue(".{}".format(self.ext))
|
|
gen_lut_node["lut1d"].setValue(self.lut_size)
|
|
gen_lut_node["style1d"].setValue(self.lut_style)
|
|
# connect
|
|
gen_lut_node.setInput(0, self.previous_node)
|
|
self._temp_nodes.append(gen_lut_node)
|
|
self.log.debug("GenerateLUT... `{}`".format(self._temp_nodes))
|
|
|
|
# ---------- end nodes creation
|
|
|
|
# Export lut file
|
|
nuke.execute(
|
|
gen_lut_node.name(),
|
|
int(self.first_frame),
|
|
int(self.first_frame))
|
|
|
|
self.log.info("Exported...")
|
|
|
|
# ---------- generate representation data
|
|
self.get_representation_data()
|
|
|
|
self.log.debug("Representation... `{}`".format(self.data))
|
|
|
|
# ---------- Clean up
|
|
self.clean_nodes()
|
|
|
|
return self.data
|
|
|
|
|
|
class ExporterReviewMov(ExporterReview):
|
|
"""
|
|
Metaclass for generating review mov files
|
|
|
|
Args:
|
|
klass (pyblish.plugin): pyblish plugin parent
|
|
instance (pyblish.instance): instance of pyblish context
|
|
|
|
"""
|
|
|
|
def __init__(self,
|
|
klass,
|
|
instance,
|
|
name=None,
|
|
ext=None,
|
|
):
|
|
# initialize parent class
|
|
ExporterReview.__init__(self, klass, instance)
|
|
|
|
# passing presets for nodes to self
|
|
if hasattr(klass, "nodes"):
|
|
self.nodes = klass.nodes
|
|
else:
|
|
self.nodes = {}
|
|
|
|
# deal with now lut defined in viewer lut
|
|
self.viewer_lut_raw = klass.viewer_lut_raw
|
|
self.bake_colorspace_fallback = klass.bake_colorspace_fallback
|
|
self.bake_colorspace_main = klass.bake_colorspace_main
|
|
self.write_colorspace = instance.data["colorspace"]
|
|
|
|
self.name = name or "baked"
|
|
self.ext = ext or "mov"
|
|
|
|
# set frame start / end and file name to self
|
|
self.get_file_info()
|
|
|
|
self.log.info("File info was set...")
|
|
|
|
self.file = self.fhead + self.name + ".{}".format(self.ext)
|
|
self.path = os.path.join(
|
|
self.staging_dir, self.file).replace("\\", "/")
|
|
|
|
def render(self, render_node_name):
|
|
self.log.info("Rendering... ")
|
|
# Render Write node
|
|
nuke.execute(
|
|
render_node_name,
|
|
int(self.first_frame),
|
|
int(self.last_frame))
|
|
|
|
self.log.info("Rendered...")
|
|
|
|
def save_file(self):
|
|
import shutil
|
|
with anlib.maintained_selection():
|
|
self.log.info("Saving nodes as file... ")
|
|
# create nk path
|
|
path = os.path.splitext(self.path)[0] + ".nk"
|
|
# save file to the path
|
|
shutil.copyfile(self.instance.context.data["currentFile"], path)
|
|
|
|
self.log.info("Nodes exported...")
|
|
return path
|
|
|
|
def generate_mov(self, farm=False):
|
|
# ---------- start nodes creation
|
|
|
|
# Read node
|
|
r_node = nuke.createNode("Read")
|
|
r_node["file"].setValue(self.path_in)
|
|
r_node["first"].setValue(self.first_frame)
|
|
r_node["origfirst"].setValue(self.first_frame)
|
|
r_node["last"].setValue(self.last_frame)
|
|
r_node["origlast"].setValue(self.last_frame)
|
|
r_node["colorspace"].setValue(self.write_colorspace)
|
|
|
|
# connect
|
|
self._temp_nodes.append(r_node)
|
|
self.previous_node = r_node
|
|
self.log.debug("Read... `{}`".format(self._temp_nodes))
|
|
|
|
# View Process node
|
|
ipn = self.get_view_process_node()
|
|
if ipn is not None:
|
|
# connect
|
|
ipn.setInput(0, self.previous_node)
|
|
self._temp_nodes.append(ipn)
|
|
self.previous_node = ipn
|
|
self.log.debug("ViewProcess... `{}`".format(self._temp_nodes))
|
|
|
|
if not self.viewer_lut_raw:
|
|
colorspaces = [
|
|
self.bake_colorspace_main, self.bake_colorspace_fallback
|
|
]
|
|
|
|
if any(colorspaces):
|
|
# OCIOColorSpace with controled output
|
|
dag_node = nuke.createNode("OCIOColorSpace")
|
|
self._temp_nodes.append(dag_node)
|
|
for c in colorspaces:
|
|
test = dag_node["out_colorspace"].setValue(str(c))
|
|
if test:
|
|
self.log.info(
|
|
"Baking in colorspace... `{}`".format(c))
|
|
break
|
|
|
|
if not test:
|
|
dag_node = nuke.createNode("OCIODisplay")
|
|
else:
|
|
# OCIODisplay
|
|
dag_node = nuke.createNode("OCIODisplay")
|
|
|
|
# connect
|
|
dag_node.setInput(0, self.previous_node)
|
|
self._temp_nodes.append(dag_node)
|
|
self.previous_node = dag_node
|
|
self.log.debug("OCIODisplay... `{}`".format(self._temp_nodes))
|
|
|
|
# Write node
|
|
write_node = nuke.createNode("Write")
|
|
self.log.debug("Path: {}".format(self.path))
|
|
write_node["file"].setValue(self.path)
|
|
write_node["file_type"].setValue(self.ext)
|
|
|
|
# Knobs `meta_codec` and `mov64_codec` are not available on centos.
|
|
# TODO change this to use conditions, if possible.
|
|
try:
|
|
write_node["meta_codec"].setValue("ap4h")
|
|
except Exception:
|
|
self.log.info("`meta_codec` knob was not found")
|
|
|
|
try:
|
|
write_node["mov64_codec"].setValue("ap4h")
|
|
except Exception:
|
|
self.log.info("`mov64_codec` knob was not found")
|
|
write_node["mov64_write_timecode"].setValue(1)
|
|
write_node["raw"].setValue(1)
|
|
# connect
|
|
write_node.setInput(0, self.previous_node)
|
|
self._temp_nodes.append(write_node)
|
|
self.log.debug("Write... `{}`".format(self._temp_nodes))
|
|
# ---------- end nodes creation
|
|
|
|
# ---------- render or save to nk
|
|
if farm:
|
|
nuke.scriptSave()
|
|
path_nk = self.save_file()
|
|
self.data.update({
|
|
"bakeScriptPath": path_nk,
|
|
"bakeWriteNodeName": write_node.name(),
|
|
"bakeRenderPath": self.path
|
|
})
|
|
else:
|
|
self.render(write_node.name())
|
|
# ---------- generate representation data
|
|
self.get_representation_data(
|
|
tags=["review", "delete"],
|
|
range=True
|
|
)
|
|
|
|
self.log.debug("Representation... `{}`".format(self.data))
|
|
|
|
# ---------- Clean up
|
|
self.clean_nodes()
|
|
nuke.scriptSave()
|
|
return self.data
|
|
|
|
|
|
def get_dependent_nodes(nodes):
|
|
"""Get all dependent nodes connected to the list of nodes.
|
|
|
|
Looking for connections outside of the nodes in incoming argument.
|
|
|
|
Arguments:
|
|
nodes (list): list of nuke.Node objects
|
|
|
|
Returns:
|
|
connections_in: dictionary of nodes and its dependencies
|
|
connections_out: dictionary of nodes and its dependency
|
|
"""
|
|
|
|
connections_in = dict()
|
|
connections_out = dict()
|
|
node_names = [n.name() for n in nodes]
|
|
for node in nodes:
|
|
inputs = node.dependencies()
|
|
outputs = node.dependent()
|
|
# collect all inputs outside
|
|
test_in = [(i, n) for i, n in enumerate(inputs)
|
|
if n.name() not in node_names]
|
|
if test_in:
|
|
connections_in.update({
|
|
node: test_in
|
|
})
|
|
# collect all outputs outside
|
|
test_out = [i for i in outputs if i.name() not in node_names]
|
|
if test_out:
|
|
# only one dependent node is allowed
|
|
connections_out.update({
|
|
node: test_out[-1]
|
|
})
|
|
|
|
return connections_in, connections_out
|
|
|
|
|
|
def find_free_space_to_paste_nodes(
|
|
nodes,
|
|
group=nuke.root(),
|
|
direction="right",
|
|
offset=300):
|
|
"""
|
|
For getting coordinates in DAG (node graph) for placing new nodes
|
|
|
|
Arguments:
|
|
nodes (list): list of nuke.Node objects
|
|
group (nuke.Node) [optional]: object in which context it is
|
|
direction (str) [optional]: where we want it to be placed
|
|
[left, right, top, bottom]
|
|
offset (int) [optional]: what offset it is from rest of nodes
|
|
|
|
Returns:
|
|
xpos (int): x coordinace in DAG
|
|
ypos (int): y coordinace in DAG
|
|
"""
|
|
if len(nodes) == 0:
|
|
return 0, 0
|
|
|
|
group_xpos = list()
|
|
group_ypos = list()
|
|
|
|
# get local coordinates of all nodes
|
|
nodes_xpos = [n.xpos() for n in nodes] + \
|
|
[n.xpos() + n.screenWidth() for n in nodes]
|
|
|
|
nodes_ypos = [n.ypos() for n in nodes] + \
|
|
[n.ypos() + n.screenHeight() for n in nodes]
|
|
|
|
# get complete screen size of all nodes to be placed in
|
|
nodes_screen_width = max(nodes_xpos) - min(nodes_xpos)
|
|
nodes_screen_heigth = max(nodes_ypos) - min(nodes_ypos)
|
|
|
|
# get screen size (r,l,t,b) of all nodes in `group`
|
|
with group:
|
|
group_xpos = [n.xpos() for n in nuke.allNodes() if n not in nodes] + \
|
|
[n.xpos() + n.screenWidth() for n in nuke.allNodes()
|
|
if n not in nodes]
|
|
group_ypos = [n.ypos() for n in nuke.allNodes() if n not in nodes] + \
|
|
[n.ypos() + n.screenHeight() for n in nuke.allNodes()
|
|
if n not in nodes]
|
|
|
|
# calc output left
|
|
if direction in "left":
|
|
xpos = min(group_xpos) - abs(nodes_screen_width) - abs(offset)
|
|
ypos = min(group_ypos)
|
|
return xpos, ypos
|
|
# calc output right
|
|
if direction in "right":
|
|
xpos = max(group_xpos) + abs(offset)
|
|
ypos = min(group_ypos)
|
|
return xpos, ypos
|
|
# calc output top
|
|
if direction in "top":
|
|
xpos = min(group_xpos)
|
|
ypos = min(group_ypos) - abs(nodes_screen_heigth) - abs(offset)
|
|
return xpos, ypos
|
|
# calc output bottom
|
|
if direction in "bottom":
|
|
xpos = min(group_xpos)
|
|
ypos = max(group_ypos) + abs(offset)
|
|
return xpos, ypos
|
|
|
|
|
|
def launch_workfiles_app():
|
|
'''Function letting start workfiles after start of host
|
|
'''
|
|
# get state from settings
|
|
open_at_start = get_current_project_settings()["nuke"].get(
|
|
"general", {}).get("open_workfile_at_start")
|
|
|
|
# return if none is defined
|
|
if not open_at_start:
|
|
return
|
|
|
|
if not opnl.workfiles_launched:
|
|
opnl.workfiles_launched = True
|
|
workfiles.show(os.environ["AVALON_WORKDIR"])
|
|
|
|
|
|
def open_last_workfile():
|
|
# get state from settings
|
|
open_last_version = get_current_project_settings()["nuke"].get(
|
|
"general", {}).get("create_initial_workfile")
|
|
|
|
log.info("Opening last workfile...")
|
|
last_workfile_path = os.environ.get("AVALON_LAST_WORKFILE")
|
|
|
|
if not os.path.exists(last_workfile_path):
|
|
# return if none is defined
|
|
if not open_last_version:
|
|
return
|
|
|
|
save_file(last_workfile_path)
|
|
else:
|
|
# to avoid looping of the callback, remove it!
|
|
nuke.removeOnCreate(open_last_workfile, nodeClass="Root")
|
|
|
|
# open workfile
|
|
open_file(last_workfile_path)
|