"""
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),
+# ('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.
+default. TODO: autodiscovery
"""
+# Plugins and settings
+
def Plugin (object):
"""The pluggable collection of Hooke commands.
obj.module_name = plugin_modname
PLUGINS[p.name] = p
-PLUGIN_GRAPH = Graph([Node(
- [PLUGINS[name] for name in p.dependencies()]
- )])
+PLUGIN_GRAPH = Graph([Node([PLUGINS[name] for name in p.dependencies()])
+ for p in PLUGINS.values()])
PLUGIN_GRAPH.topological_sort()
settings.extend(plugin.default_settings())
return settings
-class Command (object):
+
+# Commands and arguments
+
+class CommandExit (Exception):
+ def __str__(self):
+ return self.__class__.__name__
+
+class Success (CommandExit):
pass
+
+class Failure (CommandExit):
+ pass
+
+
+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
+
+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
+ <BLANKLINE>
+ Arguments:
+ help HELP (bool) Print a help message.
+ <BLANKLINE>
+ 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 <Argument help> 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])
projects / hooke.git / commitdiff