Created by Eric Hermelin 2017-09-14

action_memory_test.py

      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()

    

Comments (3)

HTTPS

SSH

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