feat(celaction): prelaunch hool with Anatomy last version workfile

pype/api.py

@@ -25,6 +25,7 @@ from .lib import (
     get_hierarchy,
     get_subsets,
     get_version_from_path,
+    get_last_version_from_path,
     modified_environ,
     add_tool_to_environment
 )
@@ -56,6 +57,7 @@ __all__ = [
     "get_asset",
     "get_subsets",
     "get_version_from_path",
+    "get_last_version_from_path",
     "modified_environ",
     "add_tool_to_environment",
 

pype/celaction/cli.py

@@ -99,19 +99,6 @@ def main():
     # Registers pype's Global pyblish plugins
     pype.install()
 
-    host_import_str = f"pype.{publish_host}"
-
-    try:
-        host_module = importlib.import_module(host_import_str)
-    except ModuleNotFoundError:
-        log.error((
-            f"Host \"{publish_host}\" can't be imported."
-            f" Import string \"{host_import_str}\" failed."
-        ))
-        return False
-
-    # avalon.api.install(host_module)
-
     for path in PUBLISH_PATHS:
         path = os.path.normpath(path)
 

pype/hooks/celaction/prelaunch.py

@@ -2,7 +2,10 @@ import logging
 import os
 import winreg
 from pype.lib import PypeHook
-from pypeapp import Logger
+from pype.api import get_last_version_from_path
+from pypeapp import Anatomy, Logger
+
+from avalon import io, api, lib
 
 log = logging.getLogger(__name__)
 
@@ -14,6 +17,7 @@ class CelactionPrelaunchHook(PypeHook):
     path to the project by environment variable to Unreal launcher
     shell script.
     """
+    workfile_ext = "scn"
 
     def __init__(self, logger=None):
         if not logger:
@@ -25,21 +29,33 @@ class CelactionPrelaunchHook(PypeHook):
 
     def execute(self, *args, env: dict = None) -> bool:
         if not env:
-            env = os.environ
-        project = env["AVALON_PROJECT"]
-        asset = env["AVALON_ASSET"]
-        task = env["AVALON_TASK"]
-        app = "celaction_publish"
-        workdir = env["AVALON_WORKDIR"]
-        project_name = f"{asset}_{task}"
-        version = "v001"
+            self.env = os.environ
+        else:
+            self.env = env
 
-        self.log.info(f"{self.signature}")
+        self._S = api.Session
+        project = self._S["AVALON_PROJECT"] = self.env["AVALON_PROJECT"]
+        asset = self._S["AVALON_ASSET"] = self.env["AVALON_ASSET"]
+        task = self._S["AVALON_TASK"] = self.env["AVALON_TASK"]
+        workdir = self._S["AVALON_WORKDIR"] = self.env["AVALON_WORKDIR"]
+
+        anatomy_filled = self.get_anatomy_filled()
+
+        app = "celaction_publish"
+        workfile = anatomy_filled["work"]["file"]
+        version = anatomy_filled["version"]
 
         os.makedirs(workdir, exist_ok=True)
         self.log.info(f"Work dir is: `{workdir}`")
 
-        project_file = os.path.join(workdir, f"{project_name}_{version}.scn")
+        # get last version if any
+        workfile_last = get_last_version_from_path(
+            workdir, workfile.split(version))
+
+        if workfile_last:
+            workfile = workfile_last
+
+        project_file = os.path.join(workdir, workfile)
         env["PYPE_CELACTION_PROJECT_FILE"] = project_file
 
         self.log.info(f"Workfile is: `{project_file}`")
@@ -73,7 +89,7 @@ class CelactionPrelaunchHook(PypeHook):
             "--resolutionWidth *X*",
             "--resolutionHeight *Y*",
             "--programDir \"'*PROGPATH*'\""
-            ]
+        ]
         winreg.SetValueEx(hKey, "SubmitParametersTitle", 0, winreg.REG_SZ,
                           " ".join(parameters))
 
@@ -105,3 +121,64 @@ class CelactionPrelaunchHook(PypeHook):
         winreg.SetValueEx(hKey, "Valid", 0, winreg.REG_DWORD, 1)
 
         return True
+
+    def get_anatomy_filled(self):
+        root_path = api.registered_root()
+        project_name = self._S["AVALON_PROJECT"]
+        asset_name = self._S["AVALON_ASSET"]
+
+        io.install()
+        project_entity = io.find_one({
+            "type": "project",
+            "name": project_name
+        })
+        assert project_entity, (
+            "Project '{0}' was not found."
+        ).format(project_name)
+        log.debug("Collected Project \"{}\"".format(project_entity))
+
+        asset_entity = io.find_one({
+            "type": "asset",
+            "name": asset_name,
+            "parent": project_entity["_id"]
+        })
+        assert asset_entity, (
+            "No asset found by the name '{0}' in project '{1}'"
+        ).format(asset_name, project_name)
+
+        project_name = project_entity["name"]
+
+        log.info(
+            "Anatomy object collected for project \"{}\".".format(project_name)
+        )
+
+        hierarchy_items = asset_entity["data"]["parents"]
+        hierarchy = ""
+        if hierarchy_items:
+            hierarchy = os.path.join(*hierarchy_items)
+
+        template_data = {
+            "root": root_path,
+            "project": {
+                "name": project_name,
+                "code": project_entity["data"].get("code")
+            },
+            "asset": asset_entity["name"],
+            "hierarchy": hierarchy.replace("\\", "/"),
+            "task": self._S["AVALON_TASK"],
+            "ext": self.workfile_ext,
+            "version": 1,
+            "username": os.getenv("PYPE_USERNAME", "").strip()
+        }
+
+        avalon_app_name = os.environ.get("AVALON_APP_NAME")
+        if avalon_app_name:
+            application_def = lib.get_application(avalon_app_name)
+            app_dir = application_def.get("application_dir")
+            if app_dir:
+                template_data["app"] = app_dir
+
+        anatomy = Anatomy(project_name)
+        anatomy_filled = anatomy.format_all(template_data).get_solved()
+
+        return anatomy_filled

pype/lib.py

@@ -469,6 +469,40 @@ def get_version_from_path(file):
         )
 
 
+def get_last_version_from_path(path_dir, filter=None):
+    """
+    Finds last version of given directory content
+
+    Args:
+        path_dir (string): directory path
+        filter (list): list of strings used as file name filter
+
+    Returns:
+        string: file name with last version
+
+    """
+    assert os.path.isdir(path_dir), "`path_dir` argument needs to be directory"
+
+    filtred_files = list()
+
+    # form regex for filtering
+    patern = r".*"
+
+    if filter:
+        patern = patern.join(filter)
+
+    for f in os.listdir(path_dir):
+        if not re.findall(patern, f):
+            continue
+        filtred_files.append(f)
+
+    if filtred_files:
+        sorted(filtred_files)
+        return filtred_files[-1]
+    else:
+        return None
+
+
 def get_avalon_database():
     if io._database is None:
         set_io_database()