15
\$\begingroup\$

I needed a tool to monitor VM's running on my server in relatively real-time (similar to top or the many variants out there). The main things I need to keep track of are:

  • All VM's listed via virsh list --all;

     Id    Name                           State
    ----------------------------------------------------
     13    Experiments-Proxy              running
     -     Experiments-PHP                shut off
     -     Experiments-Python             shut off
    
  • All networks listed via virsh net-list --all;

     Name                 State      Autostart     Persistent
    ----------------------------------------------------------
     default              inactive   yes           yes
     net_10_1_1_0         active     yes           yes
     net_10_1_2_0         active     yes           yes
     net_10_1_3_0         active     yes           yes
    
  • All storage pools listed via virsh pool-list --all;

     Name                 State      Autostart
    -------------------------------------------
     Experiments          active     yes
     images               active     yes
    

To do this, I built a small Python script using curses that effectively does three things:

  1. Lists all the aforementioned components;
  2. Updates the list on a regular basis (every 2 seconds, basically);
  3. Allows basic management of the aforementioned components (start, stop);

All of this is rather simple, if long and convoluted.

To start with, I built a function that runs the virsh command with the arguments I need to. I discarded the error output because I honestly don't care about it for this tool.

def virsh(command, arg):
    out, _ = subprocess.Popen(['virsh', command, arg], stdout=subprocess.PIPE, stderr=subprocess.STDOUT).communicate()
    out = re.split('[\r\n]+', out.decode("utf-8"))
    return list(map(lambda line: list(map(lambda x: x.strip(), re.split('\\s{2,}', line))), out))

This allows me to do something like the following further on in the script:

vms = virsh('list', '--all')[2:][:-1]
nets = virsh('net-list', '--all')[2:][:-1]
pools = virsh('pool-list', '--all')[2:][:-1]

Next, I needed a way to print a table in curses. This went relatively smooth as well, as all I did was push a list of columns and items into a function, with a few extra parameters:

def print_table(stdscr, head_color, sel_color, sel_i, x, y, cols, gray_sel, items):
    total_len = sum(list(map(lambda col: col[1] + 1, cols)))
    stdscr.insstr(y, x, ' ' * total_len, head_color)
    col_offset = 0

    if sel_i > -1:
        stdscr.addstr(y + sel_i + 1, x, ' ' * total_len, sel_color)

    c = 0
    for (name, minsize, gray) in cols:
        stdscr.addstr(y, x + col_offset, name, head_color)

        i = 1
        for item in items:
            color_offset = 1 if sel_i == (i - 1) else 0
            color = curses.color_pair(color_offset)
            gray_color = curses.color_pair(color_offset + (3 if gray_sel(item) else 0))
            stdscr.addstr(y + i, x + col_offset, item[c], gray_color if gray else color)
            i += 1
        col_offset += minsize + 1
        c += 1

Next, I needed to print a "help" at the bottom of the screen. For this I simply list each keystroke / command, and a single word about what it does. I might have a list like [("TAB", "Next"), ("F1", "Start"), ("F2", "Stop"), ("F10", "Quit")]:

def print_help(stdscr, help_color, helps):
    height, width = stdscr.getmaxyx()
    stdscr.insstr(height - 1, 0, ' ' * width, help_color)
    max_len = max(list(map(lambda x: len(x[1]), helps))) + 1
    offset = 0
    for (key, name) in helps:
        stdscr.insstr(height - 1, offset, key)
        stdscr.insstr(height - 1, offset + len(key), name, help_color)
        offset += len(key) + max_len

The next step is running all the logic to render the screen. For this, I built a render function that takes all the parameters I need:

def set_x_for_yes(x): return 'X' if x == 'yes' else ' '


def render(stdscr, vms, nets, pools, sel, sel_i):
    pool_diff = 2
    longest_net = max(list(map(lambda net: len(net[0]), nets)))
    longest_pool = max(list(map(lambda pool: len(pool[0]), pools)))
    longest_net = max(longest_net, longest_pool - pool_diff)
    height, width = stdscr.getmaxyx()
    net_offset = width - longest_net - 9 - pool_diff - 3
    vm_width = net_offset - 3 - 9 - 1 - 2

    vm_table = [("ID", 3, False), ("VM", vm_width - 1, True), ("STATUS", 9, False)]
    net_table = [("NET", longest_net, True), ("STATUS", 8, False), ("A", 1, False), ("P", 1, False)]
    pool_table = [("POOL", longest_net + pool_diff, True), ("STATUS", 8, False), ("A", 1, False)]
    nets = list(map(lambda net: [net[0], net[1], set_x_for_yes(net[2]), set_x_for_yes(net[3])], nets))
    pools = list(map(lambda pool: [pool[0], pool[1], set_x_for_yes(pool[2])], pools))

    tables = [
        (0, 0, 0, vm_table, lambda vm: vm[2] != "running", vms),
        (1, net_offset, 0, net_table, lambda net: net[1] != "active", nets),
        (2, net_offset, len(nets) + 2, pool_table, lambda pool: pool[1] != "active", pools)
    ]

    head_color = curses.color_pair(2)
    sel_color = curses.color_pair(1)
    for (sel_c, x, y, table, sel_test, items) in tables:
        print_table(stdscr, head_color, sel_color, sel_i if sel == sel_c else -1, x, y, table, sel_test, items)

    print_help(
        stdscr,
        curses.color_pair(1),
        [("TAB", "Next"), ("F1", "Start"), ("F2", "Stop"), ("F10", "Quit")])

This builds up all the components to pass to the rendering functions.

Lastly, I have a main function that I use curses.wrapper to run. This allows curses to setup all the screen components, and clean the screen up when it ends (either with success or failure):

def main(stdscr):
    curses.curs_set(0)
    curses.halfdelay(20)
    curses.start_color()
    curses.use_default_colors()
    curses.init_pair(1, 0, 6)
    curses.init_pair(2, 0, 2)
    curses.init_pair(3, 8, -1)
    curses.init_pair(4, 8, 6)
    sel = 0
    sel_i = 0

    start_commands = ['start', 'net-start', 'pool-start']
    stop_commands = ['destroy', 'net-destroy', 'pool-destroy']

    while True:
        vms = virsh('list', '--all')[2:][:-1]
        nets = virsh('net-list', '--all')[2:][:-1]
        pools = virsh('pool-list', '--all')[2:][:-1]

        args = [vms, nets, pools]
        arg_indexes = [1, 0, 0]

        stdscr.clear()
        render(stdscr, vms, nets, pools, sel, sel_i)
        stdscr.refresh()
        c = stdscr.getch()

        if c == curses.KEY_F10:
            exit()
        elif c == ord('\t'):
            sel = 0 if sel == 2 else sel + 1
        elif c == curses.KEY_DOWN or c == curses.KEY_UP:
            sel_i += -1 if c == curses.KEY_UP else 1
        elif (c == curses.KEY_F1 or c == curses.KEY_F2) and sel_i < len(args[sel]):
            commands = stop_commands if c == curses.KEY_F2 else start_commands
            virsh(commands[sel], args[sel][sel_i][arg_indexes[sel]])

        if sel_i == -1:
            sel_i += 1
        if sel_i >= len(args[sel]):
            sel_i = len(args[sel]) - 1


curses.wrapper(main)

This also has all the key-handling logic to adjust the scene.

I have yet to set up scrolling on each table, but that is beyond the scope of this question.

Once all is said and done, running the script gives me an output of the following:

virsh-monitor result

Any and all comments welcome. I don't have any PEP-8 flags in PyChar, so I'm thinking I'm already off to a good start here.

\$\endgroup\$
  • \$\begingroup\$ Not using the libvirt Python API? \$\endgroup\$ – Michael Hampton Jun 24 at 11:59
  • 1
    \$\begingroup\$ @MichaelHampton Didn't know there was one, but that would make a valid answer. ;) \$\endgroup\$ – Der Kommissar Jun 24 at 12:13
  • 2
    \$\begingroup\$ @MichaelHampton If you're interested, I posted an answer converting this to the libvirt API \$\endgroup\$ – ljrk Jun 24 at 22:41
  • 1
    \$\begingroup\$ Is this something you're willing to release in a public repo? I'd be interested in using it if you do \$\endgroup\$ – Canadian Luke Jul 6 at 16:18
  • 1
    \$\begingroup\$ @CanadianLuke I actually already did ;) github.com/ellersoft/virsh-monitor \$\endgroup\$ – Der Kommissar Jul 6 at 16:19
10
\$\begingroup\$

Bugs

There was one bug if no networks or pools existed, then calculation of longest_net and longest_pool respectively would fail, since max() would be called on an empty list. The solution is to add a default kw-arg

-    longest_net = max(len(net.name()) for net in nets)
-    longest_pool = max(len(pool.name()) for pool in pools)
+    longest_net = max((len(net.name()) for net in nets), default=0)
+    longest_pool = max((len(pool.name()) for pool in pools), default=0)

Use libvirt API

Based on @MichaelHampton's advice I moved the code to the libvirt API, basing off of the changes made in the answer by @Peilonrayz.

The crucial difference is to make a connection to libvirt in __main__ (otherwise we'd run into problems with interactive authentication on the console if curses already initialised):

 if __name__ == '__main__':
-    curses.wrapper(main)
+    conn = libvirt.open(None)
+    curses.wrapper(main, conn)

Then in main(stdscr, conn):

     while True:
-        vms = virsh('list', '--all')[2:][:-1]
-        nets = virsh('net-list', '--all')[2:][:-1]
-        pools = virsh('pool-list', '--all')[2:][:-1]
+        vms = conn.listAllDomains()
+        nets = conn.listAllNetworks()
+        pools = conn.listAllStoragePools()

Other than that it's just moving away from array-of-strings to method calls on libvirt objects, e.g.:

-    longest_net = max(len(net[0]) for net in nets)
-    longest_pool = max(len(pool[0]) for pool in pools)
+    longest_net = max((len(net.name()) for net in nets))
+    longest_pool = max((len(pool.name()) for pool in pools))

I also needed to create a 'vms' array just like the 'pools' and 'nets' array for print_table now. As this function however operates much on strings while the API returns integer constants, the least-effort approach taken by me was to convert all integers to strings via dictionaries and not touch print_table itself. Particularly virDomain.state() returns [state, reason] with both being integers; in order to pretty-print this I created a dictionary which then can be used like this:

state_string = {
    libvirt.VIR_DOMAIN_NOSTATE: 'nostate',
    libvirt.VIR_DOMAIN_RUNNING: 'running',
    libvirt.VIR_DOMAIN_BLOCKED: 'blocked',
    libvirt.VIR_DOMAIN_PAUSED: 'paused',
    libvirt.VIR_DOMAIN_SHUTDOWN: 'shutdown',
    libvirt.VIR_DOMAIN_SHUTOFF: 'shutoff',
    libvirt.VIR_DOMAIN_CRASHED: 'crashed',
    libvirt.VIR_DOMAIN_PMSUSPENDED: 'pmsuspended',
}
print(state_string[vm.state()[0]])

Similarly, the start/stop is handled via objects which reads much better:

-            commands = stop_commands if c == curses.KEY_F2 else start_commands
-            virsh(commands[sel], args[sel][sel_i][arg_indexes[sel]])
+            if c == curses.KEY_F2:
+                args[sel][sel_i].destroy()
+            else:
+                args[sel][sel_i].create()

Add other hypervisors

As we are now using the libvirt API, it's quite easy to add support for accessing other hypervisors via URI. I used getopt to parse a -c URI CLI argument:

 if __name__ == '__main__':
-    conn = libvirt.open(None)
+    import sys
+    import getopt
+    try:
+        opts, args = getopt.getopt(sys.argv[1:], 'c:')
+    except getopt.GetoptError as err:
+        print(err)
+        sys.exit(1)
+
+    uri = None
+    for o, a in opts:
+        if o == '-c':
+            uri = a
+
+    try:
+        conn = libvirt.open(uri)
+    except libvirt.libvirtError:
+        print('Failed to open connection to the hypervisor')
+        sys.exit(1)
+
     curses.wrapper(main, conn)

This allows to monitor remote hypervisor instances or the system one, e.g.:

$ ./virtop.py -c 'qemu+ssh://username@ip.of.vm.host/system' 

Final code

#! /usr/bin/env python3

import libvirt
import curses

state_string = {
    libvirt.VIR_DOMAIN_NOSTATE: 'nostate',
    libvirt.VIR_DOMAIN_RUNNING: 'running',
    libvirt.VIR_DOMAIN_BLOCKED: 'blocked',
    libvirt.VIR_DOMAIN_PAUSED: 'paused',
    libvirt.VIR_DOMAIN_SHUTDOWN: 'shutdown',
    libvirt.VIR_DOMAIN_SHUTOFF: 'shutoff',
    libvirt.VIR_DOMAIN_CRASHED: 'crashed',
    libvirt.VIR_DOMAIN_PMSUSPENDED: 'pmsuspended',
}

active_string = {
    0: 'inactive',
    1: 'active',
}


def print_table(stdscr, head_color, sel_color, sel_i, x, y, cols, gray_sel, items):
    total_len = sum(col[1] + 1 for col in cols)
    stdscr.insstr(y, x, ' ' * total_len, head_color)
    if sel_i > -1:
        stdscr.addstr(y + sel_i + 1, x, ' ' * total_len, sel_color)

    for c, (name, minsize, gray) in enumerate(cols):
        stdscr.addstr(y, x, name, head_color)
        for i, item in enumerate(items, 1):
            color = curses.color_pair(
                (sel_i == (i - 1))
                + (3 if gray and gray_sel(item) else 0)
            )
            stdscr.addstr(y + i, x, item[c], color)
        x += minsize + 1


def print_help(stdscr, help_color, helps):
    height, width = stdscr.getmaxyx()
    stdscr.insstr(height - 1, 0, ' ' * width, help_color)
    max_len = max(len(h[1]) for h in helps) + 1
    offset = 0
    for key, name in helps:
        stdscr.insstr(height - 1, offset, key)
        stdscr.insstr(height - 1, offset + len(key), name, help_color)
        offset += len(key) + max_len


def set_x_if_true(x):
    return 'X' if x else ' '


def render(stdscr, vms, nets, pools, sel, sel_i):
    pool_diff = 2
    longest_net = max((len(net.name()) for net in nets), default=0)
    longest_pool = max((len(pool.name()) for pool in pools), default=0)
    longest_net = max(longest_net, longest_pool - pool_diff)
    height, width = stdscr.getmaxyx()
    net_offset = width - longest_net - 9 - pool_diff - 3
    vm_width = net_offset - 3 - 9 - 1 - 2

    vm_table = [("ID", 3, False), ("VM", vm_width - 1, True), ("STATUS", 9, False)]
    net_table = [("NET", longest_net, True), ("STATUS", 8, False), ("A", 1, False), ("P", 1, False)]
    pool_table = [("POOL", longest_net + pool_diff, True), ("STATUS", 8, False), ("A", 1, False)]
    vms = [
        ['-' if vm.ID() == -1 else str(vm.ID()), vm.name(), state_string[vm.state()[0]]]
        for vm in vms
    ]
    nets = [
        [net.name(), active_string[net.isActive()], set_x_if_true(net.autostart()), set_x_if_true(net.isPersistent())]
        for net in nets
    ]
    pools = [
        [pool.name(), active_string[pool.isActive()], set_x_if_true(pool.autostart())]
        for pool in pools
    ]

    tables = [
        (0, 0, 0, vm_table, lambda vm: vm[2] != state_string[libvirt.VIR_DOMAIN_RUNNING], vms),
        (1, net_offset, 0, net_table, lambda net: net[1] != active_string[1], nets),
        (2, net_offset, len(nets) + 2, pool_table, lambda pool: pool[1] != active_string[1], pools)
    ]

    head_color = curses.color_pair(2)
    sel_color = curses.color_pair(1)
    for (sel_c, x, y, table, sel_test, items) in tables:
        print_table(stdscr, head_color, sel_color, sel_i if sel == sel_c else -1, x, y, table, sel_test, items)

    print_help(
        stdscr,
        curses.color_pair(1),
        [("TAB", "Next"), ("F1", "Start"), ("F2", "Stop"), ("F10", "Quit")]
    )


def main(stdscr, conn):
    curses.curs_set(0)
    curses.halfdelay(20)
    curses.start_color()
    curses.use_default_colors()
    curses.init_pair(1, 0, 6)
    curses.init_pair(2, 0, 2)
    curses.init_pair(3, 8, -1)
    curses.init_pair(4, 8, 6)
    sel = 0
    sel_i = 0

    while True:
        vms = conn.listAllDomains()
        nets = conn.listAllNetworks()
        pools = conn.listAllStoragePools()

        args = [vms, nets, pools]
        arg_indexes = [1, 0, 0]

        stdscr.clear()
        render(stdscr, vms, nets, pools, sel, sel_i)
        stdscr.refresh()
        c = stdscr.getch()

        if c == curses.KEY_F10:
            exit()
        elif c == ord('\t'):
            sel = 0 if sel == 2 else sel + 1
        elif c == curses.KEY_DOWN or c == curses.KEY_UP:
            sel_i += -1 if c == curses.KEY_UP else 1
        elif (c == curses.KEY_F1 or c == curses.KEY_F2) and sel_i < len(args[sel]):
            if c == curses.KEY_F2:
                args[sel][sel_i].destroy()
            else:
                args[sel][sel_i].create()

        if sel_i == -1:
            sel_i += 1
        if sel_i >= len(args[sel]):
            sel_i = len(args[sel]) - 1


if __name__ == '__main__':
    import sys
    import getopt
    try:
        opts, args = getopt.getopt(sys.argv[1:], 'c:')
    except getopt.GetoptError as err:
        print(err)
        sys.exit(1)

    uri = None
    for o, a in opts:
        if o == '-c':
            uri = a

    try:
        conn = libvirt.open(uri)
    except libvirt.libvirtError:
        print('Failed to open connection to the hypervisor')
        sys.exit(1)

    curses.wrapper(main, conn)

Remarks

This code is now with almost no error handling and since the libvirt functions may throw exceptions quite often (e.g. when starting if it's already started), this needs to be addressed. Also a usage() function documenting the -c option would be nice, I was too lazy for that. :-)

I'm personally not so proficient with python and more a C person, so the code might not be the most pythonic.

Also I can only recommend you looking into virt-manager which is basically what you did here. While it's a GUI solution it allows connecting to remote instances, so your server does not need to run X or Wayland, although a virt-manager-tui would be cool as well.

| improve this answer | |
\$\endgroup\$
  • 1
    \$\begingroup\$ There's also virt-top, which is much less colorful but displays some more information. I tried this code and found that it was slow to respond visually to keyboard input when connected to a remote hypervisor (on another continent). I expect you're loading information about whatever is being selected before updating the display. Other than that it looks really good. \$\endgroup\$ – Michael Hampton Jun 24 at 23:08
  • \$\begingroup\$ @MichaelHampton Oh, I didn't know about virt-top, but it's not packaged for my distribution (as some dependencies such as libvirt-ocaml aren't either) and seems to be unmaintained anyhow. Ideally one should split input-handling from the model/information a bit better, right now it's difficult to fix that issue. Perhaps someone eventually will pickup on this again. virt-top for reference: git.annexia.org/?p=virt-top.git;a=summary \$\endgroup\$ – ljrk Jun 25 at 10:37
  • \$\begingroup\$ @MichaelHampton That issue is because I use curses.halfdelay(20) as my timing loop (it waits until 2 seconds or keypress on every .getch() call), and my stdscr.getch() is only called once, then regardless of whether a use typed a key or not it reloads all the VMs, Networks, and Storage Pools. Going to play with that today, see if I can't speed the rendering up. \$\endgroup\$ – Der Kommissar Jun 25 at 12:28
14
\$\begingroup\$

Subprocess

out, _ = subprocess.Popen(['virsh', command, arg], stdout=subprocess.PIPE, stderr=subprocess.STDOUT).communicate()

is a little clunky; consider

def virsh(*args):
    out = subprocess.check_output(('virsh', *args))

This will also:

  • check for the error level after execution
  • allow for an arbitrary number of command-line arguments

Comprehensions

Let's see if we can translate this:

list(
    map(
        lambda line: list(
            map(
                lambda x: x.strip(), re.split('\\s{2,}', line)
            )
        ), 
        out
    )
)

from the old functional style to the new comprehension style. I also had to expand the above because it was a golfed nightmare.

[
    [
        x.strip()
        for x in re.split(r'\s{2,}', line)
    ]
    for line in out
]

Also note the use of a raw string for your regex.

Similarly, this:

sum(list(map(lambda col: col[1] + 1, cols)))

can be

sum(col[1] + 1 for col in cols)

Type hints

This:

def print_table(stdscr, head_color, sel_color, sel_i, x, y, cols, gray_sel, items):

could really benefit from them. For instance, maybe x and y are x: int, y: int.

Enumerate

    i = 1
    for item in items:
        # ...
        i += 1

should be

for i, item in enumerate(items):

Implicit tuple unpack

for (name, minsize, gray) in cols:

does not need parens.

Else-after-exit

        exit()
    elif c == ord('\t'):

does not need an elif; an if will suffice.

| improve this answer | |
\$\endgroup\$
  • \$\begingroup\$ 2 nitpicks—1. nothing wrong with the functional style except forcing the generators to materialize as lists; I would let everything be lazy where possible (and formatting helps, too). 2. The elif does a far better job conveying the structure of the program, even if an if would suffice. Just at a glance, and elif chain tells me I dont have to examine in detail every branch to find the control flow. \$\endgroup\$ – D. Ben Knoble Jun 24 at 15:49
  • \$\begingroup\$ @D.BenKnoble Re. lazy - I was aiming for direct equivalence, where a generator would not be. The OP uses the outputs of that comprehension in render in a manner incompatible with a generator since they're consumed more than once. \$\endgroup\$ – Reinderien Jun 24 at 15:54
  • \$\begingroup\$ fair enough: didn’t read that part too closely. \$\endgroup\$ – D. Ben Knoble Jun 24 at 16:09
10
\$\begingroup\$

Style

Your style is quite good, but you can tell that the code isn't written by a Pythonista.

  • Whilst line length can be a touchy subject, it's mostly left at 79 if you follow PEP 8 or 90 if you're running Black.
    This is causing me to have a suboptimal experience editing your code.

  • Defining functions on one line, like set_x_for_yes, are normally big no-nos.

  • I'm not a fan of your single letter variables. But I'm also not entirely sure what I'd replace most of them with.

  • (Potential religious war) The 'Pythonic' form of list(map(...)) is a list comprehension. For example in virsh we can use:

    return list(map(lambda line: list(map(lambda x: x.strip(), re.split('\\s{2,}', line))), out))
    
    return [
        [x.strip() for x in re.split('\\s{2,}', line)]
        for line in out
    ]
    
  • Like most interpreted languages there is no 'main' entry point. As the code is interpreted from top to bottom. However sometimes we don't want code to run if it is not the 'main' script. To deal with this we can use an if __name__ == '__main__' guard to prevent this code running if you import it.

  • Python is quite allergic to chaining, and so it's common for the subprocess.Popen and .communicate() chain to be split across two assignments.

    proc = subprocess.Popen(
        ['virsh', command, arg],
        stdout=subprocess.PIPE,
        stderr=subprocess.STDOUT
    )
    out, _ = proc.communicate()
    
  • Having unneeded parentheses are really discouraged as they impede readability.

Changes

  • In print_table, converting the iterator returned from map to a list is unneeded. Additionally we can opt to use a generator expression instead. This is the same as the list comprehension before except it is wrapped in parentheses () and builds a generator. Python has some sugar when a generator expression is the only argument to a function and lets you drop the double parentheses ().

    sum(list(map(lambda col: col[1] + 1, cols)))
    
    sum(col[1] + 1 for col in cols)
    
  • In print_table, it's nice to see you using the ' ' * total_len sugar.

  • In print_table, we can use enumerate rather than manually looping through c and i.

    for i, item in enumerate(items, 1):
    
  • In print_table, rather than using a turnery to build 1 or 0, you can just use int. I would also be surprised if the functions don't support taking a bool in-place for an integer.

  • In print_table, col_offset is only ever used as x + col_offset. At which point you might as well just update x.

  • In print_table, you can merge the if grey else turnery into the gray_color line to build the correct colour with less lines of code.

Additional changes not made to the below code:

  • It would be nice to add an Enum to make building the colour pairs easier. By using an IntFlag we can get the benefits of it acting like an int and act like flags. However your current mapping makes this hard. I would change it so the last bit is to change if the colour is grey.

    class Colours(enum.IntFlag):
        DEFAULT = 0
        GRAY = 1
        SELECT = 2
        HEAD = 4
    

    This has a couple of benefits:

    1. If you decide to later change what the values are it is easier.
    2. We can use Colours.DEFAULT or Colours.SELECT | Colours.GRAY to select the wanted colours.
    3. It means we can change print_help to not use magic numbers.
  • In render, I would rearrange a lot of the table information.

    The following values never change:

    • Headers.
    • Which columns can be grey.
    • Mutations (set_x_for_yes) to the items.
    • Selecting grey rows, gray_sel / sel_test.

    Values that can change each run:

    • The x position.
    • The y position.
    • The items.
    • The width of each column.

    And so I would move all the constants outside of the function. We can join these two tables together with zip.

  • In print_table, you can remove the need for the two calls to stdscr.insstr with the value ' ' * total_len if you pad the values.

    >>> '{1:<{0}}|{2:^{0}}|{3:>{0}}'.format(5, 1, 2, 3)
    '1    |  2  |    3'
    
  • A lot of print_table is not actually about printing the table it's about colouring it correctly. I would build another function that correctly colours everything.

    • If we change each item to a tuple of the item's string and the item's colour than it is easier to print the entire table.
    • If we include the headers in this function we can format everything correctly, and make print_table a very simple nested for loop.
    def select_colors(values, sel_i, gray_sel, grays):
        for i, row in enumerate(values):
            gray_row = gray_sel(row)
            new_row = []
            for item, gray in zip(row, grays):
                color = Colours.SELECT if sel_i == i else Colours.DEFAULT
                if gray_row and gray:
                    color |= Colours.GRAY
                if i == 0:
                    color = Colours.HEAD
                new_row.append((item, curses.color_pair(color)))
            yield new_row
    
import subprocess
import re
import time
import curses


def virsh(command, arg):
    proc = subprocess.Popen(
        ['virsh', command, arg],
        stdout=subprocess.PIPE,
        stderr=subprocess.STDOUT
    )
    out, _ = proc.communicate()
    return [
        [
            x.strip()
            for x in re.split('\\s{2,}', line)
        ]
        for line in re.split('[\r\n]+', out.decode("utf-8"))
    ]


def print_table(stdscr, head_color, sel_color, sel_i, x, y, cols, gray_sel, items):
    total_len = sum(col[1] + 1 for col in cols)
    stdscr.insstr(y, x, ' ' * total_len, head_color)
    if sel_i > -1:
        stdscr.addstr(y + sel_i + 1, x, ' ' * total_len, sel_color)

    for c, (name, minsize, gray) in enumerate(cols):
        stdscr.addstr(y, x, name, head_color)
        for i, item in enumerate(items, 1):
            color = curses.color_pair(
                sel_i == (i - 1)
                + (3 if gray and gray_sel(item) else 0)
            )
            stdscr.addstr(y + i, x, item[c], color)
        x += minsize + 1


def print_help(stdscr, help_color, helps):
    height, width = stdscr.getmaxyx()
    stdscr.insstr(height - 1, 0, ' ' * width, help_color)
    max_len = max(len(h[1]) for h in helps) + 1
    offset = 0
    for key, name in helps:
        stdscr.insstr(height - 1, offset, key)
        stdscr.insstr(height - 1, offset + len(key), name, help_color)
        offset += len(key) + max_len


def set_x_for_yes(x):
    return 'X' if x == 'yes' else ' '


def render(stdscr, vms, nets, pools, sel, sel_i):
    pool_diff = 2
    longest_net = max(len(net[0]) for net in nets)
    longest_pool = max(len(pool[0]) for pool in pools)
    longest_net = max(longest_net, longest_pool - pool_diff)
    height, width = stdscr.getmaxyx()
    net_offset = width - longest_net - 9 - pool_diff - 3
    vm_width = net_offset - 3 - 9 - 1 - 2

    vm_table = [("ID", 3, False), ("VM", vm_width - 1, True), ("STATUS", 9, False)]
    net_table = [("NET", longest_net, True), ("STATUS", 8, False), ("A", 1, False), ("P", 1, False)]
    pool_table = [("POOL", longest_net + pool_diff, True), ("STATUS", 8, False), ("A", 1, False)]
    nets = [
        [net[0], net[1], set_x_for_yes(net[2]), set_x_for_yes(net[3])]
        for net in nets
    ]
    pools = [
        [pool[0], pool[1], set_x_for_yes(pool[2])]
        for pool in pools
    ]

    tables = [
        (0, 0, 0, vm_table, lambda vm: vm[2] != "running", vms),
        (1, net_offset, 0, net_table, lambda net: net[1] != "active", nets),
        (2, net_offset, len(nets) + 2, pool_table, lambda pool: pool[1] != "active", pools)
    ]

    head_color = curses.color_pair(2)
    sel_color = curses.color_pair(1)
    for (sel_c, x, y, table, sel_test, items) in tables:
        print_table(stdscr, head_color, sel_color, sel_i if sel == sel_c else -1, x, y, table, sel_test, items)

    print_help(
        stdscr,
        curses.color_pair(1),
        [("TAB", "Next"), ("F1", "Start"), ("F2", "Stop"), ("F10", "Quit")]
    )


def main(stdscr):
    curses.curs_set(0)
    curses.halfdelay(20)
    curses.start_color()
    curses.use_default_colors()
    curses.init_pair(1, 0, 6)
    curses.init_pair(2, 0, 2)
    curses.init_pair(3, 8, -1)
    curses.init_pair(4, 8, 6)
    sel = 0
    sel_i = 0

    start_commands = ['start', 'net-start', 'pool-start']
    stop_commands = ['destroy', 'net-destroy', 'pool-destroy']

    while True:
        vms = virsh('list', '--all')[2:][:-1]
        nets = virsh('net-list', '--all')[2:][:-1]
        pools = virsh('pool-list', '--all')[2:][:-1]

        args = [vms, nets, pools]
        arg_indexes = [1, 0, 0]

        stdscr.clear()
        render(stdscr, vms, nets, pools, sel, sel_i)
        stdscr.refresh()
        c = stdscr.getch()

        if c == curses.KEY_F10:
            exit()
        elif c == ord('\t'):
            sel = 0 if sel == 2 else sel + 1
        elif c == curses.KEY_DOWN or c == curses.KEY_UP:
            sel_i += -1 if c == curses.KEY_UP else 1
        elif (c == curses.KEY_F1 or c == curses.KEY_F2) and sel_i < len(args[sel]):
            commands = stop_commands if c == curses.KEY_F2 else start_commands
            virsh(commands[sel], args[sel][sel_i][arg_indexes[sel]])

        if sel_i == -1:
            sel_i += 1
        if sel_i >= len(args[sel]):
            sel_i = len(args[sel]) - 1


if __name__ == '__main__':
    curses.wrapper(main)
| improve this answer | |
\$\endgroup\$
  • \$\begingroup\$ Potential religious war - Based on the (infamous?) relegation of reduce to functools and the reasoning for that change, I think you, most of the Python community and I are on the right side of history when it comes to discouraging map. \$\endgroup\$ – Reinderien Jun 23 at 18:14
  • 1
    \$\begingroup\$ @Reinderien Indeed. I am trying to to start to mark 'problematic' suggestions. I'm hoping that acknowledging that it's a source for confrontation will help reduce the amount of confrontation I get when a user disregards the advice :) \$\endgroup\$ – Peilonrayz Jun 23 at 18:18
  • \$\begingroup\$ At least on my machine this fails to highlight the correct 'machine' (see asciinema.org/a/jOrs6moLiwB2Vf2QI6pFaPLWh?t=7). I found the faulty code and reverted it in my answer below. \$\endgroup\$ – ljrk Jun 24 at 19:58
  • 1
    \$\begingroup\$ @larkey Yeah I noticed that after posting but during changing to string.Formatter. I thought I'd added it here, but I didn't. The fix is quite simple really (sel_i == (i - 1)) because I got the precedence wrong. \$\endgroup\$ – Peilonrayz Jun 24 at 20:01
  • \$\begingroup\$ @Peilonrayz Ah, I guessed something like that but was too tired and don't have enough python skill to dig deeper :) I'll update my answer then! \$\endgroup\$ – ljrk Jun 24 at 22:36
5
\$\begingroup\$

Custom Formatter

Half way through my previous answer I decided to integrate Python's Format Specification Mini-Language. I had originally thought there was a lot more formatting going on, but this was not the case. It has a few benefits, but also a few deficits.

Pros:

  • It's using syntax that should be in every Python programmers toolbox.
  • It forced me to split print_table into two functions. Because the formatting was moved inside the class. And then later I moved it out into select_colors.
  • If you are building more tables it's really quite powerful.

Cons:

  • You are unlikely to know this mini-language.
  • You're not really using any of the power it brings.
  • The method _cformat is long and filled with boilerplate.
  • You can definitely write the code in fewer lines of code without it.

Whilst it is probably not the best solution for this code it's at least interesting. And can help if you need more advanced formats.

import curses
import enum
import re
import string
import subprocess
import time


class Colours(enum.IntFlag):
    DEFAULT = 0
    GRAY = 1
    SELECT = 2
    HEAD = 4


class CursedFormatter(string.Formatter):
    def __init__(self, stdscr, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self._stdscr = stdscr
    
    def _cformat(self, format_string, args, kwargs, index=0):
        result = []
        for pre, name, spec, conversion in self.parse(format_string):
            if name is None:
                result.append((pre,))
            else:
                if name == '':
                    if index is False:
                        raise ValueError('cannot switch from manual field specification to automatic field numbering')
                    name = str(index)
                    index += 1
                elif name.isdigit():
                    if index:
                        raise ValueError('cannot switch from manual field specification to automatic field numbering')
                    index = False
                obj, _ = self.get_field(name, args, kwargs)
                if isinstance(obj, tuple):
                    obj, *a = obj
                else:
                    a = ()
                obj = self.convert_field(obj, conversion)
                spec, index = super()._vformat(spec, args, kwargs, set(), 1, auto_arg_index=index)
                result.append((self.format_field(obj, spec),) + tuple(a))
        return result, index

    def vformat(self, fmt, args, kwargs):
        return ''.join(
            value
            for value, *_ in self._cformat(fmt, args, kwargs)[0]
        )

    def _makestr(self, fn, fmt, args, kwargs):
        values, _ = self._cformat(fmt, args, kwargs)
        x = kwargs.get('x', 0)
        y = kwargs.get('y', 0)
        result = []
        for value in values:
            self._stdscr.insstr(y, x, *value)
            x += len(value[0])
            result.append(value[0])
        return ''.join(result)

    def insstr(self, fmt, *args, **kwargs):
        return self._makestr(self._stdscr.insstr, fmt, args, kwargs)
    
    def addstr(self, fmt, *args, **kwargs):
        return self._makestr(self._stdscr.addstr, fmt, args, kwargs)


def virsh(command, arg):
    proc = subprocess.Popen(
        ['virsh', command, arg],
        stdout=subprocess.PIPE,
        stderr=subprocess.STDOUT
    )
    out, _ = proc.communicate()
    return [
        [
            x.strip()
            for x in re.split('\\s{2,}', line)
        ]
        for line in re.split('[\r\n]+', out.decode("utf-8"))
    ]


def select_colors(values, sel_i, gray_sel, grays):
    for i, row in enumerate(values):
        gray_row = gray_sel(row)
        new_row = []
        for item, gray in zip(row, grays):
            color = Colours.SELECT if sel_i == i else Colours.DEFAULT
            if gray_row and gray:
                color |= Colours.GRAY
            if i == 0:
                color = Colours.HEAD
            new_row.append((item, curses.color_pair(color)))
        yield new_row


def print_table(stdscr, fmt, values, x, y):
    for i, row in enumerate(values):
        CursedFormatter(stdscr).addstr(fmt, *row, x=x, y=y + i)


def print_help(stdscr, helps):
    height, width = stdscr.getmaxyx()
    help_color = curses.color_pair(Colours.SELECT)
    CF = CursedFormatter(stdscr)
    CF.insstr('{}', (' ' * width, help_color), x=0, y=height - 1)
    max_len = max(len(h[1]) for h in helps) + 1
    offset = 0
    for key, name in helps:
        CF.insstr('{}{:<{}}', key, (name, help_color), max_len, x=offset, y=height - 1)
        offset += len(key) + max_len


def set_x_for_yes(x):
    return 'X' if x == 'yes' else ' '


def echo(x):
    return x


TABLES = [
    (
        ['ID', 'VM', 'STATUS'],
        [False, True, False],
        [echo, echo, echo],
        lambda vm: vm[2] != 'running',
    ),
    (
        ['NET', 'STATUS', 'A', 'P'],
        [True, False, False, False],
        [echo, echo, set_x_for_yes, set_x_for_yes],
        lambda net: net[1] != "active",
    ),
    (
        ['POOL', 'STATUS', 'A'],
        [True, False, False],
        [echo, echo, set_x_for_yes],
        lambda pool: pool[1] != "active",
    ),
]


def render(stdscr, vms, nets, pools, sel, sel_i):
    pool_diff = 2
    longest_net = max(len(net[0]) for net in nets)
    longest_pool = max(len(pool[0]) for pool in pools)
    longest_net = max(longest_net, longest_pool - pool_diff)
    height, width = stdscr.getmaxyx()
    net_offset = width - longest_net - 9 - pool_diff - 3
    vm_width = net_offset - 3 - 9 - 1 - 2

    tables = [
        (
            0,
            0,
            vms,
            (4, vm_width, 10)
        ),
        (
            net_offset,
            0,
            nets,
            (longest_net + 1, 9, 2, 2)
        ),
        (
            net_offset,
            len(nets) + 2,
            pools,
            (longest_net + pool_diff + 1, 9, 2)
        ),
    ]
    for (
        i,
        (
            (x, y, items, widths),
            (header, grays, maps, gray_test)
        ),
    ) in enumerate(zip(tables, TABLES)):
        values = (
            [header]
            + [
                [tran(item) for tran, item in zip(maps, row)]
                for row in items
            ]
        )
        selected = sel_i + 1 if sel == i else -1
        values = select_colors(values, selected, gray_test, grays)
        fmt = ''.join(f'{{:<{width}}}' for width in widths)
        print_table(stdscr, fmt, values, x, y)

    print_help(
        stdscr,
        [("TAB", "Next"), ("F1", "Start"), ("F2", "Stop"), ("F10", "Quit")]
    )


def main(stdscr):
    curses.curs_set(0)
    curses.halfdelay(20)
    curses.start_color()
    curses.use_default_colors()
    curses.init_pair(Colours.GRAY, 8, -1)
    curses.init_pair(Colours.SELECT, 0, 6)
    curses.init_pair(Colours.SELECT | Colours.GRAY, 8, 6)
    curses.init_pair(Colours.HEAD, 0, 2)
    curses.init_pair(Colours.HEAD | Colours.GRAY, 8, 2)
    sel = 0
    sel_i = 0

    start_commands = ['start', 'net-start', 'pool-start']
    stop_commands = ['destroy', 'net-destroy', 'pool-destroy']

    while True:
        vms = virsh('list', '--all')[2:][:-1]
        nets = virsh('net-list', '--all')[2:][:-1]
        pools = virsh('pool-list', '--all')[2:][:-1]

        args = [vms, nets, pools]
        arg_indexes = [1, 0, 0]

        stdscr.clear()
        render(stdscr, vms, nets, pools, sel, sel_i)
        stdscr.refresh()
        c = stdscr.getch()

        if c == curses.KEY_F10:
            exit()
        elif c == ord('\t'):
            sel = 0 if sel == 2 else sel + 1
        elif c == curses.KEY_DOWN or c == curses.KEY_UP:
            sel_i += -1 if c == curses.KEY_UP else 1
        elif (c == curses.KEY_F1 or c == curses.KEY_F2) and sel_i < len(args[sel]):
            commands = stop_commands if c == curses.KEY_F2 else start_commands
            virsh(commands[sel], args[sel][sel_i][arg_indexes[sel]])

        if sel_i == -1:
            sel_i += 1
        if sel_i >= len(args[sel]):
            sel_i = len(args[sel]) - 1


if __name__ == '__main__':
    curses.wrapper(main)
| improve this answer | |
\$\endgroup\$
  • \$\begingroup\$ I don't think I would use the formatter, but it's interesting to see the approach here. \$\endgroup\$ – Der Kommissar Jun 24 at 12:13
  • \$\begingroup\$ @DerKommissar Understandable, you should be able to replace it with a very simple function since the format segments, values and colours are all well formed. \$\endgroup\$ – Peilonrayz Jun 24 at 12:19

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.