code_gallery

Web News Blog Press Pricing

Learn

Training

Profiles

Executive

Templates

VFX

Code

Code Gallery Designer

v5.20

Maya

Scene Assembly

Jorge Hernandez Ibañez 08/08/2025

Tags

Scene Assembly

Jorge Herna…

Action v1

08/08/2025

Maya

Pseudo-Code

Parameters

allowed_dccs	List of Dcc from where the assets can be imported (Task dcc)
allowed_maintasks	List of maintasks names from where the assets can be imported.

Talk with TD Support

Import to my Pipeline

Code

import maya.cmds as cmds
import os
import pipelinepro as pipeline
import requests
import sys
import krakenPlugin.maya.assets

# PySide imports with version detection
try:
    import dcc_utils
    pyside_version = dcc_utils.get_pySide_version()
except ImportError:
    pyside_version = 2  # Default fallback

if pyside_version == 2:
    from PySide2 import QtWidgets, QtCore
    from PySide2.QtWidgets import QProgressDialog
    from shiboken2 import wrapInstance
else:
    from PySide6 import QtWidgets, QtCore
    from PySide6.QtWidgets import QProgressDialog
    from shiboken6 import wrapInstance

from maya import OpenMayaUI as omui

assetTypes = {
    "CH": "COLOR_02",
    "PR": "COLOR_06",
    "FX": "COLOR_03",
    "CA": "COLOR_05",
    "SE": "COLOR_04",
}

allowed_dccs = ["Maya", "Zbrush", "Houdini", "Blender"]
allowed_maintasks = ["Model", "Rig"]

def get_maya_main_window():
    """Get Maya's main window for proper dialog parenting."""
    maya_main_window_ptr = omui.MQtUtil.mainWindow()
    if maya_main_window_ptr:
        return wrapInstance(int(maya_main_window_ptr), QtWidgets.QWidget)
    return None

def create_progress_dialog(title, maximum_value, parent=None):
    """Create and configure a progress dialog for the scene assembly process.
    
    Args:
        title (str): The title for the progress dialog
        maximum_value (int): Maximum value for the progress bar
        parent (QWidget): Parent widget (Maya main window)
    
    Returns:
        QProgressDialog: Configured progress dialog
    """
    if parent is None:
        parent = get_maya_main_window()
    
    progress_dialog = QProgressDialog(parent)
    progress_dialog.setWindowTitle(title)
    progress_dialog.setLabelText("Initializing scene assembly...")
    progress_dialog.setMinimum(0)
    progress_dialog.setMaximum(maximum_value)
    progress_dialog.setValue(0)
    progress_dialog.setAutoClose(True)
    progress_dialog.setAutoReset(False)
    progress_dialog.setModal(True)
    
    # Make dialog stay on top and be visible
    progress_dialog.show()
    QtWidgets.QApplication.processEvents()
    
    return progress_dialog

def exec_query(query):
    """Execute a pipeline database query and return the response."""
    if "token:" not in query:
        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 None
    return response

def get_assets_breakdown(filename):
    """Query the pipeline for all assets assigned to the shot based on the filename."""
    query = """
    query {
        file(fileName: "{filename}", token: "{token}") {
            shot {
                assets {
                    name
                    key
                    tecnicalName
                    assetType {
                        name
                        initials
                    }
                    tasks {
                        maintask {
                            name
                            key
                            dcc{
                                name
                            }
                        }
                    }
                }
            }
            asset{
                breakdown{
                    name
                    key
                    tecnicalName
                    assetType {
                        name
                        initials
                    }
                    tasks {
                        maintask {
                            name
                            key
                            dcc{
                                name
                            }
                        }
                    }
                }
            }
            task{
                name
            }
        }
    }
    """
    query = query.replace("{filename}", filename)
    response = exec_query(query)
    if not response:
        return []
    file_data = response.get("file")
    if not file_data:
        return []
    shot = file_data.get("shot", {})
    asset = file_data.get("asset", {})
    if not shot and not asset:
        return []
    if shot:  # Shot breakdown
        assets = shot.get("assets", [])
    else:  # Asset breakdown
        assets = asset.get("breakdown", [])
    return assets if assets else [], "shot" if shot else "asset", file_data.get("task", {}).get("name", "")

def get_asset_files(asset_key, maintask_key):
    """Query the pipeline for published files of a given asset and maintask."""
    query_asset_files = """
        query{
          task(token:"{token}", assetKey: "{asset_key}", maintask:"{maintask_key}"){
            publishFolder{
              files{
                fileName
                fileLocal
                link
                pathToFile
                folder {
                    name
                    type
                    dependence {
                        name
                        type
                    }
                }
                task {
                    name
                    maintask {
                        tecnicalName
                        name
                    }
                }
                asset {
                    tecnicalName
                    assetType {
                        initials
                    }
                }
                shot{
                    name
                    tecnicalName
                }
              }
            }
            folderActive{
              files{
                fileName
                fileLocal
                link
                pathToFile
                folder {
                    name
                    type
                    dependence {
                        name
                        type
                    }
                }
                task {
                    name
                    maintask {
                        tecnicalName
                        name
                    }
                }
                asset {
                    tecnicalName
                    assetType {
                        initials
                    }
                }
                shot{
                    name
                    tecnicalName
                }
              }
            }
          }
        }
    """
    query_asset_files = query_asset_files.replace(
        "{maintask_key}", maintask_key
    ).replace("{asset_key}", asset_key)
    response = exec_query(query_asset_files)
    if (
        response
        and "task" in response
        and response["task"] is not None
    ):
        if "publishFolder" in response["task"]:
            folder_active = response["task"].get("publishFolder", None)
            if folder_active and "files" in folder_active and folder_active["files"]:
                return folder_active["files"]
        if "folderActive" in response["task"]:
            folder_active = response["task"].get("folderActive", None)
            if folder_active and "files" in folder_active:
                return folder_active["files"]
    return []

def load_asset(asset, maintask_key, asset_type, parent=None):
    """Load and reference the asset into the Maya scene."""
    asset_key = asset.get("key")
    files = get_asset_files(asset_key, maintask_key)
    asset_name = asset.get("tecnicalName", "").replace(" ", "_")
    asset_initials = asset_type.get("initials", "")
    
    if not files:
        print(f'Asset {asset_name} has no files')
        return False
    
    for file in files:
        print(f'Processing: {file["fileName"]}')
        if not file.get("fileName") or not (file.get("fileName").endswith(".ma") or file.get("fileName").endswith(".mb")):
            continue
        
        filename = file["fileName"]
        asset_path = file.get("fileLocal", None)

# If fileLocal does not exist -> download
        if not os.path.isfile(asset_path):
            file_link = file["link"]
            download_file(file_link, asset_path)
        
        # Import the asset file using the improved import function
        result = krakenPlugin.maya.assets.import_asset_file(file, asset_key, parent=parent)
        if result:
            print(f"Asset '{asset_name}' imported successfully.")
        else:
            print(f"Failed to import asset '{asset_name}'.")
    return True

def download_file(download_link, file_local_path):
    """Download the Maya file from the cloud and save it to the correct path.
    Args:
        download_link (str): The link to download the file.
        file_local_path (str): The local path to save the file.
    """
    print(f"Downloading file to: {file_local_path}")
    file_local_path = os.path.normpath(file_local_path)
    file_local_dir = os.path.dirname(file_local_path)
    try:
        print("Starting download request ----")
        response = requests.get(download_link)
        response.raise_for_status()
        print("Download request ended ----")
        os.makedirs(file_local_dir, exist_ok=True)
        with open(file_local_path, "wb") as f:
            f.write(response.content)
        print(f"File downloaded successfully in: {file_local_path}")
    except requests.exceptions.RequestException as e:
        print(f"Error on file download: {e}")

def refresh_maya_outliner():
    """Refresh the Maya outliner to show updated scene hierarchy."""
    try:
        # TODO: Implement Maya outliner refresh if needed
        # For now, we'll just print a message as Maya typically updates automatically
        print("Maya scene hierarchy updated.")
    except Exception as e:
        print(f"Error refreshing Maya outliner: {e}")

def get_filename():
    """Return the current Maya file name."""
    try:
        scene_path = cmds.file(query=True, sceneName=True)
        if scene_path:
            return os.path.basename(scene_path)
        else:
            print("No Maya scene file is currently open.")
            return ""
    except Exception as e:
        print(f"Error getting Maya scene filename: {e}")
        return ""

def main():
    """Main function to assemble the scene by importing all shot/asset assets and organizing them by asset type."""
    print('SCENE ASSEMBLY')
    filename = get_filename()
    print(filename)
    assets, entity_type, task_name = get_assets_breakdown(filename)
    if not assets:
        print(f"No assets assigned to this shot/asset - file name: {filename}")
        return
    
    # Display asset names for console logging
    for asset in assets:
        print(asset["name"])
    
    # Create progress dialog
    progress_dialog = None
    try:
        total_assets = len(assets)
        progress_dialog = create_progress_dialog("Scene Assembly", total_assets)
        progress_dialog.setLabelText(f"Preparing to import {total_assets} assets...")
        QtWidgets.QApplication.processEvents()
        
        # Process each asset with progress tracking
        for i, asset in enumerate(assets):
            # Check if user cancelled the operation
            if progress_dialog.wasCanceled():
                print("Scene assembly cancelled by user.")
                break
            
            # Update progress dialog with current asset
            asset_name = asset.get("name", "Unknown Asset")
            progress_dialog.setLabelText(f"Processing asset {i + 1} of {total_assets}: {asset_name}")
            progress_dialog.setValue(i)
            QtWidgets.QApplication.processEvents()
            
            refresh_maya_outliner()
            print("Loading asset: " + asset.get("name") + " - key: " + asset.get("key"))
            asset_tasks = asset.get("tasks", [])
            # Filter tasks to keep only those with specific DCC applications
            if allowed_dccs:
                asset_tasks = [task for task in asset_tasks if task.get("maintask", {}).get("dcc", {}).get("name") in allowed_dccs]
            # Filter tasks to keep only those with specific maintask names
            if allowed_maintasks:
                asset_tasks = [task for task in asset_tasks if task.get("maintask", {}).get("name") in allowed_maintasks]

if not asset_tasks:
                print(f"The asset {asset.get('name')} does not have tasks.")
                continue
            for task in reversed(asset_tasks):
                maintask = task.get("maintask", {})
                dcc_name = maintask.get("dcc", {}).get("name", "Unknown DCC")
                print(f"Processing task: {maintask.get('name')} for DCC: {dcc_name}")
                if entity_type == "shot":
                    success = load_asset(asset, maintask.get("key"), asset.get("assetType", {}))
                else:
                    success = load_asset(asset, maintask.get("key"), asset.get("assetType", {}), parent=task_name)
                if success:
                    break
        
        # Complete the progress
        if not progress_dialog.wasCanceled():
            progress_dialog.setValue(total_assets)
            progress_dialog.setLabelText("Scene assembly completed successfully!")
            QtWidgets.QApplication.processEvents()
            print("Scene assembly completed successfully!")
    
    except Exception as e:
        print(f"Error during scene assembly: {e}")
        if progress_dialog:
            progress_dialog.setLabelText(f"Error occurred: {str(e)}")
            QtWidgets.QApplication.processEvents()
    
    finally:
        # Clean up progress dialog
        if progress_dialog:
            progress_dialog.close()
            QtWidgets.QApplication.processEvents()

main()

Import Action

1. Copy Key

Copy

2. Go to your Action pannel

Go