diff --git a/client/ayon_core/pipeline/actions/__init__.py b/client/ayon_core/pipeline/actions/__init__.py index bda9b50ede..188414bdbe 100644 --- a/client/ayon_core/pipeline/actions/__init__.py +++ b/client/ayon_core/pipeline/actions/__init__.py @@ -1,3 +1,13 @@ +from .loader import ( + LoaderActionForm, + LoaderActionResult, + LoaderActionItem, + LoaderActionPlugin, + LoaderActionSelection, + LoaderActionsContext, + SelectionEntitiesCache, +) + from .launcher import ( LauncherAction, LauncherActionSelection, @@ -18,6 +28,14 @@ from .inventory import ( __all__= ( + "LoaderActionForm", + "LoaderActionResult", + "LoaderActionItem", + "LoaderActionPlugin", + "LoaderActionSelection", + "LoaderActionsContext", + "SelectionEntitiesCache", + "LauncherAction", "LauncherActionSelection", "discover_launcher_actions", diff --git a/client/ayon_core/pipeline/actions/loader.py b/client/ayon_core/pipeline/actions/loader.py new file mode 100644 index 0000000000..33f48b195c --- /dev/null +++ b/client/ayon_core/pipeline/actions/loader.py @@ -0,0 +1,546 @@ +from __future__ import annotations + +import os +import collections +import copy +from abc import ABC, abstractmethod +from typing import Optional, Any, Callable +from dataclasses import dataclass + +import ayon_api + +from ayon_core import AYON_CORE_ROOT +from ayon_core.lib import StrEnum, Logger, AbstractAttrDef +from ayon_core.addon import AddonsManager, IPluginPaths +from ayon_core.settings import get_studio_settings, get_project_settings +from ayon_core.pipeline.plugin_discover import discover_plugins + + +class EntityType(StrEnum): + """Selected entity type.""" + # folder = "folder" + # task = "task" + version = "version" + representation = "representation" + + +class SelectionEntitiesCache: + def __init__( + self, + project_name: str, + project_entity: Optional[dict[str, Any]] = None, + folders_by_id: Optional[dict[str, dict[str, Any]]] = None, + tasks_by_id: Optional[dict[str, dict[str, Any]]] = None, + products_by_id: Optional[dict[str, dict[str, Any]]] = None, + versions_by_id: Optional[dict[str, dict[str, Any]]] = None, + representations_by_id: Optional[dict[str, dict[str, Any]]] = None, + task_ids_by_folder_id: Optional[dict[str, str]] = None, + product_ids_by_folder_id: Optional[dict[str, str]] = None, + version_ids_by_product_id: Optional[dict[str, str]] = None, + version_id_by_task_id: Optional[dict[str, str]] = None, + representation_id_by_version_id: Optional[dict[str, str]] = None, + ): + self._project_name = project_name + self._project_entity = project_entity + self._folders_by_id = folders_by_id or {} + self._tasks_by_id = tasks_by_id or {} + self._products_by_id = products_by_id or {} + self._versions_by_id = versions_by_id or {} + self._representations_by_id = representations_by_id or {} + + self._task_ids_by_folder_id = task_ids_by_folder_id or {} + self._product_ids_by_folder_id = product_ids_by_folder_id or {} + self._version_ids_by_product_id = version_ids_by_product_id or {} + self._version_id_by_task_id = version_id_by_task_id or {} + self._representation_id_by_version_id = ( + representation_id_by_version_id or {} + ) + + def get_project(self) -> dict[str, Any]: + if self._project_entity is None: + self._project_entity = ayon_api.get_project(self._project_name) + return copy.deepcopy(self._project_entity) + + def get_folders( + self, folder_ids: set[str] + ) -> list[dict[str, Any]]: + return self._get_entities( + folder_ids, + self._folders_by_id, + "folder_ids", + ayon_api.get_folders, + ) + + def get_tasks( + self, task_ids: set[str] + ) -> list[dict[str, Any]]: + return self._get_entities( + task_ids, + self._tasks_by_id, + "task_ids", + ayon_api.get_tasks, + ) + + def get_products( + self, product_ids: set[str] + ) -> list[dict[str, Any]]: + return self._get_entities( + product_ids, + self._products_by_id, + "product_ids", + ayon_api.get_products, + ) + + def get_versions( + self, version_ids: set[str] + ) -> list[dict[str, Any]]: + return self._get_entities( + version_ids, + self._versions_by_id, + "version_ids", + ayon_api.get_versions, + ) + + def get_representations( + self, representation_ids: set[str] + ) -> list[dict[str, Any]]: + return self._get_entities( + representation_ids, + self._representations_by_id, + "representation_ids", + ayon_api.get_representations, + ) + + def get_folders_tasks( + self, folder_ids: set[str] + ) -> list[dict[str, Any]]: + task_ids = self._fill_parent_children_ids( + folder_ids, + "folderId", + "folder_ids", + self._task_ids_by_folder_id, + ayon_api.get_tasks, + ) + return self.get_tasks(task_ids) + + def get_folders_products( + self, folder_ids: set[str] + ) -> list[dict[str, Any]]: + product_ids = self._get_folders_products_ids(folder_ids) + return self.get_products(product_ids) + + def get_tasks_versions( + self, task_ids: set[str] + ) -> list[dict[str, Any]]: + folder_ids = { + task["folderId"] + for task in self.get_tasks(task_ids) + } + product_ids = self._get_folders_products_ids(folder_ids) + output = [] + for version in self.get_products_versions(product_ids): + task_id = version["taskId"] + if task_id in task_ids: + output.append(version) + return output + + def get_products_versions( + self, product_ids: set[str] + ) -> list[dict[str, Any]]: + version_ids = self._fill_parent_children_ids( + product_ids, + "productId", + "product_ids", + self._version_ids_by_product_id, + ayon_api.get_versions, + ) + return self.get_versions(version_ids) + + def get_versions_representations( + self, version_ids: set[str] + ) -> list[dict[str, Any]]: + repre_ids = self._fill_parent_children_ids( + version_ids, + "versionId", + "version_ids", + self._representation_id_by_version_id, + ayon_api.get_representations, + ) + return self.get_representations(repre_ids) + + def get_tasks_folders(self, task_ids: set[str]) -> list[dict[str, Any]]: + folder_ids = { + task["folderId"] + for task in self.get_tasks(task_ids) + } + return self.get_folders(folder_ids) + + def get_products_folders( + self, product_ids: set[str] + ) -> list[dict[str, Any]]: + folder_ids = { + product["folderId"] + for product in self.get_products(product_ids) + } + return self.get_folders(folder_ids) + + def get_versions_products( + self, version_ids: set[str] + ) -> list[dict[str, Any]]: + product_ids = { + version["productId"] + for version in self.get_versions(version_ids) + } + return self.get_products(product_ids) + + def get_versions_tasks( + self, version_ids: set[str] + ) -> list[dict[str, Any]]: + task_ids = { + version["taskId"] + for version in self.get_versions(version_ids) + if version["taskId"] + } + return self.get_tasks(task_ids) + + def get_representations_versions( + self, representation_ids: set[str] + ) -> list[dict[str, Any]]: + version_ids = { + repre["versionId"] + for repre in self.get_representations(representation_ids) + } + return self.get_versions(version_ids) + + def _get_folders_products_ids(self, folder_ids: set[str]) -> set[str]: + return self._fill_parent_children_ids( + folder_ids, + "folderId", + "folder_ids", + self._product_ids_by_folder_id, + ayon_api.get_products, + ) + + def _fill_parent_children_ids( + self, + entity_ids: set[str], + parent_key: str, + filter_attr: str, + parent_mapping: dict[str, set[str]], + getter: Callable, + ) -> set[str]: + if not entity_ids: + return set() + children_ids = set() + missing_ids = set() + for entity_id in entity_ids: + _children_ids = parent_mapping.get(entity_id) + if _children_ids is None: + missing_ids.add(entity_id) + else: + children_ids.update(_children_ids) + if missing_ids: + entities_by_parent_id = collections.defaultdict(set) + for entity in getter( + self._project_name, + fields={"id", parent_key}, + **{filter_attr: missing_ids}, + ): + child_id = entity["id"] + children_ids.add(child_id) + entities_by_parent_id[entity[parent_key]].add(child_id) + + for entity_id in missing_ids: + parent_mapping[entity_id] = entities_by_parent_id[entity_id] + + return children_ids + + def _get_entities( + self, + entity_ids: set[str], + cache_var: dict[str, Any], + filter_arg: str, + getter: Callable, + ) -> list[dict[str, Any]]: + if not entity_ids: + return [] + + output = [] + missing_ids: set[str] = set() + for entity_id in entity_ids: + entity = cache_var.get(entity_id) + if entity_id not in cache_var: + missing_ids.add(entity_id) + cache_var[entity_id] = None + elif entity: + output.append(entity) + + if missing_ids: + for entity in getter( + self._project_name, + **{filter_arg: missing_ids} + ): + output.append(entity) + cache_var[entity["id"]] = entity + return output + + +class LoaderActionSelection: + def __init__( + self, + project_name: str, + selected_ids: set[str], + selected_type: EntityType, + *, + project_anatomy: Optional["Anatomy"] = None, + project_settings: Optional[dict[str, Any]] = None, + entities_cache: Optional[SelectionEntitiesCache] = None, + ): + self._project_name = project_name + self._selected_ids = selected_ids + self._selected_type = selected_type + + self._project_anatomy = project_anatomy + self._project_settings = project_settings + + if entities_cache is None: + entities_cache = SelectionEntitiesCache(project_name) + self._entities_cache = entities_cache + + def get_entities_cache(self) -> SelectionEntitiesCache: + return self._entities_cache + + def get_project_name(self) -> str: + return self._project_name + + def get_selected_ids(self) -> set[str]: + return set(self._selected_ids) + + def get_selected_type(self) -> str: + return self._selected_type + + def get_project_settings(self) -> dict[str, Any]: + if self._project_settings is None: + self._project_settings = get_project_settings(self._project_name) + return copy.deepcopy(self._project_settings) + + def get_project_anatomy(self) -> dict[str, Any]: + if self._project_anatomy is None: + from ayon_core.pipeline import Anatomy + + self._project_anatomy = Anatomy( + self._project_name, + project_entity=self.get_entities_cache().get_project(), + ) + return self._project_anatomy + + project_name = property(get_project_name) + selected_ids = property(get_selected_ids) + selected_type = property(get_selected_type) + project_settings = property(get_project_settings) + project_anatomy = property(get_project_anatomy) + entities = property(get_entities_cache) + + +@dataclass +class LoaderActionItem: + identifier: str + entity_ids: set[str] + entity_type: EntityType + label: str + group_label: Optional[str] = None + # Is filled automatically + plugin_identifier: str = None + + +@dataclass +class LoaderActionForm: + title: str + fields: list[AbstractAttrDef] + submit_label: Optional[str] = "Submit" + submit_icon: Optional[str] = None + cancel_label: Optional[str] = "Cancel" + cancel_icon: Optional[str] = None + + +@dataclass +class LoaderActionResult: + message: Optional[str] = None + success: bool = True + form: Optional[LoaderActionForm] = None + + +class LoaderActionPlugin(ABC): + """Plugin for loader actions. + + Plugin is responsible for getting action items and executing actions. + + + """ + def __init__(self, studio_settings: dict[str, Any]): + self.apply_settings(studio_settings) + + def apply_settings(self, studio_settings: dict[str, Any]) -> None: + """Apply studio settings to the plugin. + + Args: + studio_settings (dict[str, Any]): Studio settings. + + """ + pass + + @property + def identifier(self) -> str: + """Identifier of the plugin. + + Returns: + str: Plugin identifier. + + """ + return self.__class__.__name__ + + @abstractmethod + def get_action_items( + self, selection: LoaderActionSelection + ) -> list[LoaderActionItem]: + """Action items for the selection. + + Args: + selection (LoaderActionSelection): Selection. + + Returns: + list[LoaderActionItem]: Action items. + + """ + pass + + @abstractmethod + def execute_action( + self, + identifier: str, + entity_ids: set[str], + entity_type: str, + selection: LoaderActionSelection, + form_values: dict[str, Any], + ) -> Optional[LoaderActionResult]: + """Execute an action. + + Args: + identifier (str): Action identifier. + entity_ids: (set[str]): Entity ids stored on action item. + entity_type: (str): Entity type stored on action item. + selection (LoaderActionSelection): Selection wrapper. Can be used + to get entities or get context of original selection. + form_values (dict[str, Any]): Attribute values. + + Returns: + Optional[LoaderActionResult]: Result of the action execution. + + """ + pass + + +class LoaderActionsContext: + def __init__( + self, + studio_settings: Optional[dict[str, Any]] = None, + addons_manager: Optional[AddonsManager] = None, + ) -> None: + self._log = Logger.get_logger(self.__class__.__name__) + + self._addons_manager = addons_manager + + self._studio_settings = studio_settings + self._plugins = None + + def reset( + self, studio_settings: Optional[dict[str, Any]] = None + ) -> None: + self._studio_settings = studio_settings + self._plugins = None + + def get_addons_manager(self) -> AddonsManager: + if self._addons_manager is None: + self._addons_manager = AddonsManager( + settings=self._get_studio_settings() + ) + return self._addons_manager + + def get_action_items( + self, selection: LoaderActionSelection + ) -> list[LoaderActionItem]: + output = [] + for plugin in self._get_plugins().values(): + try: + for action_item in plugin.get_action_items(selection): + action_item.identifier = plugin.identifier + output.append(action_item) + + except Exception: + self._log.warning( + "Failed to get action items for" + f" plugin '{plugin.identifier}'", + exc_info=True, + ) + return output + + def execute_action( + self, + plugin_identifier: str, + action_identifier: str, + entity_ids: set[str], + entity_type: EntityType, + selection: LoaderActionSelection, + attribute_values: dict[str, Any], + ) -> None: + plugins_by_id = self._get_plugins() + plugin = plugins_by_id[plugin_identifier] + plugin.execute_action( + action_identifier, + entity_ids, + entity_type, + selection, + attribute_values, + ) + + def _get_studio_settings(self) -> dict[str, Any]: + if self._studio_settings is None: + self._studio_settings = get_studio_settings() + return copy.deepcopy(self._studio_settings) + + def _get_plugins(self) -> dict[str, LoaderActionPlugin]: + if self._plugins is None: + addons_manager = self.get_addons_manager() + all_paths = [ + os.path.join(AYON_CORE_ROOT, "plugins", "loader") + ] + for addon in addons_manager.addons: + if not isinstance(addon, IPluginPaths): + continue + paths = addon.get_loader_action_plugin_paths() + if paths: + all_paths.extend(paths) + + studio_settings = self._get_studio_settings() + result = discover_plugins(LoaderActionPlugin, all_paths) + result.log_report() + plugins = {} + for cls in result.plugins: + try: + plugin = cls(studio_settings) + plugin_id = plugin.identifier + if plugin_id not in plugins: + plugins[plugin_id] = plugin + continue + + self._log.warning( + f"Duplicated plugins identifier found '{plugin_id}'." + ) + + except Exception: + self._log.warning( + f"Failed to initialize plugin '{cls.__name__}'", + exc_info=True, + ) + self._plugins = plugins + return self._plugins