class CommandExit (Exception):
- def __str__(self):
- return self.__class__.__name__
+ pass
class Success (CommandExit):
pass
class UncaughtException (Failure):
def __init__(self, exception):
- super(UncaughtException, self).__init__(exception)
self.exception = exception
self.exc_string = traceback.format_exc()
sys.exc_clear()
+ super(UncaughtException, self).__init__(self.exc_string)
class Command (object):
"""One-line command description here.
self._run(hooke, inqueue, outqueue, params)
except CommandExit, e:
if isinstance(e, Failure):
- outqueue.put(str(e))
outqueue.put(e)
return 1
+ # other CommandExit subclasses fall through to the end
except Exception, e:
x = UncaughtException(e)
- outqueue.put(x.exc_string)
outqueue.put(x)
return 1
+ else:
+ e = Success()
outqueue.put(e)
return 0
self.load_ui()
self.command = engine.CommandEngine()
- self.playlists = playlist.PlaylistManager()
+ self.playlists = playlist.NoteIndexList()
def load_plugins(self):
self.plugins = plugin_mod.load_graph(
from . import curve as curve
-class Playlist (list):
- """A list of :class:`hooke.curve.Curve`\s.
- Keeps a list of :attr:`drivers` for loading curves, the
- :attr:`index` (i.e. "bookmark") of the currently active curve, and
- a :class:`dict` of additional informtion (:attr:`info`).
+class NoteIndexList (list):
+ """A list that keeps track of a "current" item and additional notes.
+
+ :attr:`index` (i.e. "bookmark") is the index of the currently
+ current curve. Also keep a :class:`dict` of additional information
+ (:attr:`info`).
"""
- def __init__(self, drivers, name=None):
- super(Playlist, self).__init__()
- self.drivers = drivers
+ def __init__(self, name=None):
+ super(NoteIndexList, self).__init__()
self.name = name
self.info = {}
self._index = 0
- def append_curve_by_path(self, path, info=None, identify=True):
- if self.path != None:
- path = os.path.join(self.path, path)
- path = os.path.normpath(path)
- c = curve.Curve(path, info=info)
- if identify == True:
- c.identify(self.drivers)
- self.append(c)
- return c
-
- def active_curve(self):
+ def current(self):
+ if len(self) == 0:
+ return None
return self[self._index]
- def has_curves(self):
- return len(self) > 0
-
def jump(self, index):
if len(self) == 0:
self._index = 0
def previous(self):
self.jump(self._index - 1)
- def filter(self, keeper_fn=lambda curve:True):
- playlist = copy.deepcopy(self)
- for curve in reversed(playlist.curves):
- if keeper_fn(curve) != True:
- playlist.curves.remove(curve)
- try: # attempt to maintain the same active curve
- playlist._index = playlist.index(self.active_curve())
+ def filter(self, keeper_fn=lambda item:True):
+ c = copy.deepcopy(self)
+ for item in reversed(c):
+ if keeper_fn(item) != True:
+ c.remove(item)
+ try: # attempt to maintain the same current item
+ c._index = c.index(self.current())
except ValueError:
- playlist._index = 0
- return playlist
+ c._index = 0
+ return c
+
+class Playlist (NoteIndexList):
+ """A :class:`NoteIndexList` of :class:`hooke.curve.Curve`\s.
+
+ Keeps a list of :attr:`drivers` for loading curves.
+ """
+ def __init__(self, drivers, name=None):
+ super(Playlist, self).__init__(name=name)
+ self.drivers = drivers
+ def append_curve_by_path(self, path, info=None, identify=True):
+ if self.path != None:
+ path = os.path.join(self.path, path)
+ path = os.path.normpath(path)
+ c = curve.Curve(path, info=info)
+ if identify == True:
+ c.identify(self.drivers)
+ self.append(c)
+ return c
class FilePlaylist (Playlist):
version = '0.1'
"""Defines :class:`CutPlugin` and :class:`CutCommand`.
"""
-from ..command import Command, Argument
+from ..command import Command, Argument, Failure
from ..plugin import Plugin
def commands(self):
return [CutCommand()]
+
+# Define common or complicated arguments
+
+def current_curve_callback(hooke, command, argument, value):
+ playlist = hooke.playlists.current()
+ if playlist == None:
+ raise Failure('No playlists loaded')
+ curve = playlist.current()
+ if curve == None:
+ raise Failure('No curves in playlist %s' % playlist.name)
+ return curve
+
+CurveArgument = Argument(
+ name='curve', type='curve', callback=current_curve_callback,
+ help="""
+:class:`hooke.curve.Curve` to cut from. Defaults to the current curve.
+""".strip())
+
+
+# Define commands
+
class CutCommand (Command):
"""Cut the selected signal between two points and write it to a file.
super(CutCommand, self).__init__(
name='cut',
arguments=[
- Argument(name='curve', type='curve', help="""
-:class:`hooke.curve.Curve` to cut from. Defaults to the current curve.
-""".strip()),
+ CurveArgument,
Argument(name='block', aliases=['set'], type='int', default=0,
help="""
Data block to save. For an approach/retract force curve, `0` selects
import glob
-from ..command import Command, Argument
+from ..command import Command, Argument, Failure
from ..playlist import FilePlaylist
from ..plugin import Builtin
AddCommand(), AddGlobCommand(),
RemoveCommand(), FilterCommand()]
+
# Define common or complicated arguments
def current_playlist_callback(hooke, command, argument, value):
- return hooke.playlists.current()
+ playlist = hooke.playlists.current()
+ if playlist == None:
+ raise Failure('No playlists loaded')
+ return playlist
PlaylistArgument = Argument(
name='playlist', type='playlist', callback=current_playlist_callback,
help="""
:class:`hooke.plugin.playlist.Playlist` to act on. Defaults to the
current playlist.
-""".strip()),
+""".strip())
def playlist_name_callback(hooke, command, argument, value):
return hooke.playlists.free_name()
""" % version()).strip()
def _playlist_status(self, playlist):
- if playlist.has_curves():
+ if len(playlist) > 0:
return '%s (%s/%s)' % (playlist.name, playlist._index + 1,
len(playlist))
return 'The playlist %s does not contain any valid force curve data.' \
import shlex
from ..command import CommandExit, Command, Argument
+from ..compat.odict import odict
from ..ui import UserInterface, CommandMessage
-# Define a few helper classes. The .__call__ methods of these
-# functions will provide the do_*, help_*, and complete_* methods of
-# HookeCmd.
+# Define a few helper classes.
+
+class CommandLineParser (optparse.OptionParser):
+ """Implement a command line syntax for a
+ :class:`hooke.command.Command`.
+ """
+ def __init__(self, command):
+ optparse.OptionParser.__init__(self, prog=command.name)
+ self.command = command
+ self.command_opts = odict()
+ self.command_args = odict()
+ for a in command.arguments:
+ if a.name == 'help':
+ continue # 'help' is a default OptionParser option
+ name = a.name.replace('_', '-')
+ if a.optional == True:
+ self.add_option('--%s' % name, dest=a.name, default=a.default)
+ self.command_opts[name] = a
+ else:
+ self.command_args[name] = a
+
+ def exit(self, status=0, msg=None):
+ """Override :meth:`optparse.OptionParser.exit` which calls
+ :func:`sys.exit`.
+ """
+ if msg:
+ raise optparse.OptParseError(msg)
+ raise optparse.OptParseError('OptParse EXIT')
class CommandMethod (object):
+ """Base class for method replacer.
+
+ The .__call__ methods of `CommandMethod` subclasses functions will
+ provide the `do_*`, `help_*`, and `complete_*` methods of
+ :class:`HookeCmd`.
+ """
def __init__(self, cmd, command):
self.cmd = cmd
self.command = command
def __call__(self, *args, **kwargs):
raise NotImplementedError
-def command_parser(command):
- p = optparse.OptionParser()
- opts = []
- args = []
- for a in command.arguments:
- if a.name == 'help':
- continue # 'help' is a default OptionParser option
- name = a.name.replace('_', '-')
- if a.optional == True:
- p.add_option('--%s' % name, dest=a.name, default=a.default)
- opts.append((name, a))
- else:
- args.append((name, a))
- return (p, opts, args)
-
class DoCommand (CommandMethod):
def __init__(self, *args, **kwargs):
super(DoCommand, self).__init__(*args, **kwargs)
- self.parser,self.opts,self.args = command_parser(self.command)
+ self.parser = CommandLineParser(self.command)
def __call__(self, args):
- args = self._parse_args(self.command, args)
+ try:
+ args = self._parse_args(self.command, args)
+ except optparse.OptParseError, e:
+ self.cmd.stdout.write(str(e))
+ self.cmd.stdout.write('Failure\n')
+ return
self.cmd.inqueue.put(CommandMessage(self.command, args))
while True:
msg = self.cmd.outqueue.get()
if isinstance(msg, CommandExit):
+ self.cmd.stdout.write(msg.__class__.__name__+'\n')
+ self.cmd.stdout.write(str(msg).rstrip()+'\n')
break
self.cmd.stdout.write(str(msg).rstrip()+'\n')
argv = shlex.split(args, comments=True, posix=True)
options,args = self.parser.parse_args(argv)
params = {}
- for namearg in self.opts:
- name,argument = namearg
+ for name,argument in self.parser.command_opts.items():
params[name] = getattr(options, name)
- for namearg,value in zip(self.args, args):
- name,argument = namearg
+ for name,argument in self.parser.command_args.items():
params[name] = value
return params
class HelpCommand (CommandMethod):
def __init__(self, *args, **kwargs):
super(HelpCommand, self).__init__(*args, **kwargs)
- self.parser,self.opts,self.args = command_parser(self.command)
+ self.parser = CommandLineParser(self.command)
def __call__(self):
blocks = [self.command.help(),
return self.command.help()
def _usage_string(self):
- if len(self.args) == len(self.command.arguments):
+ if len(self.parser.command_opts) == 0:
options_string = ''
else:
options_string = '[options]'
- arg_string = ' '.join([name for name,arg in self.args])
+ arg_string = ' '.join([name for name,arg
+ in self.parser.command_args.items()])
return ' '.join([x for x in [self.command.name,
options_string,
arg_string]
def __init__(self):
super(CommandLine, self).__init__(name='command line')
- def run(self, hooke, ui_to_command_queue, command_to_ui_queue):
- cmd = HookeCmd(hooke,
+ def run(self, commands, ui_to_command_queue, command_to_ui_queue):
+ cmd = HookeCmd(commands,
inqueue=ui_to_command_queue,
outqueue=command_to_ui_queue)
cmd.cmdloop(self._splash_text())
projects / hooke.git / commitdiff