# Copyright 2011-2014 Biomedical Imaging Group Rotterdam, Departments of
# Medical Informatics and Radiology, Erasmus MC, Rotterdam, The Netherlands
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""
The base class for all Plugins in the fastr system
"""
from abc import ABCMeta
from enum import Enum
import gzip
import StringIO
import fastr
from colorama import Fore, Back, Style
from fastr import exceptions
from fastr.utils.classproperty import classproperty
[docs]class PluginState(Enum):
""" Plugin status Enum. """
uninitialized = Back.CYAN + 'UnInitialized' + Style.RESET_ALL
preload = Back.LIGHTGREEN_EX + 'PreLoad' + Style.RESET_ALL
unloaded = Back.CYAN + 'UnLoaded' + Style.RESET_ALL
failed = Fore.WHITE + Back.RED + Style.BRIGHT + 'Failed' + Style.RESET_ALL
loaded = Fore.WHITE + Back.GREEN + Style.BRIGHT + 'Loaded' + Style.RESET_ALL
[docs]class BasePlugin(object):
"""
Base class for Plugins in the fastr system.
"""
__metaclass__ = PluginMeta
#: The status of the plugin
_status = (PluginState.uninitialized, 'Plugin object created', None)
_source_code = None
module = None
_instantiate = False
[docs] def __init__(self):
"""
The BasePlugin constructor.
:return: the created plugin
:rtype: BasePlugin
:raises FastrPluginNotLoaded: if the plugin did not load correctly
"""
if self.status not in (PluginState.preload, PluginState.loaded):
raise exceptions.FastrPluginNotLoaded('Plugin was not properly loaded: [{}] {}'.format(self._status[0], self._status[1]))
[docs] def __str__(self):
"""
Creare string representation of the plugin.
:return: string represenation
:rtype: str
"""
return '<Plugin: {}>'.format(self.__class__.__name__)
[docs] def __repr__(self):
return '<{}: {} object [{}]>'.format(type(self).__bases__[0].__name__, self.id, getattr(self.status, 'value', 'NA'))
@classproperty
def id(cls):
"""
The id of this plugin
"""
return cls.__name__
@classproperty
def configuration_fields(cls):
"""
Plugins can register the fields they need from the configuration
by adding them here. The format of the data structure is::
{
"var_name": (type, default, description),
}
So for example::
{
"some_string": (str, "spam", "some string used for demonstration purposes"),
"some_int": (str, 42, "the answer to life etc"),
}
"""
return {}
@classmethod
[docs] def register_configuration(cls):
"""
Register and test the configuation fields of the plugin
"""
# Register fields
fastr.config.register_fields(cls.configuration_fields)
# Validate that current values are correct
for field_name, field_spec in cls.configuration_fields.items():
fastr.log.debug('Tested config value: {}'.format(getattr(fastr.config, field_name)))
@classproperty
def fullid(cls):
"""
The full id of this plugin
"""
return 'fastr://plugins/{}'.format(cls.id)
@classproperty
def source_code(cls):
"""
The source code of this plugin
"""
code_gzip = StringIO.StringIO(cls._source_code)
with gzip.GzipFile(fileobj=code_gzip, mode='rb') as gzip_stream:
return gzip_stream.read()
@classproperty
[docs] def status(cls):
"""
The status of the plugin.
"""
return cls._status[0]
@classproperty
def instantiate(cls):
"""
Flag if the plugin should be instantiated
"""
return cls._instantiate
@classproperty
def status_message(cls):
"""
The message explaining the status of the plugin.
"""
return cls._status[1]
@classmethod
[docs] def set_status(cls, status, message, exception=None):
"""
Update the status of the plugin
:param str status: the new status
:param str message: message explaining the status change
:param str exception: stacktrace of the exception causing the failed load
"""
if not isinstance(status, PluginState):
raise exceptions.FastrTypeError('Plugin status should be of type PluginState')
if not isinstance(message, basestring):
raise exceptions.FastrTypeError('Plugin status message should be a string')
cls._status = (status, message, exception)
@classmethod
[docs] def set_code(cls, source_code):
"""
Set the filename and source code of the plugin
:param str source_code: the source code of the plugin
"""
#if cls._source_code is not None:
# raise exceptions.FastrCannotChangeAttributeError('Filename and/or source code are already set!')
# We gzip the source code before storing
#code_gzip = StringIO.StringIO()
#with gzip.GzipFile(fileobj=code_gzip, mode='w') as gzip_stream:
# gzip_stream.write(source_code)
#cls._source_code = code_gzip.getvalue()
cls._source_code = source_code
[docs] def cleanup(self):
"""
Perform any cleanup action needed when the plugin use ended. This can
be closing files/streams etc.
"""
pass
@classmethod
[docs] def test(cls):
"""
Test the plugin, default behaviour is just to instantiate the plugin
"""
obj = cls()
obj.cleanup()
del obj
[docs]class Plugin(BasePlugin):
pass