General: Move change context functions

openpype/hosts/fusion/scripts/fusion_switch_shot.py

@@ -5,11 +5,12 @@
 
 # Pipeline imports
 import avalon.api
-from avalon import io, pipeline
+from avalon import io
 
 from openpype.lib import version_up
 from openpype.hosts.fusion import api
 from openpype.hosts.fusion.api import lib
+from openpype.lib.avalon_context import get_workdir_from_session
 
 log = logging.getLogger("Update Slap Comp")
 
@@ -46,12 +47,8 @@ def _format_version_folder(folder):
 
 def _get_work_folder(session):
     """Convenience function to get the work folder path of the current asset"""
-
     # Get new filename, create path based on asset and work template
-    template_work = self._project["config"]["template"]["work"]
-    work_path = pipeline._format_work_template(template_work, session)
-
-    return os.path.normpath(work_path)
+    return get_workdir_from_session(session)
 
 
 def _get_fusion_instance():

openpype/hosts/fusion/utility_scripts/switch_ui.py

@@ -5,11 +5,12 @@
 from Qt import QtWidgets, QtCore
 
 import avalon.api
-from avalon import io, pipeline
+from avalon import io
 from avalon.vendor import qtawesome as qta
 
 from openpype import style
 from openpype.hosts.fusion import api
+from openpype.lib.avalon_context import get_workdir_from_session
 
 log = logging.getLogger("Fusion Switch Shot")
 
@@ -158,15 +159,7 @@ def _on_switch(self):
         switch_shot.switch(asset_name=asset, filepath=file_name, new=True)
 
     def _get_context_directory(self):
-
-        project = io.find_one({"type": "project",
-                               "name": avalon.api.Session["AVALON_PROJECT"]},
-                              projection={"config": True})
-
-        template = project["config"]["template"]["work"]
-        dir = pipeline._format_work_template(template, avalon.api.Session)
-
-        return dir
+        return get_workdir_from_session(avalon.api.Session)
 
     def collect_slap_comps(self, directory):
         items = glob.glob("{}/*.comp".format(directory))

openpype/lib/avalon_context.py

@@ -644,6 +644,161 @@ def get_workdir(
     )
 
 
+def template_data_from_session(session):
+    """ Return dictionary with template from session keys.
+
+    Args:
+        session (dict, Optional): The Session to use. If not provided use the
+            currently active global Session.
+    Returns:
+        dict: All available data from session.
+    """
+    from avalon import io
+
+    if session is None:
+        session = avalon.api.Session
+
+    project_name = session["AVALON_PROJECT"]
+    project_doc = io._database[project_name].find_one({"type": "project"})
+    asset_doc = io._database[project_name].find_one({
+        "type": "asset",
+        "name": session["AVALON_ASSET"]
+    })
+    task_name = session["AVALON_TASK"]
+    host_name = session["AVALON_APP"]
+    return get_workdir_data(project_doc, asset_doc, task_name, host_name)
+
+
+def compute_session_changes(
+    session, task=None, asset=None, app=None, template_key=None
+):
+    """Compute the changes for a Session object on asset, task or app switch
+
+    This does *NOT* update the Session object, but returns the changes
+    required for a valid update of the Session.
+
+    Args:
+        session (dict): The initial session to compute changes to.
+            This is required for computing the full Work Directory, as that
+            also depends on the values that haven't changed.
+        task (str, Optional): Name of task to switch to.
+        asset (str or dict, Optional): Name of asset to switch to.
+            You can also directly provide the Asset dictionary as returned
+            from the database to avoid an additional query. (optimization)
+        app (str, Optional): Name of app to switch to.
+
+    Returns:
+        dict: The required changes in the Session dictionary.
+
+    """
+    changes = dict()
+
+    # If no changes, return directly
+    if not any([task, asset, app]):
+        return changes
+
+    # Get asset document and asset
+    asset_document = None
+    asset_tasks = None
+    if isinstance(asset, dict):
+        # Assume asset database document
+        asset_document = asset
+        asset_tasks = asset_document.get("data", {}).get("tasks")
+        asset = asset["name"]
+
+    if not asset_document or not asset_tasks:
+        from avalon import io
+
+        # Assume asset name
+        asset_document = io.find_one(
+            {
+                "name": asset,
+                "type": "asset"
+            },
+            {"data.tasks": True}
+        )
+        assert asset_document, "Asset must exist"
+
+    # Detect any changes compared session
+    mapping = {
+        "AVALON_ASSET": asset,
+        "AVALON_TASK": task,
+        "AVALON_APP": app,
+    }
+    changes = {
+        key: value
+        for key, value in mapping.items()
+        if value and value != session.get(key)
+    }
+    if not changes:
+        return changes
+
+    # Compute work directory (with the temporary changed session so far)
+    _session = session.copy()
+    _session.update(changes)
+
+    changes["AVALON_WORKDIR"] = get_workdir_from_session(_session)
+
+    return changes
+
+
+def get_workdir_from_session(session, template_key=None):
+    project_name = session["AVALON_PROJECT"]
+    host_name = session["AVALON_APP"]
+    anatomy = Anatomy(project_name)
+    template_data = template_data_from_session(session)
+    anatomy_filled = anatomy.format(template_data)
+
+    if not template_key:
+        task_type = template_data["task"]["type"]
+        template_key = get_workfile_template_key(
+            task_type,
+            host_name,
+            project_name=project_name
+        )
+    return anatomy_filled[template_key]["folder"]
+
+
+def update_current_task(task=None, asset=None, app=None, template_key=None):
+    """Update active Session to a new task work area.
+
+    This updates the live Session to a different `asset`, `task` or `app`.
+
+    Args:
+        task (str): The task to set.
+        asset (str): The asset to set.
+        app (str): The app to set.
+
+    Returns:
+        dict: The changed key, values in the current Session.
+
+    """
+    import avalon.api
+    from avalon.pipeline import emit
+
+    changes = compute_session_changes(
+        avalon.api.Session,
+        task=task,
+        asset=asset,
+        app=app,
+        template_key=template_key
+    )
+
+    # Update the Session and environments. Pop from environments all keys with
+    # value set to None.
+    for key, value in changes.items():
+        avalon.api.Session[key] = value
+        if value is None:
+            os.environ.pop(key, None)
+        else:
+            os.environ[key] = value
+
+    # Emit session change
+    emit("taskChanged", changes.copy())
+
+    return changes
+
+
 @with_avalon
 def get_workfile_doc(asset_id, task_name, filename, dbcon=None):
     """Return workfile document for entered context.

openpype/scripts/fusion_switch_shot.py

@@ -4,13 +4,15 @@
 import logging
 
 # Pipeline imports
-from avalon import api, io, pipeline
+from avalon import api, io
 import avalon.fusion
 
 # Config imports
 import openpype.lib as pype
 import openpype.hosts.fusion.lib as fusion_lib
 
+from openpype.lib.avalon_context import get_workdir_from_session
+
 log = logging.getLogger("Update Slap Comp")
 
 self = sys.modules[__name__]
@@ -48,10 +50,7 @@ def _get_work_folder(session):
     """Convenience function to get the work folder path of the current asset"""
 
     # Get new filename, create path based on asset and work template
-    template_work = self._project["config"]["template"]["work"]
-    work_path = pipeline._format_work_template(template_work, session)
-
-    return os.path.normpath(work_path)
+    return get_workdir_from_session(session)
 
 
 def _get_fusion_instance():

openpype/tools/assetcreator/app.py

@@ -4,9 +4,11 @@
 
 import ftrack_api
 from Qt import QtWidgets, QtCore
+from openpype import style
 from openpype.api import get_current_project_settings
+from openpype.lib.avalon_context import update_current_task
 from openpype.tools.utils.lib import qt_app_context
-from avalon import io, api, style, schema
+from avalon import io, api, schema
 from . import widget, model
 
 module = sys.modules[__name__]
@@ -463,12 +465,12 @@ def create_asset(self):
                 return
             task_name = task_model.itemData(index)[0]
             try:
-                api.update_current_task(task=task_name, asset=asset_name)
+                update_current_task(task=task_name, asset=asset_name)
                 self.open_app()
 
             finally:
                 if origin_task is not None and origin_asset is not None:
-                    api.update_current_task(
+                    update_current_task(
                         task=origin_task, asset=origin_asset
                     )
 

openpype/tools/workfiles/app.py

@@ -29,6 +29,10 @@
     create_workdir_extra_folders,
     get_system_general_anatomy_data
 )
+from openpype.lib.avalon_context import (
+    update_current_task,
+    compute_session_changes
+)
 from .model import FilesModel
 from .view import FilesView
 
@@ -667,7 +671,7 @@ def _get_session(self):
             session["AVALON_APP"],
             project_name=session["AVALON_PROJECT"]
         )
-        changes = pipeline.compute_session_changes(
+        changes = compute_session_changes(
             session,
             asset=self._get_asset_doc(),
             task=self._task_name,
@@ -681,7 +685,7 @@ def _enter_session(self):
         """Enter the asset and task session currently selected"""
 
         session = api.Session.copy()
-        changes = pipeline.compute_session_changes(
+        changes = compute_session_changes(
             session,
             asset=self._get_asset_doc(),
             task=self._task_name,
@@ -692,7 +696,7 @@ def _enter_session(self):
             # to avoid any unwanted Task Changed callbacks to be triggered.
             return
 
-        api.update_current_task(
+        update_current_task(
             asset=self._get_asset_doc(),
             task=self._task_name,
             template_key=self.template_key