diff --git a/info.yml b/info.yml index 6636a69a..e13f7d82 100644 --- a/info.yml +++ b/info.yml @@ -131,6 +131,24 @@ configuration: default_value: {} allows_empty: true + pipeline_step_dependencies: + type: dict + description: "Maps a pipeline step to the upstream pipeline step whose workfile must + be published before a new scene is built for it. Keys and values must + match the Flow Production Tracking Step name exactly as it appears in + the task's Pipeline Step, for example {Rig: Model, Texture: Model} + (the names are project-specific - use whatever your site calls the + modelling, surfacing and rigging steps). Used by the Flow Asset + Management 'Build New Scene' action to warn the artist when the + upstream step has nothing published to reference yet. Steps absent + from this mapping have no upstream requirement and are never validated. + Only Maya scenes are checked, so this setting has no effect in other + DCCs." + default_value: + Rig: Model + Texture: Model + allows_empty: true + entities: default_value: - caption: Project diff --git a/python/tk_multi_loader/constants.py b/python/tk_multi_loader/constants.py index ca94e8ee..22337463 100644 --- a/python/tk_multi_loader/constants.py +++ b/python/tk_multi_loader/constants.py @@ -103,3 +103,8 @@ # FlowAM versions starts with 0 and FlowPT versions starts with 1. # This is used to identify a FlowAM draft version in the UI. DRAFT_VERSION_IDENTIFIER = -1 + +# Schema type of the Maya workfile, mirroring FlowHost.WORKFILE_TYPE in tk-maya, +# which this app cannot import. Only the Maya scene publish is validated when +# building a new scene, so hosts building other workfile types skip that check. +MAYA_WORKFILE_TYPE = "type.workfile.maya" diff --git a/python/tk_multi_loader/flowam/flowam_actions.py b/python/tk_multi_loader/flowam/flowam_actions.py index 51a0a4ab..c5e082f2 100644 --- a/python/tk_multi_loader/flowam/flowam_actions.py +++ b/python/tk_multi_loader/flowam/flowam_actions.py @@ -19,7 +19,7 @@ from ..build_asset_dialog import BuildAssetDialog from ..build_template_dialog import BuildTemplateDialog -from ..constants import DRAFT_VERSION_IDENTIFIER +from ..constants import DRAFT_VERSION_IDENTIFIER, MAYA_WORKFILE_TYPE from .create import ( CreateInputs, CreateTemplateInputs, @@ -33,6 +33,7 @@ open_draft, ) from .reference import copy_reference_link, reference_revision +from .step_validation import find_unpublished_upstream_step class FlowAMActions: @@ -178,6 +179,13 @@ def _on_build_scene_dialog_accepted( prep_scene_callback=functools.partial(self._prep_scene, sg_publish_data), ) + if not self._confirm_upstream_step_published(create_inputs): + self._app.log_info( + "Build new scene cancelled: the previous pipeline step has not " + "been published." + ) + return + try: asset = create_dcc_workfile(create_inputs) self._app.log_debug(f"Created a DCC workfile asset: {asset}") @@ -190,6 +198,57 @@ def _on_build_scene_dialog_accepted( str(exc), ) + def _confirm_upstream_step_published(self, create_inputs: CreateInputs) -> bool: + """ + Warn the artist when the previous pipeline step has nothing published yet. + + The upstream publish is what gets referenced into the new scene, so + building without it leaves an empty scene. The artist is given the choice + to continue regardless. + + Only the Maya scene publish is in scope. Nuke and Houdini reach this + method through their own build actions, but a downstream step references + the Maya scene rather than the current host's workfile type, so those + hosts are left alone. + + :param create_inputs: Inputs describing the scene about to be built. + :returns: True when the build should go ahead. + """ + host = getattr(sgtk.platform.current_engine(), "flow_host", None) + workfile_type = getattr(host, "WORKFILE_TYPE", "") + if workfile_type != MAYA_WORKFILE_TYPE: + return True + + upstream_step = find_unpublished_upstream_step( + am_project_id=create_inputs.am_project_id, + sg_entity_name=create_inputs.sg_entity_name, + sg_entity_type=create_inputs.sg_entity_type, + sg_pipeline_step=create_inputs.sg_pipeline_step, + step_dependencies=self._app.get_setting("pipeline_step_dependencies", {}), + workfile_type=workfile_type, + ) + if not upstream_step: + return True + + message = ( + f'The "{upstream_step}" step has no published scene for ' + f'"{create_inputs.sg_entity_name}" yet, so there is nothing to ' + f'reference into a new "{create_inputs.sg_pipeline_step}" scene.' + "\n\nBuild an empty scene anyway?" + ) + self._app.log_warning(message) + + response = QtGui.QMessageBox.warning( + self._get_dialog_parent(), + "Previous step not published", + message, + buttons=QtGui.QMessageBox.StandardButtons( + QtGui.QMessageBox.StandardButton.Yes + | QtGui.QMessageBox.StandardButton.Cancel + ), + ) + return response == QtGui.QMessageBox.StandardButton.Yes + def _prep_scene(self, sg_publish_data: dict) -> None: """ Let clients run set-up scripts when building a new scene/asset. diff --git a/python/tk_multi_loader/flowam/step_validation.py b/python/tk_multi_loader/flowam/step_validation.py new file mode 100644 index 00000000..2b032a4d --- /dev/null +++ b/python/tk_multi_loader/flowam/step_validation.py @@ -0,0 +1,167 @@ +# Copyright (c) 2026 Shotgun Software Inc. +# +# CONFIDENTIAL AND PROPRIETARY +# +# This work is provided "AS IS" and subject to the Shotgun Pipeline Toolkit +# Source Code License included in this distribution package. See LICENSE. +# By accessing, using, copying or modifying this work you indicate your +# agreement to the Shotgun Pipeline Toolkit Source Code License. All rights +# not expressly granted therein are reserved by Shotgun Software Inc. + +"""Pipeline step validation helpers for Flow Asset Management. + +Building a new scene for a downstream department only makes sense once the +upstream department has published its workfile, because that publish is what +gets referenced into the new scene. +""" + +from __future__ import annotations # needed for Houdini support + +from typing import Dict, Optional + +import sgtk +from sgtk.flowam.create import ASSET_FOLDER, ASSET_TYPE, SHOT_TYPE +from tank_vendor.flow_integration_sdk import exceptions, objects, schema + +logger = sgtk.platform.get_logger(__name__) + + +def get_upstream_step( + pipeline_step: str, step_dependencies: Dict[str, str] +) -> Optional[str]: + """Return the pipeline step that must be published before *pipeline_step*. + + :param pipeline_step: Name of the step a new scene is being built for. + :param step_dependencies: Mapping of step name to upstream step name, as + provided by the ``pipeline_step_dependencies`` app setting. + :returns: Upstream step name, or ``None`` when the step has no configured + upstream requirement. + """ + if not pipeline_step or not step_dependencies: + return None + + return step_dependencies.get(pipeline_step) or None + + +def find_unpublished_upstream_step( + am_project_id: str, + sg_entity_type: str, + sg_entity_name: str, + sg_pipeline_step: str, + workfile_type: str, + step_dependencies: Dict[str, str], +) -> Optional[str]: + """Return the upstream pipeline step that still needs to be published. + + :param am_project_id: Id of the Flow AM project holding the asset. + :param sg_entity_type: FPTR entity type of the asset, e.g. ``"Asset"``. + :param sg_entity_name: FPTR entity name of the asset. + :param sg_pipeline_step: Step the new scene is being built for. + :param workfile_type: Schema type name of the workfile to look for, e.g. + ``"type.workfile.maya"`` from ``FlowHost.WORKFILE_TYPE``. + :param step_dependencies: Mapping of step name to upstream step name. + :returns: Name of the upstream step when it is configured but has no + published workfile, otherwise ``None``. + """ + upstream_step = get_upstream_step(sg_pipeline_step, step_dependencies) + if not upstream_step: + return None + + if has_published_workfile( + am_project_id=am_project_id, + pipeline_step=upstream_step, + sg_entity_name=sg_entity_name, + sg_entity_type=sg_entity_type, + workfile_type=workfile_type, + ): + return None + + return upstream_step + + +def has_published_workfile( + am_project_id: str, + pipeline_step: str, + sg_entity_name: str, + sg_entity_type: str, + workfile_type: str, +) -> bool: + """Return ``True`` when *pipeline_step* has a published workfile for the asset. + + A workfile asset only exists in Flow AM once it has been published: + ``sandbox.create_asset_in_sandbox()`` writes a local draft and defers the + Flow AM asset creation to publish time. Finding a workfile-typed child is + therefore enough to prove the step was published, whereas the hierarchy + enclosing it may well exist for a step nobody has published yet. + + When the answer cannot be determined this returns ``True``, so a transient + Flow AM error never blocks a build behind a misleading "not published" + message. + + :param am_project_id: Id of the Flow AM project holding the asset. + :param pipeline_step: Step to look for a published workfile under. + :param sg_entity_name: FPTR entity name of the asset. + :param sg_entity_type: FPTR entity type of the asset, e.g. ``"Asset"``. + :param workfile_type: Schema type name of the workfile to look for. + :returns: ``True`` when a published workfile exists or cannot be ruled out. + """ + root_folder_name = _get_root_folder_name(sg_entity_type) + if not root_folder_name: + logger.warning( + f'Cannot locate Flow AM assets for entity type "{sg_entity_type}". ' + f'Skipping the publish check for pipeline step "{pipeline_step}".' + ) + return True + + workfile_type_id = schema.get_schema_id(workfile_type) + if not workfile_type_id: + # An unresolved type id would disable the type filter in find_children() + # and match every child, so skip the check rather than trust it. + logger.warning( + f'Could not resolve the schema id for workfile type "{workfile_type}". ' + f'Skipping the publish check for pipeline step "{pipeline_step}".' + ) + return True + + try: + node = objects.FlowProject(am_project_id) + # Walk down to the "root asset" grouping the workfiles of this step: + # Assets///. See get_or_create_workfile_parent() + # in tk-core's tank/flowam/create.py for the hierarchy this mirrors. + for name in ( + root_folder_name, + sg_entity_name, + pipeline_step, + sg_entity_name, + ): + node = node.find_child(name) + if node is None: + return False + + return bool(node.find_children(type_id=workfile_type_id)) + except exceptions.FlowError as exc: + logger.warning( + f'Could not verify whether pipeline step "{pipeline_step}" has a ' + f'published workfile for "{sg_entity_name}". Allowing the build to ' + f"proceed. ({exc})" + ) + return True + + +def _get_root_folder_name(sg_entity_type: str) -> Optional[str]: + """Return the name of the top-level folder holding assets of *sg_entity_type*. + + Mirrors ``get_or_create_root_folder()`` in tk-core's ``tank/flowam/create.py``, + where the two folder names are asymmetric: assets live under ``ASSET_FOLDER`` + ("Assets") while shots live under a folder named after ``SHOT_TYPE`` ("Shot"). + + :param sg_entity_type: FPTR entity type of the asset. + :returns: Folder name, or ``None`` when the entity type has no such folder. + """ + if sg_entity_type == ASSET_TYPE: + return ASSET_FOLDER + + if sg_entity_type == SHOT_TYPE: + return SHOT_TYPE + + return None diff --git a/tests/test_step_validation.py b/tests/test_step_validation.py new file mode 100644 index 00000000..ad154b24 --- /dev/null +++ b/tests/test_step_validation.py @@ -0,0 +1,203 @@ +# Copyright (c) 2026 Shotgun Software Inc. +# +# CONFIDENTIAL AND PROPRIETARY +# +# This work is provided "AS IS" and subject to the Shotgun Pipeline Toolkit +# Source Code License included in this distribution package. See LICENSE. +# By accessing, using, copying or modifying this work you indicate your +# agreement to the Shotgun Pipeline Toolkit Source Code License. All rights +# not expressly granted therein are reserved by Shotgun Software Inc. + +"""Unit tests for the Flow AM pipeline step validation helpers.""" + +import importlib.util +import pathlib +import types + +import pytest + +MODULE_PATH = ( + pathlib.Path(__file__).resolve().parent.parent + / "python" + / "tk_multi_loader" + / "flowam" + / "step_validation.py" +) + +try: + # step_validation only depends on sgtk and the Flow Integration SDK, so load + # it straight from disk rather than through the flowam package, whose other + # modules pull in Qt and a live engine. + import sgtk # noqa: F401 + from tank_vendor.flow_integration_sdk import exceptions # noqa: F401 + + _spec = importlib.util.spec_from_file_location("step_validation", MODULE_PATH) + step_validation = importlib.util.module_from_spec(_spec) + _spec.loader.exec_module(step_validation) +except ImportError: + # Flow AM features need a tk-core that ships the Flow Integration SDK. + pytestmark = pytest.mark.skip() + + +ENTITY_NAME = "Hero" +MAYA_TYPE = "type.workfile.maya" +MAYA_TYPE_ID = "schema-id-maya-workfile" + + +class StubWorkfile: + """Stand-in for a workfile asset carrying a single schema type.""" + + def __init__(self, name, type_id): + self.name = name + self.type_id = type_id + + +class StubNode: + """Minimal stand-in for a ``FlowProject`` or ``FlowAsset``.""" + + def __init__(self, name, children=(), workfiles=()): + self.name = name + self._children = {child.name: child for child in children} + self._workfiles = list(workfiles) + + def find_child(self, name, force_query=False): + return self._children.get(name) + + def find_children(self, name="", type_id="", force_query=False): + return [w for w in self._workfiles if w.type_id == type_id] + + +def build_project(steps, root_folder="Assets", entity_name=ENTITY_NAME): + """Build a stub ``///`` hierarchy. + + :param steps: Mapping of pipeline step name to whether that step has a + published workfile for the entity. + :param root_folder: Name of the project's top-level folder. + :param entity_name: Name of the entity the steps belong to. + :returns: The stub project node. + """ + step_nodes = [] + for step_name, is_published in steps.items(): + workfiles = ( + [StubWorkfile(f"{entity_name} - MAYA", MAYA_TYPE_ID)] + if is_published + else [] + ) + asset_root = StubNode(entity_name, workfiles=workfiles) + step_nodes.append(StubNode(step_name, children=[asset_root])) + + container = StubNode(entity_name, children=step_nodes) + return StubNode("project", children=[StubNode(root_folder, children=[container])]) + + +@pytest.fixture +def flow_am(monkeypatch): + """Return a callable pointing ``step_validation`` at a stub project.""" + + def _install(project, resolve_type_id=True): + monkeypatch.setattr( + step_validation, + "objects", + types.SimpleNamespace(FlowProject=lambda _project_id: project), + ) + monkeypatch.setattr( + step_validation, + "schema", + types.SimpleNamespace( + get_schema_id=lambda name: ( + MAYA_TYPE_ID if resolve_type_id and name == MAYA_TYPE else None + ) + ), + ) + + return _install + + +def find_unpublished(step="Surfacing", entity_type="Asset", dependencies=None): + """Call the module under test with the common set of arguments.""" + return step_validation.find_unpublished_upstream_step( + am_project_id="am-project-1", + sg_entity_type=entity_type, + sg_entity_name=ENTITY_NAME, + sg_pipeline_step=step, + workfile_type=MAYA_TYPE, + step_dependencies=( + {"Surfacing": "Model"} if dependencies is None else dependencies + ), + ) + + +@pytest.mark.parametrize( + "pipeline_step,step_dependencies,expected", + [ + ("Surfacing", {"Surfacing": "Model"}, "Model"), + ("Rigging", {"Rigging": "Model", "Surfacing": "Model"}, "Model"), + ("Model", {"Surfacing": "Model"}, None), + ("", {"Surfacing": "Model"}, None), + ("Surfacing", {}, None), + ("Surfacing", {"Surfacing": ""}, None), + ], +) +def test_get_upstream_step(pipeline_step, step_dependencies, expected): + """Only steps mapped to a non-empty upstream step resolve to one.""" + assert ( + step_validation.get_upstream_step(pipeline_step, step_dependencies) == expected + ) + + +def test_upstream_published_allows_build(flow_am): + """No warning when the upstream step has a published workfile.""" + flow_am(build_project({"Model": True})) + assert find_unpublished() is None + + +def test_upstream_present_but_unpublished_warns(flow_am): + """The step folder existing is not proof of a publish.""" + flow_am(build_project({"Model": False})) + assert find_unpublished() == "Model" + + +def test_upstream_step_missing_warns(flow_am): + """A step nobody has touched yet has nothing published.""" + flow_am(build_project({"Layout": True})) + assert find_unpublished() == "Model" + + +def test_step_without_configured_upstream_is_not_checked(flow_am): + """Steps absent from the mapping are never validated.""" + flow_am(build_project({"Model": False})) + assert find_unpublished(step="Model") is None + + +def test_unsupported_entity_type_is_skipped(flow_am): + """An entity type with no Flow AM folder cannot be checked, so it passes.""" + flow_am(build_project({"Model": False})) + assert find_unpublished(entity_type="CustomEntity01") is None + + +def test_shot_entity_uses_shot_folder(flow_am): + """Shots live under "Shot" rather than "Assets".""" + flow_am(build_project({"Model": True}, root_folder="Shot")) + assert find_unpublished(entity_type="Shot") is None + + flow_am(build_project({"Model": True}, root_folder="Assets")) + assert find_unpublished(entity_type="Shot") == "Model" + + +def test_unresolved_workfile_type_is_skipped(flow_am): + """An unresolved schema id would match every child, so the check is skipped.""" + flow_am(build_project({"Model": False}), resolve_type_id=False) + assert find_unpublished() is None + + +def test_flow_am_error_allows_build(monkeypatch, flow_am): + """A Flow AM outage must not block a build behind a misleading message.""" + + def raise_error(_project_id): + raise step_validation.exceptions.FlowError("simulated Flow AM outage") + + flow_am(build_project({"Model": False})) + monkeypatch.setattr( + step_validation, "objects", types.SimpleNamespace(FlowProject=raise_error) + ) + assert find_unpublished() is None