Archipelago » qemu

#!/usr/bin/python

#

# Low-level QEMU shell on top of QMP.

#

# Copyright (C) 2009, 2010 Red Hat Inc.

#

# Authors:

#  Luiz Capitulino <lcapitulino@redhat.com>

#

# This work is licensed under the terms of the GNU GPL, version 2.  See

# the COPYING file in the top-level directory.

#

# Usage:

#

# Start QEMU with:

#

# # qemu [...] -qmp unix:./qmp-sock,server

#

# Run the shell:

#

# $ qmp-shell ./qmp-sock

#

# Commands have the following format:

#

#    < command-name > [ arg-name1=arg1 ] ... [ arg-nameN=argN ]

#

# For example:

#

# (QEMU) device_add driver=e1000 id=net1

# {u'return': {}}

# (QEMU)

import qmp

import readline

import sys

class QMPCompleter(list):

    def complete(self, text, state):

        for cmd in self:

            if cmd.startswith(text):

                if not state:

                    return cmd

                else:

                    state -= 1

class QMPShellError(Exception):

    pass

class QMPShellBadPort(QMPShellError):

    pass

# TODO: QMPShell's interface is a bit ugly (eg. _fill_completion() and

#       _execute_cmd()). Let's design a better one.

class QMPShell(qmp.QEMUMonitorProtocol):

    def __init__(self, address):

        qmp.QEMUMonitorProtocol.__init__(self, self.__get_address(address))

        self._greeting = None

        self._completer = None

    def __get_address(self, arg):

        """

        Figure out if the argument is in the port:host form, if it's not it's

        probably a file path.

        """

        addr = arg.split(':')

        if len(addr) == 2:

            try:

                port = int(addr[1])

            except ValueError:

                raise QMPShellBadPort

            return ( addr[0], port )

        # socket path

        return arg

    def _fill_completion(self):

        for cmd in self.cmd('query-commands')['return']:

            self._completer.append(cmd['name'])

    def __completer_setup(self):

        self._completer = QMPCompleter()

        self._fill_completion()

        readline.set_completer(self._completer.complete)

        readline.parse_and_bind("tab: complete")

        # XXX: default delimiters conflict with some command names (eg. query-),

        # clearing everything as it doesn't seem to matter

        readline.set_completer_delims('')

    def __build_cmd(self, cmdline):

        """

        Build a QMP input object from a user provided command-line in the

        following format:

            < command-name > [ arg-name1=arg1 ] ... [ arg-nameN=argN ]

        """

        cmdargs = cmdline.split()

        qmpcmd = { 'execute': cmdargs[0], 'arguments': {} }

        for arg in cmdargs[1:]:

            opt = arg.split('=')

            try:

                value = int(opt[1])

            except ValueError:

                value = opt[1]

            qmpcmd['arguments'][opt[0]] = value

        return qmpcmd

    def _execute_cmd(self, cmdline):

        try:

            qmpcmd = self.__build_cmd(cmdline)

        except:

            print 'command format: <command-name> ',

            print '[arg-name1=arg1] ... [arg-nameN=argN]'

            return True

        resp = self.cmd_obj(qmpcmd)

        if resp is None:

            print 'Disconnected'

            return False

        print resp

        return True

    def connect(self):

        self._greeting = qmp.QEMUMonitorProtocol.connect(self)

        self.__completer_setup()

    def show_banner(self, msg='Welcome to the QMP low-level shell!'):

        print msg

        version = self._greeting['QMP']['version']['qemu']

        print 'Connected to QEMU %d.%d.%d\n' % (version['major'],version['minor'],version['micro'])

    def read_exec_command(self, prompt):

        """

        Read and execute a command.

        @return True if execution was ok, return False if disconnected.

        """

        try:

            cmdline = raw_input(prompt)

        except EOFError:

            print

            return False

        if cmdline == '':

            for ev in self.get_events():

                print ev

            self.clear_events()

            return True

        else:

            return self._execute_cmd(cmdline)

class HMPShell(QMPShell):

    def __init__(self, address):

        QMPShell.__init__(self, address)

        self.__cpu_index = 0

    def __cmd_completion(self):

        for cmd in self.__cmd_passthrough('help')['return'].split('\r\n'):

            if cmd and cmd[0] != '[' and cmd[0] != '\t':

                name = cmd.split()[0] # drop help text

                if name == 'info':

                    continue

                if name.find('|') != -1:

                    # Command in the form 'foobar|f' or 'f|foobar', take the

                    # full name

                    opt = name.split('|')

                    if len(opt[0]) == 1:

                        name = opt[1]

                    else:

                        name = opt[0]

                self._completer.append(name)

                self._completer.append('help ' + name) # help completion

    def __info_completion(self):

        for cmd in self.__cmd_passthrough('info')['return'].split('\r\n'):

            if cmd:

                self._completer.append('info ' + cmd.split()[1])

    def __other_completion(self):

        # special cases

        self._completer.append('help info')

    def _fill_completion(self):

        self.__cmd_completion()

        self.__info_completion()

        self.__other_completion()

    def __cmd_passthrough(self, cmdline, cpu_index = 0):

        return self.cmd_obj({ 'execute': 'human-monitor-command', 'arguments':

                              { 'command-line': cmdline,

                                'cpu-index': cpu_index } })

    def _execute_cmd(self, cmdline):

        if cmdline.split()[0] == "cpu":

            # trap the cpu command, it requires special setting

            try:

                idx = int(cmdline.split()[1])

                if not 'return' in self.__cmd_passthrough('info version', idx):

                    print 'bad CPU index'

                    return True

                self.__cpu_index = idx

            except ValueError:

                print 'cpu command takes an integer argument'

                return True

        resp = self.__cmd_passthrough(cmdline, self.__cpu_index)

        if resp is None:

            print 'Disconnected'

            return False

        assert 'return' in resp or 'error' in resp

        if 'return' in resp:

            # Success

            if len(resp['return']) > 0:

                print resp['return'],

        else:

            # Error

            print '%s: %s' % (resp['error']['class'], resp['error']['desc'])

        return True

    def show_banner(self):

        QMPShell.show_banner(self, msg='Welcome to the HMP shell!')

def die(msg):

    sys.stderr.write('ERROR: %s\n' % msg)

    sys.exit(1)

def fail_cmdline(option=None):

    if option:

        sys.stderr.write('ERROR: bad command-line option \'%s\'\n' % option)

    sys.stderr.write('qemu-shell [ -H ] < UNIX socket path> | < TCP address:port >\n')

    sys.exit(1)

def main():

    addr = ''

    try:

        if len(sys.argv) == 2:

            qemu = QMPShell(sys.argv[1])

            addr = sys.argv[1]

        elif len(sys.argv) == 3:

            if sys.argv[1] != '-H':

                fail_cmdline(sys.argv[1])

            qemu = HMPShell(sys.argv[2])

            addr = sys.argv[2]

        else:

                fail_cmdline()

    except QMPShellBadPort:

        die('bad port number in command-line')

    try:

        qemu.connect()

    except qmp.QMPConnectError:

        die('Didn\'t get QMP greeting message')

    except qmp.QMPCapabilitiesError:

        die('Could not negotiate capabilities')

    except qemu.error:

        die('Could not connect to %s' % addr)

    qemu.show_banner()

    while qemu.read_exec_command('(QEMU) '):

        pass

    qemu.close()

if __name__ == '__main__':

    main()