Snippets

ftrack action_memory_test.py

Created by Eric Hermelin
View revision

File action_memory_test.py Added

  • Ignore whitespace
  • Hide word diff 
+import os
+import psutil
+import logging
+
+import ftrack_api
+
+class MemoryAction(object):
+
+    label = 'Memory'
+    identifier = 'MemoryAction'
+
+    def __init__(self, session):
+        super(MemoryAction, self).__init__()
+
+        self.session = session
+
+    def register(self):
+        '''Register action.'''
+        self.session.event_hub.subscribe(
+            'topic=ftrack.action.discover',
+            self.discover
+        )
+
+        self.session.event_hub.subscribe(
+            'topic=ftrack.action.launch and data.actionIdentifier={0}'.format(
+                self.identifier
+            ),
+            self.launch
+        )
+
+    def discover(self, event):
+
+        return {
+            'items':[{
+                'label':self.label,
+                'actionIdentifier':self.identifier
+            }]
+        }
+
+    def _translate_event(self, session, event):
+        '''Return *event* translated structure to be used with the API.'''
+
+        _selection = event['data'].get('selection', [])
+
+        _entities = list()
+        for entity in _selection:
+            _entities.append(
+                (
+                    self._get_entity_type(entity), entity.get('entityId')
+                )
+            )
+
+
+        return [
+            _entities,
+            event
+        ]
+
+    def _get_entity_type(self, entity):
+        '''Return translated entity type tht can be used with API.'''
+        # Get entity type and make sure it is lower cased. Most places except
+        # the component tab in the Sidebar will use lower case notation.
+        entity_type = entity.get('entityType').replace('_', '').lower()
+
+        for schema in self.session.schemas:
+            alias_for = schema.get('alias_for')
+
+        for schema in self.session.schemas:
+            alias_for = schema.get('alias_for')
+
+            if (
+                alias_for and isinstance(alias_for, basestring) and
+                alias_for.lower() == entity_type
+            ):
+                return schema['id']
+
+        for schema in self.session.schemas:
+            if schema['id'].lower() == entity_type:
+                    return schema['id']
+
+        raise ValueError(
+            'Unable to translate entity type: {0}.'.format(entity_type)
+        )
+
+    def launch(self, event):
+        '''Callback method for custom action.'''
+
+        entities, _event = self._translate_event(
+            self.session, event
+        )
+
+        for entity_type, entity_id in entities:
+            self.session.get(
+                entity_type, entity_id
+            )
+
+        logging.warning(
+            'got entities : {0}'.format(
+                entities
+            )
+        )
+
+        logging.warning(
+            'entities in cache : {0}'.format(
+                len(self.session.cache.caches[0].keys())
+            )
+        )
+
+        logging.warning(
+            'memory information launch: {0}'.format(
+                psutil.Process().memory_full_info()
+            )
+        )
+
+def main():
+    logging.basicConfig(
+        level=logging.INFO,
+        format='%(funcName)20s() ] %(message)s'
+    )
+
+    session = ftrack_api.Session()
+    action = MemoryAction(session)
+
+    action.register()
+
+    session.event_hub.wait()
+
+
+if __name__ == '__main__':
+    main()
HTTPS SSH

You can clone a snippet to your computer for local editing. Learn more.