root/cly/tags/final-dictionary-based-version/cly/util.py

from crash.console import *
from cly.symbols import *
import cly.parser, cly.rlext
from cly.parser import Parser, Result
import os
import os.path
import inspect
import pydoc
import sys
import types
import re
import readline
from fnmatch import fnmatch

"""

A collection of functions useful in conjunction with CLY.

"""


__all__ = [ "file_completer", "introspect_grammar", "format_help", "execute_result", "execute_command", "interact", "interact_loop", ]

# XXX Used by the interactive functions XXX
__cli_inject_text = ''
__completion_candidates = []
__parser = None

class file_completer:
    """File tab completion for CLY.
    Usage:
        ... COMPLETE : cly.util.file_completer('*.py')
    """
    def __init__(self, *args):
        self.__args = args

    def __call__(self, ctx, sofar):
        return self.__complete(sofar, *self.__args)

    def __complete(self, sofar, pattern = '*', show_hidden = False):
        base = sofar
        remainder = ''
        files = []
        # Is the current text a directory?
        if not os.path.exists(base):
            remainder = os.path.basename(base)
            base = os.path.dirname(base)
        if os.path.isdir(base or '.'):
            files = os.listdir(base or '.')
            files.sort()
            # Filter file list
            files = [os.path.join(base, file) for file in files if \
                (not show_hidden and file[0] != '.') \
                and file.startswith(remainder) \
                and (os.path.isdir(os.path.join(base, file)) or fnmatch(file, pattern)) \
                ]
            files = [os.path.isdir(file) and file + '/' or file for file in files]
        if len(files) == 1 and not os.path.isdir(files[0]):
            return [files[0] + ' ']
        return files

def dump_traceback(exception):
    import traceback
    from StringIO import StringIO
    out = StringIO()
    traceback.print_exc(file = out)
    error(str(exception), "\n^N^1", out.getvalue())
    raise exception

def introspect_grammar(instance, split_names = True):
    """ Generate a CLY grammar from a class instance. 

        Methods in the form "def some_command(self, arg): ..." are converted to
        the grammar:
            
            'some' : {
                HELP : 'some',
                'command' : {
                    HELP : pydoc.getdoc(instance.some_command),
                    '.+' : {
                        VAR : 's:arg',
                        ACTION : instance.some_command,
                    },
                },
            },

        CamelCase function names are also translated in a similar fashion.

        In addition, class members that are also instances of classes, will be
        inserted into the grammar in a similar fashion.
    """
    grammar = {}
    namere = re.compile(r'''([A-Z]{2,}|[A-Z]+[a-z]*|[a-z]+)''')
    for fullname, method in inspect.getmembers(instance, lambda m: type(m) is types.InstanceType or inspect.ismethod(m)):
        if fullname[0] == '_':
            continue

        if split_names:
            name = [x.lower() for x in namere.findall(' '.join(fullname.split('_')))]
        else:
            name = [ fullname ]
        # Merge generated name into grammar
        def merge(grammar, name):
            if not name:
                return grammar
            return merge(grammar.setdefault(name[0], {
                HELP : name[0],
                }), name[1:])
        g = merge(grammar, name)

        label = 'execute_%s' % fullname

        g[HELP] = pydoc.getdoc(method) or name[-1]
        g[LABEL] = label

        if type(method) is types.InstanceType:
            g.update(introspect_grammar(method, split_names = split_names))
            continue

        # Insert args
        argspec = inspect.getargspec(method)
        args = argspec[0]
        if args and args[0] == 'self':
            args = args[1:]
        if argspec[3] is not None:
            defaults = dict(zip(args[-len(argspec[3]):], argspec[3]))
        else:
            defaults = {}

        def generate_argg(arg):
            help = '%s (^B%s^N)' % (arg, arg in defaults and '%s' % str(defaults[arg]) or '^Urequired')
            return help, {
                UNLESS_VAR : arg,
                VAR : arg,
                HELP : ('<%s>' % arg, help),
                JUMP : label,
            }

        if len(args) == 1:
            arg = args[0]
            help, argg = generate_argg(arg)
            g['.*'] = argg
        else:
            for arg in args:
                help, argg = generate_argg(arg)
                g[arg] = {
                    UNLESS_VAR : arg,
                    HELP : help,
                    '.*' : argg,
                }

        class validate_args:
            def __init__(self, args, defaults):
                from copy import copy
                self.args = args
                self.defaults = defaults

            def __call__(self, ctx):
                for arg in self.args:
                    if arg not in self.defaults and arg not in ctx:
                        return False
                return True
        
        g[ACTION] = {
            IF : validate_args(args, defaults),
            ACTION : method,
            HELP : pydoc.getdoc(method) or 'Execute command',
        }

    return grammar

def __cli_completion(text, state):
    global __completion_candidates
    line = readline.get_line_buffer()[0:readline.get_begidx()]
    ctx = None
    try:
        result = __parser.parse(line)
        if not state:
            __completion_candidates = result.candidates(text)
        if __completion_candidates:
            return __completion_candidates.pop()
        return None
    except cly.parser.Error:
        return None
    except Exception, e:
        print
        dump_traceback(e)
        cly.rlext.force_redisplay()

def __cli_injector():
    try:
        global __cli_inject_text
        readline.insert_text(__cli_inject_text)
        __cli_inject_text = ''
    except e:
        dump_traceback(e)

def __generate_help(parser, line, cursor_offset, need_linefeed = False):
    try:
        result = parser.parse(line)
        # We are not at the end of the line
        if result.token:
            readline.insert_text("?")
            return False
        if result.context.parsed_tokens and result.context.parsed_tokens[-1].start() >= cursor_offset:
            print result.context.parsed_tokens[-1].start()
            return False
        if need_linefeed:
            print
        format_help(result.help())
        return True
    except Exception, e:
        dump_traceback(e)

def __cli_help(key, count):
    global __parser
    try:
        if __generate_help(__parser, readline.get_line_buffer(), cly.rlext.cursor(), True):
            cly.rlext.force_redisplay()
    except Exception, e:
        print
        try:
            dump_traceback(e)
        finally:
            cly.rlext.force_redisplay()
            raise

def format_help(help):
    """ Format the output of cly.parser.Result.help() in a useful way. """
    header, help, footer = help
    length = 0
    for group in help:
        for h in help[group]:
            if len(h[0]) > length:
                length = len(h[0])
    if header:
        cprint(wraptoterm(header))
        print
    count = 0
    groups = len(help.keys())
    last_group = last_order = None
    for order, group in sorted(help, lambda a, b: cmp(a[0], b[0])):
        subcount = 0
        for h in help[(order, group)]:
            if last_order != None and order != last_order:
                last_order = order

            cprint(wraptoterm("  ^B%s^N %s" % (h[0] + (length - len(h[0])) * ' ', h[1]), subsequent_indent = ' ' * (length + 3)))
            last_order = order
        count += 1
        if count != groups:
            print
    if footer:
        print
        cprint(wraptoterm(footer))

def execute_result(result):
    """ Execute the Result object passed. """
    if result.state == Result.NOP:
        pass
    elif result.state == Result.OK:
        result()
    else:
        if result.token:
            error("%s (near '%s')^N" % (result.message, result.token.group(0)))
        else:
            error(result.message)

def execute_command(parser, command):
    """ Parse and execute the given command, returning the Result. """
    __parser = parser
    command = re.sub(r'^\s+|\s+$', '', command)
    if command == '':
        return Result(Result.NOP)

    # Looking for help?
    if command[-1] == '?':
        command = re.sub(r'\?\s*$', '', command)
        __generate_help(parser, command, len(command))
        __cli_inject_text = command
        return Result(Result.NOP)
    else:
        result =  parser.parse(command)
        execute_result(result)
        return result

def interact(parser, prompt = 'cly> ', default_text=''):
    """ Input one command from the user and return the Result of the executed
        command. """
    global __parser, __cli_inject_text

    __cli_inject_text = default_text
    # - and : are acceptable "word" characters
    readline.set_completer_delims('`~!#$%^&*()=+[{]}\|;\'",<>? \t')
    readline.set_completer(__cli_completion)
    readline.set_startup_hook(__cli_injector)

    # Use custom readline extensions not in 2.4
    cly.rlext.bind_key(ord('?'), __cli_help)

    while True:
        __parser = parser
        command = ''
        try:
            command = raw_input(prompt)
        except KeyboardInterrupt:
            print
            continue
        except EOFError:
            print
            return Result(Result.ENDOFINPUT)

        return execute_command(parser, command)

def interact_loop(grammar, prompt='cly> ',
                  history=os.path.expanduser('~/.clyhistory'),
                  user_context=None, history_length=500):
    """ Repeatedly read and execute commands from the user. """
    try:
        readline.set_history_length(history_length)
        readline.read_history_file(history)
    except:
        pass
    parser = Parser(grammar, user_context)
    try:
        while interact(parser, prompt).state != Result.ENDOFINPUT:
            pass
    finally:
        try:
            readline.write_history_file(history)
        except:
            pass