EXPERIMENTAL Add top-like monitoring tool

Inspiration and code from qui-domains.

Still some rough edges:

- no docstrings
- missing window restoration on exit
- not scrollable when screen is too small for the amount of running
  qubes, "xl top" is scrollable
- missing stubdomains, we have it's vm-stats event from my other libvirt
  cache PR in core-admin, but here we are only registering qubes that
  are in the VMCollection
- header background is taking more space than necessary

Author: ben-grande

doc/conf.py

@@ -366,6 +366,8 @@
         u'Manage tags on a qube', _man_pages_author, 1),
     ('manpages/qvm-template', 'qvm-template',
         u'Manage templates', _man_pages_author, 1),
+    ('manpages/qvm-top', 'qvm-top',
+        u'Top-like monitoring tool', _man_pages_author, 1),
     ('manpages/qvm-unpause', 'qvm-unpause',
         u'Pause a qube', _man_pages_author, 1),
     ('manpages/qvm-volume', 'qvm-volume',

doc/manpages/qvm-top.rst

@@ -0,0 +1,22 @@
+.. program:: qvm-top
+
+:program:`qvm-top` -- top-like monitoring tool
+==============================================
+
+Synopsis
+--------
+
+:command:`qvm-top`
+
+Notes
+-----
+
+Quit by pressing `q`, `Q`, or `ESC`.
+
+Authors
+-------
+
+| Benjamin Grande <ben.grande.b at gmail dot com>
+| For complete author list see: https://github.com/QubesOS/qubes-core-admin-client.git
+
+.. vim: ts=3 sw=3 et tw=80

doc/qubesadmin.tools.rst

@@ -196,6 +196,14 @@ qubesadmin\.tools\.qvm\_template\_postprocess module
     :undoc-members:
     :show-inheritance:
 
+qubesadmin\.tools\.qvm\_top module
+----------------------------------------------------
+
+.. automodule:: qubesadmin.tools.qvm_top
+    :members:
+    :undoc-members:
+    :show-inheritance:
+
 qubesadmin\.tools\.qvm\_unpause module
 --------------------------------------
 

qubesadmin/tools/qvm_top.py

@@ -0,0 +1,259 @@
+#!/usr/bin/env python3
+
+## SPDX-FileCopyrightText: 2025 Benjamin Grande M. S. <ben.grande.b@gmail.com>
+##
+## SPDX-License-Identifier: GPL-2.0-only
+
+"""
+Top-like info for Qubes.
+"""
+
+import asyncio
+import curses
+import sys
+import time
+import traceback
+
+import qubesadmin
+from qubesadmin.events import EventsDispatcher
+
+
+STATE_DICTIONARY = {
+    "domain-pre-start": "Transient",
+    "domain-start": "Running",
+    "domain-start-failed": "Halted",
+    "domain-paused": "Paused",
+    "domain-unpaused": "Running",
+    "domain-shutdown": "Halted",
+    "domain-pre-shutdown": "Transient",
+    "domain-shutdown-failed": "Running",
+}
+
+
+class Stats:
+    # pylint: disable=missing-class-docstring
+
+    def __init__(self, vm, state=None):
+        super().__init__()
+        self.vm = vm
+        self.memory = 0
+        self.cpu = 0
+        self.state = state
+
+    def update_stats(self, memory_kb, cpu_usage):
+        # pylint: disable=missing-function-docstring
+        self.memory = int(int(memory_kb) / 1024)
+        self.cpu = int(cpu_usage)
+
+    def update_state(self, state):
+        # pylint: disable=missing-function-docstring
+        self.state = state
+
+
+class Monitor:
+    # pylint: disable=missing-class-docstring
+
+    def __init__(self, app, dispatcher, stats_dispatcher, stdscr):
+        self.entries = {}
+        self.stdscr = stdscr
+        self.app = app
+        self.dispatcher = dispatcher
+        self.stats_dispatcher = stats_dispatcher
+        self.register_events()
+        self.skip_draw = False
+
+    def run(self):
+        # pylint: disable=missing-function-docstring
+        self.initialize_entries()
+
+    def initialize_entries(self):
+        # pylint: disable=missing-function-docstring
+        # Add AdminVMS
+        self.add_domain_item(
+            submitter=None, vm=self.app.domains["dom0"], event=None
+        )
+        # and the rest of them
+        for vm in sorted(
+            [vm for vm in self.app.domains if vm not in self.entries]
+        ):
+            self.add_domain_item(submitter=None, vm=vm, event=None)
+
+        for item in self.entries.values():
+            try:
+                if item.vm and item.vm.is_running():
+                    item.state = item.vm.get_power_state()
+                else:
+                    item.state = "Halted"
+            except qubesadmin.exc.QubesPropertyAccessError:
+                item.state = "Halted"
+        self.draw_table()
+
+    def update_stats(self, vm, _event, **kwargs):
+        # pylint: disable=missing-function-docstring
+        if vm not in self.entries:
+            return
+        self.entries[vm].update_stats(
+            kwargs["memory_kb"], kwargs["cpu_usage"]
+        )
+        self.draw_table()
+
+    def refresh_all_items(self, _subject, _event, **_kwargs):
+        # pylint: disable=missing-function-docstring,unused-argument
+        self.skip_draw = True
+        items_to_delete = [
+            vm for vm in self.entries if vm not in self.app.domains
+        ]
+        for vm in items_to_delete:
+            self.remove_domain_item(submitter=None, vm=vm, event=None)
+        for vm in self.app.domains:
+            self.update_domain_item(vm, "")
+        self.skip_draw = False
+        self.draw_table()
+
+    def add_domain_item(self, submitter, event, vm, **_kwargs):
+        # pylint: disable=missing-function-docstring,unused-argument
+        try:
+            vm = self.app.domains[str(vm)]
+        except KeyError:
+            # the VM was not created successfully or was deleted before the
+            # event was fully handled
+            return
+        if vm in self.entries:
+            return
+
+        state = self.get_state_from_event(event=event, vm=vm, default="Halted")
+        entry = Stats(vm, state=state)
+        self.entries[vm] = entry
+        self.draw_table()
+
+    def remove_domain_item(self, submitter, event, vm, **_kwargs):
+        # pylint: disable=missing-function-docstring,unused-argument
+        if vm not in self.entries:
+            return
+        del self.entries[vm]
+        self.draw_table()
+
+    def update_domain_item(self, vm, event, **kwargs):
+        # pylint: disable=missing-function-docstring
+        try:
+            item = self.entries[vm]
+        except qubesadmin.exc.QubesPropertyAccessError:
+            print("Unexpected property access error", file=sys.stderr)
+            traceback.print_exc()
+            self.remove_domain_item(
+                submitter=None, vm=vm, event=event, **kwargs
+            )
+            return
+        except KeyError:
+            self.add_domain_item(submitter=None, vm=vm, event=event)
+            if not vm in self.entries:
+                # VM not added - already removed?
+                return
+            item = self.entries[vm]
+
+        state = self.get_state_from_event(
+            event=event, vm=vm, default="Transient"
+        )
+        if state is None:
+            return
+        item.update_state(state)
+        self.draw_table()
+
+    def get_state_from_event(self, event, vm, default) -> str | None:
+        # pylint: disable=missing-function-docstring
+        if event in STATE_DICTIONARY:
+            return STATE_DICTIONARY[event]
+        try:
+            state = vm.get_power_state()
+        except Exception:  # pylint: disable=broad-except
+            if vm not in self.app.domains:
+                return None
+            state = default
+        return state
+
+    def draw_table(self):
+        """
+        Draw a top-like table about qubes statuses.
+        """
+        if self.skip_draw:
+            return
+        self.stdscr.clear()
+        height, _ = self.stdscr.getmaxyx()
+
+        self.stdscr.attron(curses.A_BOLD)
+        self.stdscr.attron(curses.A_REVERSE)
+        self.stdscr.addstr(0, 0, "Qube".ljust(40))
+        self.stdscr.addstr(0, 40, "State".ljust(50))
+        self.stdscr.addstr(0, 51, "MEM(%)".ljust(57))
+        self.stdscr.addstr(0, 59, "CPU(%)".ljust(64))
+        self.stdscr.attroff(curses.A_BOLD)
+        self.stdscr.attroff(curses.A_REVERSE)
+
+        non_halted = [
+            stats for stats in self.entries.values() if stats.state != "Halted"
+        ]
+        for i, stats in enumerate(non_halted, start=1):
+            if i >= height - 1:
+                break
+            self.stdscr.addstr(i, 0, stats.vm.name.ljust(40))
+            self.stdscr.addstr(i, 40, stats.state.ljust(50))
+            self.stdscr.addstr(i, 51, str(stats.memory).rjust(6))
+            self.stdscr.addstr(i, 59, str(stats.cpu).rjust(6))
+
+        current_time = time.strftime("%Y-%m-%d %H:%M:%S")
+        self.stdscr.addstr(height - 1, 0, f"qvm-top - {current_time}")
+        self.stdscr.refresh()
+
+    def register_events(self):
+        # pylint: disable=missing-function-docstring
+        self.stats_dispatcher_handlers = [("vm-stats", self.update_stats)]
+        self.dispatcher_handlers = [
+            ("connection-established", self.refresh_all_items),
+            ("domain-add", self.add_domain_item),
+            ("domain-delete", self.remove_domain_item),
+        ]
+        for event in STATE_DICTIONARY:
+            self.dispatcher_handlers.append((event, self.update_domain_item))
+        for handler in self.stats_dispatcher_handlers:
+            self.stats_dispatcher.add_handler(*handler)
+        for handler in self.dispatcher_handlers:
+            self.dispatcher.add_handler(*handler)
+
+    def unregister_events(self):
+        # pylint: disable=missing-function-docstring
+        for handler in self.stats_dispatcher_handlers:
+            self.stats_dispatcher.remove_handler(*handler)
+        for handler in self.dispatcher_handlers:
+            self.dispatcher.remove_handler(*handler)
+
+
+async def main_async(stdscr) -> int:
+    # pylint:disable=missing-function-docstring
+    curses.curs_set(0)
+    stdscr.clear()
+    stdscr.timeout(0)
+
+    app = qubesadmin.Qubes()
+    dispatcher = EventsDispatcher(app)
+    stats_dispatcher = EventsDispatcher(app, api_method="admin.vm.Stats")
+    tasks = [
+        asyncio.ensure_future(dispatcher.listen_for_events()),
+        asyncio.ensure_future(stats_dispatcher.listen_for_events()),
+    ]
+    top = Monitor(
+        app=app,
+        dispatcher=dispatcher,
+        stats_dispatcher=stats_dispatcher,
+        stdscr=stdscr,
+    )
+    top.run()
+    return await asyncio.wait(tasks, return_when=asyncio.FIRST_EXCEPTION)
+
+
+def main() -> int:
+    # pylint:disable=missing-function-docstring
+    return asyncio.run(curses.wrapper(main_async))
+
+
+if __name__ == "__main__":
+    sys.exit(main())