"""The plugin module provides optional submodules that add new Hooke commands. All of the science happens in here. """ 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 class Plugin (object): """The pluggable collection of Hooke commands. Fulfills the same role for Hooke that a software package does for an operating system. """ def __init__(self, name): self.name = name def dependencies(self): """Return a list of :class:`Plugin`\s we require.""" return [] def default_settings(self): """Return a list of :class:`hooke.config.Setting`\s for any configurable plugin settings. The suggested section setting is:: Setting(section='%s plugin' % self.name, help=self.__doc__) """ return [] def commands(self): """Return a list of :class:`Commands` provided.""" return [] # 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, aliases=None, arguments=[], help=''): self.name = name if aliases == None: aliases = [] self.aliases = aliases self.arguments = [ Argument(name='help', type='bool', default=False, count=1, callback=StoreValue(True), help='Print a help message.'), ] + arguments 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, aliases=None, type='string', metavar=None, default=None, optional=True, count=1, completion_callback=None, callback=None, help=''): self.name = name if aliases == None: aliases = [] self.aliases = aliases self.type = type if metavar == None: metavar = type.upper() self.metavar = metavar self.default = default self.optional = optional self.count = count self.completion_callback = completion_callback self.callback = callback 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 # Construct plugin dependency graph and load default plugins. PLUGINS = {} """(name,instance) :class:`dict` of all possible :class:`Plugin`\s. """ for plugin_modname,default_include in PLUGIN_MODULES: assert len([mod_name for mod_name,di in PLUGIN_MODULES]) == 1, \ 'Multiple %s entries in PLUGIN_MODULES' % mod_name 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) try: subclass = issubclass(obj, Plugin) except TypeError: continue if subclass == True and obj != Plugin: p = obj() if p.name != plugin_modname: raise Exception('Plugin name %s does not match module name %s' % (p.name, plugin_modname)) PLUGINS[p.name] = p PLUGIN_GRAPH = Graph([Node([PLUGINS[name] for name in p.dependencies()], data=p) for p in PLUGINS.values()]) PLUGIN_GRAPH.topological_sort() def default_settings(): settings = [Setting( 'plugins', help='Enable/disable default plugins.')] for pnode in PLUGIN_GRAPH: plugin = pnode.data default_include = [di for mod_name,di in PLUGIN_MODULES if mod_name == plugin.name][0] help = 'Commands: ' + ', '.join([c.name for c in p.commands()]) settings.append(Setting( section='plugins', option=plugin.name, value=str(default_include), help=help, )) for pnode in PLUGIN_GRAPH: plugin = pnode.data settings.extend(plugin.default_settings()) return settings