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

@@ -32,8 +32,12 @@
     download_revision,
     open_draft,
 )
-from .reference import copy_reference_link, reference_revision
-from .step_validation import find_unpublished_upstream_step
+from .reference import (
+    copy_reference_link,
+    reference_published_workfile,
+    reference_revision,
+)
+from .step_validation import find_unpublished_upstream_step, find_upstream_workfile
 
 
 class FlowAMActions:
@@ -176,7 +180,11 @@ def _on_build_scene_dialog_accepted(
             am_project_id=flow_am_id,
             create_mode=dialog.build,
             source_path=template_path,
-            prep_scene_callback=functools.partial(self._prep_scene, sg_publish_data),
+        )
+        # Bound after construction so the callback can reference create_inputs
+        # (which needs the fully built object to resolve the upstream scene).
+        create_inputs.prep_scene_callback = functools.partial(
+            self._prepare_build_scene, create_inputs, sg_publish_data
         )
 
         if not self._confirm_upstream_step_published(create_inputs):
@@ -262,6 +270,77 @@ def _prep_scene(self, sg_publish_data: dict) -> None:
         # TDs can override this method to add custom scene prep logic
         pass
 
+    def _prepare_build_scene(
+        self, create_inputs: CreateInputs, sg_publish_data: dict
+    ) -> None:
+        """
+        Prep callback run while a new scene is being built.
+
+        This fires after the host has created/loaded the scene and before it is
+        saved into the draft, which is exactly when the upstream reference must
+        exist so it gets baked into the built scene. Standard referencing runs
+        first, then the TD-overridable prep hook.
+
+        :param create_inputs: Inputs describing the scene being built.
+        :param sg_publish_data: FPTR data dictionary for the task being built from.
+        """
+        self._reference_upstream_step(create_inputs)
+        self._prep_scene(sg_publish_data)
+
+    def _reference_upstream_step(self, create_inputs: CreateInputs) -> None:
+        """
+        Reference the previous pipeline step's published Maya scene into the
+        scene being built.
+
+        This is the referencing counterpart of
+        `_confirm_upstream_step_published`. It is a no-op unless the current host
+        is Maya (only the Maya scene publish is referenced) and the upstream step
+        actually has a published workfile. When nothing is published - for
+        instance when the artist chose to build an empty scene from that 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.
+
+        :param create_inputs: Inputs describing the scene being built.
+        """
+        host = getattr(sgtk.platform.current_engine(), "flow_host", None)
+        workfile_type = getattr(host, "WORKFILE_TYPE", "")
+        if workfile_type != MAYA_WORKFILE_TYPE:
+            return
+
+        workfile = find_upstream_workfile(
+            am_project_id=create_inputs.am_project_id,
+            sg_entity_type=create_inputs.sg_entity_type,
+            sg_entity_name=create_inputs.sg_entity_name,
+            sg_pipeline_step=create_inputs.sg_pipeline_step,
+            workfile_type=workfile_type,
+            step_dependencies=self._app.get_setting("pipeline_step_dependencies", {}),
+        )
+        if workfile is None:
+            return
+
+        try:
+            file_path = reference_published_workfile(workfile.revision_id)
+        except exceptions.FlowError as exc:
+            message = (
+                f"Could not reference the previous step's published scene for "
+                f'"{create_inputs.sg_entity_name}". The new scene was built '
+                f"without it. ({exc})"
+            )
+            self._app.log_error(message)
+            QtGui.QMessageBox.warning(
+                self._get_dialog_parent(),
+                "Reference failed",
+                message,
+            )
+            return
+
+        self._app.log_info(
+            f"Referenced the previous step's published scene into the new "
+            f'"{create_inputs.sg_pipeline_step}" scene: {file_path}'
+        )
+
     def _discard_draft(self, sg_publish_data: dict) -> None:
         """
         Discard the local draft for the given PublishedFile.

python/tk_multi_loader/flowam/reference.py

@@ -89,6 +89,70 @@ def reference_revision(revision_id: str) -> str:
     return depdata.file_path
 
 
+def reference_published_workfile(revision_id: str) -> str:
+    """Reference a published workfile's source into the current scene.
+
+    Intended for the "Build New Scene" flow, where a downstream step's fresh
+    scene should open with the upstream step's published output already
+    referenced in. It intentionally mirrors :func:`reference_revision` but omits
+    its ``flow_draft_id`` guard: at build time the scene is brand new and has no
+    asset/draft context yet, which is exactly the state that guard rejects.
+
+    Args:
+        revision_id: The id of the asset revision to be referenced.
+                     This can also be a version id.
+
+    Returns:
+        File path of referenced file.
+
+    Raises:
+        CreateReferenceError
+    """
+    engine = sgtk.platform.current_engine()
+
+    if not hasattr(engine.flow_host, "create_reference"):
+        msg = "Referencing is not supported in current execution."
+        raise CreateReferenceError(input_id=revision_id, details=msg)
+
+    try:
+        if objects.FlowVersion.is_version_id(revision_id):
+            input_type = "version"
+            revision = objects.FlowVersion(revision_id).revision
+        else:
+            input_type = "revision"
+            revision = objects.FlowRevision.get_revision(revision_id)
+    except exceptions.FlowError as exc:
+        msg = f"Could not retrieve {input_type} object."
+        raise CreateReferenceError(input_id=revision_id, details=msg) from exc
+
+    # Fetch source component of revision
+    revision.fetch(component_purpose=globals.SOURCE_PURPOSE, fetch_dependencies=True)
+
+    # Get path to source path of revision in local storage
+    file_path = revision.get_storage_component_path(
+        component_purpose=globals.SOURCE_PURPOSE
+    )
+    if file_path is None:
+        msg = "Revision does not have a source component to be referenced."
+        raise CreateReferenceError(input_id=revision_id, details=msg)
+    file_seq_comp = revision.find_component(
+        type_id=schema.get_schema_id(globals.FILE_SEQ_TYPE)
+    )
+    if not file_seq_comp and not os.path.exists(file_path):
+        msg = f"Source file does not exist in storage: {file_path}. "
+        msg += "Fetching the revision was not successful!"
+        raise CreateReferenceError(input_id=revision_id, details=msg)
+    elif file_seq_comp:
+        # Return a file path with embedded frame padding
+        file_path = utils.cleanpath(
+            revision.get_storage_dir(), file_seq_comp.properties["fileFormat"]
+        )
+
+    # Create reference
+    depdata = engine.flow_host.create_reference(file_path, namespace=revision.name)
+    return depdata.file_path
+
+
 def copy_reference_link(revision_id: str) -> str:
     """Copy the reference link (file path) to the source component
     the of given revision to application clipboard.

python/tk_multi_loader/flowam/step_validation.py

@@ -124,21 +124,13 @@ def has_published_workfile(
         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))
+        workfile = _find_workfile_asset(
+            am_project_id=am_project_id,
+            root_folder_name=root_folder_name,
+            sg_entity_name=sg_entity_name,
+            pipeline_step=pipeline_step,
+            workfile_type_id=workfile_type_id,
+        )
     except exceptions.FlowError as exc:
         logger.warning(
             f'Could not verify whether pipeline step "{pipeline_step}" has a '
@@ -147,6 +139,99 @@ def has_published_workfile(
         )
         return True
 
+    return workfile is not None
+
+
+def find_upstream_workfile(
+    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[objects.FlowAsset]:
+    """Return the upstream step's published workfile asset for the entity.
+
+    This is the referencing counterpart of
+    :func:`find_unpublished_upstream_step`. Where that helper answers "should we
+    warn?", this one answers "what should we reference?". Every unresolved case -
+    no configured upstream, an unsupported entity type, an unresolved workfile
+    schema id, nothing published, or a Flow AM query error - yields ``None`` so
+    the caller simply skips referencing rather than surfacing an error while
+    building a scene.
+
+    :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 reference.
+    :param step_dependencies: Mapping of step name to upstream step name.
+    :returns: The upstream workfile ``FlowAsset``, or ``None``.
+    """
+    upstream_step = get_upstream_step(sg_pipeline_step, step_dependencies)
+    if not upstream_step:
+        return None
+
+    root_folder_name = _get_root_folder_name(sg_entity_type)
+    if not root_folder_name:
+        return None
+
+    workfile_type_id = schema.get_schema_id(workfile_type)
+    if not workfile_type_id:
+        logger.warning(
+            f'Could not resolve the schema id for workfile type "{workfile_type}". '
+            f'Skipping referencing of pipeline step "{upstream_step}".'
+        )
+        return None
+
+    try:
+        return _find_workfile_asset(
+            am_project_id=am_project_id,
+            root_folder_name=root_folder_name,
+            sg_entity_name=sg_entity_name,
+            pipeline_step=upstream_step,
+            workfile_type_id=workfile_type_id,
+        )
+    except exceptions.FlowError as exc:
+        logger.warning(
+            f"Could not resolve a published workfile for pipeline step "
+            f'"{upstream_step}" of "{sg_entity_name}". Skipping referencing. ({exc})'
+        )
+        return None
+
+
+def _find_workfile_asset(
+    am_project_id: str,
+    root_folder_name: str,
+    sg_entity_name: str,
+    pipeline_step: str,
+    workfile_type_id: str,
+) -> Optional[objects.FlowAsset]:
+    """Return the workfile asset published under *pipeline_step* for the entity.
+
+    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 root_folder_name: Name of the project's top-level folder.
+    :param sg_entity_name: FPTR entity name of the asset.
+    :param pipeline_step: Step to look under.
+    :param workfile_type_id: Resolved schema id of the workfile type.
+    :returns: The workfile ``FlowAsset``, or ``None`` when the hierarchy is
+        incomplete or the step has no published workfile.
+    :raises exceptions.FlowError: If a Flow AM query fails.
+    """
+    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)
+    return workfiles[0] if workfiles else None
+
 
 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*.