SG-44786 Reference previous pipeline step publish when building a new scene (Rig, Texture)

hooks/tk-maya_actions.py

@@ -230,7 +230,19 @@ def generate_actions(self, sg_publish_data, actions, ui_area):
                         "name": "build_new_scene",
                         "params": None,
                         "caption": "Build New Scene",
-                        "description": "This will create a new scene in the current project.",
+                        "description": (
+                            "<nobr>Create a new Maya scene for this task in the "
+                            "current Flow AM project.</nobr><br><br>"
+                            "When this task's pipeline step depends on an upstream "
+                            "step - configured through the "
+                            "<b>pipeline_step_dependencies</b> setting (for "
+                            "example Rig and Texture depend on Model) - the "
+                            "upstream step's published Maya scene is automatically "
+                            "referenced into the new scene.<br><br>"
+                            "If that upstream step has not been published yet, you "
+                            "are warned and can still choose to build an empty "
+                            "scene."
+                        ),
                     }
                 )
 

python/tk_multi_loader/dialog.py

@@ -2245,6 +2245,12 @@ def on_action_click(act):
                 )
 
             action = QtGui.QAction(entity_action["caption"], view)
+            description = entity_action.get("description")
+            if description:
+                # QMenu does not show action tooltips on its own, so drive it
+                # from the hovered signal like the built-in actions above.
+                action.setToolTip(description)
+                action.hovered.connect(partial(action_hovered, action))
             action.triggered.connect(partial(on_action_click, act=entity_action))
             view.addAction(action)
             self._dynamic_widgets.append(action)

python/tk_multi_loader/flowam/flowam_actions.py

@@ -33,7 +33,11 @@
     open_draft,
 )
 from .reference import copy_reference_link, reference_revision
-from .step_validation import find_unpublished_upstream_step
+from .step_validation import (
+    find_unpublished_upstream_step,
+    find_workfile_asset,
+    get_upstream_step,
+)
 
 
 class FlowAMActions:
@@ -253,16 +257,82 @@ def _confirm_upstream_step_published(self, create_inputs: CreateInputs) -> bool:
 
     def _prep_scene(self, sg_publish_data: dict) -> None:
         """
-        Let clients run set-up scripts when building a new scene/asset.
+        Default scene-prep opinion for a freshly built scene: reference the
+        previous pipeline step's published Maya scene into it.
 
-        :param sg_publish_data: Shotgun data dictionary with all the standard publish fields.
+        Runs from ``create_dcc_workfile()``'s ``prep_scene_callback`` - after the
+        host has created/loaded the scene and before it is saved into the draft -
+        so the reference is baked into the built scene. It is a no-op unless the
+        host is Maya (only the Maya scene publish is referenced) and the upstream
+        step has a published workfile. When nothing is published - for instance
+        when the artist chose to build an empty scene from the publish warning -
+        there is simply nothing to reference.
+
+        A referencing failure is surfaced as a warning but never aborts the
+        build: the artist still gets their new scene, just without the reference.
+
+        TDs can override this method to change or replace this default behavior.
+
+        :param sg_publish_data: FPTR Task data the new scene is built from.
         """
+        host = getattr(sgtk.platform.current_engine(), "flow_host", None)
+        workfile_type = getattr(host, "WORKFILE_TYPE", "")
+        if workfile_type != MAYA_WORKFILE_TYPE:
+            return
+
+        entity = sg_publish_data.get("entity") or {}
+        task = (
+            self._app.shotgun.find_one(
+                "Task",
+                filters=[["id", "is", sg_publish_data["id"]]],
+                fields=["step"],
+            )
+            or {}
+        )
+        pipeline_step = (task.get("step") or {}).get("name", "")
+        upstream_step = get_upstream_step(
+            pipeline_step,
+            self._app.get_setting("pipeline_step_dependencies", {}),
+        )
+        if not upstream_step:
+            return
+
+        try:
+            workfile = find_workfile_asset(
+                am_project_id=self._get_flowam_id(),
+                sg_entity_type=entity.get("type", ""),
+                sg_entity_name=entity.get("name", ""),
+                pipeline_step=upstream_step,
+                workfile_type=workfile_type,
+            )
+        except exceptions.FlowError as exc:
+            self._app.log_warning(
+                f"Could not resolve a published workfile for pipeline step "
+                f'"{upstream_step}". Building without a reference. ({exc})'
+            )
+            return
+        if workfile is None:
+            return
+
+        try:
+            file_path = reference_revision(
+                workfile.revision_id, require_asset_context=False
+            )
+        except exceptions.FlowError as exc:
+            message = (
+                f"Could not reference the previous step's published scene for "
+                f'"{entity.get("name", "")}". The new scene was built without '
+                f"it. ({exc})"
+            )
+            self._app.log_error(message)
+            if host:
+                host.dialog("Reference failed", message, buttons=["OK"])
+            return
 
         self._app.log_info(
-            f"prep_scene() called with sg_publish_data: {sg_publish_data}"
+            f"Referenced the previous step's published scene into the new "
+            f'"{pipeline_step}" scene: {file_path}'
         )
-        # TDs can override this method to add custom scene prep logic
-        pass
 
     def _discard_draft(self, sg_publish_data: dict) -> None:
         """

python/tk_multi_loader/flowam/reference.py

@@ -26,12 +26,17 @@ def __init__(self, *args, input_id: str, **kwargs):
         super().__init__(f"Could not create reference to {input_id}.", *args, **kwargs)
 
 
-def reference_revision(revision_id: str) -> str:
+def reference_revision(revision_id: str, require_asset_context: bool = True) -> str:
     """Reference the source component of the given revision into the current scene.
 
     Args:
         revision_id: The id of the asset revision to be referenced.
                      This can also be a version id.
+        require_asset_context: When True (the default, used by the interactive
+                     reference action) referencing is only allowed from an open
+                     asset scene. Building a new scene sets this to False: the
+                     scene is freshly created and has no draft context yet, which
+                     is precisely the state the asset-context check rejects.
 
     Returns:
         File path of referenced file.
@@ -46,7 +51,7 @@ def reference_revision(revision_id: str) -> str:
         raise CreateReferenceError(input_id=revision_id, details=msg)
 
     # We will disallow referencing into a non-asset scene
-    if engine.context.flow_draft_id is None:
+    if require_asset_context and engine.context.flow_draft_id is None:
         msg = "Please open an asset from the loader before doing a reference operation."
         raise CreateReferenceError(input_id=revision_id, details=msg)
 

python/tk_multi_loader/flowam/step_validation.py

@@ -67,85 +67,85 @@ def find_unpublished_upstream_step(
     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,
-    ):
+    try:
+        workfile = find_workfile_asset(
+            am_project_id=am_project_id,
+            sg_entity_type=sg_entity_type,
+            sg_entity_name=sg_entity_name,
+            pipeline_step=upstream_step,
+            workfile_type=workfile_type,
+        )
+    except exceptions.FlowError as exc:
+        # Cannot determine the publish state (unknown entity type, unresolved
+        # workfile type, or a Flow AM query error). Never block a build behind a
+        # misleading "not published" message.
+        logger.warning(
+            f'Could not verify whether pipeline step "{upstream_step}" has a '
+            f'published workfile for "{sg_entity_name}". Allowing the build to '
+            f"proceed. ({exc})"
+        )
         return None
 
-    return upstream_step
+    return upstream_step if workfile is None else None
 
 
-def has_published_workfile(
+def find_workfile_asset(
     am_project_id: str,
-    pipeline_step: str,
-    sg_entity_name: str,
     sg_entity_type: str,
+    sg_entity_name: str,
+    pipeline_step: str,
     workfile_type: str,
-) -> bool:
-    """Return ``True`` when *pipeline_step* has a published workfile for the asset.
+) -> Optional[objects.FlowAsset]:
+    """Return the workfile asset published under *pipeline_step* for the entity.
 
     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.
+    Walks down to the "root asset" that groups the workfiles of a step -
+    ``<root_folder>/<entity>/<step>/<entity>`` - and returns its first
+    workfile-typed child. See ``get_or_create_workfile_parent()`` in tk-core's
+    ``tank/flowam/create.py`` for the hierarchy this mirrors.
 
     :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.
+    :param sg_entity_name: FPTR entity name of the asset.
+    :param pipeline_step: Step to look for a published workfile under.
+    :param workfile_type: Schema type name of the workfile, e.g.
+        ``"type.workfile.maya"`` from ``FlowHost.WORKFILE_TYPE``.
+    :returns: The workfile ``FlowAsset``, or ``None`` when the step has no
+        published workfile.
+    :raises exceptions.FlowError: When the lookup cannot be performed - an
+        unknown entity type, an unresolved workfile schema id, or a failed Flow
+        AM query - so callers can tell "not published" apart from "not checked".
     """
     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}".'
+        raise exceptions.FlowError(
+            f'No Flow AM root folder for entity type "{sg_entity_type}".'
         )
-        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}".'
+        # and match every child, so refuse to run the query rather than trust it.
+        raise exceptions.FlowError(
+            f'Could not resolve the schema id for workfile type "{workfile_type}".'
         )
-        return True
 
-    try:
-        node = objects.FlowProject(am_project_id)
-        # Walk down to the "root asset" grouping the workfiles of this step:
-        # Assets/<entity>/<step>/<entity>. 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
+    node = objects.FlowProject(am_project_id)
+    for name in (root_folder_name, sg_entity_name, pipeline_step, sg_entity_name):
+        node = node.find_child(name)
+        if node is None:
+            return None
+
+    workfiles = node.find_children(type_id=workfile_type_id)
+    # NOTE: an asset root may hold several workfiles of the same type; picking
+    # the first is a deliberate simplification until we define what "multiple
+    # Maya workfiles under one step" should mean (see PR #163 discussion).
+    return workfiles[0] if workfiles else None
 
 
 def _get_root_folder_name(sg_entity_type: str) -> Optional[str]:

tests/test_step_validation.py

@@ -47,9 +47,10 @@
 class StubWorkfile:
     """Stand-in for a workfile asset carrying a single schema type."""
 
-    def __init__(self, name, type_id):
+    def __init__(self, name, type_id, revision_id="rev-1"):
         self.name = name
         self.type_id = type_id
+        self.revision_id = revision_id
 
 
 class StubNode:
@@ -79,7 +80,13 @@ def build_project(steps, root_folder="Assets", entity_name=ENTITY_NAME):
     step_nodes = []
     for step_name, is_published in steps.items():
         workfiles = (
-            [StubWorkfile(f"{entity_name} - MAYA", MAYA_TYPE_ID)]
+            [
+                StubWorkfile(
+                    f"{entity_name} - MAYA",
+                    MAYA_TYPE_ID,
+                    revision_id=f"rev-{step_name}",
+                )
+            ]
             if is_published
             else []
         )
@@ -201,3 +208,56 @@ def raise_error(_project_id):
         step_validation, "objects", types.SimpleNamespace(FlowProject=raise_error)
     )
     assert find_unpublished() is None
+
+
+def find_workfile(step="Model", entity_type="Asset"):
+    """Call the single workfile resolver with the common set of arguments."""
+    return step_validation.find_workfile_asset(
+        am_project_id="am-project-1",
+        sg_entity_type=entity_type,
+        sg_entity_name=ENTITY_NAME,
+        pipeline_step=step,
+        workfile_type=MAYA_TYPE,
+    )
+
+
+def test_find_workfile_asset_returns_published_asset(flow_am):
+    """The step's published workfile asset is returned for referencing."""
+    flow_am(build_project({"Model": True}))
+    workfile = find_workfile()
+    assert workfile is not None
+    assert workfile.revision_id == "rev-Model"
+
+
+def test_find_workfile_asset_none_when_unpublished(flow_am):
+    """The step folder existing is not proof of a publish."""
+    flow_am(build_project({"Model": False}))
+    assert find_workfile() is None
+
+
+def test_find_workfile_asset_raises_for_unsupported_entity(flow_am):
+    """An entity type with no Flow AM folder cannot be checked."""
+    flow_am(build_project({"Model": True}))
+    with pytest.raises(step_validation.exceptions.FlowError):
+        find_workfile(entity_type="CustomEntity01")
+
+
+def test_find_workfile_asset_raises_when_type_unresolved(flow_am):
+    """An unresolved schema id would match every child, so refuse the query."""
+    flow_am(build_project({"Model": True}), resolve_type_id=False)
+    with pytest.raises(step_validation.exceptions.FlowError):
+        find_workfile()
+
+
+def test_find_workfile_asset_propagates_flow_am_error(monkeypatch, flow_am):
+    """A Flow AM query error surfaces so callers can tell it from 'not found'."""
+
+    def raise_error(_project_id):
+        raise step_validation.exceptions.FlowError("simulated Flow AM outage")
+
+    flow_am(build_project({"Model": True}))
+    monkeypatch.setattr(
+        step_validation, "objects", types.SimpleNamespace(FlowProject=raise_error)
+    )
+    with pytest.raises(step_validation.exceptions.FlowError):
+        find_workfile()