"""The plugin module provides optional submodules that add new Hooke commands. All of the science happens in here. """ import os.path import Queue as queue from ..config import Setting from ..util.graph import Node, Graph PLUGIN_MODULES = [ # ('autopeak', True), # ('curvetools', True), # ('cut', True), # ('fit', True), # ('flatfilts-rolf', True), # ('flatfilts', True), # ('generalclamp', True), # ('generaltccd', True), # ('generalvclamp', True), # ('jumpstat', True), # ('macro', True), # ('massanalysis', True), # ('multidistance', True), # ('multifit', True), # ('pcluster', True), # ('procplots', True), # ('review', True), # ('showconvoluted', True), # ('superimpose', True), # ('tutorial', True), # ('viewer', True), ] """List of plugin modules and whether they should be included by default. TODO: autodiscovery """ # Plugins and settings def Plugin (object): """The pluggable collection of Hooke commands. Fulfills the same role for Hooke that a software package does for an operating system. """ name = "base plugin" def dependencies(self): """Return a list of Plugins we require.""" return [] def default_settings(self): """Return a list of hooke.config.Settings() for any configurable module settings.""" return [] def commands(self): """Return a list of Commands provided.""" return [] PLUGINS = {} """(name,instance) :class:`dict` of all possible :class:`Plugin`\s. """ print __name__ for plugin_modname,value in PLUGIN_MODULES: this_mod = __import__(__name__, fromlist=[plugin_modname]) plugin_mod = getattr(this_mod, plugin_modname) for objname in dir(plugin_mod): obj = getattr(plugin_mod, objname) if type(obj) == Plugin: obj.module_name = plugin_modname PLUGINS[p.name] = p PLUGIN_GRAPH = Graph([Node([PLUGINS[name] for name in p.dependencies()]) for p in PLUGINS.values()]) PLUGIN_GRAPH.topological_sort() def default_settings(self): settings = [Setting( 'plugins', help='Enable/disable default plugins.')] for pnode in PLUGIN_GRAPH: settings.append(Setting(p.name, str(PLUGIN_MODULES[p.module_name][1]))) for pnode in PLUGIN_GRAPH: plugin = pnode.data settings.extend(plugin.default_settings()) return settings # Commands and arguments class CommandExit (Exception): def __str__(self): return self.__class__.__name__ class Success (CommandExit): pass class Failure (CommandExit): pass class Command (object): """One-line command description here. >>> c = Command(name='test', help='An example Command.') >>> status = c.run(NullQueue(), PrintQueue(), help=True) ITEM: Command: test Arguments: help HELP (bool) Print a help message. An example Command. ITEM: Success """ def __init__(self, name, arguments=[], aliases=None, help=''): self.name = name self.arguments = [ Argument(name='help', type='bool', default=False, count=1, callback=StoreValue(True), help='Print a help message.'), ] + arguments if aliases == None: aliases = [] self.aliases = aliases self._help = help def run(self, inqueue=None, outqueue=None, **kwargs): """`Normalize inputs and handle before punting to :meth:`_run`. """ if inqueue == None: inqueue = NullQueue() if outqueue == None: outqueue = NullQueue() try: params = self.handle_arguments(inqueue, outqueue, kwargs) if params['help'] == True: outqueue.put(self.help()) raise(Success()) self._run(inqueue, outqueue, params) except CommandExit, e: if isinstance(e, Failure): outqueue.put(e.message) outqueue.put(e) return 1 outqueue.put(e) return 0 def _run(self, inqueue, outqueue, params): """This is where the command-specific magic will happen. """ pass def handle_arguments(self, inqueue, outqueue, params): """Normalize and validate input parameters (:class:`Argument` values). """ for argument in self.arguments: names = [argument.name] + argument.aliases settings = [(name,v) for name,v in params.items() if name in names] if len(settings) == 0: if argument.optional == True or argument.count == 0: settings = [(argument.name, argument.default)] else: raise Failure('Required argument %s not set.' % argument.name) if len(settings) > 1: raise Failure('Multiple settings for %s:\n %s' % (argument.name, '\n '.join(['%s: %s' % (name,value) for name,value in sorted(settings)]))) name,value = settings[0] if name != argument.name: params.remove(name) params[argument.name] = value if argument.callback != None: value = argument.callback(self, argument, value) params[argument.name] = value argument.validate(value) return params def help(self, *args): name_part = 'Command: %s' % self.name if len(self.aliases) > 0: name_part += ' (%s)' % ', '.join(self.aliases) argument_part = ['Arguments:'] + [a.help() for a in self.arguments] argument_part = '\n'.join(argument_part) help_part = self._help return '\n\n'.join([name_part, argument_part, help_part]) class Argument (object): """Structured user input for :class:`Command`\s. TODO: ranges for `count`? """ def __init__(self, name, type='string', metavar=None, default=None, optional=True, count=1, completion_callback=None, callback=None, aliases=None, help=''): self.name = name self.type = type if metavar == None: metavar = name.upper() self.metavar = metavar self.default = default self.optional = optional self.count = count self.completion_callback = completion_callback self.callback = callback if aliases == None: aliases = [] self.aliases = aliases self._help = help def __str__(self): return '<%s %s>' % (self.__class__.__name__, self.name) def __repr__(self): return self.__str__() def help(self): parts = ['%s ' % self.name] if self.metavar != None: parts.append('%s ' % self.metavar) parts.extend(['(%s) ' % self.type, self._help]) return ''.join(parts) def validate(self, value): """If `value` is not appropriate, raise `ValueError`. """ pass # TODO: validation # TODO: type conversion # TODO: type extensions? # Useful callbacks class StoreValue (object): def __init__(self, value): self.value = value def __call__(self, command, argument, fragment=None): return self.value class NullQueue (queue.Queue): """The :class:`queue.Queue` equivalent of `/dev/null`. This is a bottomless pit. Items go in, but never come out. """ def get(self, block=True, timeout=None): """Raise queue.Empty. There's really no need to override the base Queue.get, but I want to know if someone tries to read from a NullQueue. With the default implementation they would just block silently forever :(. """ raise queue.Empty def put(self, item, block=True, timeout=None): """Dump an item into the void. Block and timeout are meaningless, because there is always a free slot available in a bottomless pit. """ pass class PrintQueue (NullQueue): """Debugging :class:`NullQueue` that prints items before dropping them. """ def put(self, item, block=True, timeout=None): """Print `item` and then dump it into the void. """ print 'ITEM:\n%s' % item