Merge a9280bd109 into f9bbab9944

client/ayon_core/lib/__init__.py

@@ -20,6 +20,7 @@ from .cache import (
     CacheItem,
     NestedCacheItem,
 )
+from .track_changes import TrackDictChangesItem
 from .events import (
     emit_event,
     register_event_callback
@@ -161,6 +162,8 @@ __all__ = [
     "CacheItem",
     "NestedCacheItem",
 
+    "TrackDictChangesItem",
+
     "emit_event",
     "register_event_callback",
 

client/ayon_core/lib/track_changes.py

@@ -3,14 +3,14 @@ import copy
 _EMPTY_VALUE = object()
 
 
-class TrackChangesItem:
+class TrackDictChangesItem:
     """Helper object to track changes in data.
 
     Has access to full old and new data and will create deep copy of them,
     so it is not needed to create copy before passed in.
 
     Can work as a dictionary if old or new value is a dictionary. In
-    that case received object is another object of 'TrackChangesItem'.
+    that case received object is another object of 'TrackDictChangesItem'.
 
     Goal is to be able to get old or new value as was or only changed values
     or get information about removed/changed keys, and all of that on
@@ -39,7 +39,7 @@ class TrackChangesItem:
     ...     "key_3": "value_3"
     ... }
 
-    >>> changes = TrackChangesItem(old_value, new_value)
+    >>> changes = TrackDictChangesItem(old_value, new_value)
     >>> changes.changed
     True
 
@@ -280,7 +280,7 @@ class TrackChangesItem:
         old_value = self.old_value
         if self._old_is_dict and self._new_is_dict:
             for key in self.available_keys:
-                item = TrackChangesItem(
+                item = TrackDictChangesItem(
                     old_value.get(key), new_value.get(key)
                 )
                 sub_items[key] = item
@@ -294,7 +294,7 @@ class TrackChangesItem:
             for key in available_keys:
                 # NOTE Use '_EMPTY_VALUE' because old value could be 'None'
                 #   which would result in "unchanged" item
-                sub_items[key] = TrackChangesItem(
+                sub_items[key] = TrackDictChangesItem(
                     old_value.get(key), _EMPTY_VALUE
                 )
 
@@ -305,7 +305,7 @@ class TrackChangesItem:
             for key in available_keys:
                 # NOTE Use '_EMPTY_VALUE' because new value could be 'None'
                 #   which would result in "unchanged" item
-                sub_items[key] = TrackChangesItem(
+                sub_items[key] = TrackDictChangesItem(
                     _EMPTY_VALUE, new_value.get(key)
                 )
 

client/ayon_core/pipeline/create/context.py

@@ -22,7 +22,7 @@ import pyblish.api
 import ayon_api
 
 from ayon_core.settings import get_project_settings
-from ayon_core.lib import is_func_signature_supported
+from ayon_core.lib import is_func_signature_supported, TrackDictChangesItem
 from ayon_core.lib.events import QueuedEventSystem
 from ayon_core.lib.attribute_definitions import get_default_values
 from ayon_core.host import IWorkfileHost, IPublishHost
@@ -41,7 +41,6 @@ from .exceptions import (
     UnavailableSharedData,
     HostMissRequiredMethod,
 )
-from .changes import TrackChangesItem
 from .structures import (
     PublishAttributes,
     ConvertorItem,
@@ -1138,10 +1137,10 @@ class CreateContext:
             "publish_attributes": self._publish_attributes.data_to_store()
         }
 
-    def context_data_changes(self) -> TrackChangesItem:
+    def context_data_changes(self) -> TrackDictChangesItem:
         """Changes of attributes."""
 
-        return TrackChangesItem(
+        return TrackDictChangesItem(
             self._original_context_data, self.context_data_to_store()
         )
 

client/ayon_core/pipeline/create/structures.py

@@ -5,6 +5,7 @@ from enum import Enum
 import typing
 from typing import Optional, Dict, List, Any
 
+from ayon_core.lib import TrackDictChangesItem
 from ayon_core.lib.attribute_definitions import (
     AbstractAttrDef,
     UnknownDef,
@@ -19,7 +20,6 @@ from ayon_core.pipeline import (
 )
 
 from .exceptions import ImmutableKeyError
-from .changes import TrackChangesItem
 
 if typing.TYPE_CHECKING:
     from .creator_plugins import BaseCreator
@@ -825,7 +825,7 @@ class CreatedInstance:
     def changes(self):
         """Calculate and return changes."""
 
-        return TrackChangesItem(self.origin_data, self.data_to_store())
+        return TrackDictChangesItem(self.origin_data, self.data_to_store())
 
     def mark_as_stored(self):
         """Should be called when instance data are stored.

client/ayon_core/pipeline/load/context.py

@@ -0,0 +1,392 @@
+from __future__ import annotations
+
+import copy
+import typing
+from typing import Any, Optional, Iterable
+
+from ayon_core.lib import TrackDictChangesItem, Logger
+
+if typing.TYPE_CHECKING:
+    from .plugins import LoadPlugin
+
+
+class RepresentationContext:
+    """Representation context used for loading.
+
+    Attributes:
+        project_entity (dict[str, Any]): Project entity.
+        folder_entity (dict[str, Any]): Folder entity.
+        product_entity (dict[str, Any]): Product entity.
+        version_entity (dict[str, Any]): Version entity.
+        representation_entity (dict[str, Any]): Representation entity.
+
+    """
+    project_entity: dict[str, Any]
+    folder_entity: dict[str, Any]
+    product_entity: dict[str, Any]
+    version_entity: dict[str, Any]
+    representation_entity: dict[str, Any]
+
+
+class ContainerItem:
+    """Container item of loaded content.
+
+    Args:
+        scene_identifier (str): Unique container id.
+        project_name (str): Project name.
+        representation_id (str): Representation id.
+        label (str): Label of container for UI purposes.
+        namespace (str): Group label of container for UI purposes.
+        version_locked (bool): Version is locked to ignore
+            the last version checks.
+        parent_scene_identifier (Optional[str]): Parent container id. For visual
+            purposes.
+        scene_data (Optional[dict[str, Any]]): Additional data stored to the
+            scene.
+        transient_data (Optional[dict[str, Any]]): Internal load plugin data
+            related to the container. Could be any object e.g. node.
+
+    """
+    def __init__(
+        self,
+        scene_identifier: str,
+        project_name: str,
+        representation_id: str,
+        label: str,
+        namespace: str,
+        load_plugin: LoadPlugin,
+        *,
+        version_locked: bool = False,
+        is_dirty: bool = False,
+        parent_scene_identifier: Optional[str] = None,
+        scene_data: Optional[dict[str, Any]] = None,
+        transient_data: Optional[dict[str, Any]] = None,
+    ) -> None:
+        self._scene_identifier = scene_identifier
+        self._project_name = project_name
+        self._representation_id = representation_id
+        self._label = label
+        self._namespace = namespace
+        self._load_plugin_identifier = load_plugin.identifier
+        self._version_locked = version_locked
+        self._is_dirty = is_dirty
+        self._parent_scene_identifier = parent_scene_identifier
+
+        if transient_data is None:
+            transient_data = {}
+
+        if scene_data is None:
+            scene_data = {}
+
+        self._orig_generic_data = {
+            "scene_identifier": self._scene_identifier,
+            "project_name": self._project_name,
+            "representation_id": self._representation_id,
+            "label": self._label,
+            "namespace": self._namespace,
+            "load_plugin_identifier": self._load_plugin_identifier,
+            "version_locked": self._version_locked,
+            "is_dirty": self._is_dirty,
+            "parent_scene_identifier": self._parent_scene_identifier,
+        }
+        self._scene_data = scene_data
+        self._origin_scene_data = copy.deepcopy(scene_data)
+        self._transient_data = transient_data
+        self._load_plugin = load_plugin
+
+    # --- Dictionary like methods ---
+    def __getitem__(self, key: str) -> Any:
+        return self._scene_data[key]
+
+    def __contains__(self, key: str) -> bool:
+        return key in self._scene_data
+
+    def __setitem__(self, key: str, value: Any) -> None:
+        if key in self._scene_data and self._scene_data[key] == value:
+            return
+
+        self._scene_data[key] = value
+
+    def get(self, key: str, default: Any = None) -> Any:
+        return self._scene_data.get(key, default)
+
+    def pop(self, key: str, *args, **kwargs) -> Any:
+        return self._scene_data.pop(key, *args, **kwargs)
+
+    def keys(self) -> Iterable[str]:
+        return self._scene_data.keys()
+
+    def values(self) -> Iterable[Any]:
+        return self._scene_data.values()
+
+    def items(self) -> Iterable[tuple[str, Any]]:
+        return self._scene_data.items()
+    # ------
+
+    def get_scene_identifier(self) -> str:
+        return self._scene_identifier
+
+    def get_project_name(self) -> str:
+        return self._project_name
+
+    def get_representation_id(self) -> str:
+        return self._representation_id
+
+    def get_is_dirty(self) -> bool:
+        return self._is_dirty
+
+    def set_is_dirty(self, dirty: bool) -> None:
+        if dirty is self._is_dirty:
+            return
+        self._is_dirty = dirty
+        # TODO trigger event
+
+    def get_version_locked(self) -> bool:
+        return self._version_locked
+
+    def set_version_locked(self, version_locked: bool) -> None:
+        if self._version_locked == version_locked:
+            return
+        self._version_locked = version_locked
+        # TODO trigger event
+
+    def get_load_plugin_identifier(self) -> str:
+        return self._load_plugin_identifier
+
+    def get_scene_data(self) -> dict[str, Any]:
+        return copy.deepcopy(self._scene_data)
+
+    def get_origin_scene_data(self) -> dict[str, Any]:
+        return copy.deepcopy(self._origin_scene_data)
+
+    def get_transient_data(self) -> dict[str, Any]:
+        """Transient data are manager by load plugin.
+
+        Should be used for any arbitrary data needed for a container
+            management.
+
+        """
+        return self._transient_data
+
+    def get_changes(self) -> TrackDictChangesItem:
+        """Calculate and return changes.
+
+        Returns:
+            TrackDictChangesItem: Calculated changes on container.
+
+        """
+        new_data = {
+            "scene_identifier": self._scene_identifier,
+            "project_name": self._project_name,
+            "representation_id": self._representation_id,
+            "label": self._label,
+            "namespace": self._namespace,
+            "load_plugin_identifier": self._load_plugin_identifier,
+            "version_locked": self._version_locked,
+            "is_dirty": self._is_dirty,
+            "parent_scene_identifier": self._parent_scene_identifier,
+            "scene_data": self.get_scene_data(),
+        }
+        orig_data = copy.deepcopy(self._orig_generic_data)
+        orig_data["scene_data"] = self.get_origin_scene_data()
+        return TrackDictChangesItem(orig_data, new_data)
+
+    id: str = property(get_scene_identifier)
+    scene_identifier: str = property(get_scene_identifier)
+    project_name: str = property(get_project_name)
+    load_plugin_identifier: str = property(get_load_plugin_identifier)
+    representation_id: str = property(get_representation_id)
+    scene_data: dict[str, Any] = property(get_scene_data)
+    origin_scene_data: dict[str, Any] = property(get_origin_scene_data)
+    transient_data: dict[str, Any] = property(get_transient_data)
+    changes: TrackDictChangesItem = property(get_changes)
+
+
+class LoadContext:
+    """Context of logic related to loading.
+
+    To be able to load anything in a DCC using AYON is to have load plugins.
+    Load plugin is responsible for loading representation. To maintain the
+    loaded content it is usually necessary to store some metadata in workfile.
+
+    Loaded content is refered to as a 'container' which is a helper wrapper
+    to manage loaded the content, to be able to switch versions or switch to
+    different representation (png -> exr), or to remove them from the scene.
+    """
+    def __init__(self) -> None:
+        self._shared_data = {}
+        self._plugins = None
+        self._containers = {}
+        self._collect_containers()
+        self._log = Logger.get_logger(self.__class__.__name__)
+
+    def reset(self) -> None:
+        self._shared_data = {}
+        self._plugins = None
+        self._containers = {}
+
+        self._collect_plugins()
+        self._collect_containers()
+
+    def get_plugins(self) -> dict[str, LoadPlugin]:
+        if self._plugins is None:
+            self._collect_plugins()
+        return self._plugins
+
+    def get_plugin(self, identifier: str) -> Optional[LoadPlugin]:
+        """Get plugin by identifier.
+
+        Args:
+            identifier (str): Plugin identifier.
+
+        Returns:
+            Optional[LoadPlugin]: Load plugin or None if not found.
+
+        """
+        return self._get_plugin_by_identifier(identifier, validate=False)
+
+    def add_containers(self, containers: list[ContainerItem]) -> None:
+        """Called by load plugins.
+
+        Args:
+            containers (list[ContainerItem]): Containers to add.
+
+        """
+        for container in containers:
+            if container.id in self._containers:
+                self._log.warning()
+                continue
+            self._containers[container.id] = container
+
+    def get_container_by_id(
+        self, scene_identifier: str
+    ) -> Optional[ContainerItem]:
+        return self._containers.get(scene_identifier)
+
+    def get_containers(self) -> dict[str, ContainerItem]:
+        return self._containers
+
+    @property
+    def shared_data(self) -> dict[str, Any]:
+        """Access to shared data of load plugins.
+
+        It is common that load plugins do store data the same way for all
+        containers. This helps to share data between the plugins.
+
+        Returns:
+            dict[str, Any]: Shared data.
+
+        """
+        return self._shared_data
+
+    def load_representations(
+        self,
+        identifier: str,
+        representation_contexts: list[RepresentationContext],
+    ) -> list[ContainerItem]:
+        """Load representations.
+
+        Args:
+            identifier (str): Load plugin identifier.
+            representation_contexts (list[RepresentationContext]): List of
+                representation contexts.
+
+        Returns:
+            list[ContainerItem]: List of loaded containers.
+
+        """
+        plugin = self._get_plugin_by_identifier(identifier, validate=True)
+        return plugin.load_representations(representation_contexts)
+
+    def change_representations(
+        self,
+        identifier: str,
+        items: list[tuple[ContainerItem, RepresentationContext]],
+    ) -> None:
+        """Change representations of loaded containers.
+
+        Args:
+            identifier (str): Load plugin identifier.
+            items (list[tuple[ContainerItem, RepresentationContext]]): List
+                of containers and their new representation contexts.
+
+        """
+        plugin = self._get_plugin_by_identifier(identifier, validate=True)
+        return plugin.change_representations(items)
+
+    def remove_containers(
+        self,
+        identifier: str,
+        containers: list[ContainerItem],
+    ) -> None:
+        """Remove containers content with metadata from scene.
+
+        Args:
+            identifier (str): Load plugin identifier.
+            containers (list[ContainerItem]): Containers to remove.
+
+        """
+        plugin = self._get_plugin_by_identifier(identifier, validate=True)
+        return plugin.remove_containers(containers)
+
+    def can_switch_container(
+        self,
+        identifier: str,
+        container: ContainerItem,
+    ) -> bool:
+        """Check if container can be switched.
+
+        Args:
+            identifier: Load plugin identifier.
+            container (ContainerItem): Container to check.
+
+        Returns:
+            bool: True if container can be switched, False otherwise.
+
+        """
+        plugin = self._get_plugin_by_identifier(identifier, validate=True)
+        return plugin.can_switch_container(container)
+
+    def switch_containers(
+        self,
+        identifier: str,
+        containers: list[ContainerItem],
+    ) -> list[ContainerItem]:
+        """Switch containers of other load plugins.
+
+        Args:
+            identifier: Load plugin identifier.
+            containers (list[ContainerItem]): Containers to switch.
+
+        Raises:
+            UnsupportedSwitchError: If switching is not supported.
+
+        Returns:
+            list[ContainerItem]: New containers after switching.
+
+        """
+        plugin = self._get_plugin_by_identifier(identifier, validate=True)
+        raise plugin.switch_containers(containers)
+
+    def _collect_plugins(self) -> None:
+        # TODO implement
+        self._plugins = {}
+
+    def _get_plugin_by_identifier(
+        self, identifier: str, validate: bool,
+    ) -> Optional[LoadPlugin]:
+        if self._plugins is None:
+            self._collect_plugins()
+        plugin = self._plugins.get(identifier)
+        if validate and plugin is None:
+            # QUESTION: Use custom exception?
+            raise ValueError(
+                f"Plugin with identifier '{identifier}' not found."
+            )
+        return plugin
+
+    def _collect_containers(self) -> None:
+        for plugin in sorted(
+            self.get_plugins().values(), key=lambda p: p.order
+        ):
+            plugin.collect_containers()

client/ayon_core/pipeline/load/exceptions.py

@@ -0,0 +1,3 @@
+class UnsupportedSwitchError(Exception):
+    """Raised when load plugin does not support switching containers."""
+    pass

client/ayon_core/pipeline/load/plugins.py

@@ -1,7 +1,8 @@
 """Plugins for loading representations and products into host applications."""
 from __future__ import annotations
 
-from abc import abstractmethod
+from abc import ABC, abstractmethod
+import logging
 import os
 from typing import Any, Optional, Type
 
@@ -15,7 +16,13 @@ from ayon_core.pipeline.plugin_discover import (
 )
 from ayon_core.settings import get_project_settings
 
+from .exceptions import UnsupportedSwitchError
 from .utils import get_representation_path_from_context
+from .context import (
+    LoadContext,
+    RepresentationContext,
+    ContainerItem,
+)
 
 
 class LoaderPlugin(list):
@@ -440,6 +447,187 @@ def add_hooks_to_loader(
             wrap_method(method)
 
 
+class LoadPlugin(ABC):
+    """Base class for load plugins.
+
+    Load plugin is responsible for loading representation into
+        a host application and storing metadata in workfile
+        to be able to manage them.
+
+    Attributes:
+        order (int): Order of plugins in which will be executed
+            collection and in which will be shown in the load menu.
+        settings_category (str): Settings category (addon name) used to
+            auto-apply settings.
+        settings_name (str): Key under 'load' used to auto-apply settings.
+        identifier (str): Plugin identifier. Must be unique.
+
+    """
+    order: int = 100
+    settings_category: str = None
+    settings_name: str = None
+
+    def __init__(
+        self,
+        load_context: LoadContext,
+        project_settings: dict[str, Any],
+    ) -> None:
+        self._log = None
+        self._load_context = load_context
+        self.apply_settings(project_settings)
+
+    @property
+    @abstractmethod
+    def identifier(self) -> str:
+        """Plugin identifier."""
+        pass
+
+    @property
+    def log(self) -> logging.Logger:
+        if self._log is None:
+            self._log = Logger.get_logger(self.__class__.__name__)
+        return self._log
+
+    def apply_settings(self, project_settings: dict[str, Any]) -> None:
+        """Apply project settings to the plugin.
+
+        Args:
+            project_settings (dict[str, Any]): Project settings.
+
+        """
+        cls_name = self.__class__.__name__
+        settings_name = self.settings_name or cls_name
+
+        settings = project_settings
+        for key in (
+            self.settings_category,
+            "load",
+            settings_name,
+        ):
+            settings = settings.get(key)
+            if settings is None:
+                self.log.debug(f"No settings found for {cls_name}")
+                return
+
+        for key, value in settings.items():
+            # Log out attributes that are not defined on plugin object
+            # - those may be potential dangerous typos in settings
+            if not hasattr(self, key):
+                self.log.debug(
+                    "Applying settings to unknown attribute"
+                    f" '{key}' on '{cls_name}'."
+                )
+            setattr(self, key, value)
+
+    @abstractmethod
+    def collect_containers(self) -> None:
+        """Collect containers from the current workfile.
+
+        This method is called by LoadContext on initialization and on reset.
+        """
+        pass
+
+    @abstractmethod
+    def is_representation_compatible(
+        self,
+        representation_context: RepresentationContext,
+    ) -> bool:
+        """Check if representation is compatible with the plugin.
+
+        Args:
+            representation_context (RepresentationContext): Representation
+                context.
+
+        Returns:
+            bool: True if compatible, False otherwise.
+
+        """
+        pass
+
+    @abstractmethod
+    def load_representations(
+        self,
+        representation_contexts: list[RepresentationContext],
+    ) -> list[ContainerItem]:
+        """Load representations.
+
+        Method still has to call 'add_containers' method on 'LoadContext'.
+
+        Args:
+            representation_contexts (list[RepresentationContext]): List of
+                representation contexts.
+
+        Returns:
+            list[ContainerItem]: List of loaded containers.
+
+        """
+        pass
+
+    @abstractmethod
+    def change_representations(
+        self,
+        items: list[tuple[ContainerItem, RepresentationContext]],
+    ) -> None:
+        """Change representations of loaded containers.
+
+        Can be used to switch between versions or different representations.
+
+        Args:
+            items (list[tuple[ContainerItem, RepresentationContext]]): List of
+                containers and their new representation contexts.
+
+        """
+        pass
+
+    @abstractmethod
+    def remove_containers(
+        self,
+        containers: list[ContainerItem]
+    ) -> None:
+        """Remove containers from the workfile.
+
+        Args:
+            containers (list[ContainerItem]): Containers to remove.
+
+        """
+
+    def can_switch_container(self, container: ContainerItem) -> bool:
+        """Check if container can be switched.
+
+        Args:
+            container (ContainerItem): Container to check.
+
+        Returns:
+            bool: True if container can be switched, False otherwise.
+
+        """
+        return False
+
+    def switch_containers(
+        self, containers: list[ContainerItem]
+    ) -> list[ContainerItem]:
+        """Switch containers of other load plugins.
+
+        Args:
+            containers (list[ContainerItem]): Containers to switch.
+
+        Raises:
+            UnsupportedSwitchError: If switching is not supported.
+
+        Returns:
+            list[ContainerItem]: New containers after switching.
+
+        """
+        raise UnsupportedSwitchError()
+
+    def _add_containers_to_context(
+        self,
+        containers: list[ContainerItem],
+    ) -> None:
+        """Helper method to add containers to load context."""
+        self._load_context.add_containers(containers)
+
+
 def register_loader_plugin(plugin):
     return register_plugin(LoaderPlugin, plugin)