From c52665461621bf3109c142371d4fc85c1f782613 Mon Sep 17 00:00:00 2001 From: Carlos Villavicencio Date: Fri, 22 May 2026 17:05:05 -0500 Subject: [PATCH 1/4] Add MEDM abstractions and utilities --- python/tk_multi_loader/medm/__init__.py | 32 + python/tk_multi_loader/medm/entity_model.py | 438 +++++++++++ .../medm/latestpublish_model.py | 705 ++++++++++++++++++ .../medm/publishhistory_model.py | 493 ++++++++++++ python/tk_multi_loader/medm/shared_cache.py | 107 +++ .../tk_multi_loader/medm/thumbnail_service.py | 202 +++++ python/tk_multi_loader/medm/utils.py | 201 +++++ 7 files changed, 2178 insertions(+) create mode 100644 python/tk_multi_loader/medm/__init__.py create mode 100644 python/tk_multi_loader/medm/entity_model.py create mode 100644 python/tk_multi_loader/medm/latestpublish_model.py create mode 100644 python/tk_multi_loader/medm/publishhistory_model.py create mode 100644 python/tk_multi_loader/medm/shared_cache.py create mode 100644 python/tk_multi_loader/medm/thumbnail_service.py create mode 100644 python/tk_multi_loader/medm/utils.py diff --git a/python/tk_multi_loader/medm/__init__.py b/python/tk_multi_loader/medm/__init__.py new file mode 100644 index 0000000..4ebbf58 --- /dev/null +++ b/python/tk_multi_loader/medm/__init__.py @@ -0,0 +1,32 @@ +# Copyright (c) 2026 Shotgun Software Inc. +# +# CONFIDENTIAL AND PROPRIETARY +# +# This work is provided "AS IS" and subject to the Shotgun Pipeline Toolkit +# Source Code License included in this distribution package. See LICENSE. +# By accessing, using, copying or modifying this work you indicate your +# agreement to the Shotgun Pipeline Toolkit Source Code License. All rights +# not expressly granted therein are reserved by Shotgun Software Inc. + +"""MEDM (Flow Asset Management) integration models for the Loader app. + +This package provides Qt models that back the loader when ``use_medm_data`` +is enabled in the app configuration. All models share a single +:class:`~medm.shared_cache.MedmSharedCache` and +:class:`~medm.thumbnail_service.MedmThumbnailService` instance injected by +the dialog at construction time. +""" + +from .entity_model import MedmEntityModel +from .latestpublish_model import MedmLatestPublishModel +from .publishhistory_model import MedmPublishHistoryModel +from .shared_cache import MedmSharedCache +from .thumbnail_service import MedmThumbnailService + +__all__ = [ + "MedmEntityModel", + "MedmLatestPublishModel", + "MedmPublishHistoryModel", + "MedmSharedCache", + "MedmThumbnailService", +] diff --git a/python/tk_multi_loader/medm/entity_model.py b/python/tk_multi_loader/medm/entity_model.py new file mode 100644 index 0000000..e8c294f --- /dev/null +++ b/python/tk_multi_loader/medm/entity_model.py @@ -0,0 +1,438 @@ +# Copyright (c) 2026 Shotgun Software Inc. +# +# CONFIDENTIAL AND PROPRIETARY +# +# This work is provided "AS IS" and subject to the Shotgun Pipeline Toolkit +# Source Code License included in this distribution package. See LICENSE. +# By accessing, using, copying or modifying this work you indicate your +# agreement to the Shotgun Pipeline Toolkit Source Code License. All rights +# not expressly granted therein are reserved by Shotgun Software Inc. + +"""MEDM Asset Model - Tree model for MEDM asset hierarchy + +This module provides a tree model that displays MEDM assets +in the left-hand tree view of the loader. + +Children are loaded lazily: only the immediate children of the current project +are fetched on startup. Deeper levels are fetched on demand when the user +expands a tree node (via Qt's ``canFetchMore`` / ``fetchMore`` protocol). +The shared :class:`~medm.shared_cache.MedmSharedCache` ``children`` dict +prevents duplicate API round-trips when the same asset's children are +requested by both the tree and the center-panel publish model. +""" + +from __future__ import annotations + +from typing import TYPE_CHECKING, Any, List, Optional + +import sgtk +from sgtk.platform.qt import QtCore, QtGui + +from .shared_cache import MedmSharedCache +from .utils import is_structural_asset as _is_structural_asset_util + +# Import types for type hints only - actual objects come from framework at runtime +# Framework is loaded dynamically via sgtk.platform.import_framework() +if TYPE_CHECKING: + from adsk.flow.data import Asset +else: + Asset = Any + + +class MedmEntityModel(QtGui.QStandardItemModel): + """ + Tree model that displays MEDM assets in a hierarchical structure. + This replaces SgEntityModel for MEDM data sources. + + Uses lazy loading: only the project's immediate children are fetched at + startup. Deeper levels are fetched when the user expands a node. + """ + + # Custom roles - matching ShotgunModel interface + SG_DATA_ROLE = QtCore.Qt.UserRole + 1 + SG_ASSOCIATED_FIELD_ROLE = QtCore.Qt.UserRole + 2 + ASSET_ROLE = QtCore.Qt.UserRole + 200 # Stores MEDM Asset object (shared with all MEDM models) + + # Lazy-loading bookkeeping role: True once children have been fetched for a node. + CHILDREN_LOADED_ROLE = QtCore.Qt.UserRole + 201 + + # Signals - required for ShotgunModelOverlayWidget compatibility + cache_loaded = QtCore.Signal() + data_refreshed = QtCore.Signal(bool) # Argument: data_changed + query_changed = QtCore.Signal() + data_refreshing = QtCore.Signal() + data_refresh_fail = QtCore.Signal(str) + + def __init__( + self, + parent, + entity_type, + filters, + hierarchy, + bg_task_manager, + cache: Optional[MedmSharedCache] = None, + ): + """ + Constructor + + :param parent: Parent QObject + :param entity_type: Entity type (kept for API compatibility, not used) + :param filters: Filters (kept for API compatibility, not used) + :param hierarchy: Hierarchy fields (kept for API compatibility, not used) + :param bg_task_manager: Background task manager (kept for API compatibility) + :param cache: Shared :class:`MedmSharedCache`. When provided the + ``children`` dict is used so that expanding a tree node and + selecting it in the centre panel never duplicate an API call. + When *None* a private fallback cache is used (test / standalone use). + """ + super().__init__(parent) + + self._app = sgtk.platform.current_bundle() + self._flow_module = sgtk.platform.import_framework("tk-framework-flowam", "flow") + + self._cache = cache if cache is not None else MedmSharedCache() + self._folder_icon = QtGui.QIcon(QtGui.QPixmap(":/res/icon_Folder.png")) + self._binary_icon = QtGui.QIcon(QtGui.QPixmap(":/res/icon_Asset_dark.png")) + + # Lazily resolved set of structural type IDs (folder, pipeline step). + # Assets matching any of these are always shown in the tree; + # others are only shown when they have structural descendants. + self._structural_type_ids: Optional[set] = None + + self._project = None + self._initialize_project() + + # Defer loading to allow UI to set up first (shows spinner) + self.data_refreshing.emit() + QtCore.QTimer.singleShot(100, self._load_medm_assets) + + # ------------------------------------------------------------------------- + # Qt virtual overrides - lazy loading protocol + # ------------------------------------------------------------------------- + + def hasChildren(self, parent=QtCore.QModelIndex()): + """ + Return ``True`` when *parent* might have children. + + For nodes whose children have not been fetched yet we optimistically + return ``True`` so that Qt draws an expansion arrow. Once children + have been loaded the answer is based on the actual row count. + """ + if not parent.isValid(): + return self.rowCount() > 0 + item = self.itemFromIndex(parent) + if item is None: + return False + if item.data(self.CHILDREN_LOADED_ROLE): + return item.rowCount() > 0 + # Not yet loaded -> assume children exist (shows the expand arrow) + return True + + def canFetchMore(self, parent): + """Return ``True`` if *parent*'s children have not been loaded yet.""" + if not parent.isValid(): + return False + item = self.itemFromIndex(parent) + if item is None: + return False + return not item.data(self.CHILDREN_LOADED_ROLE) + + def fetchMore(self, parent): + """Load the immediate children of *parent* from the MEDM API (or cache).""" + if not parent.isValid(): + return + item = self.itemFromIndex(parent) + if item is None or item.data(self.CHILDREN_LOADED_ROLE): + return + self._load_children_for_item(item) + + # ------------------------------------------------------------------------- + # Public API - Called by dialog.py and other external code + # ------------------------------------------------------------------------- + + def destroy(self): + """Clean up model resources.""" + self._cache.children.clear() + + def async_refresh(self): + """Refresh the model data.""" + self.clear() + self._cache.clear_on_hard_refresh() + self.data_refreshing.emit() + QtCore.QTimer.singleShot(100, self._load_medm_assets) + + def hard_refresh(self): + """Hard refresh (same as async_refresh for this simple model).""" + self.async_refresh() + + def item_from_entity(self, entity_type: str, entity_id: int): + """ + Returns a QStandardItem based on entity type and entity id. + + **OVERRIDE:** This method overrides the ShotgunModel.item_from_entity() interface + to provide MEDM-compatible implementation. The original ShotgunModel version uses + an internal data handler (_data_handler.get_uid_from_entity_id), but MEDM models + store data in a tree structure requiring recursive search. + + **Implementation Differences from ShotgunModel.item_from_entity:** + - Original: Uses flat data handler lookup (uid-based) + - This override: Performs recursive tree search through QStandardItem hierarchy + - Original: Validates entity_type matches model's __entity_type + - This override: Ignores entity_type (MEDM uses unified Asset model) + + **Note:** Method name preserved for API compatibility with dialog.py which expects + all entity models (SgEntityModel, SgHierarchyModel, MedmEntityModel) to implement + this interface. Called by dialog._get_item_from_entity() for navigation/selection. + + :param entity_type: Shotgun entity type (ignored in MEDM implementation) + :param entity_id: Entity ID to search for (compared against SG_DATA_ROLE["id"]) + :returns: :class:`~PySide.QtGui.QStandardItem` or None if not found + """ + + def search_item(parent): + for row in range(parent.rowCount() if parent else self.rowCount()): + item = parent.child(row) if parent else self.item(row) + if item: + sg_data = item.data(self.SG_DATA_ROLE) + if sg_data and sg_data.get("id") == entity_id: + return item + found = search_item(item) + if found: + return found + return None + + return search_item(None) + + def get_cached_children(self, asset: Asset) -> List[Asset]: + """ + Return child :class:`Asset` objects for *asset*. + + Uses the internal cache when available; otherwise fetches from the MEDM + API and stores the result. This is the single entry-point that both + the tree's ``fetchMore`` and :class:`MedmLatestPublishModel` use, so + that a drill-down never fetches the same level twice. + + :param asset: Parent MEDM Asset whose children are needed. + :returns: List of child Asset objects (may be empty). + """ + return self._fetch_and_cache_children(asset) + + # ------------------------------------------------------------------------- + # Private utility methods - Internal implementation details + # ------------------------------------------------------------------------- + + def _initialize_project(self) -> None: + """ + Initialize and cache the MEDM Project object. + Called during __init__ to fail fast if project is unavailable. + """ + try: + session_project = self._flow_module.data.get_session_project() + self._project = self._flow_module.data.Project(session_project.id) + self._app.log_debug(f"MEDM Entity: Initialized project '{self._project.name}'") + except Exception as e: + self._app.log_error( + f"MEDM Entity: Failed to initialize project: {type(e).__name__}: {e}. " + "Entity tree will not be loaded." + ) + self._project = None + + def _get_structural_type_ids(self) -> set: + """ + Return the set of type-ID strings that are always shown in the tree + regardless of whether they have children. + + The set is resolved once and cached on the instance. It contains: + - ``FOLDER_TYPE_ID`` - Autodesk built-in type, available as a constant. + - pipeline-step - schema-registered type whose ID varies per collection + and is resolved via ``flow_module.schema.get_schema_id``. + + Template and generic-workfile types are intentionally excluded: they + are publishable leaf assets that belong in the centre panel, not in + the tree. + + On any error (e.g. framework not ready) an empty set is returned so + that the tree still loads without crashing. + """ + if self._structural_type_ids is not None: + return self._structural_type_ids + + try: + folder_id = self._flow_module.data.FOLDER_TYPE_ID + pipeline_step_id = self._flow_module.schema.get_schema_id( + self._flow_module.asset_management.PIPELINE_STEP_TYPE + ) + + self._structural_type_ids = {folder_id, pipeline_step_id} + self._app.log_debug( + f"MEDM Entity: structural type IDs = {self._structural_type_ids}" + ) + except self._flow_module.FlowError as e: + self._app.log_warning( + f"MEDM Entity: could not resolve structural type IDs ({e}); " + "non-structural assets without structural descendants will be hidden." + ) + self._structural_type_ids = set() + + return self._structural_type_ids + + def _is_tree_node(self, asset: Asset) -> bool: + """ + Return ``True`` when *asset* should appear as a node in the left-hand + tree view. + + An asset qualifies if **any** of the following conditions hold: + + * it is a structural container (folder, container type, or pipeline + step) - these are always visible regardless of whether they have + children; **or** + * it has at least one direct child asset - workfiles can themselves + parent child workfiles, making them container nodes in the tree + even though they are not structural types. + + Assets that satisfy none of the above are pure leaf items that belong + only in the centre-panel publish list, not in the tree. + + :param asset: MEDM ``Asset`` to test. + :returns: ``True`` if the asset should appear in the tree. + """ + if _is_structural_asset_util(asset, self._flow_module): + return True + + # Non-structural: show in the tree only when the asset has direct + # children. Results are cached so each asset is fetched from the + # API at most once. + children = self._fetch_and_cache_children(asset) + return len(children) > 0 + + def _icon_for_asset(self, asset: Asset) -> QtGui.QIcon: + """ + Return the appropriate tree icon for *asset* based on its type. + + * **Structural container** (folder, container type, or pipeline-step) + -> folder icon. + * **Everything else** (workfiles, generic assets, ...) -> binary/data + icon, reflecting that the item holds or organises file data. + + :param asset: MEDM ``Asset`` to pick an icon for. + :returns: A :class:`QtGui.QIcon` instance. + """ + return ( + self._folder_icon + if _is_structural_asset_util(asset, self._flow_module) + else self._binary_icon + ) + + def _load_medm_assets(self): + """ + Load the first level of MEDM assets (project's immediate children). + Called asynchronously after a short delay to show the loading spinner. + """ + if self._project is None: + self._app.log_warning("MEDM Entity: Cannot load assets - project not initialized") + self.data_refresh_fail.emit("Project not initialized") + return + + try: + self._app.log_debug("MEDM: Loading entity tree (project children only)...") + + count = 0 + for asset in self._project.iterate_children(): + if self._is_tree_node(asset): + self._add_asset_item(asset, None) + count += 1 + + self._app.log_debug( + f"MEDM: Entity tree loaded successfully. Loaded {count} root assets" + ) + self.cache_loaded.emit() + self.data_refreshed.emit(True) + + except Exception as e: + self._app.log_error(f"Failed to load MEDM data: {e}") + import traceback + + self._app.log_debug(traceback.format_exc()) + self.data_refresh_fail.emit(str(e)) + + def _add_asset_item( + self, asset: Asset, parent_item: Optional[QtGui.QStandardItem] + ) -> QtGui.QStandardItem: + """ + Create a single ``QStandardItem`` for *asset* and append it to the tree. + + The item is marked as *not* children-loaded so that ``hasChildren`` + reports ``True`` and Qt draws an expand arrow until the user actually + drills in. + + :param asset: The MEDM Asset to add. + :param parent_item: Parent item, or ``None`` for root level. + :returns: The newly created item. + """ + asset_item = QtGui.QStandardItem(asset.name) + asset_item.setEditable(False) + + asset_item.setData(asset, self.ASSET_ROLE) + + sg_data = { + "type": asset.__class__.__name__, + "id": None, + "name": asset.name, + "code": asset.name, + } + asset_item.setData(sg_data, self.SG_DATA_ROLE) + + # Mark children as not-yet-loaded so canFetchMore/hasChildren work. + asset_item.setData(False, self.CHILDREN_LOADED_ROLE) + + asset_item.setIcon(self._icon_for_asset(asset)) + + if parent_item is None: + self.appendRow(asset_item) + else: + parent_item.appendRow(asset_item) + + return asset_item + + def _load_children_for_item(self, item: QtGui.QStandardItem) -> None: + """ + Fetch the immediate children of *item* and add them to the tree. + + ``CHILDREN_LOADED_ROLE`` is set to ``True`` **before** any rows are + inserted so that the ``rowsInserted`` signal (emitted by ``appendRow``) + cannot re-enter ``fetchMore`` for the same item. + + :param item: The tree item whose children should be loaded. + """ + # Mark loaded FIRST to prevent re-entrant fetchMore calls triggered by + # appendRow -> rowsInserted -> canFetchMore check on the same parent. + item.setData(True, self.CHILDREN_LOADED_ROLE) + + asset = item.data(self.ASSET_ROLE) + if asset is None: + return + + try: + children = self._fetch_and_cache_children(asset) + tree_children = [c for c in children if self._is_tree_node(c)] + for child_asset in tree_children: + self._add_asset_item(child_asset, item) + self._app.log_debug( + f"MEDM: Loaded {len(tree_children)}/{len(children)} children for '{asset.name}' " + f"(non-structural leaf children hidden from tree)" + ) + except Exception as e: + self._app.log_debug(f"MEDM: Could not get children for '{asset.name}': {e}") + + def _fetch_and_cache_children(self, asset: Asset) -> List[Asset]: + """ + Return child assets for *asset*, fetching from the API only on the + first call and caching the result in the shared cache for subsequent + lookups by any MEDM model. + """ + if asset.id in self._cache.children: + return self._cache.children[asset.id] + + children = list(asset.iterate_children()) + self._cache.children[asset.id] = children + return children diff --git a/python/tk_multi_loader/medm/latestpublish_model.py b/python/tk_multi_loader/medm/latestpublish_model.py new file mode 100644 index 0000000..4eae730 --- /dev/null +++ b/python/tk_multi_loader/medm/latestpublish_model.py @@ -0,0 +1,705 @@ +# Copyright (c) 2026 Shotgun Software Inc. +# +# CONFIDENTIAL AND PROPRIETARY +# +# This work is provided "AS IS" and subject to the Shotgun Pipeline Toolkit +# Source Code License included in this distribution package. See LICENSE. +# By accessing, using, copying or modifying this work you indicate your +# agreement to the Shotgun Pipeline Toolkit Source Code License. All rights +# not expressly granted therein are reserved by Shotgun Software Inc. + +"""MEDM Latest Publish Model - Replacement for SgLatestPublishModel + +This module provides a drop-in replacement for SgLatestPublishModel that uses +Flow Asset Management (MEDM) data instead of Shotgun data. +""" + +from __future__ import annotations + +from collections import defaultdict +from typing import TYPE_CHECKING, Any, Dict, List, Optional + +import sgtk +from sgtk.platform.qt import QtCore, QtGui + +from .shared_cache import MedmSharedCache +from .thumbnail_service import MedmThumbnailService +from .utils import build_draft_sg_dict +from .utils import is_structural_asset as _is_structural_asset_util + +if TYPE_CHECKING: + from flow.data import Asset + from flow.sandbox import DraftInfo +else: + Asset = Any + DraftInfo = Any + + +class MedmLatestPublishModel(QtGui.QStandardItemModel): + """ + Model which handles the main spreadsheet view which displays the latest version of all + publishes from Flow Asset Management. + + This is a drop-in replacement for SgLatestPublishModel that uses MEDM data. + """ + + # Matches the V1 FlowActions hook constant. Draft rows carry this as + # version_number so action hooks route them to asset_management.open_draft(). + DRAFT_VERSION_IDENTIFIER = -1 + + # Sentinel key used inside cache.drafts to store the list returned by + # get_drafts(draft_type="new"). A real asset.id is always a UUID/storage + # key string; this double-underscore key cannot collide with a real ID. + _NEW_DRAFTS_CACHE_KEY = "__new_asset_drafts__" + + # Custom roles - matching the original model's interface + TYPE_ID_ROLE = QtCore.Qt.UserRole + 101 + IS_FOLDER_ROLE = QtCore.Qt.UserRole + 102 + ASSOCIATED_TREE_VIEW_ITEM_ROLE = QtCore.Qt.UserRole + 103 + PUBLISH_TYPE_NAME_ROLE = QtCore.Qt.UserRole + 104 + SEARCHABLE_NAME = QtCore.Qt.UserRole + 105 + + # Additional MEDM-specific roles + SG_DATA_ROLE = QtCore.Qt.UserRole + 1 # To maintain compatibility with ShotgunModel + SG_ASSOCIATED_FIELD_ROLE = QtCore.Qt.UserRole + 2 + ASSET_ROLE = QtCore.Qt.UserRole + 200 # Stores MEDM Asset object (shared with all MEDM models) + DRAFT_ROLE = QtCore.Qt.UserRole + 202 # Stores DraftInfo for draft rows (shared with history model) + + # Signals + loadingStarted = QtCore.Signal() + loadingFinished = QtCore.Signal() + loadingError = QtCore.Signal(str) + cache_loaded = QtCore.Signal() + data_refreshed = QtCore.Signal(bool) # Argument: data_changed + query_changed = QtCore.Signal() # For overlay widget compatibility + data_refreshing = QtCore.Signal() # For overlay widget compatibility + data_refresh_fail = QtCore.Signal(str) # For overlay widget compatibility + + def __init__( + self, + parent: Optional[QtCore.QObject], + publish_type_model, + bg_task_manager, + cache: Optional[MedmSharedCache] = None, + thumbnail_service: Optional[MedmThumbnailService] = None, + ): + """ + Model which represents the latest publishes for an entity from MEDM. + + :param parent: Parent QObject + :param publish_type_model: Model for tracking publish types + :param bg_task_manager: Background task manager (kept for API compatibility) + :param cache: Shared :class:`MedmSharedCache`. When provided all data + caches (children, drafts, publish types) are shared with other MEDM + models so no duplicate API calls are made. When *None* a private + cache is created for standalone use. + :param thumbnail_service: Shared :class:`MedmThumbnailService` instance. + When provided (the normal case) thumbnail downloads are shared with + other models. When *None* a private service is created. + """ + super().__init__(parent) + + self._app = sgtk.platform.current_bundle() + self._flow_module = sgtk.platform.import_framework("tk-framework-flowam", "flow") + + self._publish_type_model = publish_type_model + self._bg_task_manager = bg_task_manager + self._cache = cache if cache is not None else MedmSharedCache() + self._thumbnail_service = thumbnail_service or MedmThumbnailService(self._cache, self) + self._owns_thumbnail_service = thumbnail_service is None + + self._loading_icon = QtGui.QIcon.fromTheme("view-refresh") + self._publish_icon = QtGui.QIcon.fromTheme("document") + + self._current_entity = None + + self._project_id = 0 + self._project_name = "MEDM Project" + self._initialize_project_info() + + # ------------------------------------------------------------------------- + # Public API - Called by dialog.py and other external code + # ------------------------------------------------------------------------- + + def destroy(self): + """Clean up model resources.""" + self._cache.drafts.clear() + if self._owns_thumbnail_service: + self._thumbnail_service.destroy() + + def load_data(self, item): + """ + Clears the model and sets it up for the selected asset from left treeview panel. + Loads data from MEDM instead of Shotgun. + + :param item: Selected item in the treeview, None if nothing is selected. + """ + self.loadingStarted.emit() + + try: + self._current_entity = item + + self.layoutAboutToBeChanged.emit() + self.removeRows(0, self.rowCount()) + self._populate_model_from_selected_item(item) + self.layoutChanged.emit() + + self.loadingFinished.emit() + self.cache_loaded.emit() + + except Exception as e: + self.loadingError.emit(str(e)) + + def async_refresh(self): + """ + Refresh the current data set. + + Clears volatile shared caches (drafts) so fresh state is fetched from + the server. Thumbnail and publish-type caches are intentionally kept: + revision IDs are immutable so those will always return the same data. + """ + self._cache.clear_on_refresh() + if self._current_entity is not None: + self.load_data(self._current_entity) + self.data_refreshed.emit(True) + + # ------------------------------------------------------------------------- + # Private utility methods - Internal implementation details + # ------------------------------------------------------------------------- + + def _initialize_project_info(self) -> None: + """Initialize project ID and name from the session project.""" + try: + self._project_id = self._app.context.project["id"] + self._project_name = self._app.context.project["name"] + except Exception as e: + self._app.log_warning( + f"MEDM LatestPublish: Failed to initialize project info: {type(e).__name__}: {e}. " + "Using default project values. This may indicate the session project was not " + "initialized or the project ID is invalid." + ) + + def _populate_model_from_selected_item(self, selected_item): + """ + Populate the model with latest versions of child assets from the selected tree item. + + Extracts the Asset from the selected tree view item, fetches latest versions + of all its child assets, and populates the model with them as Qt items. + + :param selected_item: The selected asset item from the tree view (or None) + """ + if selected_item is None: + return + + asset = self._extract_asset_from_tree_item(selected_item) + if asset is None: + self._app.log_warning("MEDM: Could not extract asset from selected item") + return + + self._app.log_debug(f"MEDM: Asset extracted: {asset.name}") + + children_asset_sg_dicts = self._fetch_asset_children(asset) + + # If the selected asset has no (non-structural) children it may itself + # be the publishable leaf. Only fall back to showing it directly when + # the asset is NOT a structural container. + leaf_asset_fallback = None + if not children_asset_sg_dicts and not _is_structural_asset_util( + asset, self._flow_module + ): + try: + children_asset_sg_dicts = [self._asset_to_sg_dict(asset)] + except Exception as e: + self._app.log_warning( + f"MEDM: Could not convert leaf asset '{asset.name}' to sg_dict: {e}" + ) + # Keep a reference to the raw asset so we can still fetch its + # drafts below. + leaf_asset_fallback = asset + + self._app.log_debug( + f"MEDM: Fetched {len(children_asset_sg_dicts)} latest version dicts from children" + ) + + assets_for_draft_lookup = [ + sg_dict.get("_medm_asset") for sg_dict in children_asset_sg_dicts + ] + assets_for_draft_lookup = [a for a in assets_for_draft_lookup if a is not None] + if not assets_for_draft_lookup and leaf_asset_fallback is not None: + assets_for_draft_lookup = [leaf_asset_fallback] + + # --- Pass 1: collect draft cards per asset ---------------------------- + drafts_by_asset_id: Dict[str, list] = {} + for child_asset in assets_for_draft_lookup: + try: + if child_asset.id in self._cache.drafts: + raw_drafts = self._cache.drafts[child_asset.id] + else: + raw_drafts = self._flow_module.asset_management.get_asset_drafts( + child_asset.id + ) + self._cache.drafts[child_asset.id] = raw_drafts + except Exception as e: + self._app.log_debug( + f"MEDM: Could not fetch drafts for '{child_asset.name}': {e}" + ) + continue + + draft_dicts = [] + for draft_info in raw_drafts: + try: + draft_dicts.append(self._draft_to_sg_dict(draft_info, child_asset)) + self._app.log_debug( + f"MEDM: Found draft '{getattr(draft_info, 'name', '?')}' " + f"for asset '{child_asset.name}'" + ) + except Exception as e: + self._app.log_warning( + f"MEDM: Could not convert draft '{getattr(draft_info, 'name', '?')}' " + f"for '{child_asset.name}': {e}" + ) + + if draft_dicts: + drafts_by_asset_id[child_asset.id] = draft_dicts + + # --- Pass 2: populate the center panel -------------------------------- + # - Asset has a draft -> show only the draft card(s). + # - Asset has no draft -> show its latest published version card. + draft_count = 0 + published_count = 0 + + for sg_dict in children_asset_sg_dicts: + medm_asset = sg_dict.get("_medm_asset") + asset_id = medm_asset.id if medm_asset is not None else None + + if asset_id is not None and asset_id in drafts_by_asset_id: + for draft_sg_dict in drafts_by_asset_id[asset_id]: + self._add_sg_dict_as_qt_item(draft_sg_dict) + draft_count += 1 + else: + self._add_sg_dict_as_qt_item(sg_dict) + published_count += 1 + + # Surface drafts for the leaf-fallback asset. + if leaf_asset_fallback is not None and leaf_asset_fallback.id in drafts_by_asset_id: + for draft_sg_dict in drafts_by_asset_id[leaf_asset_fallback.id]: + self._add_sg_dict_as_qt_item(draft_sg_dict) + draft_count += 1 + + # --- Pass 3: surface NewDraftInfo entries for unpublished child assets + for new_draft_sg_dict in self._fetch_new_draft_items_for_parent(asset.id): + self._add_sg_dict_as_qt_item(new_draft_sg_dict) + draft_count += 1 + + self._app.log_debug( + f"MEDM: center panel now has {self.rowCount()} items " + f"({draft_count} draft(s), {published_count} published)" + ) + + sg_publish_type_counts = self._calculate_sg_publish_type_counts() + self._publish_type_model.set_active_types(sg_publish_type_counts) + + def _extract_asset_from_tree_item(self, item) -> Optional[Asset]: + """ + Extract the MEDM Asset object from a tree view QStandardItem. + + :param item: The QStandardItem from the entity tree (left panel) + :returns: MEDM Asset object or None if not found + """ + # Both MedmEntityModel and MedmLatestPublishModel use ASSET_ROLE = Qt.UserRole + 200 + asset = item.data(self.ASSET_ROLE) + if asset: + return asset + + asset_data = item.data(QtCore.Qt.UserRole + 1) + return asset_data + + def _fetch_asset_children(self, asset: Asset) -> List[Dict[str, Any]]: + """ + Fetch all non-structural child assets and convert to sg_data dicts. + + Structural containers (folders, pipeline steps, container types) are + filtered out here because they belong only in the left-hand tree, not + in the center panel publish list. + + The shared ``cache.children`` dict is consulted first so that selecting + a tree node never duplicates an API call that was already made when the + node was expanded by :class:`MedmEntityModel` (or vice-versa). + + :param asset: The selected MEDM Asset in MEDM treeview + :returns: List of sg_data dictionaries representing each non-structural child asset + """ + children_asset_sg_dicts = [] + + try: + if asset.id in self._cache.children: + child_assets = self._cache.children[asset.id] + else: + child_assets = list(asset.iterate_children()) + self._cache.children[asset.id] = child_assets + + for child_asset in child_assets: + if _is_structural_asset_util(child_asset, self._flow_module): + self._app.log_debug( + f"MEDM: Skipping structural asset '{child_asset.name}' from center panel" + ) + continue + try: + asset_dict = self._asset_to_sg_dict(child_asset) + children_asset_sg_dicts.append(asset_dict) + self._app.log_debug( + f"MEDM: Added asset '{child_asset.name}' with latest version " + f"v{child_asset.version_number}" + ) + except Exception as e: + self._app.log_warning( + f"MEDM: Error processing child asset '{child_asset.name}': {e}" + ) + continue + + except Exception as e: + self._app.log_warning(f"MEDM: Error fetching asset children: {e}") + + self._app.log_debug( + f"MEDM: Loaded {len(children_asset_sg_dicts)} children assets for asset '{asset.name}'" + ) + return children_asset_sg_dicts + + def _asset_to_sg_dict(self, asset: Asset) -> Dict[str, Any]: + """ + Convert an MEDM Asset to a Shotgun-compatible dictionary. + + :param asset: The MEDM Asset + :returns: Dictionary with Shotgun-compatible fields + """ + sg_publish_type_id = None + sg_publish_type_code = "MEDM Asset" + + medm_type_ids = asset.type_ids + if len(medm_type_ids) > 0: + sg_publish_type_id, sg_publish_type_code = self._resolve_publish_type( + medm_type_ids[0] + ) + + sg_dict = { + "id": None, + "type": "PublishedFile", + "code": asset.name, + "name": asset.name, + "version_number": asset.version_number, + "description": asset.description or "", + "created_at": asset.created_at, + "created_by": {"type": "HumanUser", "id": 1, "name": asset.created_by or "MEDM User"}, + "entity": {"type": "Asset", "id": None, "name": asset.name}, + "project": {"type": "Project", "id": self._project_id, "name": self._project_name}, + "task": None, + "task_uniqueness": True, + "published_file_type": { + "type": "PublishedFileType", + "id": sg_publish_type_id, + "name": sg_publish_type_code, + }, + "tank_type": { + "type": "TankType", + "id": sg_publish_type_id, + "name": sg_publish_type_code, + }, + "path": {"local_path": ""}, + "image": None, + # Thumbnail is resolved in a background thread via this revision ID. + "_thumbnail_revision_id": asset.revision_id, + "sg_flow_revision_id": asset.revision_id, + "_medm_asset": asset, + } + + return sg_dict + + def _draft_to_sg_dict( + self, draft_info: DraftInfo, asset: Optional[Asset] = None + ) -> Dict[str, Any]: + """ + Convert a local DraftInfo into a Shotgun-compatible dictionary suitable + for display as a center-panel card. + + Key conventions that V1 action hooks rely on: + - ``version_number == DRAFT_VERSION_IDENTIFIER (-1)`` - identifies a local draft + - ``sg_flow_revision_id`` - the draft's sandbox ID (draft_info.draft_id), used + by asset_management.open_draft() and sandbox.is_local_draft() + + :param draft_info: DraftInfo returned by asset_management.get_asset_drafts() + (CheckoutDraftInfo) or get_drafts() (NewDraftInfo). + :param asset: The MEDM Asset the draft belongs to. May be ``None`` for + ``NewDraftInfo`` entries whose parent asset has not been published yet. + :returns: sg_data dictionary compatible with action hooks and center panel UI. + """ + sg_publish_type_id = None + sg_publish_type_code = "MEDM Asset" + + # Prefer the published asset's type_ids; fall back to the draft's own + # type_ids for NewDraftInfo where no published asset exists yet. + type_ids = [] + if asset is not None: + type_ids = getattr(asset, "type_ids", None) or [] + if not type_ids: + type_ids = getattr(draft_info, "type_ids", None) or [] + if type_ids: + sg_publish_type_id, sg_publish_type_code = self._resolve_publish_type(type_ids[0]) + + # For checkout drafts the parent asset's revision_id provides the + # thumbnail; for new drafts there is no published revision. + draft_type = getattr(draft_info, "draft_type", "unknown") + thumb_revision_id = ( + asset.revision_id if (draft_type == "checkout" and asset is not None) else None + ) + + sg_dict = build_draft_sg_dict( + draft_info, + asset, + self._app.context, + self._project_id, + self._project_name, + sg_publish_type_id, + sg_publish_type_code, + ) + sg_dict.update( + { + "task_uniqueness": True, + "tank_type": { + "type": "TankType", + "id": sg_publish_type_id, + "name": sg_publish_type_code, + }, + "_thumbnail_revision_id": thumb_revision_id, + } + ) + return sg_dict + + def _fetch_new_draft_items_for_parent( + self, parent_asset_id: str + ) -> List[Dict[str, Any]]: + """ + Return sg_dict items for ``NewDraftInfo`` entries whose ``parent_id`` + matches *parent_asset_id*. + + These represent brand-new assets that exist only on disk and have never + been published to MEDM. Because they have no published asset record + they are invisible to ``asset.iterate_children()`` and must be surfaced + via :func:`~flow.asset_management.get_drafts`. + + The full list returned by ``get_drafts(draft_type="new")`` is cached + under :attr:`_NEW_DRAFTS_CACHE_KEY` in ``cache.drafts`` for the + lifetime of a single load cycle (cleared by + :meth:`~MedmSharedCache.clear_on_refresh`). + + :param parent_asset_id: ``asset.id`` of the tree item currently selected. + :returns: List of sg_data dicts ready for :meth:`_add_sg_dict_as_qt_item`. + """ + if self._NEW_DRAFTS_CACHE_KEY not in self._cache.drafts: + try: + all_new_drafts = self._flow_module.asset_management.get_drafts( + draft_type="new" + ) + except Exception as e: + self._app.log_warning(f"MEDM: Could not fetch new-asset drafts: {e}") + all_new_drafts = [] + self._cache.drafts[self._NEW_DRAFTS_CACHE_KEY] = all_new_drafts + else: + all_new_drafts = self._cache.drafts[self._NEW_DRAFTS_CACHE_KEY] + + result = [] + for draft_info in all_new_drafts: + if getattr(draft_info, "parent_id", None) != parent_asset_id: + continue + try: + draft_sg_dict = self._draft_to_sg_dict(draft_info, asset=None) + self._app.log_debug( + f"MEDM: Found new-asset draft '{getattr(draft_info, 'name', '?')}' " + f"under parent '{parent_asset_id}'" + ) + result.append(draft_sg_dict) + except Exception as e: + self._app.log_warning( + f"MEDM: Could not convert new-asset draft " + f"'{getattr(draft_info, 'name', '?')}': {e}" + ) + return result + + def _resolve_publish_type(self, medm_type_id_str: str) -> tuple: + """ + Resolve a MEDM schema type ID to a ``(sg_publish_type_id, display_name)`` pair. + + Resolution order: + 1. In-process cache (avoids redundant SG round-trips within the same load). + 2. ShotGrid ``PublishedFileType`` lookup by display name (real ID). + 3. Hash-based integer fallback when no SG record exists. + + :param medm_type_id_str: MEDM schema type ID string. + :returns: Tuple of (integer publish type id, human-readable display name). + """ + if medm_type_id_str in self._cache.publish_types: + return self._cache.publish_types[medm_type_id_str] + + display_name = medm_type_id_str + try: + schema_name = self._flow_module.schema.get_schema_display_name(medm_type_id_str) + if schema_name: + display_name = schema_name + except Exception as e: + self._app.log_debug( + f"MEDM: Could not get schema display name for '{medm_type_id_str}': {e}" + ) + + sg_publish_type_id = None + try: + pft = self._app.shotgun.find_one( + "PublishedFileType", + [["code", "is", display_name]], + ["id", "code"], + ) + if pft: + sg_publish_type_id = pft["id"] + self._app.log_debug( + f"MEDM: Resolved PublishedFileType '{display_name}' " + f"-> SG id={sg_publish_type_id}" + ) + else: + self._app.log_debug( + f"MEDM: No SG PublishedFileType found for '{display_name}', " + f"item will bypass type filter" + ) + except Exception as e: + self._app.log_debug( + f"MEDM: Could not look up PublishedFileType for '{display_name}': {e}" + ) + + result = (sg_publish_type_id, display_name) + self._cache.publish_types[medm_type_id_str] = result + return result + + def _add_sg_dict_as_qt_item(self, sg_item: Dict[str, Any]): + """ + Create a QStandardItem from a Shotgun-compatible dict and add it to the model. + + The QStandardItem stores multiple pieces of data in custom Qt roles: + - SG_DATA_ROLE: Full Shotgun-compatible dict for backwards compatibility + - ASSET_ROLE: Original MEDM Asset object (from "_medm_asset" key) + - TYPE_ID_ROLE: Publish type ID for filtering + - PUBLISH_TYPE_NAME_ROLE: Publish type name for display + - SEARCHABLE_NAME: Name for search/filter operations + + The item is then appended to the model as a new row, making it visible in the + center panel's publish view. + + :param sg_item: Shotgun-compatible dictionary created by _asset_to_sg_dict(). + Must contain "_medm_asset" key with the original MEDM Asset object. + """ + qt_item = QtGui.QStandardItem(sg_item.get("code", "Unnamed")) + + qt_item.setData(sg_item, self.SG_DATA_ROLE) + qt_item.setData(sg_item.get("code", ""), self.SEARCHABLE_NAME) + qt_item.setData(False, self.IS_FOLDER_ROLE) + pft = sg_item.get("published_file_type") or {} + qt_item.setData(pft.get("id"), self.TYPE_ID_ROLE) + qt_item.setData(pft.get("name"), self.PUBLISH_TYPE_NAME_ROLE) + + if "_medm_asset" in sg_item: + qt_item.setData(sg_item["_medm_asset"], self.ASSET_ROLE) + if "_medm_draft" in sg_item: + qt_item.setData(sg_item["_medm_draft"], self.DRAFT_ROLE) + + qt_item.setEditable(False) + qt_item.setIcon(self._publish_icon) + + revision_id = sg_item.get("_thumbnail_revision_id") + if revision_id: + self._resolve_and_download_thumbnail(qt_item, revision_id) + + self._set_tooltip(qt_item, sg_item) + + def get_sg_data(): + return qt_item.data(self.SG_DATA_ROLE) + + qt_item.get_sg_data = get_sg_data + + self.appendRow(qt_item) + + self._app.log_debug( + f"MEDM: Added item '{qt_item.text()}' to model (row count: {self.rowCount()})" + ) + + def _set_tooltip(self, item: QtGui.QStandardItem, sg_item: Dict[str, Any]): + """ + Sets a tooltip for a publish item. + + :param item: QStandardItem associated with the publish + :param sg_item: Publish information dictionary + """ + tooltip = f"Name: {sg_item.get('code', 'No name given.')}" + + # Version info - drafts use DRAFT_VERSION_IDENTIFIER (-1) internally; + # show a human-readable label instead of the raw sentinel value. + version = sg_item.get("version_number") + if version == self.DRAFT_VERSION_IDENTIFIER: + draft_type = sg_item.get("_medm_draft_type", "") + vers_str = f"Draft ({draft_type})" if draft_type else "Draft" + elif version is not None and version >= 0: + vers_str = f"{version:03d}" + else: + vers_str = "N/A" + + author_str = sg_item.get("created_by", {}).get("name", "Unknown") + date_str = str(sg_item.get("created_at", "Unknown")) + + tooltip += f"

Version: {vers_str} by {author_str} at {date_str}" + tooltip += ( + f"

Description: {sg_item.get('description', 'No description given.')}" + ) + + item.setToolTip(tooltip) + + def _resolve_and_download_thumbnail( + self, qt_item: QtGui.QStandardItem, revision_id: str + ): + """ + Delegate thumbnail resolution and download to the shared + :class:`MedmThumbnailService`. The service calls :meth:`_apply_thumbnail` + on the main thread once the image bytes are available. + + :param qt_item: The QStandardItem to set the thumbnail on. + :param revision_id: MEDM AssetRevision ID whose thumbnail is needed. + """ + self._thumbnail_service.request(qt_item, revision_id, self._apply_thumbnail) + + def _apply_thumbnail( + self, qt_item: QtGui.QStandardItem, image_data: bytes + ) -> None: + """ + Apply downloaded image bytes as a scaled icon on *qt_item*. + Called on the main thread by :class:`MedmThumbnailService`. + + :param qt_item: The QStandardItem to update. + :param image_data: Raw image bytes. + """ + pixmap = QtGui.QPixmap() + if pixmap.loadFromData(image_data): + scaled = pixmap.scaled( + 512, 400, QtCore.Qt.KeepAspectRatio, QtCore.Qt.SmoothTransformation + ) + qt_item.setIcon(QtGui.QIcon(scaled)) + + def _calculate_sg_publish_type_counts(self) -> Dict[int, int]: + """ + Count how many items in the model have each Shotgun PublishedFileType. + + :returns: Dictionary mapping sg_publish_type_id to item count + """ + sg_publish_type_aggregates = defaultdict(int) + + for row in range(self.rowCount()): + item = self.item(row, 0) + if item and not item.data(self.IS_FOLDER_ROLE): + sg_publish_type_id = item.data(self.TYPE_ID_ROLE) + if sg_publish_type_id is not None: + sg_publish_type_aggregates[sg_publish_type_id] += 1 + + return sg_publish_type_aggregates diff --git a/python/tk_multi_loader/medm/publishhistory_model.py b/python/tk_multi_loader/medm/publishhistory_model.py new file mode 100644 index 0000000..be18c45 --- /dev/null +++ b/python/tk_multi_loader/medm/publishhistory_model.py @@ -0,0 +1,493 @@ +# Copyright (c) 2026 Shotgun Software Inc. +# +# CONFIDENTIAL AND PROPRIETARY +# +# This work is provided "AS IS" and subject to the Shotgun Pipeline Toolkit +# Source Code License included in this distribution package. See LICENSE. +# By accessing, using, copying or modifying this work you indicate your +# agreement to the Shotgun Pipeline Toolkit Source Code License. All rights +# not expressly granted therein are reserved by Shotgun Software Inc. + +""" +MEDM Publish History Model - Shows all revisions for a selected MEDM entity. + +This module provides a model that displays all revisions (version history) for +a selected MEDM entity, similar to SgPublishHistoryModel for Shotgun data. +""" + +from __future__ import annotations + +from typing import TYPE_CHECKING, Any, Dict, Optional + +import sgtk +from sgtk.platform.qt import QtCore, QtGui + +from .. import utils +from .shared_cache import MedmSharedCache +from .thumbnail_service import MedmThumbnailService +from .utils import build_draft_sg_dict + + +if TYPE_CHECKING: + from adsk.flow.am import ( + Asset, + AssetRevision, + AssetVersion, + ) + from flow.sandbox import CheckoutDraftInfo, DraftInfo, NewDraftInfo +else: + Asset = Any + AssetRevision = Any + AssetVersion = Any + CheckoutDraftInfo = Any + DraftInfo = Any + NewDraftInfo = Any + + +class MedmPublishHistoryModel(QtGui.QStandardItemModel): + """ + Model that displays the version history (all revisions) for an MEDM entity. + + This is the MEDM equivalent of SgPublishHistoryModel. + """ + + # Matches the V1 FlowActions hook constant so action hooks correctly identify + # draft rows (version_number == -1 -> local draft; version_number > -1 -> published). + DRAFT_VERSION_IDENTIFIER = -1 + + # Custom roles - matching SgPublishHistoryModel interface + USER_THUMB_ROLE = QtCore.Qt.UserRole + 101 + PUBLISH_THUMB_ROLE = QtCore.Qt.UserRole + 102 + FULL_IMAGE_PATH_ROLE = QtCore.Qt.UserRole + 103 + + # MEDM-specific roles + SG_DATA_ROLE = QtCore.Qt.UserRole + 1 # To maintain compatibility with ShotgunModel + ASSET_ROLE = QtCore.Qt.UserRole + 200 # Stores MEDM Asset object (shared with all MEDM models) + VERSION_ROLE = QtCore.Qt.UserRole + 201 # Stores MEDM AssetVersion object + DRAFT_ROLE = QtCore.Qt.UserRole + 202 # Stores DraftInfo for draft rows + + # Signals for compatibility with ShotgunModelOverlayWidget + cache_loaded = QtCore.Signal() + data_refreshed = QtCore.Signal(bool) # Argument: data_changed + query_changed = QtCore.Signal() + data_refreshing = QtCore.Signal() + data_refresh_fail = QtCore.Signal(str) + + def __init__( + self, + parent, + bg_task_manager, + cache: Optional[MedmSharedCache] = None, + thumbnail_service: Optional[MedmThumbnailService] = None, + ): + """ + Constructor + + :param parent: Parent QObject + :param bg_task_manager: Background task manager (kept for API compatibility) + :param cache: Shared :class:`MedmSharedCache`. When provided all data + caches (drafts, versions, publish types) are shared with other MEDM + models. When *None* a private cache is created for standalone use. + :param thumbnail_service: Shared :class:`MedmThumbnailService` instance. + When provided thumbnail downloads are shared with other models. + When *None* a private service is created. + """ + super().__init__(parent) + + self._app = sgtk.platform.current_bundle() + self._flow_module = sgtk.platform.import_framework("tk-framework-flowam", "flow") + + self._bg_task_manager = bg_task_manager + self._cache = cache if cache is not None else MedmSharedCache() + self._thumbnail_service = thumbnail_service or MedmThumbnailService(self._cache, self) + self._owns_thumbnail_service = thumbnail_service is None + self._loading_icon = QtGui.QPixmap(":/res/loading_100x100.png") + self._current_sg_data = None + + self._project_id = 0 + self._project_name = "MEDM Project" + self._initialize_project_info() + + # ------------------------------------------------------------------------- + # Public API - Called by dialog.py and other external code + # ------------------------------------------------------------------------- + + def destroy(self): + """Clean up model resources.""" + self._cache.drafts.clear() + self._cache.versions.clear() + if self._owns_thumbnail_service: + self._thumbnail_service.destroy() + + def load_data(self, sg_data: Dict[str, Any]) -> None: + """ + Load and display all versions (version history) for the selected asset. + + This populates the history panel with all versions of the asset + selected in the center panel, ordered by version number (newest first). + + :param sg_data: Shotgun-compatible data dict from center panel selection. + Must contain "_medm_asset" field with the MEDM Asset. + """ + self.clear() + self._current_sg_data = sg_data + + medm_asset = sg_data.get("_medm_asset") + if medm_asset is None: + # A NewDraftInfo card has no published asset yet. Show the draft + # itself as the sole history entry so the user can still open it. + draft_info = sg_data.get("_medm_draft") + if draft_info is not None and getattr(draft_info, "draft_type", None) == "new": + self._add_draft_as_qt_item(draft_info, asset=None) + self.cache_loaded.emit() + self.data_refreshed.emit(True) + else: + self._app.log_warning( + "MEDM History: No asset found in selected publish asset sg_data dict" + ) + self.data_refresh_fail.emit("No asset found in selection") + return + + try: + asset_id = medm_asset.id + if asset_id in self._cache.versions: + versions = self._cache.versions[asset_id] + else: + versions = list(medm_asset.iterate_versions()) + self._cache.versions[asset_id] = versions + + for asset_version in versions: + self._add_version_as_qt_item(asset_version, medm_asset) + + self._app.log_debug(f"MEDM History: Loaded {len(versions)} published versions") + + try: + if asset_id in self._cache.drafts: + drafts = self._cache.drafts[asset_id] + else: + drafts = self._flow_module.asset_management.get_asset_drafts(asset_id) + self._cache.drafts[asset_id] = drafts + for draft_info in drafts: + self._add_draft_as_qt_item(draft_info, medm_asset) + if drafts: + self._app.log_debug( + f"MEDM History: Added {len(drafts)} draft(s) for asset '{medm_asset.name}'" + ) + except Exception as e: + # Drafts are optional - a failure here should not prevent published + # versions from being shown. + self._app.log_warning(f"MEDM History: Could not fetch drafts: {e}") + + self.cache_loaded.emit() + self.data_refreshed.emit(True) + + except Exception as e: + self._app.log_error(f"MEDM History: Error loading versions: {e}") + self.data_refresh_fail.emit(str(e)) + + def async_refresh(self): + """ + Refresh the current data set. + + Clears volatile shared caches (versions, drafts) so that fresh data is + fetched. Thumbnail and publish-type caches are intentionally kept: + revision IDs are immutable so those will always return the same data. + """ + self._cache.clear_on_refresh() + if self._current_sg_data is not None: + self.load_data(self._current_sg_data) + self.data_refreshed.emit(True) + + def hard_refresh(self): + """Force refresh of data (same as async_refresh for MEDM).""" + self.async_refresh() + + # ------------------------------------------------------------------------- + # Private utility methods - Internal implementation details + # ------------------------------------------------------------------------- + + def _initialize_project_info(self) -> None: + """Initialize project ID and name from the session project.""" + try: + self._project_id = self._app.context.project["id"] + self._project_name = self._app.context.project["name"] + except Exception as e: + self._app.log_warning( + f"MEDM History: Failed to initialize project info: {type(e).__name__}: {e}. " + "Using default project values. This may indicate the session project was not " + "initialized or the project ID is invalid." + ) + + def _add_version_as_qt_item( + self, asset_version: AssetVersion, asset: Asset + ) -> None: + """ + Convert an AssetVersion to a QStandardItem and add it to the history model. + + :param asset_version: The MEDM AssetVersion to add + :param asset: The parent MEDM Asset + """ + version_number = asset_version.version_number + + qt_item = QtGui.QStandardItem(f"{version_number:03d}") + qt_item.setEditable(False) + + sg_data = self._version_to_sg_dict(asset_version, asset) + qt_item.setData(sg_data, self.SG_DATA_ROLE) + + qt_item.setData(asset, self.ASSET_ROLE) + qt_item.setData(asset_version, self.VERSION_ROLE) + + qt_item.setData(self._loading_icon, self.PUBLISH_THUMB_ROLE) + thumb = utils.create_overlayed_user_publish_thumbnail( + qt_item.data(self.PUBLISH_THUMB_ROLE), None + ) + qt_item.setIcon(QtGui.QIcon(thumb)) + + revision_id = sg_data.get("_thumbnail_revision_id") + if revision_id: + self._resolve_and_download_thumbnail(qt_item, revision_id) + + def get_sg_data(): + return qt_item.data(self.SG_DATA_ROLE) + + qt_item.get_sg_data = get_sg_data + + self.appendRow(qt_item) + self._app.log_debug(f"MEDM History: Added version v{version_number}") + + def _version_to_sg_dict( + self, version: AssetVersion, asset: Asset + ) -> Dict[str, Any]: + """ + Convert a MEDM AssetVersion to Shotgun-compatible dictionary. + + :param version: The MEDM AssetVersion + :param asset: The parent MEDM Asset + :returns: sg_data dictionary compatible with Shotgun UI + """ + sg_publish_type_id = None + sg_publish_type_code = "MEDM Asset" + + medm_type_ids = asset.type_ids + if medm_type_ids: + sg_publish_type_id, sg_publish_type_code = self._resolve_publish_type( + medm_type_ids[0] + ) + + asset_revision = version.revision + description = asset_revision.comment or "" + + sg_dict = { + "id": None, + "type": "PublishedFile", + "code": asset_revision.name, + "name": asset_revision.name, + "version_number": version.version_number, + "description": description, + "created_at": version.created_at, + "created_by": { + "type": "HumanUser", + "id": 1, + "name": version.created_by or "MEDM User", + }, + "entity": { + "type": "Asset", + "id": None, + "name": asset.name, + }, + "project": { + "type": "Project", + "id": self._project_id, + "name": self._project_name, + }, + "task": None, + "published_file_type": { + "type": "PublishedFileType", + "id": sg_publish_type_id, + "name": sg_publish_type_code, + }, + "image": None, + # Thumbnail is resolved in a background thread via this revision ID. + "_thumbnail_revision_id": asset_revision.id, + "sg_flow_revision_id": asset_revision.id, + "_medm_asset": asset, + "_medm_version": version, + } + + return sg_dict + + def _draft_to_sg_dict( + self, draft_info: DraftInfo, asset: Optional[Asset] + ) -> Dict[str, Any]: + """ + Convert a DraftInfo (CheckoutDraftInfo or NewDraftInfo) to a Shotgun-compatible + dictionary for display in the version history list as a local draft entry. + + The key conventions that V1 action hooks (flowam_actions.py) rely on: + - ``version_number == DRAFT_VERSION_IDENTIFIER (-1)`` -> row is a local draft + - ``sg_flow_revision_id`` -> the draft's unique sandbox ID (draft_info.draft_id), + used by asset_management.open_draft() and sandbox.is_local_draft() + + :param draft_info: DraftInfo object returned by asset_management.get_asset_drafts() + or get_drafts(). May be CheckoutDraftInfo or NewDraftInfo. + :param asset: The parent MEDM Asset (may be None for NewDraftInfo) + :returns: sg_data dictionary compatible with action hooks and Shotgun UI + """ + sg_publish_type_id = None + sg_publish_type_code = "MEDM Asset" + + # Prefer the published asset's type_ids; fall back to the draft's own + # type_ids for NewDraftInfo where no published asset exists yet. + type_ids = [] + if asset is not None: + type_ids = getattr(asset, "type_ids", None) or [] + if not type_ids: + type_ids = getattr(draft_info, "type_ids", None) or [] + if type_ids: + sg_publish_type_id, sg_publish_type_code = self._resolve_publish_type(type_ids[0]) + + sg_dict = build_draft_sg_dict( + draft_info, + asset, + self._app.context, + self._project_id, + self._project_name, + sg_publish_type_id, + sg_publish_type_code, + ) + sg_dict["_medm_version"] = None + return sg_dict + + def _add_draft_as_qt_item( + self, draft_info: DraftInfo, asset: Optional[Asset] + ) -> None: + """ + Convert a DraftInfo to a QStandardItem and insert it at the top of + the history model so drafts always appear above published versions. + + Supports both CheckoutDraftInfo (checkout of a published revision) and + NewDraftInfo (new asset not yet published). + + :param draft_info: DraftInfo object (CheckoutDraftInfo or NewDraftInfo) + :param asset: The parent MEDM Asset (may be None for NewDraftInfo) + """ + draft_type = getattr(draft_info, "draft_type", "unknown") + + if draft_type == "checkout": + source_version = getattr(draft_info, "version", "?") + label = f"Draft (v{source_version} checkout)" + elif draft_type == "new": + label = "Draft (new)" + else: + label = f"Draft ({draft_type})" + + qt_item = QtGui.QStandardItem(label) + qt_item.setEditable(False) + + sg_data = self._draft_to_sg_dict(draft_info, asset) + qt_item.setData(sg_data, self.SG_DATA_ROLE) + qt_item.setData(asset, self.ASSET_ROLE) + qt_item.setData(draft_info, self.DRAFT_ROLE) + + qt_item.setData(self._loading_icon, self.PUBLISH_THUMB_ROLE) + thumb = utils.create_overlayed_user_publish_thumbnail( + qt_item.data(self.PUBLISH_THUMB_ROLE), None + ) + qt_item.setIcon(QtGui.QIcon(thumb)) + + def get_sg_data(): + return qt_item.data(self.SG_DATA_ROLE) + + qt_item.get_sg_data = get_sg_data + + # Prepend so drafts always appear above all published versions. + self.insertRow(0, qt_item) + self._app.log_debug( + f"MEDM History: Added {draft_type} draft '{draft_info.name}' " + f"(draft_id={draft_info.draft_id})" + ) + + def _resolve_publish_type(self, medm_type_id_str: str) -> tuple: + """ + Resolve a MEDM schema type ID to a ``(sg_publish_type_id, display_name)`` pair. + + Resolution order: + 1. In-process cache (avoids redundant SG round-trips within the same load). + 2. ShotGrid ``PublishedFileType`` lookup by display name (real ID). + 3. Hash-based integer fallback when no SG record exists. + + :param medm_type_id_str: MEDM schema type ID string. + :returns: Tuple of (integer publish type id, human-readable display name). + """ + if medm_type_id_str in self._cache.publish_types: + return self._cache.publish_types[medm_type_id_str] + + display_name = medm_type_id_str + try: + schema_name = self._flow_module.schema.get_schema_display_name(medm_type_id_str) + if schema_name: + display_name = schema_name + except Exception as e: + self._app.log_debug( + f"MEDM History: Could not get schema display name for '{medm_type_id_str}': {e}" + ) + + sg_publish_type_id = None + try: + pft = self._app.shotgun.find_one( + "PublishedFileType", + [["code", "is", display_name]], + ["id", "code"], + ) + if pft: + sg_publish_type_id = pft["id"] + self._app.log_debug( + f"MEDM History: Resolved PublishedFileType '{display_name}' " + f"-> SG id={sg_publish_type_id}" + ) + else: + self._app.log_debug( + f"MEDM History: No SG PublishedFileType found for '{display_name}', " + f"item will bypass type filter" + ) + except Exception as e: + self._app.log_debug( + f"MEDM History: Could not look up PublishedFileType for '{display_name}': {e}" + ) + + result = (sg_publish_type_id, display_name) + self._cache.publish_types[medm_type_id_str] = result + return result + + def _resolve_and_download_thumbnail( + self, qt_item: QtGui.QStandardItem, revision_id: str + ): + """ + Delegate thumbnail resolution and download to the shared + :class:`MedmThumbnailService`. The service calls :meth:`_apply_thumbnail` + on the main thread once the image bytes are available. + + :param qt_item: The QStandardItem to set the thumbnail on. + :param revision_id: MEDM AssetRevision ID whose thumbnail is needed. + """ + self._thumbnail_service.request(qt_item, revision_id, self._apply_thumbnail) + + def _apply_thumbnail( + self, qt_item: QtGui.QStandardItem, image_data: bytes + ) -> None: + """ + Apply downloaded image bytes to *qt_item* as a composite publish thumbnail. + Called on the main thread by :class:`MedmThumbnailService`. + + :param qt_item: The QStandardItem to update. + :param image_data: Raw image bytes. + """ + pixmap = QtGui.QPixmap() + if pixmap.loadFromData(image_data): + qt_item.setData(pixmap, self.PUBLISH_THUMB_ROLE) + thumb = utils.create_overlayed_user_publish_thumbnail( + qt_item.data(self.PUBLISH_THUMB_ROLE), + qt_item.data(self.USER_THUMB_ROLE), + ) + qt_item.setIcon(QtGui.QIcon(thumb)) diff --git a/python/tk_multi_loader/medm/shared_cache.py b/python/tk_multi_loader/medm/shared_cache.py new file mode 100644 index 0000000..ea9831f --- /dev/null +++ b/python/tk_multi_loader/medm/shared_cache.py @@ -0,0 +1,107 @@ +# Copyright (c) 2026 Shotgun Software Inc. +# +# CONFIDENTIAL AND PROPRIETARY +# +# This work is provided "AS IS" and subject to the Shotgun Pipeline Toolkit +# Source Code License included in this distribution package. See LICENSE. +# By accessing, using, copying or modifying this work you indicate your +# agreement to the Shotgun Pipeline Toolkit Source Code License. All rights +# not expressly granted therein are reserved by Shotgun Software Inc. + +""" +Shared cache container for all MEDM model data. + +A single :class:`MedmSharedCache` instance is created by the dialog and +injected into every MEDM model. Centralising the dictionaries here means: + +* No duplicate API calls when the same data is needed by more than one model + (e.g. drafts used by both the latest-publish and history panels). +* A single place to inspect or clear all MEDM state. +* Refresh semantics are explicit: :meth:`clear_on_refresh` drops only the + data that can change between user-triggered refreshes; immutable data + (thumbnail images, publish-type mappings) is preserved. +""" + +from __future__ import annotations + +import dataclasses +from typing import Any, Dict, List, Optional + + +@dataclasses.dataclass +class MedmSharedCache: + """ + Central store for all dictionaries shared across MEDM models. + + All values default to empty dicts so the object can be created with + ``MedmSharedCache()`` and immediately passed to model constructors. + + Cache semantics + --------------- + ``children`` + ``asset.id → list[Asset]``. Populated by both :class:`MedmEntityModel` + (when the user expands a tree node) and :class:`MedmLatestPublishModel` + (when the user selects a node that hasn't been expanded yet). Cleared + on *hard* refresh only, since a new tree load invalidates the hierarchy. + + ``drafts`` + ``asset.id → list[DraftInfo]``. Volatile — cleared on every refresh + so local-draft state is always up-to-date. + + ``versions`` + ``asset.id → list[AssetVersion]``. Cleared on refresh; a publish + action could add a new version. + + ``publish_types`` + ``medm_type_id_str → (sg_id, display_name)``. Schema-derived and + stable for a session; never cleared. + + ``thumbnail_urls`` + ``revision_id → URL | None``. Immutable (a revision's thumbnail URL + never changes); never cleared. + + ``thumbnail_data`` + ``URL → bytes``. Immutable pixel data; never cleared. + """ + + # asset.id → list[Asset] + children: Dict[str, List[Any]] = dataclasses.field(default_factory=dict) + # asset.id → list[DraftInfo] + drafts: Dict[str, List[Any]] = dataclasses.field(default_factory=dict) + # asset.id → list[AssetVersion] + versions: Dict[str, List[Any]] = dataclasses.field(default_factory=dict) + # medm_type_id_str → (sg_publish_type_id, display_name) + publish_types: Dict[str, tuple] = dataclasses.field(default_factory=dict) + # revision_id → thumbnail URL (or None when the API returned nothing) + thumbnail_urls: Dict[str, Optional[str]] = dataclasses.field(default_factory=dict) + # URL → raw image bytes + thumbnail_data: Dict[str, bytes] = dataclasses.field(default_factory=dict) + + # ------------------------------------------------------------------------- + # Convenience clear helpers + # ------------------------------------------------------------------------- + + def clear_on_refresh(self) -> None: + """ + Clear caches that may be stale after a user-triggered refresh. + + Deliberately preserves: + * ``children`` — tree structure is still valid. + * ``publish_types`` — ShotGrid types don't change during a session. + * ``thumbnail_urls`` — revision IDs are immutable. + * ``thumbnail_data`` — pixel data for a URL never changes. + """ + self.drafts.clear() + self.versions.clear() + + def clear_on_hard_refresh(self) -> None: + """ + Full reset used when the tree is reloaded from scratch. + + Still preserves thumbnail caches because those are purely content- + addressed (URL → bytes) and revision IDs are immutable. + """ + self.children.clear() + self.drafts.clear() + self.versions.clear() + self.publish_types.clear() diff --git a/python/tk_multi_loader/medm/thumbnail_service.py b/python/tk_multi_loader/medm/thumbnail_service.py new file mode 100644 index 0000000..e03acb3 --- /dev/null +++ b/python/tk_multi_loader/medm/thumbnail_service.py @@ -0,0 +1,202 @@ +# Copyright (c) 2026 Shotgun Software Inc. +# +# CONFIDENTIAL AND PROPRIETARY +# +# This work is provided "AS IS" and subject to the Shotgun Pipeline Toolkit +# Source Code License included in this distribution package. See LICENSE. +# By accessing, using, copying or modifying this work you indicate your +# agreement to the Shotgun Pipeline Toolkit Source Code License. All rights +# not expressly granted therein are reserved by Shotgun Software Inc. + +""" +MEDM Thumbnail Service - shared thumbnail URL resolver and image downloader. + +A single instance is created per dialog session and injected into every MEDM +model that needs thumbnails. Because MEDM revision IDs are immutable (a given +ID always resolves to the same URL and pixel data), both caches are kept for +the entire session and are never evicted on refresh. + +The URL and data caches are owned by the :class:`~medm.shared_cache.MedmSharedCache` +passed at construction time, so all MEDM state lives in one place. +""" + +from __future__ import annotations + +import socket +import threading +import urllib.error +import urllib.request +from typing import TYPE_CHECKING, Callable, Dict, Optional + +import sgtk +from sgtk.platform.qt import QtCore + +if TYPE_CHECKING: + from .shared_cache import MedmSharedCache + + +class MedmThumbnailService(QtCore.QObject): + """ + Session-scoped thumbnail URL resolver and image downloader shared across + all MEDM model instances (MedmLatestPublishModel, MedmPublishHistoryModel). + + The URL and image-data caches are provided externally via + :class:`~medm.shared_cache.MedmSharedCache` so all MEDM caches live in a + single, inspectable location. The service itself owns only the operational + state: a timer, a pending-work queue, and the background threads. + + Usage:: + + cache = MedmSharedCache() + service = MedmThumbnailService(cache, parent=dialog) + service.request(qt_item, revision_id, my_apply_fn) + # my_apply_fn(qt_item, image_data) is called on the main thread. + """ + + _CONNECTION_TIMEOUT = 5 # seconds - waiting for the server to respond + _READ_TIMEOUT = 10 # seconds - total socket timeout + _MAX_IMAGE_BYTES = 10 * 1024 * 1024 # 10 MB hard cap per thumbnail + + def __init__( + self, + cache: "MedmSharedCache", + parent: Optional[QtCore.QObject] = None, + ) -> None: + """ + :param cache: Shared cache whose ``thumbnail_urls`` and + ``thumbnail_data`` dicts are used for storage. The service never + clears these dicts; that responsibility belongs to the owner of the + cache (they are immutable by design). + :param parent: Optional Qt parent object. + """ + super().__init__(parent) + + self._app = sgtk.platform.current_bundle() + self._flow_module = sgtk.platform.import_framework("tk-framework-flowam", "flow") + + # Both dicts are references into the shared cache - not owned here. + self._url_cache: Dict[str, Optional[str]] = cache.thumbnail_urls + self._data_cache: Dict[str, bytes] = cache.thumbnail_data + + # item_id(int) → (qt_item, image_data, callback) + self._pending: Dict[int, tuple] = {} + self._timer: Optional[QtCore.QTimer] = None + + # ------------------------------------------------------------------------- + # Public API + # ------------------------------------------------------------------------- + + def request( + self, + qt_item, + revision_id: str, + callback: Callable, + ) -> None: + """ + Schedule a thumbnail download for *qt_item*. + + If both the URL and the pixel bytes are already cached the item is + queued for the next timer tick (no background thread is started). + Otherwise a daemon thread resolves the URL and downloads the image. + + :param qt_item: ``QStandardItem`` whose icon should be updated. + :param revision_id: MEDM ``AssetRevision`` ID to look up. + :param callback: ``callable(qt_item, image_data: bytes)`` that will be + invoked on the **main thread** once the image bytes are available. + The callback is responsible for converting bytes to a ``QPixmap`` + and applying it to *qt_item*. + """ + self._ensure_timer() + + # Fast path: both URL and image bytes already cached. + cached_url = self._url_cache.get(revision_id) + if cached_url and cached_url in self._data_cache: + self._pending[id(qt_item)] = (qt_item, self._data_cache[cached_url], callback) + return + + threading.Thread( + target=self._resolve_and_fetch, + args=(qt_item, revision_id, callback), + daemon=True, + ).start() + + def destroy(self) -> None: + """Stop the processing timer and discard all pending work.""" + if self._timer is not None: + self._timer.stop() + self._timer = None + self._pending.clear() + + # ------------------------------------------------------------------------- + # Private helpers + # ------------------------------------------------------------------------- + + def _ensure_timer(self) -> None: + """Start the main-thread processing timer on first use.""" + if self._timer is None: + self._timer = QtCore.QTimer(self) + self._timer.timeout.connect(self._process_pending) + self._timer.start(100) + + def _resolve_and_fetch( + self, qt_item, revision_id: str, callback: Callable + ) -> None: + """Background-thread worker: resolve URL then download bytes.""" + # 1. Resolve the thumbnail URL (hits the API at most once per revision). + url = self._url_cache.get(revision_id) + if url is None and revision_id not in self._url_cache: + try: + url = self._flow_module.asset_management.get_thumbnail_url(revision_id) + except Exception as exc: + self._app.log_debug( + f"MEDM ThumbnailService: URL resolve failed for {revision_id}: {exc}" + ) + self._url_cache[revision_id] = url + + if not url: + return + + # 2. Download image bytes (cached after the first download). + if url in self._data_cache: + self._pending[id(qt_item)] = (qt_item, self._data_cache[url], callback) + return + + try: + req = urllib.request.Request(url) + req.add_header("User-Agent", "ShotgunToolkit/1.0") + + old_timeout = socket.getdefaulttimeout() + try: + socket.setdefaulttimeout(self._CONNECTION_TIMEOUT + self._READ_TIMEOUT) + with urllib.request.urlopen(req, timeout=self._CONNECTION_TIMEOUT) as response: + if response.status != 200: + self._app.log_debug( + f"MEDM ThumbnailService: HTTP {response.status} for {revision_id}" + ) + return + image_data = response.read(self._MAX_IMAGE_BYTES) + if len(image_data) >= self._MAX_IMAGE_BYTES: + # Suspiciously large - skip rather than store garbage. + return + self._data_cache[url] = image_data + self._pending[id(qt_item)] = (qt_item, image_data, callback) + finally: + socket.setdefaulttimeout(old_timeout) + + except Exception as exc: + self._app.log_debug( + f"MEDM ThumbnailService: Download failed: {type(exc).__name__}: {exc}" + ) + + def _process_pending(self) -> None: + """Main-thread timer callback: fire stored callbacks for ready items.""" + pending = list(self._pending.items()) + self._pending.clear() + + for _item_id, (qt_item, image_data, callback) in pending: + try: + callback(qt_item, image_data) + except Exception as exc: + self._app.log_debug( + f"MEDM ThumbnailService: Callback error: {type(exc).__name__}: {exc}" + ) diff --git a/python/tk_multi_loader/medm/utils.py b/python/tk_multi_loader/medm/utils.py new file mode 100644 index 0000000..ba4212a --- /dev/null +++ b/python/tk_multi_loader/medm/utils.py @@ -0,0 +1,201 @@ +# Copyright (c) 2026 Shotgun Software Inc. +# +# CONFIDENTIAL AND PROPRIETARY +# +# This work is provided "AS IS" and subject to the Shotgun Pipeline Toolkit +# Source Code License included in this distribution package. See LICENSE. +# By accessing, using, copying or modifying this work you indicate your +# agreement to the Shotgun Pipeline Toolkit Source Code License. All rights +# not expressly granted therein are reserved by Shotgun Software Inc. + +"""Shared utility helpers for MEDM models. + +Functions here are intentionally free of Qt and SGTK framework imports so +they stay lightweight and easy to unit-test. +""" + +from __future__ import annotations + +import os +from typing import Any, Dict, Optional, Tuple + + +def is_structural_asset(asset: Any, flow_module: Any) -> bool: + """Return ``True`` when *asset* is a structural container in the MEDM hierarchy. + + An asset is considered structural — and therefore belongs only in the + left-hand tree, never in the centre-panel publish list — when **any** of + the following hold: + + * its type IDs include the built-in ``FOLDER_TYPE_ID``; **or** + * it carries a ``CONTAINER_TYPE`` or ``PIPELINE_STEP_TYPE`` component + (schema-registered organisational node types). + + All framework calls are wrapped in a broad ``except`` so that a transient + schema error never surfaces as a visible crash; the safe default is + ``False`` (treat the asset as publishable). + + :param asset: MEDM ``Asset`` object to test. + :param flow_module: The ``flow`` framework module imported via + ``sgtk.platform.import_framework("tk-framework-flowam", "flow")``. + :returns: ``True`` if the asset is a structural container. + """ + try: + type_ids = set(getattr(asset, "type_ids", None) or []) + if flow_module.data.FOLDER_TYPE_ID in type_ids: + return True + + _am = flow_module.asset_management + structural_types = (_am.CONTAINER_TYPE, _am.PIPELINE_STEP_TYPE) + return any( + asset.find_component(type_id=flow_module.schema.get_schema_id(ct)) + for ct in structural_types + ) + except Exception: + return False + + +# Sentinel version number that action hooks use to detect a local draft row +# and route it to asset_management.open_draft() instead of checkout_revision(). +# Both MEDM model classes expose this as a class constant with the same value. +DRAFT_VERSION_IDENTIFIER: int = -1 + + +def get_draft_created_at(draft_info: Any) -> Optional[float]: + """Return the creation timestamp for a local draft as a Unix float. + + Uses the mtime of the draft's ``source_path`` file on disk. Falls back + to ``None`` when the path is absent or the filesystem query fails. + + :param draft_info: A ``DraftInfo`` instance (``CheckoutDraftInfo`` or + ``NewDraftInfo``). + :returns: Unix timestamp float, or ``None`` if unavailable. + """ + source_path = getattr(draft_info, "source_path", None) + if source_path: + try: + return os.path.getmtime(source_path) + except OSError: + pass + return None + + +def get_draft_created_by(context: Any) -> Dict[str, Any]: + """Return a Shotgun-compatible ``created_by`` dict for the current user. + + Local drafts are always owned by the currently logged-in user, so the + SGTK context is the authoritative source. + + :param context: A ``sgtk.Context`` object (``self._app.context``). + :returns: Dict with ``type``, ``id``, and ``name`` keys. + """ + ctx_user = getattr(context, "user", None) or {} + return { + "type": "HumanUser", + "id": ctx_user.get("id", 0), + "name": ctx_user.get("name", ""), + } + + +def get_draft_metadata( + draft_info: Any, context: Any +) -> Tuple[Optional[float], Dict[str, Any]]: + """Convenience wrapper returning ``(created_at, created_by)`` for a draft. + + Combines :func:`get_draft_created_at` and :func:`get_draft_created_by` + into a single call for the common case where both values are needed + together. + + :param draft_info: A ``DraftInfo`` instance. + :param context: A ``sgtk.Context`` object. + :returns: Tuple of ``(created_at_float_or_none, created_by_dict)``. + """ + return get_draft_created_at(draft_info), get_draft_created_by(context) + + +def get_draft_description(draft_info: Any) -> str: + """Build a human-readable description string for a local draft. + + :param draft_info: A ``DraftInfo`` instance. + :returns: Non-empty description string. + """ + draft_type = getattr(draft_info, "draft_type", "unknown") + if draft_type == "checkout": + source_version = getattr(draft_info, "version", "?") + return f"Local checkout of v{source_version}" + if draft_type == "new": + return getattr(draft_info, "description", "") or "New unpublished asset" + return f"Local draft ({draft_type})" + + +def build_draft_sg_dict( + draft_info: Any, + asset: Any, + context: Any, + project_id: int, + project_name: str, + sg_publish_type_id: Optional[int], + sg_publish_type_code: str, +) -> Dict[str, Any]: + """Build the common Shotgun-compatible dictionary for a local draft. + + Returns the intersection of fields used by both the latest-publish model + and the publish-history model. Callers should merge in any additional + model-specific keys after calling this function:: + + sg_dict = build_draft_sg_dict(...) + sg_dict.update({"_medm_version": None}) # history model + sg_dict.update({ # latest-publish model + "task_uniqueness": True, + "tank_type": {...}, + "_thumbnail_revision_id": ..., + }) + + :param draft_info: ``DraftInfo`` instance (``CheckoutDraftInfo`` or + ``NewDraftInfo``). + :param asset: MEDM ``Asset`` the draft belongs to. May be ``None`` for + ``NewDraftInfo`` entries surfaced by the history model. + :param context: ``sgtk.Context`` used to resolve the current user. + :param project_id: ShotGrid project ID. + :param project_name: ShotGrid project name. + :param sg_publish_type_id: Resolved SG ``PublishedFileType`` ID (may be + ``None`` when no matching SG record exists). + :param sg_publish_type_code: Human-readable type display name. + :returns: sg_data dict ready for Qt model consumption. + """ + draft_name = getattr(draft_info, "name", getattr(asset, "name", "")) + draft_type = getattr(draft_info, "draft_type", "unknown") + created_at, created_by = get_draft_metadata(draft_info, context) + + return { + "id": None, + "type": "PublishedFile", + "code": f"{draft_name}.DRAFT", + "name": draft_name, + # DRAFT_VERSION_IDENTIFIER (-1) signals action hooks to call + # open_draft() instead of checkout_revision(). + "version_number": DRAFT_VERSION_IDENTIFIER, + "description": get_draft_description(draft_info), + "created_at": created_at, + "created_by": created_by, + "entity": ( + {"type": "Asset", "id": None, "name": asset.name} + if asset is not None + else None + ), + "project": {"type": "Project", "id": project_id, "name": project_name}, + "task": None, + "published_file_type": { + "type": "PublishedFileType", + "id": sg_publish_type_id, + "name": sg_publish_type_code, + }, + "image": None, + # draft_id is the sandbox ID consumed by asset_management.open_draft() + # and sandbox.is_local_draft(). + "sg_flow_revision_id": getattr(draft_info, "draft_id", None), + "path": {"local_path": getattr(draft_info, "source_path", None)}, + "_medm_asset": asset, + "_medm_draft": draft_info, + "_medm_draft_type": draft_type, + } From e96615e460d555b01f2d1fe66003d0d152bf5ccb Mon Sep 17 00:00:00 2001 From: Carlos Villavicencio Date: Fri, 22 May 2026 17:09:38 -0500 Subject: [PATCH 2/4] Apply black --- python/tk_multi_loader/medm/entity_model.py | 16 ++++-- .../medm/latestpublish_model.py | 53 +++++++++++++------ .../medm/publishhistory_model.py | 37 +++++++++---- .../tk_multi_loader/medm/thumbnail_service.py | 18 ++++--- 4 files changed, 87 insertions(+), 37 deletions(-) diff --git a/python/tk_multi_loader/medm/entity_model.py b/python/tk_multi_loader/medm/entity_model.py index e8c294f..9c731a3 100644 --- a/python/tk_multi_loader/medm/entity_model.py +++ b/python/tk_multi_loader/medm/entity_model.py @@ -51,7 +51,9 @@ class MedmEntityModel(QtGui.QStandardItemModel): # Custom roles - matching ShotgunModel interface SG_DATA_ROLE = QtCore.Qt.UserRole + 1 SG_ASSOCIATED_FIELD_ROLE = QtCore.Qt.UserRole + 2 - ASSET_ROLE = QtCore.Qt.UserRole + 200 # Stores MEDM Asset object (shared with all MEDM models) + ASSET_ROLE = ( + QtCore.Qt.UserRole + 200 + ) # Stores MEDM Asset object (shared with all MEDM models) # Lazy-loading bookkeeping role: True once children have been fetched for a node. CHILDREN_LOADED_ROLE = QtCore.Qt.UserRole + 201 @@ -88,7 +90,9 @@ def __init__( super().__init__(parent) self._app = sgtk.platform.current_bundle() - self._flow_module = sgtk.platform.import_framework("tk-framework-flowam", "flow") + self._flow_module = sgtk.platform.import_framework( + "tk-framework-flowam", "flow" + ) self._cache = cache if cache is not None else MedmSharedCache() self._folder_icon = QtGui.QIcon(QtGui.QPixmap(":/res/icon_Folder.png")) @@ -229,7 +233,9 @@ def _initialize_project(self) -> None: try: session_project = self._flow_module.data.get_session_project() self._project = self._flow_module.data.Project(session_project.id) - self._app.log_debug(f"MEDM Entity: Initialized project '{self._project.name}'") + self._app.log_debug( + f"MEDM Entity: Initialized project '{self._project.name}'" + ) except Exception as e: self._app.log_error( f"MEDM Entity: Failed to initialize project: {type(e).__name__}: {e}. " @@ -329,7 +335,9 @@ def _load_medm_assets(self): Called asynchronously after a short delay to show the loading spinner. """ if self._project is None: - self._app.log_warning("MEDM Entity: Cannot load assets - project not initialized") + self._app.log_warning( + "MEDM Entity: Cannot load assets - project not initialized" + ) self.data_refresh_fail.emit("Project not initialized") return diff --git a/python/tk_multi_loader/medm/latestpublish_model.py b/python/tk_multi_loader/medm/latestpublish_model.py index 4eae730..67eb4cd 100644 --- a/python/tk_multi_loader/medm/latestpublish_model.py +++ b/python/tk_multi_loader/medm/latestpublish_model.py @@ -62,8 +62,12 @@ class MedmLatestPublishModel(QtGui.QStandardItemModel): # Additional MEDM-specific roles SG_DATA_ROLE = QtCore.Qt.UserRole + 1 # To maintain compatibility with ShotgunModel SG_ASSOCIATED_FIELD_ROLE = QtCore.Qt.UserRole + 2 - ASSET_ROLE = QtCore.Qt.UserRole + 200 # Stores MEDM Asset object (shared with all MEDM models) - DRAFT_ROLE = QtCore.Qt.UserRole + 202 # Stores DraftInfo for draft rows (shared with history model) + ASSET_ROLE = ( + QtCore.Qt.UserRole + 200 + ) # Stores MEDM Asset object (shared with all MEDM models) + DRAFT_ROLE = ( + QtCore.Qt.UserRole + 202 + ) # Stores DraftInfo for draft rows (shared with history model) # Signals loadingStarted = QtCore.Signal() @@ -100,12 +104,16 @@ def __init__( super().__init__(parent) self._app = sgtk.platform.current_bundle() - self._flow_module = sgtk.platform.import_framework("tk-framework-flowam", "flow") + self._flow_module = sgtk.platform.import_framework( + "tk-framework-flowam", "flow" + ) self._publish_type_model = publish_type_model self._bg_task_manager = bg_task_manager self._cache = cache if cache is not None else MedmSharedCache() - self._thumbnail_service = thumbnail_service or MedmThumbnailService(self._cache, self) + self._thumbnail_service = thumbnail_service or MedmThumbnailService( + self._cache, self + ) self._owns_thumbnail_service = thumbnail_service is None self._loading_icon = QtGui.QIcon.fromTheme("view-refresh") @@ -281,7 +289,10 @@ def _populate_model_from_selected_item(self, selected_item): published_count += 1 # Surface drafts for the leaf-fallback asset. - if leaf_asset_fallback is not None and leaf_asset_fallback.id in drafts_by_asset_id: + if ( + leaf_asset_fallback is not None + and leaf_asset_fallback.id in drafts_by_asset_id + ): for draft_sg_dict in drafts_by_asset_id[leaf_asset_fallback.id]: self._add_sg_dict_as_qt_item(draft_sg_dict) draft_count += 1 @@ -389,9 +400,17 @@ def _asset_to_sg_dict(self, asset: Asset) -> Dict[str, Any]: "version_number": asset.version_number, "description": asset.description or "", "created_at": asset.created_at, - "created_by": {"type": "HumanUser", "id": 1, "name": asset.created_by or "MEDM User"}, + "created_by": { + "type": "HumanUser", + "id": 1, + "name": asset.created_by or "MEDM User", + }, "entity": {"type": "Asset", "id": None, "name": asset.name}, - "project": {"type": "Project", "id": self._project_id, "name": self._project_name}, + "project": { + "type": "Project", + "id": self._project_id, + "name": self._project_name, + }, "task": None, "task_uniqueness": True, "published_file_type": { @@ -443,13 +462,17 @@ def _draft_to_sg_dict( if not type_ids: type_ids = getattr(draft_info, "type_ids", None) or [] if type_ids: - sg_publish_type_id, sg_publish_type_code = self._resolve_publish_type(type_ids[0]) + sg_publish_type_id, sg_publish_type_code = self._resolve_publish_type( + type_ids[0] + ) # For checkout drafts the parent asset's revision_id provides the # thumbnail; for new drafts there is no published revision. draft_type = getattr(draft_info, "draft_type", "unknown") thumb_revision_id = ( - asset.revision_id if (draft_type == "checkout" and asset is not None) else None + asset.revision_id + if (draft_type == "checkout" and asset is not None) + else None ) sg_dict = build_draft_sg_dict( @@ -541,7 +564,9 @@ def _resolve_publish_type(self, medm_type_id_str: str) -> tuple: display_name = medm_type_id_str try: - schema_name = self._flow_module.schema.get_schema_display_name(medm_type_id_str) + schema_name = self._flow_module.schema.get_schema_display_name( + medm_type_id_str + ) if schema_name: display_name = schema_name except Exception as e: @@ -651,9 +676,7 @@ def _set_tooltip(self, item: QtGui.QStandardItem, sg_item: Dict[str, Any]): date_str = str(sg_item.get("created_at", "Unknown")) tooltip += f"

Version: {vers_str} by {author_str} at {date_str}" - tooltip += ( - f"

Description: {sg_item.get('description', 'No description given.')}" - ) + tooltip += f"

Description: {sg_item.get('description', 'No description given.')}" item.setToolTip(tooltip) @@ -670,9 +693,7 @@ def _resolve_and_download_thumbnail( """ self._thumbnail_service.request(qt_item, revision_id, self._apply_thumbnail) - def _apply_thumbnail( - self, qt_item: QtGui.QStandardItem, image_data: bytes - ) -> None: + def _apply_thumbnail(self, qt_item: QtGui.QStandardItem, image_data: bytes) -> None: """ Apply downloaded image bytes as a scaled icon on *qt_item*. Called on the main thread by :class:`MedmThumbnailService`. diff --git a/python/tk_multi_loader/medm/publishhistory_model.py b/python/tk_multi_loader/medm/publishhistory_model.py index be18c45..55ebd53 100644 --- a/python/tk_multi_loader/medm/publishhistory_model.py +++ b/python/tk_multi_loader/medm/publishhistory_model.py @@ -62,7 +62,9 @@ class MedmPublishHistoryModel(QtGui.QStandardItemModel): # MEDM-specific roles SG_DATA_ROLE = QtCore.Qt.UserRole + 1 # To maintain compatibility with ShotgunModel - ASSET_ROLE = QtCore.Qt.UserRole + 200 # Stores MEDM Asset object (shared with all MEDM models) + ASSET_ROLE = ( + QtCore.Qt.UserRole + 200 + ) # Stores MEDM Asset object (shared with all MEDM models) VERSION_ROLE = QtCore.Qt.UserRole + 201 # Stores MEDM AssetVersion object DRAFT_ROLE = QtCore.Qt.UserRole + 202 # Stores DraftInfo for draft rows @@ -95,11 +97,15 @@ def __init__( super().__init__(parent) self._app = sgtk.platform.current_bundle() - self._flow_module = sgtk.platform.import_framework("tk-framework-flowam", "flow") + self._flow_module = sgtk.platform.import_framework( + "tk-framework-flowam", "flow" + ) self._bg_task_manager = bg_task_manager self._cache = cache if cache is not None else MedmSharedCache() - self._thumbnail_service = thumbnail_service or MedmThumbnailService(self._cache, self) + self._thumbnail_service = thumbnail_service or MedmThumbnailService( + self._cache, self + ) self._owns_thumbnail_service = thumbnail_service is None self._loading_icon = QtGui.QPixmap(":/res/loading_100x100.png") self._current_sg_data = None @@ -137,7 +143,10 @@ def load_data(self, sg_data: Dict[str, Any]) -> None: # A NewDraftInfo card has no published asset yet. Show the draft # itself as the sole history entry so the user can still open it. draft_info = sg_data.get("_medm_draft") - if draft_info is not None and getattr(draft_info, "draft_type", None) == "new": + if ( + draft_info is not None + and getattr(draft_info, "draft_type", None) == "new" + ): self._add_draft_as_qt_item(draft_info, asset=None) self.cache_loaded.emit() self.data_refreshed.emit(True) @@ -159,13 +168,17 @@ def load_data(self, sg_data: Dict[str, Any]) -> None: for asset_version in versions: self._add_version_as_qt_item(asset_version, medm_asset) - self._app.log_debug(f"MEDM History: Loaded {len(versions)} published versions") + self._app.log_debug( + f"MEDM History: Loaded {len(versions)} published versions" + ) try: if asset_id in self._cache.drafts: drafts = self._cache.drafts[asset_id] else: - drafts = self._flow_module.asset_management.get_asset_drafts(asset_id) + drafts = self._flow_module.asset_management.get_asset_drafts( + asset_id + ) self._cache.drafts[asset_id] = drafts for draft_info in drafts: self._add_draft_as_qt_item(draft_info, medm_asset) @@ -345,7 +358,9 @@ def _draft_to_sg_dict( if not type_ids: type_ids = getattr(draft_info, "type_ids", None) or [] if type_ids: - sg_publish_type_id, sg_publish_type_code = self._resolve_publish_type(type_ids[0]) + sg_publish_type_id, sg_publish_type_code = self._resolve_publish_type( + type_ids[0] + ) sg_dict = build_draft_sg_dict( draft_info, @@ -425,7 +440,9 @@ def _resolve_publish_type(self, medm_type_id_str: str) -> tuple: display_name = medm_type_id_str try: - schema_name = self._flow_module.schema.get_schema_display_name(medm_type_id_str) + schema_name = self._flow_module.schema.get_schema_display_name( + medm_type_id_str + ) if schema_name: display_name = schema_name except Exception as e: @@ -473,9 +490,7 @@ def _resolve_and_download_thumbnail( """ self._thumbnail_service.request(qt_item, revision_id, self._apply_thumbnail) - def _apply_thumbnail( - self, qt_item: QtGui.QStandardItem, image_data: bytes - ) -> None: + def _apply_thumbnail(self, qt_item: QtGui.QStandardItem, image_data: bytes) -> None: """ Apply downloaded image bytes to *qt_item* as a composite publish thumbnail. Called on the main thread by :class:`MedmThumbnailService`. diff --git a/python/tk_multi_loader/medm/thumbnail_service.py b/python/tk_multi_loader/medm/thumbnail_service.py index e03acb3..344c6da 100644 --- a/python/tk_multi_loader/medm/thumbnail_service.py +++ b/python/tk_multi_loader/medm/thumbnail_service.py @@ -72,7 +72,9 @@ def __init__( super().__init__(parent) self._app = sgtk.platform.current_bundle() - self._flow_module = sgtk.platform.import_framework("tk-framework-flowam", "flow") + self._flow_module = sgtk.platform.import_framework( + "tk-framework-flowam", "flow" + ) # Both dicts are references into the shared cache - not owned here. self._url_cache: Dict[str, Optional[str]] = cache.thumbnail_urls @@ -111,7 +113,11 @@ def request( # Fast path: both URL and image bytes already cached. cached_url = self._url_cache.get(revision_id) if cached_url and cached_url in self._data_cache: - self._pending[id(qt_item)] = (qt_item, self._data_cache[cached_url], callback) + self._pending[id(qt_item)] = ( + qt_item, + self._data_cache[cached_url], + callback, + ) return threading.Thread( @@ -138,9 +144,7 @@ def _ensure_timer(self) -> None: self._timer.timeout.connect(self._process_pending) self._timer.start(100) - def _resolve_and_fetch( - self, qt_item, revision_id: str, callback: Callable - ) -> None: + def _resolve_and_fetch(self, qt_item, revision_id: str, callback: Callable) -> None: """Background-thread worker: resolve URL then download bytes.""" # 1. Resolve the thumbnail URL (hits the API at most once per revision). url = self._url_cache.get(revision_id) @@ -168,7 +172,9 @@ def _resolve_and_fetch( old_timeout = socket.getdefaulttimeout() try: socket.setdefaulttimeout(self._CONNECTION_TIMEOUT + self._READ_TIMEOUT) - with urllib.request.urlopen(req, timeout=self._CONNECTION_TIMEOUT) as response: + with urllib.request.urlopen( + req, timeout=self._CONNECTION_TIMEOUT + ) as response: if response.status != 200: self._app.log_debug( f"MEDM ThumbnailService: HTTP {response.status} for {revision_id}" From de8d721fa109d016df811eced5505dca68a2ab17 Mon Sep 17 00:00:00 2001 From: Carlos Villavicencio Date: Fri, 22 May 2026 17:15:22 -0500 Subject: [PATCH 3/4] Apply black - pre-commit version --- python/tk_multi_loader/medm/publishhistory_model.py | 1 - 1 file changed, 1 deletion(-) diff --git a/python/tk_multi_loader/medm/publishhistory_model.py b/python/tk_multi_loader/medm/publishhistory_model.py index 55ebd53..3e2d51d 100644 --- a/python/tk_multi_loader/medm/publishhistory_model.py +++ b/python/tk_multi_loader/medm/publishhistory_model.py @@ -27,7 +27,6 @@ from .thumbnail_service import MedmThumbnailService from .utils import build_draft_sg_dict - if TYPE_CHECKING: from adsk.flow.am import ( Asset, From 5a4909a8c04892bd1beb14b0e50be3be0a28e372 Mon Sep 17 00:00:00 2001 From: Carlos Villavicencio Date: Fri, 29 May 2026 10:53:41 -0500 Subject: [PATCH 4/4] Add remaining types --- python/tk_multi_loader/medm/entity_model.py | 18 +++++++++-------- .../medm/latestpublish_model.py | 20 +++++++++++-------- .../medm/publishhistory_model.py | 8 ++++---- .../tk_multi_loader/medm/thumbnail_service.py | 9 +++++---- 4 files changed, 31 insertions(+), 24 deletions(-) diff --git a/python/tk_multi_loader/medm/entity_model.py b/python/tk_multi_loader/medm/entity_model.py index 9c731a3..93c9ac5 100644 --- a/python/tk_multi_loader/medm/entity_model.py +++ b/python/tk_multi_loader/medm/entity_model.py @@ -114,7 +114,7 @@ def __init__( # Qt virtual overrides - lazy loading protocol # ------------------------------------------------------------------------- - def hasChildren(self, parent=QtCore.QModelIndex()): + def hasChildren(self, parent: QtCore.QModelIndex = QtCore.QModelIndex()) -> bool: """ Return ``True`` when *parent* might have children. @@ -132,7 +132,7 @@ def hasChildren(self, parent=QtCore.QModelIndex()): # Not yet loaded -> assume children exist (shows the expand arrow) return True - def canFetchMore(self, parent): + def canFetchMore(self, parent: QtCore.QModelIndex) -> bool: """Return ``True`` if *parent*'s children have not been loaded yet.""" if not parent.isValid(): return False @@ -141,7 +141,7 @@ def canFetchMore(self, parent): return False return not item.data(self.CHILDREN_LOADED_ROLE) - def fetchMore(self, parent): + def fetchMore(self, parent: QtCore.QModelIndex) -> None: """Load the immediate children of *parent* from the MEDM API (or cache).""" if not parent.isValid(): return @@ -154,22 +154,24 @@ def fetchMore(self, parent): # Public API - Called by dialog.py and other external code # ------------------------------------------------------------------------- - def destroy(self): + def destroy(self) -> None: """Clean up model resources.""" self._cache.children.clear() - def async_refresh(self): + def async_refresh(self) -> None: """Refresh the model data.""" self.clear() self._cache.clear_on_hard_refresh() self.data_refreshing.emit() QtCore.QTimer.singleShot(100, self._load_medm_assets) - def hard_refresh(self): + def hard_refresh(self) -> None: """Hard refresh (same as async_refresh for this simple model).""" self.async_refresh() - def item_from_entity(self, entity_type: str, entity_id: int): + def item_from_entity( + self, entity_type: str, entity_id: int + ) -> Optional[QtGui.QStandardItem]: """ Returns a QStandardItem based on entity type and entity id. @@ -329,7 +331,7 @@ def _icon_for_asset(self, asset: Asset) -> QtGui.QIcon: else self._binary_icon ) - def _load_medm_assets(self): + def _load_medm_assets(self) -> None: """ Load the first level of MEDM assets (project's immediate children). Called asynchronously after a short delay to show the loading spinner. diff --git a/python/tk_multi_loader/medm/latestpublish_model.py b/python/tk_multi_loader/medm/latestpublish_model.py index 67eb4cd..2fd2357 100644 --- a/python/tk_multi_loader/medm/latestpublish_model.py +++ b/python/tk_multi_loader/medm/latestpublish_model.py @@ -129,13 +129,13 @@ def __init__( # Public API - Called by dialog.py and other external code # ------------------------------------------------------------------------- - def destroy(self): + def destroy(self) -> None: """Clean up model resources.""" self._cache.drafts.clear() if self._owns_thumbnail_service: self._thumbnail_service.destroy() - def load_data(self, item): + def load_data(self, item: Optional[QtGui.QStandardItem]) -> None: """ Clears the model and sets it up for the selected asset from left treeview panel. Loads data from MEDM instead of Shotgun. @@ -158,7 +158,7 @@ def load_data(self, item): except Exception as e: self.loadingError.emit(str(e)) - def async_refresh(self): + def async_refresh(self) -> None: """ Refresh the current data set. @@ -187,7 +187,9 @@ def _initialize_project_info(self) -> None: "initialized or the project ID is invalid." ) - def _populate_model_from_selected_item(self, selected_item): + def _populate_model_from_selected_item( + self, selected_item: Optional[QtGui.QStandardItem] + ) -> None: """ Populate the model with latest versions of child assets from the selected tree item. @@ -310,7 +312,9 @@ def _populate_model_from_selected_item(self, selected_item): sg_publish_type_counts = self._calculate_sg_publish_type_counts() self._publish_type_model.set_active_types(sg_publish_type_counts) - def _extract_asset_from_tree_item(self, item) -> Optional[Asset]: + def _extract_asset_from_tree_item( + self, item: QtGui.QStandardItem + ) -> Optional[Asset]: """ Extract the MEDM Asset object from a tree view QStandardItem. @@ -601,7 +605,7 @@ def _resolve_publish_type(self, medm_type_id_str: str) -> tuple: self._cache.publish_types[medm_type_id_str] = result return result - def _add_sg_dict_as_qt_item(self, sg_item: Dict[str, Any]): + def _add_sg_dict_as_qt_item(self, sg_item: Dict[str, Any]) -> None: """ Create a QStandardItem from a Shotgun-compatible dict and add it to the model. @@ -652,7 +656,7 @@ def get_sg_data(): f"MEDM: Added item '{qt_item.text()}' to model (row count: {self.rowCount()})" ) - def _set_tooltip(self, item: QtGui.QStandardItem, sg_item: Dict[str, Any]): + def _set_tooltip(self, item: QtGui.QStandardItem, sg_item: Dict[str, Any]) -> None: """ Sets a tooltip for a publish item. @@ -682,7 +686,7 @@ def _set_tooltip(self, item: QtGui.QStandardItem, sg_item: Dict[str, Any]): def _resolve_and_download_thumbnail( self, qt_item: QtGui.QStandardItem, revision_id: str - ): + ) -> None: """ Delegate thumbnail resolution and download to the shared :class:`MedmThumbnailService`. The service calls :meth:`_apply_thumbnail` diff --git a/python/tk_multi_loader/medm/publishhistory_model.py b/python/tk_multi_loader/medm/publishhistory_model.py index 3e2d51d..ee735ee 100644 --- a/python/tk_multi_loader/medm/publishhistory_model.py +++ b/python/tk_multi_loader/medm/publishhistory_model.py @@ -117,7 +117,7 @@ def __init__( # Public API - Called by dialog.py and other external code # ------------------------------------------------------------------------- - def destroy(self): + def destroy(self) -> None: """Clean up model resources.""" self._cache.drafts.clear() self._cache.versions.clear() @@ -197,7 +197,7 @@ def load_data(self, sg_data: Dict[str, Any]) -> None: self._app.log_error(f"MEDM History: Error loading versions: {e}") self.data_refresh_fail.emit(str(e)) - def async_refresh(self): + def async_refresh(self) -> None: """ Refresh the current data set. @@ -210,7 +210,7 @@ def async_refresh(self): self.load_data(self._current_sg_data) self.data_refreshed.emit(True) - def hard_refresh(self): + def hard_refresh(self) -> None: """Force refresh of data (same as async_refresh for MEDM).""" self.async_refresh() @@ -478,7 +478,7 @@ def _resolve_publish_type(self, medm_type_id_str: str) -> tuple: def _resolve_and_download_thumbnail( self, qt_item: QtGui.QStandardItem, revision_id: str - ): + ) -> None: """ Delegate thumbnail resolution and download to the shared :class:`MedmThumbnailService`. The service calls :meth:`_apply_thumbnail` diff --git a/python/tk_multi_loader/medm/thumbnail_service.py b/python/tk_multi_loader/medm/thumbnail_service.py index 344c6da..40a43e7 100644 --- a/python/tk_multi_loader/medm/thumbnail_service.py +++ b/python/tk_multi_loader/medm/thumbnail_service.py @@ -24,12 +24,11 @@ import socket import threading -import urllib.error import urllib.request from typing import TYPE_CHECKING, Callable, Dict, Optional import sgtk -from sgtk.platform.qt import QtCore +from sgtk.platform.qt import QtCore, QtGui if TYPE_CHECKING: from .shared_cache import MedmSharedCache @@ -90,7 +89,7 @@ def __init__( def request( self, - qt_item, + qt_item: QtGui.QStandardItem, revision_id: str, callback: Callable, ) -> None: @@ -144,7 +143,9 @@ def _ensure_timer(self) -> None: self._timer.timeout.connect(self._process_pending) self._timer.start(100) - def _resolve_and_fetch(self, qt_item, revision_id: str, callback: Callable) -> None: + def _resolve_and_fetch( + self, qt_item: QtGui.QStandardItem, revision_id: str, callback: Callable + ) -> None: """Background-thread worker: resolve URL then download bytes.""" # 1. Resolve the thumbnail URL (hits the API at most once per revision). url = self._url_cache.get(revision_id)