hooke/ui/commandline.py

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