Blender Kitsu: Refactor Shot Builder #183

Merged
Nick Alberelli merged 55 commits from TinyNick/blender-studio-pipeline:feature/shot-builder-2 into main 2023-12-21 23:58:21 +01:00
3 changed files with 259 additions and 1 deletions
Showing only changes of commit af2ec3c737 - Show all commits

View File

@ -0,0 +1,51 @@
import bpy
from blender_kitsu.shot_builder_2.hooks import hook
from blender_kitsu.types import Shot, Asset
import logging
'''
Arguments to use in hooks
scene: bpy.types.Scene # current scene
shot: Shot class from blender_kitsu.types.py
prod_path: str # path to production root dir (your_project/svn/)
shot_path: str # path to shot file (your_project/svn/pro/shots/{sequence_name}/{shot_name}/{shot_task_name}.blend})
Notes
matching_task_type = ['anim', 'lighting', 'fx', 'comp'] # either use list or just one string
output_col_name = shot.get_output_collection_name(task_type="anim")
'''
logger = logging.getLogger(__name__)
# ---------- Global Hook ----------
@hook()
def set_eevee_render_engine(scene: bpy.types.Scene, **kwargs):
"""
By default, we set EEVEE as the renderer.
"""
scene.render.engine = 'BLENDER_EEVEE'
print("HOOK SET RENDER ENGINE")
# ---------- Overrides for animation files ----------
@hook(match_task_type='anim')
def test_args(
scene: bpy.types.Scene, shot: Shot, prod_path: str, shot_path: str, **kwargs
):
"""
Set output parameters for animation rendering
"""
print(f"Scene = {scene.name}")
print(f"Shot = {shot.name}")
print(f"Prod Path = {prod_path}")
print(f"Shot Path = {shot_path}")

View File

@ -0,0 +1,195 @@
import sys
import pathlib
from typing import *
import typing
import types
from collections.abc import Iterable
import importlib
from .. import prefs
import logging
logger = logging.getLogger(__name__)
class Wildcard:
pass
class DoNotMatch:
pass
MatchCriteriaType = typing.Union[
str, typing.List[str], typing.Type[Wildcard], typing.Type[DoNotMatch]
]
"""
The MatchCriteriaType is a type definition for the parameters of the `hook` decorator.
The matching parameters can use multiple types to detect how the matching criteria
would work.
* `str`: would perform an exact string match.
* `typing.Iterator[str]`: would perform an exact string match with any of the given strings.
* `typing.Type[Wildcard]`: would match any type for this parameter. This would be used so a hook
is called for any value.
* `typing.Type[DoNotMatch]`: would ignore this hook when matching the hook parameter. This is the default
value for the matching criteria and would normally not be set directly in a
production configuration.
"""
MatchingRulesType = typing.Dict[str, MatchCriteriaType]
"""
Hooks are stored as `_shot_builder_rules' attribute on the function.
The MatchingRulesType is the type definition of the `_shot_builder_rules` attribute.
"""
HookFunction = typing.Callable[[typing.Any], None]
def _match_hook_parameter(
hook_criteria: MatchCriteriaType, match_query: typing.Optional[str]
) -> bool:
if hook_criteria == None:
return True
if hook_criteria == DoNotMatch:
return match_query is None
if hook_criteria == Wildcard:
return True
if isinstance(hook_criteria, str):
return match_query == hook_criteria
if isinstance(hook_criteria, list):
return match_query in hook_criteria
logger.error(f"Incorrect matching criteria {hook_criteria}, {match_query}")
return False
class Hooks:
def __init__(self):
self._hooks: typing.List[HookFunction] = []
def matches(
self,
hook: HookFunction,
match_task_type: typing.Optional[str] = None,
match_asset_type: typing.Optional[str] = None,
**kwargs: typing.Optional[str],
) -> bool:
assert not kwargs
rules = typing.cast(MatchingRulesType, getattr(hook, '_shot_builder_rules'))
return all(
(
_match_hook_parameter(rules['match_task_type'], match_task_type),
_match_hook_parameter(rules['match_asset_type'], match_asset_type),
)
)
def filter(self, **kwargs: typing.Optional[str]) -> typing.Iterator[HookFunction]:
for hook in self._hooks:
if self.matches(hook=hook, **kwargs):
yield hook
def execute_hooks(
self, match_task_type: str = None, match_asset_type: str = None, *args, **kwargs
) -> None:
for hook in self._hooks:
if self.matches(
hook, match_task_type=match_task_type, match_asset_type=match_asset_type
):
hook(*args, **kwargs)
def load_hooks(self, context):
root_dir = prefs.project_root_dir_get(context)
shot_builder_config_dir = root_dir.joinpath("pro/assets/scripts/shot-builder")
if not shot_builder_config_dir.exists():
raise Exception("Shot Builder Hooks directory does not exist")
paths = [
shot_builder_config_dir.resolve().__str__() # TODO Make variable path
] # TODO Set path to where hooks are stored
with SystemPathInclude(paths) as _include:
try:
import hooks as production_hooks
importlib.reload(production_hooks)
self.register_hooks(production_hooks)
except ModuleNotFoundError:
raise Exception("Production has no `hooks.py` configuration file")
return False
def register(self, func: HookFunction) -> None:
logger.info(f"registering hook '{func.__name__}'")
self._hooks.append(func)
def register_hooks(self, module: types.ModuleType) -> None:
"""
Register all hooks inside the given module.
"""
for module_item_str in dir(module):
module_item = getattr(module, module_item_str)
if not isinstance(module_item, types.FunctionType):
continue
if module_item.__module__ != module.__name__:
continue
if not hasattr(module_item, "_shot_builder_rules"):
continue
self.register(module_item)
def hook(
match_task_type: MatchCriteriaType = None,
match_asset_type: MatchCriteriaType = None,
) -> typing.Callable[[types.FunctionType], types.FunctionType]:
"""
Decorator to add custom logic when building a shot.
Hooks are used to extend the configuration that would be not part of the core logic of the shot builder tool.
"""
rules = {
'match_task_type': match_task_type,
'match_asset_type': match_asset_type,
}
def wrapper(func: types.FunctionType) -> types.FunctionType:
setattr(func, '_shot_builder_rules', rules)
return func
return wrapper
class SystemPathInclude:
"""
Resource class to temporary include system paths to `sys.paths`.
Usage:
```
paths = [pathlib.Path("/home/guest/my_python_scripts")]
with SystemPathInclude(paths) as t:
import my_module
reload(my_module)
```
It is possible to nest multiple SystemPathIncludes.
"""
def __init__(self, paths_to_add: List[pathlib.Path]):
# TODO: Check if all paths exist and are absolute.
self.__paths = paths_to_add
self.__original_sys_path: List[str] = []
def __enter__(self):
self.__original_sys_path = sys.path
new_sys_path = []
for path_to_add in self.__paths:
# Do not add paths that are already in the sys path.
# Report this to the logger as this might indicate wrong usage.
path_to_add_str = str(path_to_add)
if path_to_add_str in self.__original_sys_path:
logger.warn(f"{path_to_add_str} already added to `sys.path`")
continue
new_sys_path.append(path_to_add_str)
new_sys_path.extend(self.__original_sys_path)
sys.path = new_sys_path
def __exit__(self, exc_type, exc_value, exc_traceback):
sys.path = self.__original_sys_path

View File

@ -17,6 +17,7 @@ from .editorial import editorial_export_get_latest
from .file_save import save_shot_builder_file from .file_save import save_shot_builder_file
from .template import replace_workspace_with_template from .template import replace_workspace_with_template
from .assets import get_shot_assets from .assets import get_shot_assets
from .hooks import Hooks
active_project = None active_project = None
@ -139,6 +140,7 @@ class KITSU_OT_build_new_shot(bpy.types.Operator):
shot = active_project.get_shot(self.shot_id) shot = active_project.get_shot(self.shot_id)
task_type = self._get_task_type_for_shot(context, shot) task_type = self._get_task_type_for_shot(context, shot)
task_short_name = task_type.get_short_name() task_short_name = task_type.get_short_name()
shot_file_path_str = shot.get_shot_filepath(context, task_type)
# Open Template File # Open Template File
replace_workspace_with_template(context, task_short_name) replace_workspace_with_template(context, task_short_name)
@ -151,7 +153,6 @@ class KITSU_OT_build_new_shot(bpy.types.Operator):
# File Path # File Path
# TODO Only run if saving file # TODO Only run if saving file
shot_file_path_str = shot.get_shot_filepath(context, task_type)
# Set Render Settings # Set Render Settings
if task_short_name == 'anim': # TODO get anim from a constant instead if task_short_name == 'anim': # TODO get anim from a constant instead
@ -176,6 +177,17 @@ class KITSU_OT_build_new_shot(bpy.types.Operator):
if bkglobals.LOAD_EDITORIAL_REF.get(task_short_name): if bkglobals.LOAD_EDITORIAL_REF.get(task_short_name):
editorial_export_get_latest(context, shot) editorial_export_get_latest(context, shot)
# Run Hooks
hooks_instance = Hooks()
hooks_instance.load_hooks(context)
hooks_instance.execute_hooks(
match_task_type=task_short_name,
scene=context.scene,
shot=shot,
prod_path=prefs.project_root_dir_get(context),
shot_path=shot_file_path_str,
)
# Save File # Save File
if self.save_file: if self.save_file:
save_shot_builder_file(file_path=shot_file_path_str) save_shot_builder_file(file_path=shot_file_path_str)