pipeline

def add_nuke_callbacks(project_settings: dict = None):
    """Adding all available nuke callbacks"""
    if project_settings is None:
        project_settings = get_current_project_settings()

    nuke_settings = project_settings["nuke"]

    nuke.addOnCreate(on_root_create, nodeClass="Root")

    nuke.addOnScriptLoad(on_script_load)

    # set checker for last versions on loaded containers
    nuke.addOnScriptSave(check_inventory_versions)

    if nuke_settings["dirmap"]["enabled"]:
        log.info("Added Nuke's dir-mapping callback ...")
        # Add dirmap for file paths.
        nuke.addFilenameFilter(dirmap_file_name_filter)

    log.info("Added Nuke callbacks ...")

def containerise(node,
                 name,
                 namespace,
                 context,
                 loader=None,
                 data=None):
    """Bundle `node` into an assembly and imprint it with metadata

    Containerisation enables a tracking of version, author and origin
    for loaded assets.

    Arguments:
        node (nuke.Node): Nuke's node object to imprint as container
        name (str): Name of resulting assembly
        namespace (str): Namespace under which to host container
        context (dict): Asset information
        loader (str, optional): Name of node used to produce this container.

    Returns:
        node (nuke.Node): containerised nuke's node object

    """
    data = OrderedDict(
        [
            ("schema", "ayon:container-3.0"),
            ("id", AVALON_CONTAINER_ID),
            ("name", name),
            ("namespace", namespace),
            ("loader", str(loader)),
            ("representation", context["representation"]["id"]),
            ("project_name", context["project"]["name"]),
        ],

        **data or dict()
    )

    set_avalon_knob_data(node, data)

    # set tab to first native
    node.setTab(0)

    return node

def list_instances(creator_id=None):
    """List all created instances to publish from current workfile.

    For SubsetManager

    Args:
        creator_id (Optional[str]): creator identifier

    Returns:
        (list) of dictionaries matching instances format
    """
    instances_by_order = defaultdict(list)
    product_instances = []
    instance_ids = set()

    for node in nuke.allNodes(recurseGroups=True):

        if node.Class() in ["Viewer", "Dot"]:
            continue

        try:
            if node["disable"].value():
                continue
        except NameError:
            # pass if disable knob doesn't exist
            pass

        # get data from avalon knob
        instance_data = get_node_data(
            node, INSTANCE_DATA_KNOB)

        if not instance_data:
            continue

        if instance_data["id"] not in {
            AYON_INSTANCE_ID, AVALON_INSTANCE_ID
        }:
            continue

        if creator_id and instance_data["creator_identifier"] != creator_id:
            continue

        instance_id = instance_data.get("instance_id")
        if not instance_id:
            pass
        elif instance_id in instance_ids:
            instance_data.pop("instance_id")
        else:
            instance_ids.add(instance_id)

        # node name could change, so update product name data
        _update_product_name_data(instance_data, node)

        if "render_order" not in node.knobs():
            product_instances.append((node, instance_data))
            continue

        order = int(node["render_order"].value())
        instances_by_order[order].append((node, instance_data))

    # Sort instances based on order attribute or product name.
    # TODO: remove in future Publisher enhanced with sorting
    ordered_instances = []
    for key in sorted(instances_by_order.keys()):
        instances_by_product = defaultdict(list)
        for node, data_ in instances_by_order[key]:
            product_name = data_.get("productName")
            if product_name is None:
                product_name = data_.get("subset")
            instances_by_product[product_name].append((node, data_))
        for subkey in sorted(instances_by_product.keys()):
            ordered_instances.extend(instances_by_product[subkey])

    instances_by_product = defaultdict(list)
    for node, data_ in product_instances:
        product_name = data_.get("productName")
        if product_name is None:
            product_name = data_.get("subset")
        instances_by_product[product_name].append((node, data_))
    for key in sorted(instances_by_product.keys()):
        ordered_instances.extend(instances_by_product[key])

    return ordered_instances

def ls():
    """List available containers.

    This function is used by the Container Manager in Nuke. You'll
    need to implement a for-loop that then *yields* one Container at
    a time.
    """
    all_nodes = nuke.allNodes(recurseGroups=True)

    nodes = [n for n in all_nodes]

    for n in nodes:
        container = parse_container(n)
        if container:
            yield container

def on_root_create() -> None:
    """Callback function for on root node create."""
    # adding favorites to file browser
    workfile_settings = WorkfileSettings()
    workfile_settings.set_favorites()

    # Root created callback is also triggered on scene load because that also
    # creates a new root node. So we check if current file is None to run
    # logic that we only want to apply on new scene creation.
    if current_file() is None:
        on_script_create_settings = (
            workfile_settings.project_settings
            ["nuke"]
            ["workfile_callbacks"]
            ["on_script_create"]
        )

        if on_script_create_settings["set_resolution"]:
            workfile_settings.reset_resolution()

        if on_script_create_settings["set_frame_range"]:
            workfile_settings.reset_frame_range_handles()

        if on_script_create_settings["set_colorspace"]:
            workfile_settings.set_colorspace()

        start_workfile_template_builder()

def on_script_load() -> None:
    """Callback function for on script load."""
    log.info("On script load...")
    # fix ffmpeg settings on script
    if nuke.env["LINUX"]:
        nuke.tcl('load ffmpegReader')
        nuke.tcl('load ffmpegWriter')
    else:
        nuke.tcl('load movReader')
        nuke.tcl('load movWriter')

    # set apply all workfile settings on script load and save
    workfile_settings = WorkfileSettings()
    on_script_load_settings = (
        workfile_settings.project_settings
        ["nuke"]
        ["workfile_callbacks"]
        ["on_script_open"]
    )

    if on_script_load_settings["set_resolution"]:
        workfile_settings.reset_resolution()

    if on_script_load_settings["set_frame_range"]:
        workfile_settings.reset_frame_range_handles()

    if on_script_load_settings["set_colorspace"]:
        workfile_settings.set_colorspace()

    # set checker for last versions on loaded containers
    check_inventory_versions()

def parse_container(node):
    """Returns containerised data of a node

    Reads the imprinted data from `containerise`.

    Arguments:
        node (nuke.Node): Nuke's node object to read imprinted data

    Returns:
        dict: The container schema data for this container node.

    """
    data = read_avalon_data(node)

    # If not all required data return the empty container
    required = ["schema", "id", "name",
                "namespace", "loader", "representation"]
    if not all(key in data for key in required):
        return

    # Store the node's name
    data.update({
        "objectName": node.fullName(),
        "node": node,
    })

    return data

def reload_config():
    """Attempt to reload pipeline at run-time.

    CAUTION: This is primarily for development and debugging purposes.

    """

    for module in (
        "ayon_nuke.api.actions",
        "ayon_nuke.api.menu",
        "ayon_nuke.api.plugin",
        "ayon_nuke.api.lib",
    ):
        log.info("Reloading module: {}...".format(module))

        module = importlib.import_module(module)

        try:
            importlib.reload(module)
        except AttributeError as e:
            from importlib import reload
            log.warning("Cannot reload module: {}".format(e))
            reload(module)

def remove_instance(instance):
    """Remove instance from current workfile metadata.

    For SubsetManager

    Args:
        instance (dict): instance representation from subsetmanager model
    """
    instance_node = instance.transient_data["node"]
    instance_knob = instance_node.knobs()[INSTANCE_DATA_KNOB]
    instance_node.removeKnob(instance_knob)
    nuke.delete(instance_node)

def select_instance(instance):
    """
        Select instance in Node View

        Args:
            instance (dict): instance representation from subsetmanager model
    """
    instance_node = instance.transient_data["node"]
    instance_node["selected"].setValue(True)

def start_workfile_template_builder() -> None:
    """Trigger workfile template builder when a new file is created."""
    from .workfile_template_builder import (
        trigger_on_app_launch,
        trigger_on_new_file,
    )
    if not os.getenv("AYON_NUKE_INITIALIZED"):
        log.info("Triggering workfile template builder on application launch.")
        trigger_on_app_launch()
        # Track this at process level because the host instance may be
        # recreated or reloaded during Nuke's lifetime.
        # The environment marker persists across those reloads and ensures
        # the application-launch trigger runs only once.
        os.environ["AYON_NUKE_INITIALIZED"] = "True"
    else:
        log.info("Triggering workfile template builder on new file.")
        trigger_on_new_file()

def update_container(node, keys=None):
    """Returns node with updateted containder data

    Arguments:
        node (nuke.Node): The node in Nuke to imprint as container,
        keys (dict, optional): data which should be updated

    Returns:
        node (nuke.Node): nuke node with updated container data

    Raises:
        TypeError on given an invalid container node

    """
    keys = keys or dict()

    container = parse_container(node)
    if not container:
        raise TypeError("Not a valid container node.")

    container.update(keys)
    node = set_avalon_knob_data(node, container)

    return node