Add a small utility to send pymin commands.

[software/pymin.git] / pymin / dispatcher.py

diff --git a/pymin/dispatcher.py b/pymin/dispatcher.py
index 39b4b75cf6522f64e78b2d097422a2bfc69285e1..24f0d1f14e6a52b6f97fd49664d767b07382a99a 100644 (file)
--- a/pymin/dispatcher.py
+++ b/pymin/dispatcher.py
@@ -7,6 +7,8 @@ It's based on Zope or Cherrypy dispatching (but implemented from the scratch)
 and translates commands to functions/objects/methods.
 """
 
 and translates commands to functions/objects/methods.
 """
 

+import re
+

 __ALL__ = ('Error', 'HandlerError', 'CommandNotFoundError', 'Handler',
             'Dispatcher', 'handler', 'is_handler', 'get_help')
 
 __ALL__ = ('Error', 'HandlerError', 'CommandNotFoundError', 'Handler',
             'Dispatcher', 'handler', 'is_handler', 'get_help')
 


@@ -52,6 +54,20 @@ class CommandError(Error):
     def __unicode__(self):
         return u'Error in command "%s".' % u' '.join(self.command)
 
     def __unicode__(self):
         return u'Error in command "%s".' % u' '.join(self.command)
 

+class WrongArgumentsError(CommandError):
+    r"""WrongArgumentsError(handler, message) -> WrongArgumentsError instance.
+
+    This exception is raised when an empty command string is received.
+    """
+
+    def __init__(self, handler, message):
+        r"Initialize the object, see class documentation for more info."
+        self.handler = handler
+        self.message = message
+
+    def __unicode__(self):
+        return u'Command "%s" %s.' % (self.handler.__name__, self.message)
+

 class CommandNotSpecifiedError(CommandError):
     r"""CommandNotSpecifiedError() -> CommandNotSpecifiedError instance.
 
 class CommandNotSpecifiedError(CommandError):
     r"""CommandNotSpecifiedError() -> CommandNotSpecifiedError instance.
 


@@ -186,17 +202,31 @@ class Handler:
             d = dict()
             for a in dir(self):
                 h = getattr(self, a)
             d = dict()
             for a in dir(self):
                 h = getattr(self, a)

+                if a == 'parent': continue # Skip parents in SubHandlers

                 if is_handler(h) or isinstance(h, Handler):
                     d[a] = h.handler_help
             return d
         # A command was specified
                 if is_handler(h) or isinstance(h, Handler):
                     d[a] = h.handler_help
             return d
         # A command was specified

+        if command == 'parent': # Skip parents in SubHandlers
+            raise HelpNotFoundError(command)

         if not hasattr(self, command.encode('utf-8')):
             raise HelpNotFoundError(command)
         handler = getattr(self, command.encode('utf-8'))
         if not hasattr(self, command.encode('utf-8')):
             raise HelpNotFoundError(command)
         handler = getattr(self, command.encode('utf-8'))

-        if not is_handler(handler) and not hasattr(handler):
+        if not is_handler(handler) and not hasattr(handler, 'handler_help'):

             raise HelpNotFoundError(command)
         return handler.handler_help
 
             raise HelpNotFoundError(command)
         return handler.handler_help
 

+    def handle_timer(self):
+        r"""handle_timer() -> None :: Do periodic tasks.
+
+        By default we do nothing but calling handle_timer() on subhandlers.
+        """
+        for a in dir(self):
+            if a == 'parent': continue # Skip parents in SubHandlers
+            h = getattr(self, a)
+            if isinstance(h, Handler):
+                h.handle_timer()
+

 def parse_command(command):
     r"""parse_command(command) -> (args, kwargs) :: Parse a command.
 
 def parse_command(command):
     r"""parse_command(command) -> (args, kwargs) :: Parse a command.
 


@@ -223,6 +253,8 @@ def parse_command(command):
     arguments is preserved and if there are multiple keyword arguments with
     the same key, the last value is the winner (all other values are lost).
 
     arguments is preserved and if there are multiple keyword arguments with
     the same key, the last value is the winner (all other values are lost).
 

+    The command should be a unicode string.
+

     Examples:
 
     >>> parse_command('hello world')
     Examples:
 
     >>> parse_command('hello world')


@@ -250,6 +282,12 @@ def parse_command(command):
     ([u'=hello'], {})
     >>> parse_command(r'\thello')
     ([u'\thello'], {})
     ([u'=hello'], {})
     >>> parse_command(r'\thello')
     ([u'\thello'], {})

+    >>> parse_command(r'hello \n')
+    ([u'hello', u'\n'], {})
+    >>> parse_command(r'hello \nmundo')
+    ([u'hello', u'\nmundo'], {})
+    >>> parse_command(r'test \N')
+    ([u'test', None], {})

     >>> parse_command(r'\N')
     ([None], {})
     >>> parse_command(r'none=\N')
     >>> parse_command(r'\N')
     ([None], {})
     >>> parse_command(r'none=\N')


@@ -274,9 +312,23 @@ def parse_command(command):
     escape = False
     keyword = None
     state = SEP
     escape = False
     keyword = None
     state = SEP

+    def register_token(buff, keyword, seq, dic):
+        if buff == r'\N':
+            buff = None
+        if keyword is not None:
+            dic[keyword.encode('utf-8')] = buff
+            keyword = None
+        else:
+            seq.append(buff)
+        buff = u''
+        return (buff, keyword)

     for n, c in enumerate(command):
         # Escaped character
         if escape:
     for n, c in enumerate(command):
         # Escaped character
         if escape:

+            # Not yet registered the token
+            if state == SEP and buff:
+                (buff, keyword) = register_token(buff, keyword, seq, dic)
+                state = TOKEN

             for e in escaped_chars:
                 if c == e:
                     buff += eval(u'"\\' + e + u'"')
             for e in escaped_chars:
                 if c == e:
                     buff += eval(u'"\\' + e + u'"')


@@ -301,14 +353,7 @@ def parse_command(command):
                     keyword = buff
                     buff = u''
                     continue
                     keyword = buff
                     buff = u''
                     continue

-                if buff == r'\N':
-                    buff = None
-                if keyword is not None: # Value found
-                    dic[str(keyword)] = buff
-                    keyword = None
-                else: # Normal parameter found
-                    seq.append(buff)
-                buff = u''
+                (buff, keyword) = register_token(buff, keyword, seq, dic)

             state = TOKEN
         # Getting a token
         if state == TOKEN:
             state = TOKEN
         # Getting a token
         if state == TOKEN:


@@ -350,14 +395,12 @@ def parse_command(command):
         raise ParseError(command,
                         u'keyword argument (%s) without value' % keyword)
     if buff:
         raise ParseError(command,
                         u'keyword argument (%s) without value' % keyword)
     if buff:

-        if buff == r'\N':
-            buff = None
-        if keyword is not None:
-            dic[str(keyword)] = buff
-        else:
-            seq.append(buff)
+        register_token(buff, keyword, seq, dic)

     return (seq, dic)
 
     return (seq, dic)
 

+args_re = re.compile(r'\w+\(\) takes (.+) (\d+) \w+ \((\d+) given\)')
+kw_re = re.compile(r'\w+\(\) got an unexpected keyword argument (.+)')
+

 class Dispatcher:
     r"""Dispatcher([root]) -> Dispatcher instance :: Command dispatcher.
 
 class Dispatcher:
     r"""Dispatcher([root]) -> Dispatcher instance :: Command dispatcher.
 


@@ -413,13 +456,35 @@ class Dispatcher:
                     raise CommandIsAHandlerError(command)
                 raise CommandNotFoundError(command)
             command.append(route[0])
                     raise CommandIsAHandlerError(command)
                 raise CommandNotFoundError(command)
             command.append(route[0])

+            if route[0] == 'parent':
+                raise CommandNotFoundError(command)

             if not hasattr(handler, route[0].encode('utf-8')):
                 if isinstance(handler, Handler) and len(command) > 1:
                     raise CommandNotInHandlerError(command)
                 raise CommandNotFoundError(command)
             handler = getattr(handler, route[0].encode('utf-8'))
             route = route[1:]
             if not hasattr(handler, route[0].encode('utf-8')):
                 if isinstance(handler, Handler) and len(command) > 1:
                     raise CommandNotInHandlerError(command)
                 raise CommandNotFoundError(command)
             handler = getattr(handler, route[0].encode('utf-8'))
             route = route[1:]

-        return handler(*route, **kwargs)
+        try:
+            return handler(*route, **kwargs)
+        except TypeError, e:
+            m = args_re.match(unicode(e))
+            if m:
+                (quant, n_ok, n_bad)  = m.groups()
+                n_ok = int(n_ok)
+                n_bad = int(n_bad)
+                n_ok -= 1
+                n_bad -= 1
+                pl = ''
+                if n_ok > 1:
+                    pl = 's'
+                raise WrongArgumentsError(handler, u'takes %s %s argument%s, '
+                            '%s given' % (quant, n_ok, pl, n_bad))
+            m = kw_re.match(unicode(e))
+            if m:
+                (kw,)  = m.groups()
+                raise WrongArgumentsError(handler,
+                        u'got an unexpected keyword argument %s' % kw)
+            raise

 
 
 if __name__ == '__main__':
 
 
 if __name__ == '__main__':


@@ -501,6 +566,12 @@ if __name__ == '__main__':
     assert p == ([u'=hello'], {}), p
     p = parse_command(r'\thello')
     assert p == ([u'\thello'], {}), p
     assert p == ([u'=hello'], {}), p
     p = parse_command(r'\thello')
     assert p == ([u'\thello'], {}), p

+    p = parse_command(r'hello \n')
+    assert p == ([u'hello', u'\n'], {}), p
+    p = parse_command(r'hello \nmundo')
+    assert p == ([u'hello', u'\nmundo'], {}), p
+    p = parse_command(r'test \N')
+    assert p == ([u'test', None], {}), p

     p = parse_command(r'\N')
     assert p == ([None], {}), p
     p = parse_command(r'none=\N')
     p = parse_command(r'\N')
     assert p == ([None], {}), p
     p = parse_command(r'none=\N')