General: Move formatting and workfile functions

openpype/hosts/standalonepublisher/plugins/publish/collect_texture.py

@@ -3,9 +3,10 @@
 import pyblish.api
 import json
 
-from avalon.api import format_template_with_optional_keys
-
-from openpype.lib import prepare_template_data
+from openpype.lib import (
+ prepare_template_data,
+ StringTemplate,
+)
 
 
 class CollectTextures(pyblish.api.ContextPlugin):
@@ -110,8 +111,9 @@ def process(self, context):
 
  formatting_data.update(explicit_data)
  fill_pairs = prepare_template_data(formatting_data)
- workfile_subset = format_template_with_optional_keys(
- fill_pairs, self.workfile_subset_template)
+ workfile_subset = StringTemplate.format_strict_template(
+ self.workfile_subset_template, fill_pairs
+ )
 
  asset_build = self._get_asset_build(
  repre_file,
@@ -201,8 +203,9 @@ def process(self, context):
  formatting_data.update(explicit_data)
 
  fill_pairs = prepare_template_data(formatting_data)
- subset = format_template_with_optional_keys(
- fill_pairs, self.texture_subset_template)
+ subset = StringTemplate.format_strict_template(
+ self.texture_subset_template, fill_pairs
+ )
 
  asset_build = self._get_asset_build(
  repre_file,

openpype/hosts/tvpaint/plugins/load/load_workfile.py

@@ -1,10 +1,11 @@
-import getpass
 import os
 
 from avalon import api, io
 from openpype.lib import (
+ StringTemplate,
  get_workfile_template_key_from_context,
- get_workdir_data
+ get_workdir_data,
+ get_last_workfile_with_version,
 )
 from openpype.api import Anatomy
 from openpype.hosts.tvpaint.api import lib, pipeline, plugin
@@ -67,9 +68,8 @@ def load(self, context, name, namespace, options):
 
  data = get_workdir_data(project_doc, asset_doc, task_name, host_name)
  data["root"] = anatomy.roots
- data["user"] = getpass.getuser()
 
- template = anatomy.templates[template_key]["file"]
+ file_template = anatomy.templates[template_key]["file"]
 
  # Define saving file extension
  if current_file:
@@ -81,11 +81,12 @@ def load(self, context, name, namespace, options):
 
  data["ext"] = extension
 
- work_root = api.format_template_with_optional_keys(
- data, anatomy.templates[template_key]["folder"]
+ folder_template = anatomy.templates[template_key]["folder"]
+ work_root = StringTemplate.format_strict_template(
+ folder_template, data
  )
- version = api.last_workfile_with_version(
- work_root, template, data, host.file_extensions()
+ version = get_last_workfile_with_version(
+ work_root, file_template, data, host.file_extensions()
  )[1]
 
  if version is None:
@@ -95,8 +96,8 @@ def load(self, context, name, namespace, options):
 
  data["version"] = version
 
- path = os.path.join(
- work_root,
- api.format_template_with_optional_keys(data, template)
+ filename = StringTemplate.format_strict_template(
+ file_template, data
  )
+ path = os.path.join(work_root, filename)
  host.save_file(path)

openpype/lib/__init__.py

@@ -114,6 +114,8 @@
  get_workdir_data,
  get_workdir,
  get_workdir_with_workdir_data,
+ get_last_workfile_with_version,
+ get_last_workfile,
 
  create_workfile_doc,
  save_workfile_data_to_doc,
@@ -263,6 +265,8 @@
  "get_workdir_data",
  "get_workdir",
  "get_workdir_with_workdir_data",
+ "get_last_workfile_with_version",
+ "get_last_workfile",
 
  "create_workfile_doc",
  "save_workfile_data_to_doc",

openpype/lib/applications.py

@@ -28,7 +28,8 @@
 from .avalon_context import (
  get_workdir_data,
  get_workdir_with_workdir_data,
- get_workfile_template_key
+ get_workfile_template_key,
+ get_last_workfile
 )
 
 from .python_module_tools import (
@@ -1609,7 +1610,7 @@ def _prepare_last_workfile(data, workdir):
  "ext": extensions[0]
  })
 
- last_workfile_path = avalon.api.last_workfile(
+ last_workfile_path = get_last_workfile(
  workdir, file_template, workdir_data, extensions, True
  )
 

openpype/lib/avalon_context.py

@@ -16,6 +16,7 @@
 from .anatomy import Anatomy
 from .profiles_filtering import filter_profiles
 from .events import emit_event
+from .path_templates import StringTemplate
 
 # avalon module is not imported at the top
 # - may not be in path at the time of pype.lib initialization
@@ -1735,8 +1736,6 @@ def get_custom_workfile_template_by_context(
  context. (Existence of formatted path is not validated.)
  """
 
- from openpype.lib import filter_profiles
-
  if anatomy is None:
  anatomy = Anatomy(project_doc["name"])
 
@@ -1759,7 +1758,9 @@ def get_custom_workfile_template_by_context(
  # there are some anatomy template strings
  if matching_item:
  template = matching_item["path"][platform.system().lower()]
- return template.format(**anatomy_context_data)
+ return StringTemplate.format_strict_template(
+ template, anatomy_context_data
+ )
 
  return None
 
@@ -1847,3 +1848,124 @@ def get_custom_workfile_template(template_profiles):
  io.Session["AVALON_TASK"],
  io
  )
+
+
+def get_last_workfile_with_version(
+ workdir, file_template, fill_data, extensions
+):
+ """Return last workfile version.
+
+ Args:
+ workdir(str): Path to dir where workfiles are stored.
+ file_template(str): Template of file name.
+ fill_data(dict): Data for filling template.
+ extensions(list, tuple): All allowed file extensions of workfile.
+
+ Returns:
+ tuple: Last workfile<str> with version<int> if there is any otherwise
+ returns (None, None).
+ """
+ if not os.path.exists(workdir):
+ return None, None
+
+ # Fast match on extension
+ filenames = [
+ filename
+ for filename in os.listdir(workdir)
+ if os.path.splitext(filename)[1] in extensions
+ ]
+
+ # Build template without optionals, version to digits only regex
+ # and comment to any definable value.
+ _ext = []
+ for ext in extensions:
+ if not ext.startswith("."):
+ ext = "." + ext
+ # Escape dot for regex
+ ext = "\\" + ext
+ _ext.append(ext)
+ ext_expression = "(?:" + "|".join(_ext) + ")"
+
+ # Replace `.{ext}` with `{ext}` so we are sure there is not dot at the end
+ file_template = re.sub(r"\.?{ext}", ext_expression, file_template)
+ # Replace optional keys with optional content regex
+ file_template = re.sub(r"<.*?>", r".*?", file_template)
+ # Replace `{version}` with group regex
+ file_template = re.sub(r"{version.*?}", r"([0-9]+)", file_template)
+ file_template = re.sub(r"{comment.*?}", r".+?", file_template)
+ file_template = StringTemplate.format_strict_template(
+ file_template, fill_data
+ )
+
+ # Match with ignore case on Windows due to the Windows
+ # OS not being case-sensitive. This avoids later running
+ # into the error that the file did exist if it existed
+ # with a different upper/lower-case.
+ kwargs = {}
+ if platform.system().lower() == "windows":
+ kwargs["flags"] = re.IGNORECASE
+
+ # Get highest version among existing matching files
+ version = None
+ output_filenames = []
+ for filename in sorted(filenames):
+ match = re.match(file_template, filename, **kwargs)
+ if not match:
+ continue
+
+ file_version = int(match.group(1))
+ if version is None or file_version > version:
+ output_filenames[:] = []
+ version = file_version
+
+ if file_version == version:
+ output_filenames.append(filename)
+
+ output_filename = None
+ if output_filenames:
+ if len(output_filenames) == 1:
+ output_filename = output_filenames[0]
+ else:
+ last_time = None
+ for _output_filename in output_filenames:
+ full_path = os.path.join(workdir, _output_filename)
+ mod_time = os.path.getmtime(full_path)
+ if last_time is None or last_time < mod_time:
+ output_filename = _output_filename
+ last_time = mod_time
+
+ return output_filename, version
+
+
+def get_last_workfile(
+ workdir, file_template, fill_data, extensions, full_path=False
+):
+ """Return last workfile filename.
+
+ Returns file with version 1 if there is not workfile yet.
+
+ Args:
+ workdir(str): Path to dir where workfiles are stored.
+ file_template(str): Template of file name.
+ fill_data(dict): Data for filling template.
+ extensions(list, tuple): All allowed file extensions of workfile.
+ full_path(bool): Full path to file is returned if set to True.
+
+ Returns:
+ str: Last or first workfile as filename of full path to filename.
+ """
+ filename, version = get_last_workfile_with_version(
+ workdir, file_template, fill_data, extensions
+ )
+ if filename is None:
+ data = copy.deepcopy(fill_data)
+ data["version"] = 1
+ data.pop("comment", None)
+ if not data.get("ext"):
+ data["ext"] = extensions[0]
+ filename = StringTemplate.format_strict_template(file_template, data)
+
+ if full_path:
+ return os.path.normpath(os.path.join(workdir, filename))
+
+ return filename

openpype/lib/delivery.py

@@ -5,6 +5,11 @@
 import clique
 import collections
 
+from .path_templates import (
+ StringTemplate,
+ TemplateUnsolved,
+)
+
 
 def collect_frames(files):
  """
@@ -52,8 +57,6 @@ def sizeof_fmt(num, suffix='B'):
 
 
 def path_from_representation(representation, anatomy):
- from avalon import pipeline # safer importing
-
  try:
  template = representation["data"]["template"]
 
@@ -63,12 +66,10 @@ def path_from_representation(representation, anatomy):
  try:
  context = representation["context"]
  context["root"] = anatomy.roots
- path = pipeline.format_template_with_optional_keys(
- context, template
- )
- path = os.path.normpath(path.replace("/", "\\"))
+ path = StringTemplate.format_strict_template(template, context)
+ return os.path.normpath(path)
 
- except KeyError:
+ except TemplateUnsolved:
  # Template references unavailable data
  return None
 

openpype/modules/ftrack/event_handlers_user/action_delete_old_versions.py

@@ -5,11 +5,11 @@
 import clique
 from pymongo import UpdateOne
 
-from openpype_modules.ftrack.lib import BaseAction, statics_icon
 from avalon.api import AvalonMongoDB
-from openpype.api import Anatomy
 
-import avalon.pipeline
+from openpype.api import Anatomy
+from openpype.lib import StringTemplate, TemplateUnsolved
+from openpype_modules.ftrack.lib import BaseAction, statics_icon
 
 
 class DeleteOldVersions(BaseAction):
@@ -563,18 +563,16 @@ def path_from_represenation(self, representation, anatomy):
  try:
  context = representation["context"]
  context["root"] = anatomy.roots
- path = avalon.pipeline.format_template_with_optional_keys(
- context, template
- )
+ path = StringTemplate.format_strict_template(template, context)
  if "frame" in context:
  context["frame"] = self.sequence_splitter
  sequence_path = os.path.normpath(
- avalon.pipeline.format_template_with_optional_keys(
+ StringTemplate.format_strict_template(
  context, template
  )
  )
 
- except KeyError:
+ except (KeyError, TemplateUnsolved):
  # Template references unavailable data
  return (None, None)
 

openpype/pipeline/load/utils.py

@@ -525,7 +525,7 @@ def get_representation_path(representation, root=None, dbcon=None):
 
  """
 
- from openpype.lib import StringTemplate
+ from openpype.lib import StringTemplate, TemplateUnsolved
 
  if dbcon is None:
  dbcon = io
@@ -542,13 +542,14 @@ def path_from_represenation():
  try:
  context = representation["context"]
  context["root"] = root
- template_obj = StringTemplate(template)
- path = str(template_obj.format(context))
+ path = StringTemplate.format_strict_template(
+ template, context
+ )
  # Force replacing backslashes with forward slashed if not on
  # windows
  if platform.system().lower() != "windows":
  path = path.replace("\\", "/")
- except KeyError:
+ except (TemplateUnsolved, KeyError):
  # Template references unavailable data
  return None