hooke/ui/commandline.py

   1 """Defines :class:`CommandLine` for driving Hooke from the command
   2 line.
   3 """
   4
   5 import cmd
   6 import optparse
   7 import readline # including readline makes cmd.Cmd.cmdloop() smarter
   8 import shlex
   9
  10 from ..command import CommandExit, Exit, Command, Argument, StoreValue
  11 from ..interaction import Request, BooleanRequest, ReloadUserInterfaceConfig
  12 from ..ui import UserInterface, CommandMessage
  13
  14
  15 # Define a few helper classes.
  16
  17 class Default (object):
  18     """Marker for options not given on the command line.
  19     """
  20     pass
  21
  22 class CommandLineParser (optparse.OptionParser):
  23     """Implement a command line syntax for a
  24     :class:`hooke.command.Command`.
  25     """
  26     def __init__(self, command, name_fn):
  27         optparse.OptionParser.__init__(self, prog=name_fn(command.name))
  28         self.command = command
  29         self.command_opts = []
  30         self.command_args = []
  31         for a in command.arguments:
  32             if a.name == 'help':
  33                 continue # 'help' is a default OptionParser option
  34             if a.optional == True:
  35                 name = name_fn(a.name)
  36                 self.add_option(
  37                     '--%s' % name, dest=name, default=Default)
  38                 self.command_opts.append(a)
  39             else:
  40                 self.command_args.append(a)
  41         infinite_counters = [a for a in self.command_args if a.count == -1]
  42         assert len(infinite_counters) <= 1, \
  43             'Multiple infinite counts for %s: %s\nNeed a better CommandLineParser implementation.' \
  44             % (command.name, ', '.join([a.name for a in infinite_counters]))
  45         if len(infinite_counters) == 1: # move the big counter to the end.
  46             infinite_counter = infinite_counters[0]
  47             self.command_args.remove(infinite_counter)
  48             self.command_args.append(infinite_counter)
  49
  50     def exit(self, status=0, msg=None):
  51         """Override :meth:`optparse.OptionParser.exit` which calls
  52         :func:`sys.exit`.
  53         """
  54         if msg:
  55             raise optparse.OptParseError(msg)
  56         raise optparse.OptParseError('OptParse EXIT')
  57
  58 class CommandMethod (object):
  59     """Base class for method replacer.
  60
  61     The .__call__ methods of `CommandMethod` subclasses functions will
  62     provide the `do_*`, `help_*`, and `complete_*` methods of
  63     :class:`HookeCmd`.
  64     """
  65     def __init__(self, cmd, command, name_fn):
  66         self.cmd = cmd
  67         self.command = command
  68         self.name_fn = name_fn
  69
  70     def __call__(self, *args, **kwargs):
  71         raise NotImplementedError
  72
  73 class DoCommand (CommandMethod):
  74     def __init__(self, *args, **kwargs):
  75         super(DoCommand, self).__init__(*args, **kwargs)
  76         self.parser = CommandLineParser(self.command, self.name_fn)
  77
  78     def __call__(self, args):
  79         try:
  80             args = self._parse_args(args)
  81         except optparse.OptParseError, e:
  82             self.cmd.stdout.write(str(e).lstrip()+'\n')
  83             self.cmd.stdout.write('Failure\n')
  84             return
  85         self.cmd.inqueue.put(CommandMessage(self.command, args))
  86         while True:
  87             msg = self.cmd.outqueue.get()
  88             if isinstance(msg, Exit):
  89                 return True
  90             elif isinstance(msg, CommandExit):
  91                 self.cmd.stdout.write(msg.__class__.__name__+'\n')
  92                 self.cmd.stdout.write(str(msg).rstrip()+'\n')
  93                 break
  94             elif isinstance(msg, ReloadUserInterfaceConfig):
  95                 self.cmd.ui.reload_config(msg.config)
  96                 continue
  97             elif isinstance(msg, Request):
  98                 self._handle_request(msg)
  99                 continue
 100             self.cmd.stdout.write(str(msg).rstrip()+'\n')
 101
 102     def _parse_args(self, args):
 103         argv = shlex.split(args, comments=True, posix=True)
 104         options,args = self.parser.parse_args(argv)
 105         self._check_argument_length_bounds(args)
 106         params = {}
 107         for argument in self.parser.command_opts:
 108             value = getattr(options, self.name_fn(argument.name))
 109             if value != Default:
 110                 params[argument.name] = value
 111         arg_index = 0
 112         for argument in self.parser.command_args:
 113             if argument.count == 1:
 114                 params[argument.name] = args[arg_index]
 115             elif argument.count > 1:
 116                 params[argument.name] = \
 117                     args[arg_index:arg_index+argument.count]
 118             else: # argument.count == -1:
 119                 params[argument.name] = args[arg_index:]
 120             arg_index += argument.count
 121         return params
 122
 123     def _check_argument_length_bounds(self, arguments):
 124         """Check that there are an appropriate number of arguments in
 125         `args`.
 126
 127         If not, raise optparse.OptParseError().
 128         """
 129         min_args = 0
 130         max_args = 0
 131         for argument in self.parser.command_args:
 132             if argument.optional == False and argument.count > 0:
 133                 min_args += argument.count
 134             if max_args >= 0: # otherwise already infinite
 135                 if argument.count == -1:
 136                     max_args = -1
 137                 else:
 138                     max_args += argument.count
 139         if len(arguments) < min_args \
 140                 or (max_args >= 0 and len(arguments) > max_args):
 141             if min_args == max_args:
 142                 target_string = str(min_args)
 143             elif max_args == -1:
 144                 target_string = 'more than %d' % min_args
 145             else:
 146                 target_string = '%d to %d' % (min_args, max_args)
 147             raise optparse.OptParseError(
 148                 '%d arguments given, but %s takes %s'
 149                 % (len(arguments), self.name_fn(self.command.name),
 150                    target_string))
 151
 152     def _handle_request(self, msg):
 153         """Repeatedly try to get a response to `msg`.
 154         """
 155         prompt = getattr(self, '_%s_request_prompt' % msg.type, None)
 156         if prompt == None:
 157             raise NotImplementedError('_%s_request_prompt' % msg.type)
 158         prompt_string = prompt(msg)
 159         parser = getattr(self, '_%s_request_parser' % msg.type, None)
 160         if parser == None:
 161             raise NotImplementedError('_%s_request_parser' % msg.type)
 162         error = None
 163         while True:
 164             if error != None:
 165                 self.cmd.stdout.write(''.join([
 166                         error.__class__.__name__, ': ', str(error), '\n']))
 167             self.cmd.stdout.write(prompt_string)
 168             value = parser(msg, self.cmd.stdin.readline())
 169             try:
 170                 response = msg.response(value)
 171                 break
 172             except ValueError, error:
 173                 continue
 174         self.cmd.inqueue.put(response)
 175
 176     def _boolean_request_prompt(self, msg):
 177         if msg.default == True:
 178             yn = ' [Y/n] '
 179         else:
 180             yn = ' [y/N] '
 181         return msg.msg + yn
 182
 183     def _boolean_request_parser(self, msg, response):
 184         value = response.strip().lower()
 185         if value.startswith('y'):
 186             value = True
 187         elif value.startswith('n'):
 188             value = False
 189         elif len(value) == 0:
 190             value = msg.default
 191         return value
 192
 193     def _string_request_prompt(self, msg):
 194         if msg.default == None:
 195             d = ' '
 196         else:
 197             d = ' [%s] ' % msg.default
 198         return msg.msg + d
 199
 200     def _string_request_parser(self, msg, response):
 201         return response.strip()
 202
 203     def _float_request_prompt(self, msg):
 204         return self._string_request_prompt(msg)
 205
 206     def _float_request_parser(self, msg, resposne):
 207         return float(response)
 208
 209     def _selection_request_prompt(self, msg):
 210         options = []
 211         for i,option in enumerate(msg.options):
 212             options.append('   %d) %s' % (i,option))
 213         options = ''.join(options)
 214         if msg.default == None:
 215             prompt = '? '
 216         else:
 217             prompt = '? [%d] ' % msg.default
 218         return '\n'.join([msg,options,prompt])
 219
 220     def _selection_request_parser(self, msg, response):
 221         return int(response)
 222
 223
 224 class HelpCommand (CommandMethod):
 225     def __init__(self, *args, **kwargs):
 226         super(HelpCommand, self).__init__(*args, **kwargs)
 227         self.parser = CommandLineParser(self.command, self.name_fn)
 228
 229     def __call__(self):
 230         blocks = [self.command.help(name_fn=self.name_fn),
 231                   '----',
 232                   'Usage: ' + self._usage_string(),
 233                   '']
 234         self.cmd.stdout.write('\n'.join(blocks))
 235
 236     def _message(self):
 237         return self.command.help(name_fn=self.name_fn)
 238
 239     def _usage_string(self):
 240         if len(self.parser.command_opts) == 0:
 241             options_string = ''
 242         else:
 243             options_string = '[options]'
 244         arg_string = ' '.join(
 245             [self.name_fn(arg.name) for arg in self.parser.command_args])
 246         return ' '.join([x for x in [self.parser.prog,
 247                                      options_string,
 248                                      arg_string]
 249                          if x != ''])
 250
 251 class CompleteCommand (CommandMethod):
 252     def __call__(self, text, line, begidx, endidx):
 253         pass
 254
 255
 256 # Define some additional commands
 257
 258 class LocalHelpCommand (Command):
 259     """Called with an argument, prints that command's documentation.
 260
 261     With no argument, lists all available help topics as well as any
 262     undocumented commands.
 263     """
 264     def __init__(self):
 265         super(LocalHelpCommand, self).__init__(name='help', help=self.__doc__)
 266         # We set .arguments now (vs. using th arguments option to __init__),
 267         # to overwrite the default help argument.  We don't override
 268         # :meth:`cmd.Cmd.do_help`, so `help --help` is not a valid command.
 269         self.arguments = [
 270             Argument(name='command', type='string', optional=True,
 271                      help='The name of the command you want help with.')
 272             ]
 273
 274     def _run(self, hooke, inqueue, outqueue, params):
 275         raise NotImplementedError # cmd.Cmd already implements .do_help()
 276
 277 class LocalExitCommand (Command):
 278     """Exit Hooke cleanly.
 279     """
 280     def __init__(self):
 281         super(LocalExitCommand, self).__init__(
 282             name='exit', aliases=['quit', 'EOF'], help=self.__doc__,
 283             arguments = [
 284                 Argument(name='force', type='bool', default=False,
 285                          help="""
 286 Exit without prompting the user.  Use if you save often or don't make
 287 typing mistakes ;).
 288 """.strip()),
 289                 ])
 290
 291     def _run(self, hooke, inqueue, outqueue, params):
 292         """The guts of the `do_exit/_quit/_EOF` commands.
 293
 294         A `True` return stops :meth:`.cmdloop` execution.
 295         """
 296         _exit = True
 297         if params['force'] == False:
 298             not_saved = [p.name for p in hooke.playlists
 299                          if p.is_saved() == False]
 300             msg = 'Exit?'
 301             default = True
 302             if len(not_saved) > 0:
 303                 msg = 'Unsaved playlists (%s).  %s' \
 304                     % (', '.join([str(p) for p in not_saved]), msg)
 305                 default = False
 306             outqueue.put(BooleanRequest(msg, default))
 307             result = inqueue.get()
 308             assert result.type == 'boolean'
 309             _exit = result.value
 310         if _exit == True:
 311             raise Exit()
 312
 313
 314 # Now onto the main attraction.
 315
 316 class HookeCmd (cmd.Cmd):
 317     def __init__(self, ui, commands, inqueue, outqueue):
 318         cmd.Cmd.__init__(self)
 319         self.ui = ui
 320         self.commands = commands
 321         self.local_commands = [LocalExitCommand(), LocalHelpCommand()]
 322         self.prompt = 'hooke> '
 323         self._add_command_methods()
 324         self.inqueue = inqueue
 325         self.outqueue = outqueue
 326
 327     def _name_fn(self, name):
 328         return name.replace(' ', '_')
 329
 330     def _add_command_methods(self):
 331         for command in self.commands + self.local_commands:
 332             for name in [command.name] + command.aliases:
 333                 name = self._name_fn(name)
 334                 setattr(self.__class__, 'help_%s' % name,
 335                         HelpCommand(self, command, self._name_fn))
 336                 if name != 'help':
 337                     setattr(self.__class__, 'do_%s' % name,
 338                             DoCommand(self, command, self._name_fn))
 339                     setattr(self.__class__, 'complete_%s' % name,
 340                             CompleteCommand(self, command, self._name_fn))
 341
 342
 343 class CommandLine (UserInterface):
 344     """Command line interface.  Simple and powerful.
 345     """
 346     def __init__(self):
 347         super(CommandLine, self).__init__(name='command line')
 348
 349     def run(self, commands, ui_to_command_queue, command_to_ui_queue):
 350         cmd = HookeCmd(self, commands,
 351                        inqueue=ui_to_command_queue,
 352                        outqueue=command_to_ui_queue)
 353         cmd.cmdloop(self._splash_text())