Asset Pipeline: Store Asset Catalog in String #230

Merged
Nick Alberelli merged 6 commits from TinyNick/blender-studio-pipeline:fix/asset-catalog into main 2024-02-15 22:54:08 +01:00
6 changed files with 158 additions and 41 deletions

View File

@ -1,15 +1,23 @@
import os
from pathlib import Path
import bpy
from typing import List
from .config import verify_task_layer_json_data
asset_file_cache = None
cat_data_cache = None
asset_cat_dict = {}
# TODO add refresh operator
def find_asset_cat_file(directory: str) -> str:
"""Find Asset Catalog file in directory or parent directories, recursively
Args:
directory (str): Directory to search for Asset Catalog file
def find_asset_cat_file(directory):
Returns:
str: Path to Asset Catalog file or None if not found
"""
global asset_file_cache
if asset_file_cache is not None:
return asset_file_cache
@ -24,16 +32,36 @@ def find_asset_cat_file(directory):
return find_asset_cat_file(parent_dir)
def get_asset_cat_enum_items(reload: bool = False):
def get_asset_catalog_items(reload: bool = False) -> List[str]:
"""Generate List of Asset Catalog Items, and Dictionary of
Asset Catalog UUIDs with their matching names. When this function
is called the list and dict of asset catalog items will be updated.
The List is used in the UI to populate the Asset Catalog List, and the
Dictionary is used to look up the asset catalog UUID based on the name.
Args:
reload (bool, optional): Forces reload of list/dict if True. Defaults to False.
Returns:
List[str]: Returns list of strings representing Asset Catalog Names
"""
global cat_data_cache
if cat_data_cache is not None and not reload:
return cat_data_cache
items = []
items.append(('NONE', 'None', ''))
global asset_cat_dict
asset_cat_list = []
# Return Empty List if File doesn't exist
asset_cat_file = find_asset_cat_file(Path(bpy.data.filepath).parent.__str__())
if asset_cat_file is None:
return items
return asset_cat_list
# Return Cached List if exists and reload is False
if cat_data_cache is not None and not reload:
return cat_data_cache
asset_cat_dict.clear() # Reset dict so it is in sync with name list
# Loop over items in file to find asset catalog
with (Path(asset_cat_file)).open() as file:
for line in file.readlines():
if line.startswith(("#", "VERSION", "\n")):
@ -41,6 +69,38 @@ def get_asset_cat_enum_items(reload: bool = False):
# Each line contains : 'uuid:catalog_tree:catalog_name' + eol ('\n')
name = line.split(':', 1)[1].split(":")[-1].strip("\n")
uuid = line.split(':', 1)[0]
asset_cat_dict[uuid] = name # Populate dict of uuid:name
asset_cat_list.append(name) # Populate list of asset catalogue names
items.append((uuid, name, ''))
return items
cat_data_cache = asset_cat_list # Update Cache List
return asset_cat_list
def get_asset_id(name: str) -> str:
"""Get Asset Catalog UUID based on Asset Catalog Name
Args:
name (str): Asset Catalog Name
Returns:
str: Asset Catalog UUID or None if not found
"""
global asset_cat_dict
for key, value in asset_cat_dict.items():
if value == name:
return key
def get_asset_name(id: str) -> str:
"""Get Asset Catalog UUID based on Asset Catalog Name
Args:
name (str): Asset Catalog Name
Returns:
str: Asset Catalog UUID or None if not found
"""
global asset_cat_dict
for key, value in asset_cat_dict.items():
if key == id:
return value

View File

@ -3,38 +3,53 @@ from pathlib import Path
import json
from . import constants
# TODO could refactor this into a class, but only one instance of that class will be needed
TASK_LAYER_TYPES = {}
TRANSFER_DATA_DEFAULTS = {}
ATTRIBUTE_DEFAULTS = {}
ASSET_CATALOG_ID = ""
def get_json_file():
def get_task_layer_json_filepath() -> Path:
directory = Path(bpy.data.filepath).parent
json_file_path = directory.joinpath(constants.TASK_LAYER_CONFIG_NAME)
if json_file_path.exists():
return json_file_path
return
return json_file_path
def get_task_layer_dict(file_path_str="") -> dict:
if file_path_str == "":
json_file_path = get_task_layer_json_filepath()
else:
json_file_path = Path(file_path_str)
if not json_file_path.exists():
return
return json.load(open(json_file_path))
def get_task_layer_presets_path():
return Path(__file__).parent.joinpath(constants.TASK_LAYER_CONFIG_DIR_NAME)
def verify_json_data(json_file_path=""):
def verify_task_layer_json_data(json_file_path=""):
global TASK_LAYER_TYPES
global TRANSFER_DATA_DEFAULTS
global ATTRIBUTE_DEFAULTS
directory = Path(bpy.data.filepath).parent
if json_file_path == "":
json_file_path = directory.joinpath(constants.TASK_LAYER_CONFIG_NAME)
if not json_file_path.exists():
global ASSET_CATALOG_ID
json_content = get_task_layer_dict(json_file_path)
if not json_content:
return
json_file = open(json_file_path)
json_content = json.load(json_file)
try:
TASK_LAYER_TYPES = json_content["TASK_LAYER_TYPES"]
TRANSFER_DATA_DEFAULTS = json_content["TRANSFER_DATA_DEFAULTS"]
ATTRIBUTE_DEFAULTS = json_content["ATTRIBUTE_DEFAULTS"]
# Asset Catalog is an optional value in task_layers.json and doesn't exist by default
if "ASSET_CATALOG_ID" in json_content:
ASSET_CATALOG_ID = json_content["ASSET_CATALOG_ID"]
return True
except KeyError:
return
@ -47,3 +62,10 @@ def write_json_file(asset_path: Path, source_file_path: Path):
json_dump = json.dumps(json_content, indent=4)
with open(json_file_path, "w") as config_output:
config_output.write(json_dump)
def update_task_layer_json_data(task_layer_dict: dict):
filepath = get_task_layer_json_filepath()
with filepath.open("w") as json_file:
json.dump(task_layer_dict, json_file, indent=4)
verify_task_layer_json_data()

View File

@ -25,7 +25,8 @@ from .sync import (
sync_execute_push,
)
from .asset_catalog import get_asset_cat_enum_items
from .asset_catalog import get_asset_catalog_items, get_asset_id
from .config import verify_task_layer_json_data
class ASSETPIPE_OT_create_new_asset(bpy.types.Operator):
@ -65,7 +66,7 @@ class ASSETPIPE_OT_create_new_asset(bpy.types.Operator):
# Dynamically Create Task Layer Bools
self._asset_pipe = context.scene.asset_pipeline
config.verify_json_data(Path(self._asset_pipe.task_layer_config_type))
config.verify_task_layer_json_data(self._asset_pipe.task_layer_config_type)
all_task_layers = self._asset_pipe.all_task_layers
all_task_layers.clear()
@ -434,7 +435,7 @@ class ASSETPIPE_OT_publish_new_version(bpy.types.Operator):
f"Only '{constants.REVIEW_PUBLISH_KEY}' Publish is supported when a version is staged",
)
return {'CANCELLED'}
catalog_id = context.scene.asset_pipeline.asset_catalog_id
catalog_id = get_asset_id(context.scene.asset_pipeline.asset_catalog_name)
create_next_published_file(
current_file=Path(bpy.data.filepath),
publish_type=self.publish_types,
@ -476,7 +477,7 @@ class ASSETPIPE_OT_publish_staged_as_active(bpy.types.Operator):
staged_file = find_latest_publish(current_file, publish_type=constants.STAGED_PUBLISH_KEY)
# Delete Staged File
staged_file.unlink()
catalog_id = context.scene.asset_pipeline.asset_catalog_id
catalog_id = get_asset_id(context.scene.asset_pipeline.asset_catalog_name)
create_next_published_file(current_file=current_file, catalog_id=catalog_id)
return {'FINISHED'}
@ -960,7 +961,8 @@ class ASSETPIPE_OT_refresh_asset_cat(bpy.types.Operator):
bl_description = """Refresh Asset Catalogs"""
def execute(self, context: bpy.types.Context):
get_asset_cat_enum_items()
get_asset_catalog_items(reload=True)
verify_task_layer_json_data()
self.report({'INFO'}, "Asset Catalogs Refreshed!")
return {'FINISHED'}

View File

@ -2,21 +2,30 @@ import bpy
import os
from typing import List
from . import constants
from .config import get_task_layer_presets_path
from . import config
from pathlib import Path
from .prefs import get_addon_prefs
from .asset_catalog import get_asset_cat_enum_items
from .asset_catalog import get_asset_catalog_items, get_asset_name, get_asset_id
import json
""" NOTE Items in these properties groups should be generated by a function that finds the
avaliable task layers from the task_layer.json file that needs to be created.
"""
def get_safely_string_prop(self, name: str) -> str:
"""Return Value of String Property, and return "" if value isn't set"""
try:
return self[name]
except KeyError:
return ""
def get_task_layer_presets(self, context):
prefs = get_addon_prefs()
user_tls = Path(prefs.custom_task_layers_dir)
presets_dir = get_task_layer_presets_path()
presets_dir = config.get_task_layer_presets_path()
items = []
for file in presets_dir.glob('*.json'):
@ -126,7 +135,7 @@ class AssetPipeline(bpy.types.PropertyGroup):
task_layer_config_type: bpy.props.EnumProperty(
name="Task Layer Preset",
items=get_task_layer_presets,
)
) # type: ignore
temp_file: bpy.props.StringProperty(name="Pre-Sync Backup")
source_file: bpy.props.StringProperty(name="File that started Sync")
@ -157,15 +166,31 @@ class AssetPipeline(bpy.types.PropertyGroup):
attribute_ui_bool: bpy.props.BoolProperty(name="Show/Hide Attributes", default=False)
file_parent_ui_bool: bpy.props.BoolProperty(name="Show/Hide Parent", default=False)
def get_asset_catalogs(self, context):
return get_asset_cat_enum_items()
def set_asset_catalog_name(self, input):
task_layer_dict = config.get_task_layer_dict()
task_layer_dict["ASSET_CATALOG_ID"] = get_asset_id(input)
config.update_task_layer_json_data(task_layer_dict)
self['asset_catalog_name'] = input
asset_catalog_id: bpy.props.EnumProperty(
def get_asset_catalog_name(self):
if config.ASSET_CATALOG_ID != "":
asset_name = get_asset_name(config.ASSET_CATALOG_ID)
if asset_name is None:
return ""
return asset_name
return get_safely_string_prop(self, 'asset_catalog_name')
def get_asset_catalogs_search(self, context, edit_text: str):
return get_asset_catalog_items()
asset_catalog_name: bpy.props.StringProperty(
name="Catalog",
items=get_asset_catalogs,
get=get_asset_catalog_name,
set=set_asset_catalog_name,
search=get_asset_catalogs_search,
search_options={'SORT'},
description="Select Asset Library Catalog for the current Asset, this value will be updated each time you Push to an 'Active' Publish",
)
) # type: ignore
@bpy.app.handlers.persistent
def set_asset_collection_name_post_file_load(_):
@ -179,6 +204,12 @@ def set_asset_collection_name_post_file_load(_):
del scene.asset_pipeline['asset_collection']
@bpy.app.handlers.persistent
def refresh_asset_catalog(_):
get_asset_catalog_items()
config.verify_task_layer_json_data()
classes = (
AssetTransferData,
AssetTransferDataTemp,
@ -197,6 +228,7 @@ def register():
name="Surrender Ownership", default=False
)
bpy.app.handlers.load_post.append(set_asset_collection_name_post_file_load)
bpy.app.handlers.load_post.append(refresh_asset_catalog)
def unregister():
@ -206,3 +238,4 @@ def unregister():
del bpy.types.Scene.asset_pipeline
del bpy.types.ID.asset_id_owner
bpy.app.handlers.load_post.remove(set_asset_collection_name_post_file_load)
bpy.app.handlers.load_post.remove(refresh_asset_catalog)

View File

@ -16,7 +16,7 @@ from .merge.shared_ids import get_shared_id_icon
from . import constants, config
from .hooks import Hooks
from .merge.task_layer import draw_task_layer_selection
from .asset_catalog import get_asset_id
def sync_poll(cls, context):
if any([img.is_dirty for img in bpy.data.images]):
@ -155,7 +155,7 @@ def sync_execute_push(self, context):
hooks_instance = Hooks()
hooks_instance.load_hooks(context)
temp_file_path = create_temp_file_backup(self, context)
_catalog_id = context.scene.asset_pipeline.asset_catalog_id
_catalog_id = get_asset_id(context.scene.asset_pipeline.asset_catalog_name)
file_path = self._sync_target.__str__()
bpy.ops.wm.open_mainfile(filepath=file_path)

View File

@ -3,7 +3,7 @@ import bpy
from pathlib import Path
from .merge.transfer_data.transfer_ui import draw_transfer_data
from .merge.task_layer import draw_task_layer_selection
from .config import verify_json_data
from .config import verify_task_layer_json_data
from .prefs import get_addon_prefs
from . import constants
from .merge.publish import is_staged_publish
@ -45,7 +45,7 @@ class ASSETPIPE_PT_sync(bpy.types.Panel):
return
# TODO Move this call out of the UI because we keep re-loading this file every draw
if not verify_json_data():
if not verify_task_layer_json_data():
layout.label(text="Task Layer Config is invalid", icon="ERROR")
return
@ -94,7 +94,7 @@ class ASSETPIPE_PT_sync_tools(bpy.types.Panel):
def draw(self, context: bpy.types.Context) -> None:
layout = self.layout
cat_row = layout.row(align=True)
cat_row.prop(context.scene.asset_pipeline, 'asset_catalog_id')
cat_row.prop(context.scene.asset_pipeline, 'asset_catalog_name')
cat_row.operator("assetpipe.refresh_asset_cat", icon='FILE_REFRESH', text="")
layout.operator("assetpipe.batch_ownership_change")
layout.operator("assetpipe.revert_file", icon="FILE_TICK")