Source code for engine

# Copyright (c) 2019 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.

import logging
import os
import re
import glob
import subprocess
import sys
import threading
import uuid

from contextlib import contextmanager


import sgtk
from sgtk.util.filesystem import ensure_folder_exists
from tank_vendor import six


[docs]class AfterEffectsEngine(sgtk.platform.Engine): """ An After Effects CC engine for Shotgun Toolkit. """ # the maximum size for a generated thumbnail MAX_THUMB_SIZE = 512 SHOTGUN_ADOBE_HEARTBEAT_INTERVAL = 1.0 SHOTGUN_ADOBE_HEARTBEAT_TOLERANCE = 2 SHOTGUN_ADOBE_NETWORK_DEBUG = "SHOTGUN_ADOBE_NETWORK_DEBUG" in os.environ TEST_SCRIPT_BASENAME = "run_tests.py" PY_TO_JS_LOG_LEVEL_MAPPING = { "CRITICAL": "error", "ERROR": "error", "WARNING": "warn", "INFO": "info", "DEBUG": "debug", } _SHOTGUN_ADOBE_PORT = os.environ.get("SHOTGUN_ADOBE_PORT") _SHOTGUN_ADOBE_APPID = os.environ.get("SHOTGUN_ADOBE_APPID") _COMMAND_UID_COUNTER = 0 _LOCK = threading.Lock() _FAILED_PINGS = 0 _CONTEXT_CACHE = dict() _CHECK_CONNECTION_TIMER = None _CONTEXT_CHANGES_DISABLED = False _DIALOG_PARENT = None _WIN32_AFTEREFFECTS_MAIN_HWND = None _PROXY_WIN_HWND = None _HEARTBEAT_DISABLED = False _PROJECT_CONTEXT = None _AFX_PID = None _POPUP_CACHE = None _AFX_WIN32_DIALOG_WINDOW_CLASS = "#32770" # the windows window class name used by After Effects for modal dialogs __WIN32_GW_CHILD = 5 _CONTEXT_CACHE_KEY = "aftereffects_context_cache" _HAS_CHECKED_CONTEXT_POST_LAUNCH = False __CC_VERSION_MAPPING = { 12: "2015", 13: "2016", 14: "2017", 15: "2018", 16: "2019", 17: "2020", 18: "2021", 22: "2022", 23: "2023", 24: "2024", } __IS_SEQUENCE_REGEX = re.compile("[\[]?([#@]+|[%]0\dd)[\]]?") ############################################################################ # context changing def post_context_change(self, old_context, new_context): """ Runs after a context change has occurred. This will trigger the new state to be sent to the Adobe CC host application. :param old_context: The previous context. :param new_context: The current context. """ # keep track of schema load for the current project to make sure we # aren't trying to use sg globals prior to load self.__schema_loaded = False # get the project id to supply to sg globals if new_context.project: project_id = new_context.project["id"] else: project_id = None # callback to set the schema loaded flag def _on_schema_loaded(): self.__schema_loaded = True # tell sg globals to load the schema for the current project. sg globals # will run the callback immediately if already cached so this is likely # very quick. self.__shotgun_globals.run_on_schema_loaded( _on_schema_loaded, project_id=project_id ) # go ahead and start the process of sending the current state back to js self.__send_state() # If the context is set in the environment, then we'll update it with # the new one. This will mean that a CEP extension restart will come # back up with the same context that it went down with. We have to set # this in ExtendScript, because it's the parent process of any CEP and # Python processes that get spawned. By setting it at the top, it'll be # propagated down to any of After Effects's subprocesses. if "TANK_CONTEXT" in os.environ: self.adobe.dollar.setenv("TANK_CONTEXT", new_context.serialize()) ############################################################################ # engine initialization def pre_app_init(self): """ Sets up the engine into an operational state. This method called before any apps are loaded. """ # import and keep a handle on the bundled python module self.__tk_aftereffects = self.import_module("tk_aftereffects") # constant command uid lookups for these special commands self.__jump_to_sg_command_id = self.__get_command_uid() self.__jump_to_fs_command_id = self.__get_command_uid() # get the adobe instance. it may have been initialized already by a # previous instance of the engine. if not, initialize a new one. self._adobe = self.__tk_aftereffects.AdobeBridge.get_or_create( identifier=self.instance_name, port=self._SHOTGUN_ADOBE_PORT, logger=self.logger, network_debug=self.SHOTGUN_ADOBE_NETWORK_DEBUG, ) self.logger.debug("Network debug logging is %s" % self._adobe.network_debug) self.logger.debug("%s: Initializing..." % (self,)) # connect to all the adobe bridge signals self.adobe.logging_received.connect(self._handle_logging) self.adobe.command_received.connect(self._handle_command) self.adobe.active_document_changed.connect(self._handle_active_document_change) self.adobe.run_tests_request_received.connect(self._run_tests) self.adobe.state_requested.connect(self.__send_state) # in order to use frameworks, they have to be imported via # import_module. so they're exposed in the bundled python. shotgun_data = self.__tk_aftereffects.shotgun_data settings = self.__tk_aftereffects.shotgun_settings # keep a handle for shotgun globals as they are needed in other # functions as well self.__shotgun_globals = self.__tk_aftereffects.shotgun_globals # import here since the engine is responsible for defining Qt. from sgtk.platform.qt import QtCore # create a data retriever for async querying of sg data self.__sg_data = shotgun_data.ShotgunDataRetriever( QtCore.QCoreApplication.instance() ) # get outselves a settings manager where we can store metadata. self.__settings_manager = settings.UserSettings(self) # connect the retriever signals self.__sg_data.work_completed.connect(self.__on_worker_signal) self.__sg_data.work_failure.connect(self.__on_worker_failure) # context request uids. we keep track of these to make sure we're only # processing the current requests. self.__context_find_uid = None self.__context_thumb_uid = None # keep track if sg global schema has been cached self.__schema_loaded = False # start the retriever thread self.__sg_data.start() # keep a list of handles on the launched dialogs self.__qt_dialogs = [] def post_app_init(self): """ Runs after all apps have been initialized. """ if not self.adobe.event_processor: try: from sgtk.platform.qt import QtGui self.adobe.event_processor = QtGui.QApplication.processEvents except ImportError: pass self.__setup_connection_timer() self.__send_state() # forward the log file path back to the js side. this is used to direct # clients to the file in the event of an error log_file = sgtk.LogManager().base_file_handler.baseFilename self.adobe.send_log_file_path(log_file) # If there are fewer than 2 documents open, we don't need the stored # cache, regardless of whether this is a restart situation or a fresh # launch of AE. In that case, we take the opportunity to clear anything # that might exist in the stored cache, as it's data we don't need. self.logger.debug("Single document found, clearing stored context cache.") self.__settings_manager.store( self._CONTEXT_CACHE_KEY, dict(), ) # Normally the bootstrap logic would handle the file open, but since the bootstrap logic is handled by # the adobe framework, and is generic, we should handle it here. file_to_open = os.environ.get("SGTK_FILE_TO_OPEN") if file_to_open: # open the specified script self.adobe.app.open(self.adobe.File(file_to_open)) # clear the environment variable after loading so that it doesn't get reopened on an engine restart. del os.environ["SGTK_FILE_TO_OPEN"] def destroy_engine(self): """ Called when the engine should tear down itself and all its apps. """ self.logger.debug("Destroying engine...") # Set our parent widget back to being owned by the window manager # instead of After Effects's application window. if self._PROXY_WIN_HWND and sgtk.util.is_windows(): self.__tk_aftereffects.win_32_api.SetParent(self._PROXY_WIN_HWND, 0) # No longer poll for new messages from this engine. if self._CHECK_CONNECTION_TIMER: self._CHECK_CONNECTION_TIMER.stop() # We're going to hide and force the garbage collection of any dialogs # that we know about. This will stop memory leaks, and is also prudent # since we're severing the socket.io connection that will allow them # to function properly. for dialog in self.__qt_dialogs: dialog.hide() dialog.setParent(None) dialog.deleteLater() # Gracefully stop our data retriever. This call will block until the # currently-processing request has completed. self.__sg_data.stop() # Disconnect from the server. self.adobe.disconnect() # Disconnect the signals in case there are references to this engine # out there. without disconnecting, it will still respond to signals # from the adobe bridge. self.adobe.logging_received.disconnect(self._handle_logging) self.adobe.command_received.disconnect(self._handle_command) self.adobe.active_document_changed.disconnect( self._handle_active_document_change ) self.adobe.run_tests_request_received.disconnect(self._run_tests) self.adobe.state_requested.disconnect(self.__send_state) def post_qt_init(self): """ Called externally once a ``QApplication`` has been created and completes the engine setup process. """ # We need to have the RPC API call processEvents during its response # wait loop. This will keep that loop from blocking the UI thread. from sgtk.platform.qt import QtGui self.adobe.event_processor = QtGui.QApplication.processEvents # Since this is running in our own Qt event loop, we'll use the bundled # dark look and feel. breaking encapsulation to do so. self.logger.debug("Initializing default styling...") self._initialize_dark_look_and_feel() # Sets up the heartbeat timer to run asynchronously. self.__setup_connection_timer(force=True) def register_command(self, name, callback, properties=None): """ Registers a new command with the engine. For Adobe RPC purposes, a "uid" property is added to the command's properties. """ properties = properties or dict() properties["uid"] = self.__get_command_uid() return super(AfterEffectsEngine, self).register_command( name, callback, properties, ) @property def host_info(self): """ Returns information about the application hosting this engine. :returns: A {"name": application name, "version": application version} dictionary. eg: {"name": "AfterFX", "version": "2017.1.1"} """ if not self.adobe: # Don't error out if the bridge was not yet started return {"name": "AfterFX", "version": "unknown"} version = self.adobe.app.version # app.aftereffects.AfterEffectsVersion just returns 18.1.1 which is not what users see in the UI # extract a more meaningful version from the systemInformation property # which gives something like: # Adobe After Effects Version: 2017.1.1 20170425.r.252 2017/04/25:23:00:00 CL 1113967 x64\rNumber of ..... # and use it instead if available. regex = re.compile(r"(\d+\.?\d*)") match = regex.search(six.ensure_str(version)) major = int(float(match[0])) cc_version = self.__CC_VERSION_MAPPING.get(major, version) return { "name": "AfterFX", "version": cc_version, } ############################################################################ # engine host interaction methods @property def project_path(self): """ Get the active project filepath. :returns: The current project path or an empty string :rtype: str """ doc_obj = self.adobe.app.project.file doc_path = "" # doc_obj will always be a ProxyWrapper instance so we cannot # use the `doc_obj is not None` comparison if doc_obj is not None: doc_path = doc_obj.fsName return doc_path
[docs] def save(self, path=None): """ Save the project in place or to the given file-path :param str path: The target file path. optional. """ with self.context_changes_disabled(): if path is None: self.adobe.app.project.save() else: # After Effects won't ensure that the folder is # created when saving, so we must make sure it exists ensure_folder_exists(os.path.dirname(path)) self.adobe.app.project.save(self.adobe.File(path)) new_path = self.project_path self.logger.info("Saved file to to {!r}".format(new_path))
[docs] def save_as(self): """ Launch a Qt file browser to select a file, then save the supplied project to that path. """ from sgtk.platform.qt import QtGui doc_path = self.project_path # After Effects doesn't appear to have a # "save as" dialog accessible via # python. so open our own Qt file dialog. file_dialog = QtGui.QFileDialog( parent=self._get_dialog_parent(), caption="Save As", directory=doc_path, filter="After Effects Documents (*.aep, *.aepx)", ) file_dialog.setLabelText(QtGui.QFileDialog.Accept, "Save") file_dialog.setLabelText(QtGui.QFileDialog.Reject, "Cancel") file_dialog.setOption(QtGui.QFileDialog.DontResolveSymlinks) file_dialog.setOption(QtGui.QFileDialog.DontUseNativeDialog) if not file_dialog.exec_(): return path = file_dialog.selectedFiles()[0] if path: self.save(path)
@property def AdobeItemTypes(self): """ This returns a constant class that maps constants against adobe aftereffects internal class names. :returns: AdobeItemTypes constants object. see :class:`~python.tk_aftereffects.AdobeItemTypes` :rtype: AdobeItemTypes """ afx_module = self.import_module("tk_aftereffects") return afx_module.AdobeItemTypes
[docs] def is_item_of_type(self, item, adobe_type): """ Indicates whether the given item is of the given AdobeItemType. :param item: item to be checked. :type item: `adobe.ItemObject`_ :param str adobe_type: AdobeItemType-constant. One of the constants held by :class:`~python.tk_aftereffects.AdobeItemTypes` :returns: Indicating if the given item is of the given type :rtype: bool """ return item.data.get("instanceof", "") == adobe_type
@property def selected_item(self): """ Helper getting the currently selected item in the after effects project :returns: The active item :rtype: `adobe.ItemObject`_ """ return self.adobe.app.project.activeItem
[docs] def is_adobe_sequence(self, path): """ Helper to query if an adobe-style render path is describing a sequence. The sequencepath must contain one of the following patterns in order to be recognized:: ### @@@ [###] %0xd :param str path: filepath to check :returns: True if the path describes a sequence :rtype: bool """ if re.search(self.__IS_SEQUENCE_REGEX, path): return True return False
[docs] def check_sequence(self, path, queue_item): """ Helper to query if all render files of a given queue item are actually existing. :param str path: filepath to check :param queue_item: an after effects render queue item :type queue_item: `adobe.RenderQueueItemObject`_ :returns: True if the path describes a sequence :rtype: bool """ for file_path, _ in self.get_render_files(path, queue_item): if not os.path.exists(file_path): return False return True
[docs] def iter_collection(self, collection_item): """ Helper to iter safely through an adobe-collection item as its index starts at 1, not 0. One may use this method like this:: item_collection = engine.adobe.project.items for item in engine.iter_collection(item_collection): print item.name :param collection_item: the after-effects collection item to iter :type collection_item: `adobe.ItemCollectionObject`_ :yields: the next child item of the collection :rtype: `adobe.ItemObject`_ """ for i in range(1, collection_item.length + 1): yield collection_item[i]
[docs] def get_render_files(self, path, queue_item): """ Yields all render-files and its frame number of a given after effects render queue item. The path is needed to uniquely identify the correct output_module :param str path: filepath to iter :param queue_item: an after effects render queue item :type queue_item: `adobe.RenderQueueItemObject`_ :yields: 2-item-tuple where the firstitem is the resolved path (str) of the render file and the second item the frame-number or None if the path is not an image-sequence. :rtype: tuple """ # is the given render-path a sequence? match = re.search(self.__IS_SEQUENCE_REGEX, path) if not match: # if not, we just check if the file exists yield path, None return # Exit the iterator # if yes, we check the existence of each frame frame_time = queue_item.comp.frameDuration start_time = int(round(queue_item.timeSpanStart / frame_time, 3)) frame_numbers = int(round(queue_item.timeSpanDuration / frame_time, 3)) skip_frames = queue_item.skipFrames + 1 padding = len(match.group(1)) test_path = path.replace(match.group(0), "%%0%dd" % padding) for frame_no in range(start_time, start_time + frame_numbers, skip_frames): yield test_path % frame_no, frame_no
[docs] def import_filepath(self, path): """ Helper method to import footage into the current comp. This method contains logic to determine as what the given file-path may be imported (PROJECT, FOOTAGE, COMP etc.) To influence this logic you can implement the "import_footage_hook" for this engine. It will also collect all new items and return those in case the import leads to more than one (1) Item-objects to be imported. :param str path: filepath to be imported. Sequences should give either the first frame or the frames replaced by hashes (#), at (@) or the percent formatting (%04d) syntax. :returns: All new items that were imported :rtype: list-of-`adobe.CompItemObject`_ """ file_obj = self.adobe.File(path) import_options = self.adobe.ImportOptions() import_options.file = file_obj import_options.sequence = False self.execute_hook_method( "import_footage_hook", "set_import_options", import_options=import_options ) return self.__import_file(import_options)
[docs] def add_items_to_comp(self, item_collection, comp_item): """ Helper method that adds the "best-matching" items from a given adobe.CollectionItem into a given CompItem. Where "best-matching" means, that the first CompItem found is added alternatively the first Footage item is added. In case the collection contains only FolderItems, this method will recurse into the folders until one ItemObject was found. :param item_collection: object to be analyzed :type item_collection: `adobe.ItemCollectionObject`_ :param comp_item: comp that should receive the items from item_collection :type comp_item: `adobe.CompItemObject`_ :returns: Indicating if an item was added or not. :rtype: bool """ # first we generate a list of items within # the given collection (FolderItem), because # we have different include-priorities per # type folders = [] comps = [] footage = [] for item in self.iter_collection(item_collection): if self.is_item_of_type(item, self.AdobeItemTypes.FOLDER_ITEM): folders.append(item) elif self.is_item_of_type(item, self.AdobeItemTypes.COMP_ITEM): comps.append(item) elif self.is_item_of_type(item, self.AdobeItemTypes.FOOTAGE_ITEM): footage.append(item) # if there is a comp in the given folder, # then we prefer to add this to the comp, # as it is likely to include the other layers added = False comp_and_footage = comps + footage while not added and comp_and_footage: comp_item = comp_item.layers.add(comp_and_footage.pop(0)) added = True # in case no footage was added, we recurse into # the folders until we find something or return None while not added and folders: folder = folders.pop(0) added = self.add_items_to_comp(folder.items, comp_item) return added
[docs] def render_queue_item(self, queue_item): """ renders a given queue_item, by disabling all other queued items and only enabling the given item. After rendering the state of the render-queue is reverted. :param queue_item: The render queue item to be rendered :type queue_item: `adobe.RenderQueueItemObject`_ :returns: Indicating successful rendering or not :rtype: bool """ # save the queue state for all unrendered items queue_item_state_cache = [(queue_item, queue_item.render)] for item in self.iter_collection(self.adobe.app.project.renderQueue.items): # one cannot change the status on if item.status != self.adobe.RQItemStatus.QUEUED: continue queue_item_state_cache.append((item, item.render)) item.render = False success = False queue_item.render = True self.logger.debug("Start rendering..") try: self.adobe.app.project.renderQueue.render() except Exception as e: # This catches errors thrown during the AfterEffects Render process. # Which may return various different errors. This situation never # occured during development. self.logger.error( ("Skipping item due to an error " "while rendering: {}").format(e) ) finally: acceptable_states = [ self.adobe.RQItemStatus.DONE, self.adobe.RQItemStatus.ERR_STOPPED, self.adobe.RQItemStatus.RENDERING, ] # reverting the original queued state for all # unprocessed items while queue_item_state_cache: item, status = queue_item_state_cache.pop(0) if item.status not in acceptable_states: item.render = status # we check for success if the render queue item status # has changed to DONE success = queue_item.status == self.adobe.RQItemStatus.DONE return success
[docs] def find_sequence_range(self, path): """ Parses the file name in an attempt to determine the first and last frame number of a sequence. This assumes some sort of common convention for the file names, where the frame number is an integer at the end of the basename, just ahead of the file extension, such as ``file.0001.jpg``, or ``file_001.jpg``. We also check for input file names with abstracted frame number tokens, such as ``file.####.jpg`` or ``file.%04d.jpg``. :param str path: The file path to parse. :returns: None if no range could be determined, otherwise (min, max) :rtype: tuple or None """ # This pattern will match the following at the end of a string and # retain the frame number or frame token as group(1) in the resulting # match object: # # 0001 # #### # @@@@ # [####] # %04d # # The number of digits or hashes does not matter; we match as many as # exist. frame_pattern = re.compile(r"(^|[\.\_\- ])[\[]?([0-9#@]+|[%]0\d+d)[\]]?$") root, ext = os.path.splitext(path) match = re.search(frame_pattern, root) # If we did not match, we don't know how to parse the file name, or there # is no frame number to extract. if not match: return None # We need to get all files that match the pattern from disk so that we # can determine what the min and max frame number is. glob_path = "%s%s" % ( re.sub(frame_pattern, r"\1*", root), ext, ) files = glob.glob(glob_path) # Our pattern from above matches against the file root, so we need # to chop off the extension at the end. file_roots = [os.path.splitext(f)[0] for f in files] # We know that the search will result in a match at this point, otherwise # the glob wouldn't have found the file. We can search and pull group 1 # to get the integer frame number from the file root name. frames = [int(re.search(frame_pattern, f).group(2)) for f in file_roots] return (min(frames), max(frames))
############################################################################ # RPC def _check_connection(self): """Make sure we are still connected to the adobe cc product.""" # If we're in a disabled state, then we don't do anything here. This # is controlled by the heartbeat_disabled context manager provided # by this engine. if self._HEARTBEAT_DISABLED: return try: self.adobe.ping() except Exception: if self._FAILED_PINGS >= self.SHOTGUN_ADOBE_HEARTBEAT_TOLERANCE: from sgtk.platform.qt import QtCore QtCore.QCoreApplication.instance().quit() else: self._FAILED_PINGS += 1 else: self._FAILED_PINGS = 0 # Will allow queued up messages (like logging calls) # to be handled on the Python end. self.adobe.process_new_messages() # We also have a one-time check we need to make after the timer is # started. In the event that the user opened a document before the # integration completed its initialization, we need to make sure # that the context is correct. Since we rely of After Effects sending # an event on active document change to control our context, if that # occurred before we were listening then we likely missed it and # need to make sure we're not in a stale state. # # Note: This is occurring in the timer here because the context # change can only occur once the engine initialization process has # completed. As such, we can rely on the timer to delay its execution # until we're in a state where the context change will succeed. if not self._HAS_CHECKED_CONTEXT_POST_LAUNCH: if sgtk.platform.current_engine(): self.logger.debug( "Engine initialization complete -- checking active " "document context..." ) active_document_path = self.project_path if active_document_path: self.logger.debug( "There is an active document. Checking to see if a " "context change is required." ) self._handle_active_document_change(active_document_path) else: self.logger.debug( "There is no active document, so there is no need to change context." ) self._HAS_CHECKED_CONTEXT_POST_LAUNCH = True else: self.logger.debug( "Engine initialization has not completed -- waiting to " "check the active document context..." ) def _emit_log_message(self, handler, record): """ Called by the engine whenever a new log message is available. All log messages from the toolkit logging namespace will be passed to this method. :param handler: Log handler that this message was dispatched from :type handler: :class:`~python.logging.LogHandler` :param record: Std python logging record :type record: :class:`~python.logging.LogRecord` """ # If the _adobe attribute is set, then we can forward logging calls # back to the js process via rpc. if hasattr(self, "_adobe"): level = self.PY_TO_JS_LOG_LEVEL_MAPPING[record.levelname] # log the message back to js via rpc self.adobe.log_message(level, record.message) # prior to the _adobe attribute being set, we rely on the js process # handling stdout and logging it. else: # we don't use the handler's format method here because the adobe # side expects a certain format. msg_str = "[%s]: %s" % (record.levelname, record.message) sys.stdout.write(msg_str) sys.stdout.flush() def _handle_active_document_change(self, active_document_path): """ Gets the active document from the host application, determines which context it belongs to, and changes to that context. :param str active_document_path: The path to the new active document. :returns: True if the context changed, False if it did not. """ # If the config says to not change context on active document change, # then we don't do anything here. if not self.get_setting("automatic_context_switch"): self.logger.debug( "Engine setting automatic_context_switch is false." "Not changing context." ) return # Make sure we have a properly-encoded string for the path. We can # possibly get a file path/name that contains unicode, and we don't # want to deal with that later on. if active_document_path is not None: active_document_path = six.ensure_str(active_document_path) # This will be True if the context_changes_disabled context manager is # used. We're just in a temporary state of not allowing context changes, # which is useful when an app is doing a lot of After Effects work that # might be triggering active document changes that we don't want to # result in PTR context changes. with self.heartbeat_disabled(): if self._CONTEXT_CHANGES_DISABLED: self.logger.debug( "Engine is in 'no context changes' mode. Not changing context." ) return False if active_document_path: self.logger.debug("New active document is %s" % active_document_path) cached_context = self.__get_from_context_cache(active_document_path) if cached_context: context = cached_context self.logger.debug("Document found in context cache: %r" % context) else: try: context = self.sgtk.context_from_path( active_document_path, previous_context=self.context, ) self.__add_to_context_cache(active_document_path, context) except Exception: self.logger.debug( "Unable to determine context from path. Setting the Project context." ) # clear the context finding task ids so that any tasks that # finish won't send data to js. self.__context_find_uid = None self.__context_thumb_uid = None # We go to the project context if this is a file outside of # PTR control. if self._PROJECT_CONTEXT is None: self._PROJECT_CONTEXT = sgtk.Context( tk=self.context.sgtk, project=self.context.project, ) context = self._PROJECT_CONTEXT if not context.project and not self.context.project: self.logger.debug( "New context doesn't have a Project entity. Not changing " "context." ) return False elif not context.project: context = self.sgtk.context_from_entity( self.context.project["type"], self.context.project["id"] ) if context and context != self.context: self.adobe.context_about_to_change() sgtk.platform.change_context(context) return True return False def _handle_command(self, uid): """ Handles an RPC engine command execution request. :param int uid: The unique id of the engine command to run. """ self.logger.debug("Handling command request for uid: %s" % (uid,)) with self.heartbeat_disabled(): from sgtk.platform.qt import QtGui if uid == self.__jump_to_fs_command_id: # jump to fs special command triggered self._jump_to_fs() elif uid == self.__jump_to_sg_command_id: # jump to sg special command triggered self._jump_to_sg() else: # a registered command was triggered for command in self.commands.values(): if command.get("properties", dict()).get("uid") == uid: self.logger.debug( "Executing callback for command: %s" % (command,) ) result = command["callback"]() if isinstance(result, QtGui.QWidget): # if the callback returns a widget # keep a handle on it self.__qt_dialogs.append(result) def _handle_logging(self, level, message): """ Handles an RPC logging request. :param str level: One of "debug", "info", "warning", or "error". :param str message: The log message. """ # manually create a record to log to the standard file handler. # we format it to match the regular logs, but tack on the '.js' to # indicate that it came from javascript. record = logging.makeLogRecord( { "levelname": level.upper(), "name": "%s.js" % (self.logger.name,), "msg": message, } ) # forward this message to the base file handler so that it is logged # appropriately. if sgtk.LogManager().base_file_handler: sgtk.LogManager().base_file_handler.handle(record) def _run_tests(self): """ Runs the test suite for the tk-aftereffects bundle. """ # If we don't know what the tests root directory path is # via the environment, then we shouldn't be here. try: tests_root = os.environ["SHOTGUN_ADOBE_TESTS_ROOT"] except KeyError: self.logger.error( "The SHOTGUN_ADOBE_TESTS_ROOT environment variable " "must be set to the root directory of the tests to be " "run. Not running tests!" ) return else: # Make sure we can find the run_tests.py file within the root # that was specified in the environment. self.logger.debug("Test root path found. Looking for run_tests.py.") test_module = os.path.join(tests_root, self.TEST_SCRIPT_BASENAME) if not os.path.exists(test_module): self.logger.error( "Unable to find run_tests.py in the directory " "specified by the SHOTGUN_ADOBE_TESTS_ROOT " "environment variable. Not running tests!" ) return self.logger.debug("Found run_tests.py. Importing to run tests.") try: # We need to prepend to sys.path. We'll set it back to # what it was before once we're done running the tests. original_sys_path = sys.path python_root = os.path.abspath( os.path.join(os.path.dirname(__file__), "python") ) sys.path = [tests_root, python_root] + sys.path import run_tests # The run_tests.py module should make available a run_tests # function. We need to run that, giving it the engine pointer # so that it can use that for logging purposes. run_tests.run_tests(self) except Exception as exc: # If we got an unhandled exception, then something went very # wrong in the test suite. We'll just trap that and print it # as an error without letting it bubble up any farther. import traceback self.logger.error( "Tests raised the following:\n%s" % traceback.format_exc(exc) ) finally: # Reset sys.path back to what it was before we started. sys.path = original_sys_path ############################################################################ # properties @property def adobe(self): """ The handle to the Adobe RPC API. """ return self._adobe @property def app_id(self): """ The runtime app id. This will be a string -- something like PHSP for After Effects, or AEFT for After Effect. """ return self._SHOTGUN_ADOBE_APPID @property def context_change_allowed(self): """ Specifies that context changes are allowed by the engine. """ return True ############################################################################ # context manager @contextmanager def context_changes_disabled(self): """ A context manager that disables context changes on enter, and enables them on exit. This is useful in apps that might be performing operations that require changes in the active document that don't want to trigger a context change. """ self._CONTEXT_CHANGES_DISABLED = True yield self._CONTEXT_CHANGES_DISABLED = False @contextmanager def heartbeat_disabled(self): """ A context manager that disables the heartbeat and message processing timer on enter, and restarts it on exit. """ try: self.logger.debug("Pausing heartbeat...") self._HEARTBEAT_DISABLED = True except Exception as e: self.logger.debug("Unable to pause heartbeat as requested.") self.logger.error(str(e)) else: self.logger.debug("Heartbeat paused.") yield self._HEARTBEAT_DISABLED = False self.logger.debug("Heartbeat restarted.") ############################################################################ # UI def _define_qt_base(self): """ This will be called at initialisation time and will allow a user to control various aspects of how QT is being used by Toolkit. The method should return a dictionary with a number of specific keys, outlined below. * qt_core - the QtCore module to use * qt_gui - the QtGui module to use * dialog_base - base class for to use for Toolkit's dialog factory :returns: dict """ # Just call the base implementation and monkey patch QMessageBox. base = super(AfterEffectsEngine, self)._define_qt_base() if not base: raise ImportError("Unable to find a QT Python module") QtCore = base["qt_core"] QtGui = base["qt_gui"] # tell QT4 to interpret C strings as utf-8 # note: this will be ignored on QT5 via our shim utf8 = QtCore.QTextCodec.codecForName("utf-8") QtCore.QTextCodec.setCodecForCStrings(utf8) # override message boxes self._override_qmessagebox(QtGui.QMessageBox) return base def _override_qmessagebox(self, q_message_box): """ Redefine the method calls for QMessageBox static methods. These are often called from within apps and because QT is running in a separate process, they will pop up behind the After Effects window. Wrap each of these calls in a raise method to activate the QT process. :param q_message_box: The QMessageBox class to patch. """ info_fn = q_message_box.information critical_fn = q_message_box.critical question_fn = q_message_box.question warning_fn = q_message_box.warning @staticmethod def _info_wrapper(*args, **kwargs): self.__activate_python() return info_fn(*args, **kwargs) @staticmethod def _critical_wrapper(*args, **kwargs): self.__activate_python() return critical_fn(*args, **kwargs) @staticmethod def _question_wrapper(*args, **kwargs): self.__activate_python() return question_fn(*args, **kwargs) @staticmethod def _warning_wrapper(*args, **kwargs): self.__activate_python() return warning_fn(*args, **kwargs) q_message_box.information = _info_wrapper q_message_box.critical = _critical_wrapper q_message_box.question = _question_wrapper q_message_box.warning = _warning_wrapper def __check_for_popups(self): """ Method will check if a popup dialog from aftereffects was openend and if so, it will raise the window to the front. NOTE: This is only done in windows. TODO: Implement an equivalent method on mac """ if sgtk.util.is_windows(): # to get all dialogs from After Effects, we have to query the # After Effects process id first. As this code runs inside a # separate python process, we use a subprocess for this. if self._AFX_PID == -1: return elif self._AFX_PID is None: # the windows tasklist command will provide the process id # of the After Effects executable. As there is always only one # instance of After Effects, this should be safe to determine the # process id. pid_query_process = subprocess.Popen( [ "tasklist", "/FI", "ImageName eq AfterFX.exe", "/FO", "CSV", "/NH", ], stdout=subprocess.PIPE, ) out_string, _ = pid_query_process.communicate() # The out_string will look like: # "AfterFX.ext","1234","SessionName","SessionNum","MemoryUsage" # where 1234 describes the pid of After Effects match = re.match(b"[^0-9]+([0-9]+).*", out_string, re.DOTALL) if not match: self._AFX_PID = -1 self._AFX_PID = int(match.group(1)) # with the process id of After Effects, we can get all HWNDS that point to # dialog classes. hwnds = self.__tk_aftereffects.win_32_api.find_windows( process_id=self._AFX_PID, class_name=self._AFX_WIN32_DIALOG_WINDOW_CLASS, stop_if_found=True, ) # To avoid raising a dialog, that we raised before, # we will compare the list of visible dialog-hwnds with the list # that we already raised. # As we cannot compare the hwnd-pointers directly, we need to compare # the integer value (hwnd long) of the hwnd pointers. # we build a dict that maps the hwnd longs to the current hwnd pointer all_hwnds = {} GetWindow = self.__tk_aftereffects.win_32_api.ctypes.windll.user32.GetWindow for hwnd in hwnds: # GetWindow with GW_CHILD is used to get the hwnd long that identifies the child window # at the top of the Z order, of the specified parent window pointer. all_hwnds[GetWindow(hwnd, self.__WIN32_GW_CHILD)] = hwnd # by comparing the cached hwnd longs with the current list of hwnd longs, # we find out which hwnds are actually pointing to new (unraised) dialogs. new_hwnds = set(all_hwnds.keys()) - (self._POPUP_CACHE or set([])) # in case there is a new dialog, we bring it to front and update the cache for hwnd_long in new_hwnds: self.__tk_aftereffects.win_32_api.bring_to_front(all_hwnds[hwnd_long]) self._POPUP_CACHE = set(all_hwnds.keys()) return # in case there are open dialogs, but we already raised them, # we do nothing. if all_hwnds: return # In case there is no open dialog, we will reset the cache to None self._POPUP_CACHE = None def _win32_get_aftereffects_main_hwnd(self): """ Windows specific method to find the main After Effects window handle (HWND) """ if not self._WIN32_AFTEREFFECTS_MAIN_HWND: for major in sorted(self.__CC_VERSION_MAPPING.keys()): for minor in range(10): found_hwnds = self.__tk_aftereffects.win_32_api.find_windows( class_name="AE_CApplication_{}.{}".format(major, minor), stop_if_found=True, ) if found_hwnds: self._WIN32_AFTEREFFECTS_MAIN_HWND = found_hwnds[0] return self._WIN32_AFTEREFFECTS_MAIN_HWND return self._WIN32_AFTEREFFECTS_MAIN_HWND def _win32_get_proxy_window(self): """ Windows-specific method to get the proxy window that will 'own' all Toolkit dialogs. This will be parented to the main After Effects application. :returns: A QWidget that has been parented to After Effects's window. """ # Get the main After Effects window: ps_hwnd = self._win32_get_aftereffects_main_hwnd() win32_proxy_win = None proxy_win_hwnd = None if ps_hwnd: from sgtk.platform.qt import QtGui, QtCore # Create the proxy QWidget. win32_proxy_win = QtGui.QWidget() window_title = "Flow Production Tracking Parent Widget {0}".format( uuid.uuid4().hex ) win32_proxy_win.setWindowTitle(window_title) # We have to take different approaches depending on whether # we're using Qt4 (PySide) or Qt5 (PySide2). The functionality # needed to turn a Qt5 WId into an HWND is not exposed in PySide2, # so we can't do what we did below for Qt4. if QtCore.__version__.startswith("4."): proxy_win_hwnd = ( self.__tk_aftereffects.win_32_api.qwidget_winid_to_hwnd( win32_proxy_win.winId(), ) ) else: # With PySide2, we're required to look up our proxy parent # widget's HWND the hard way, following the same logic used # to find After Effects's main window. To do that, we actually have # to show our widget so that Windows knows about it. We can make # it effectively invisible if we zero out its size, so we do that, # show the widget, and then look up its HWND by window title before # hiding it. win32_proxy_win.setGeometry(0, 0, 0, 0) win32_proxy_win.show() try: proxy_win_hwnd_found = ( self.__tk_aftereffects.win_32_api.find_windows( stop_if_found=True, window_text=window_title ) ) finally: win32_proxy_win.hide() if proxy_win_hwnd_found: proxy_win_hwnd = proxy_win_hwnd_found[0] else: self.logger.debug( "Unable to determine the HWND of After Effects itself. This means " "that we can't properly setup window parenting for Toolkit apps." ) # Parent to the After Effects application window if we found everything # we needed. If we didn't find our proxy window for some reason, we # will return None below. In that case, we'll just end up with no # window parenting, but apps will still launch. if proxy_win_hwnd is None: self.logger.warning( "Unable setup window parenting properly. Dialogs shown will " "not be parented to After Effects, but they will still function " "properly otherwise." ) else: # Set the window style/flags. We don't need or want our Python # dialogs to notify the After Effects application window when they're # opened or closed, so we'll disable that behavior. win_ex_style = self.__tk_aftereffects.win_32_api.GetWindowLong( proxy_win_hwnd, self.__tk_aftereffects.win_32_api.GWL_EXSTYLE, ) self.__tk_aftereffects.win_32_api.SetWindowLong( proxy_win_hwnd, self.__tk_aftereffects.win_32_api.GWL_EXSTYLE, win_ex_style | self.__tk_aftereffects.win_32_api.WS_EX_NOPARENTNOTIFY, ) self.__tk_aftereffects.win_32_api.SetParent(proxy_win_hwnd, ps_hwnd) self._PROXY_WIN_HWND = proxy_win_hwnd return win32_proxy_win def _get_dialog_parent(self): """ Get the QWidget parent for all dialogs created through show_dialog & show_modal. """ """ Get the QWidget parent for all dialogs created through show_dialog & show_modal. """ # determine the parent widget to use: from sgtk.platform.qt import QtGui if not self._DIALOG_PARENT: if sgtk.util.is_windows(): # for windows, we create a proxy window parented to the # main application window that we can then set as the owner # for all Toolkit dialogs self._DIALOG_PARENT = self._win32_get_proxy_window() else: self._DIALOG_PARENT = QtGui.QApplication.activeWindow() return self._DIALOG_PARENT def show_dialog(self, title, bundle, widget_class, *args, **kwargs): """ Shows a non-modal dialog window in a way suitable for this engine. The engine will attempt to parent the dialog nicely to the host application. :param title: The title of the window :param bundle: The app, engine or framework object that is associated with this window :param widget_class: The class of the UI to be constructed. This must derive from QWidget. Additional parameters specified will be passed through to the widget_class constructor. :returns: the created widget_class instance """ if not self.has_ui: self.logger.error( "Sorry, this environment does not support UI display! Cannot " "show the requested window '%s'." % title ) return None # create the dialog: dialog, widget = self._create_dialog_with_widget( title, bundle, widget_class, *args, **kwargs ) # Note - the base engine implementation will try to clean up # dialogs and widgets after they've been closed. However this # can cause a crash in After Effects as the system may try to send # an event after the dialog has been deleted. # Keeping track of all dialogs will ensure this doesn't happen self.__qt_dialogs.append(dialog) # make python active if possible self.__activate_python() # make sure the window raised so it doesn't # appear behind the main After Effects window self.logger.debug("Showing dialog: %s" % (title,)) dialog.show() dialog.raise_() dialog.activateWindow() return widget def show_modal(self, title, bundle, widget_class, *args, **kwargs): """ Shows a modal dialog window in a way suitable for this engine. The engine will attempt to integrate it as seamlessly as possible into the host application. This call is blocking until the user closes the dialog. :param title: The title of the window :param bundle: The app, engine or framework object that is associated with this window :param widget_class: The class of the UI to be constructed. This must derive from QWidget. Additional parameters specified will be passed through to the widget_class constructor. :returns: (a standard QT dialog status return code, the created widget_class instance) """ if not self.has_ui: self.logger.error( "Sorry, this environment does not support UI display! Cannot " "show the requested window '%s'." % title ) return # create the dialog: dialog, widget = self._create_dialog_with_widget( title, bundle, widget_class, *args, **kwargs ) # Note - the base engine implementation will try to clean up # dialogs and widgets after they've been closed. However this # can cause a crash in After Effects as the system may try to send # an event after the dialog has been deleted. # Keeping track of all dialogs will ensure this doesn't happen self.__qt_dialogs.append(dialog) # make python active if possible self.__activate_python() self.logger.debug("Showing modal: %s" % (title,)) dialog.raise_() dialog.activateWindow() status = dialog.exec_() return status, widget ############################################################################ # internal methods def __get_command_uid(self): """ Returns a guaranteed unique command id. """ with self._LOCK: self._COMMAND_UID_COUNTER += 1 return self._COMMAND_UID_COUNTER def __get_icon_path(self, properties): """ Processes the command properties dictionary to find the most appropriate icon path. This code looks for an `icons` dictionary of the following form:: { "dark": { "png": "/path/to/dark_icon.png", "svg": "/path/to/dark_icon.svg" }, "light": { "png": "/path/to/light_icon.png", "svg": "/path/to/light_icon.svg" } } For Adobe, the preference is dark png then light png If neither of these is found, fall back to the standard `properties["icon"]`. """ icon_path = None icons = properties.get("icons") if icons: dark = icons.get("dark") light = icons.get("light") # check for dark icon if dark: icon_path = dark.get("png", icon_path) # if no dark icon, check the light: if not icon_path and light: icon_path = dark.get("png", icon_path) # still no icon path, fall back to regular icon if not icon_path: icon_path = properties.get("icon") return icon_path def __send_state(self): """ Sends information back to javascript representing the current context. """ # alert js that the state is about to change. this allows the panel to # clear its current state and display a loading message. self.adobe.context_about_to_change() # ---- process the context for display # clear existing context requests to prevent unnecessary processing self.__context_find_uid = None self.__context_thumb_uid = None self.__sg_data.clear() # determine the best entity to show for the current context context_entity = self.__get_context_entity() # this will inspect the context and do any additional queries for fields # that are required to show it self.__request_context_display(context_entity) # ---- the engine already has access to all the commands that need to # be display for the current context. so go ahead and process those # and send them back separately # first, we'll process the menu favorites fav_lookup = {} fav_index = 0 # create a lookup of the combined app instance name with the display # name. that should be unique and provide an easy lookup to match # against. we'll remember the order processed in order to sort our # favorites list once all the registered commands are processed for fav_command in self.get_setting("shelf_favorites"): app_instance_name = fav_command["app_instance"] display_name = fav_command["name"] # build unique lookup for this combo of app instance and command fav_id = app_instance_name + display_name fav_lookup[fav_id] = fav_index # give it an index so that we can sort and maintain order later fav_index += 1 # keep a list of each type of command since they'll be displayed # differently on the adobe side. favorites = [] context_menu_cmds = [] commands = [] # iterate over all the registered commands and gather the necessary info # to display them in adobe for (command_name, command_info) in self.commands.items(): # commands come with a dict of properties that may or may not # contain certain data. properties = command_info.get("properties", {}) # ---- determine the app's instance name app_instance = properties.get("app", None) app_name = None # check this command's app against the engine's apps. if app_instance: for (app_instance_name, app_instance_obj) in self.apps.items(): if app_instance_obj == app_instance: app_name = app_instance_name cmd_type = properties.get("type", "default") # create the command dict to hand over to adobe command = dict( uid=properties.get("uid"), display_name=command_name, icon_path=self.__get_icon_path(properties), description=properties.get("description"), type=properties.get("type", "default"), ) # build the lookup string to see if this app is a favorite fav_name = str(app_name) + command_name if cmd_type == "context_menu": # these commands will show up in the panel flyout menu context_menu_cmds.append(command) elif fav_name in fav_lookup: # add the fav index to the command so that we can sort after # all favorites are identified. command["fav_index"] = fav_lookup[fav_name] favorites.append(command) else: commands.append(command) # ---- include the "jump to" commands that are common to all engines jump_commands = [] # the icon to use for the command. bundled with the engine sg_icon = os.path.join(self.disk_location, "resources", "shotgun_logo.png") jump_commands.append( dict( uid=self.__jump_to_sg_command_id, display_name="Jump to Flow Production Tracking", icon_path=sg_icon, description="Open the current context in a web browser.", type="context_menu", ) ) if self.context.filesystem_locations: # the icon to use for the command. bundled with the engine fs_icon = os.path.join( self.disk_location, "resources", "shotgun_folder.png" ) jump_commands.append( dict( uid=self.__jump_to_fs_command_id, display_name="Jump to File System", icon_path=fs_icon, description="Open the current context in a file browser.", type="context_menu", ) ) # sort the favorites based on their index favorites = sorted(favorites, key=lambda d: d["fav_index"]) # sort the other commands alphabetically by display name commands = sorted(commands, key=lambda d: d["display_name"]) # sort the context menu commands alphabetically by display name. We # force the Jump to Shotgun and Jump to Filesystem commands onto the # front of the list to match other integrations. context_menu_cmds = jump_commands + sorted( context_menu_cmds, key=lambda d: d["display_name"], ) # ---- populate the state structure to hand over to adobe all_commands = { "favorites": favorites, "commands": commands, "context_menu_cmds": context_menu_cmds, } # send the commands back to adobe self.adobe.send_commands(all_commands) def __setup_connection_timer(self, force=False): """ Sets up the connection timer that handles monitoring of the live connection, as well as the triggering of message processing. :param bool force: Forces the creation of a new connection timer, even if one already exists. When this occurs, if a timer already exists, it will be stopped. """ if self._CHECK_CONNECTION_TIMER is None or force: self.logger.debug("Creating connection timer...") if self._CHECK_CONNECTION_TIMER: self.logger.debug( "Connection timer already exists, so it will be stopped." ) try: self._CHECK_CONNECTION_TIMER.stop() except Exception: # No reason to be alarmed here. Just let it go and it'll # garbage collected when appropriate. pass from sgtk.platform.qt import QtCore timer = QtCore.QTimer( parent=QtCore.QCoreApplication.instance(), ) timer.timeout.connect(self._check_connection) timer.timeout.connect(self.__check_for_popups) # The class variable is in seconds, so multiply to get milliseconds. timer.start( self.SHOTGUN_ADOBE_HEARTBEAT_INTERVAL * 1000.0, ) self._CHECK_CONNECTION_TIMER = timer self.logger.debug("Connection timer created and started.") def _jump_to_sg(self): """ Jump to shotgun, launch web browser """ from sgtk.platform.qt import QtGui, QtCore url = self.context.shotgun_url QtGui.QDesktopServices.openUrl(QtCore.QUrl(url)) def _jump_to_fs(self): """ Jump from context to FS """ # launch one window for each location on disk paths = self.context.filesystem_locations self.logger.debug("FS paths: %s" % (str(paths),)) for disk_location in paths: # run the app if sgtk.util.is_macos(): cmd = 'open "%s"' % disk_location elif sgtk.util.is_windows(): cmd = 'cmd.exe /C start "Folder" "%s"' % disk_location elif sgtk.util.is_linux(): cmd = 'xdg-open "%s"' % disk_location else: raise Exception("Platform '%s' is not supported." % sys.platform) exit_code = os.system(cmd) if exit_code != 0: self.logger.error("Failed to launch '%s'!" % cmd) ########################################################################################## # context data methods def __add_to_context_cache(self, path, context): """ Adds the given active document path to the context cache, associating it with the given context object. This will trigger the storing of a serialized cache as a user setting for use during panel extension restarts. :param str path: The document path to add to the cache. :param context: The context object to associate with the document. """ if path not in self._CONTEXT_CACHE: # We're storing the context cache in a sgtk user setting at the project # level. This will ensure that when we read the cache back, we'll only # be getting contexts in our current project. Anything outside of that # scope would be unusable, as we don't allow context changing across # project boundaries. self._CONTEXT_CACHE[path] = context serial_cache = dict() for k, v in self._CONTEXT_CACHE.items(): serial_cache[k] = v.serialize() self.logger.debug("Storing context cache: %s" % serial_cache) self.__settings_manager.store( self._CONTEXT_CACHE_KEY, serial_cache, self.__settings_manager.SCOPE_PROJECT, ) def __get_from_context_cache(self, path): """ Gets the document path's associated context object, if one has been cached. :returns: Context object, or None """ return self._CONTEXT_CACHE.get(path) def __request_context_display(self, entity): """ Request fields to show in the context header for the given entity. Always includes the image url to display a thumbnail. """ if not entity: # no entity. this will retrieve the html to display for the site. fields_html = self.execute_hook_method( "context_fields_display_hook", "get_context_html", entity=None, sg_globals=self.__shotgun_globals, ) self.adobe.send_context_display(fields_html) # go ahead and forward the site thumbnail back to js data = dict( thumb_path="../images/default_Site_thumb_dark.png", url=self.sgtk.shotgun_url, ) self.adobe.send_context_thumbnail(data) return # get the fields to query from the hook fields = self.execute_hook_method( "context_fields_display_hook", "get_entity_fields", entity_type=entity["type"], ) # always try to query the image for the entity if "image" not in fields: fields.append("image") entity_type = entity["type"] entity_id = entity["id"] # kick off an async request to query the necessary fields self.__context_find_uid = self.__sg_data.execute_find_one( entity_type, [["id", "is", entity_id]], fields ) def __on_worker_failure(self, uid, msg): """ Asynchronous callback - the worker thread errored. """ # log a message if the worker failed to retrieve the necessary info. if uid == self.__context_find_uid: # clear the find id since we are now processing it self.__context_find_uid = None # send an error message back to the context header. self.adobe.send_context_display( """ There was an error retrieving fields for this context. Please see the logs for the specific error message. If this is a recurring error and you need further assistance, please contact our support team via <a href='{url}'> {url}</a>. """.format( url=sgtk.support_url ) ) self.logger.error("Failed to query context fields: %s" % (msg,)) elif uid == self.__context_thumb_uid: # clear the thumb id since we are now processing it self.__context_thumb_uid = None # log this. the panel will display a default thumbnail, so this # should be sufficient self.logger.error("Failed to query context thumbnail: %s" % (msg,)) def __on_worker_signal(self, uid, request_type, data): """ Signaled whenever the worker completes something. """ self.logger.debug("Worker signal: %s" % (data,)) # the find query for the context entity with the specified fields if uid == self.__context_find_uid: # clear the find id since we are now processing it self.__context_find_uid = None context_entity = data["sg"] # should have an image url now. submit a request to download the # entity's thumbnail. if "image" in context_entity and context_entity["image"]: self.__context_thumb_uid = self.__sg_data.request_thumbnail( context_entity["image"], context_entity["type"], context_entity["id"], "image", load_image=False, ) # no image, use a default image based on the entity type else: if context_entity["type"] in ["Asset", "Project", "Shot", "Task"]: thumb_path = "../images/default_%s_thumb_dark.png" % ( context_entity["type"] ) data["thumb_path"] = thumb_path else: thumb_path = "../images/default_Entity_thumb_dark.png" data = dict( thumb_path=thumb_path, url=self.get_entity_url(context_entity), ) self.adobe.send_context_thumbnail(data) # now that we have all the field values, go back to the hook and # build the html to display them. fields_html = self.execute_hook_method( "context_fields_display_hook", "get_context_html", entity=context_entity, sg_globals=self.__shotgun_globals, ) # forward the display html back to the js panel self.adobe.send_context_display(fields_html) # thumbnail download. forward the path and a url back to js elif uid == self.__context_thumb_uid: # clear the thumb id since we already processed it self.__context_thumb_uid = None context_entity = self.__get_context_entity() # add a url to allow the panel to make the thumbnail clickable data["url"] = self.get_entity_url(context_entity) self.adobe.send_context_thumbnail(data) def __get_project_id(self): """Helper method to return the project id for the current context.""" if self.context.project: return self.context.project["id"] else: return None def __get_context_entity(self): """Helper method to return an entity to display for current context.""" context = self.context # determine the best entity for displaying the thumbnail. just return # the first of task, entity, project that is defined for entity in [context.task, context.entity, context.project]: if not entity: continue return entity def get_entity_url(self, entity): """Helper method to return a PTR url for the supplied entity.""" return "%s/detail/%s/%d" % (self.sgtk.shotgun_url, entity["type"], entity["id"]) def get_panel_link(self, url, text): """ Helper method to return an html link to display in the panel and will launch the supplied url in the default browser. """ return """ <a href='#' class='sg_value_link' onclick='sg_panel.Panel.open_external_url("{url}")' >{text}</a> """.format( url=url, text=text, ) def __activate_python(self): """ Do the Os-specific thing to show this process above all others. """ if sgtk.util.is_macos(): # a little action script to activate the given python process. osx_activate_script = """ tell application "System Events" set frontmost of the first process whose unix id is {pid} to true end tell """.format( pid=os.getpid() ) # force this python process to the front cmd = ["osascript", "-e", osx_activate_script] status = subprocess.call(cmd) if status: self.logger.error("Could not activate python.") elif sgtk.util.is_windows(): pass def __import_file(self, import_options): """ Does the import and gets all new items, by caching the list of existing items before importing and comparing them to the list of existing items after importing. :param import_options: Import options object that should be imported :type import_options: `adobe.ImportOptionsObject`_ :returns: All new items that were imported :rtype: list-of-`adobe.CompItemObject`_ """ # as the return value of importFile will not # reliably return the new items, one # has to track the changes item_cache = [] for item in self.iter_collection(self.adobe.app.project.rootFolder.items): item_cache.append(item) # do the import self.adobe.app.project.importFile(import_options) new_items = [] for item in self.iter_collection(self.adobe.app.project.rootFolder.items): if item not in item_cache: new_items.append(item) return new_items