ARC Reactor, CLI Monitoring for Intel Arc (requires xpu-smi)

README.md

ARC Reactor

ARC Reactor is a simple monitoring script with an epic name for Intel ARC GPUs, it provides an interface similar to xpu-smi stats but with the blank values removed for easier viewing and will refresh every second. However, ARC Reactor still requires xpu-smi to be installed, as it uses xpu-smi dump to collect the information.

Usage

usage: arc-reactor.py [-h] [-d DEVICE] [-m METRICS] [-i I] [-n N] [-c]

options:
  -h, --help            show this help message and exit
  -d DEVICE, --device DEVICE
                        The device IDs or PCI BDF addresses to query.
  -m METRICS, --metrics METRICS
                        Metrics type to collect raw data, options. Separated by the comma. Default: All.
  -i I                  The interval (in seconds) to dump the device statistics to screen. Default value: 1 second.
  -n N                  Number of the device statistics dump to screen. The dump will never be ended if this parameter is not
                        specified.
  -c                    Enable Colourful output.

arc-reactor.py

#!/usr/bin/env python3

import subprocess
import curses
import argparse

def parse_line(line):
    """Parse a line of xpu-smi output into a list of strings. It doesn't look like xpu-smi supports -j for JSON output"""
    return [item.strip().decode('utf-8') for item in line.split(b", ")]

def gen_hr(key_len, value_len, key_title="", value_title=""):
    """Generate a horizontal rule for the table"""
    return '+-' + key_title.center(key_len, '-') + '-+-' + value_title.center(value_len, '-') + '-+'

def main(stdscr):
    # Setup curses
    curses.use_default_colors()
    stdscr = curses.newwin(curses.LINES, curses.COLS, 0, 0)
    stdscr.keypad(True)
    curses.curs_set(0)

    # Draw loading screen
    stdscr.addstr(0, 0, "--ARC-REACTOR-by-Steve-Tech--", curses.A_BOLD)
    stdscr.addstr(2, 0, "Waiting for xpu-smi..")
    stdscr.addstr(2, 21, ".", curses.A_BLINK)
    stdscr.addstr(3, 0, "Press Ctrl+C to exit.", curses.A_STANDOUT)
    stdscr.refresh()

    # Setup colors
    if args.c:
        curses.init_pair(1, curses.COLOR_WHITE, curses.COLOR_BLUE)
        curses.init_pair(2, curses.COLOR_RED, curses.COLOR_BLUE)
        stdscr.bkgd(' ', curses.color_pair(1) | curses.A_BOLD)
        colour1 = curses.color_pair(1)
        colour2 = curses.color_pair(2)
    else:
        colour1 = curses.A_NORMAL
        colour2 = curses.A_NORMAL

    # Load arguments for xpu-smi
    xpu_args = []
    for key, value in vars(args).items():
        if value is not None and key != "c":
            xpu_args.extend([("--" if len(key) > 1 else "-") + key, value])

    # Run xpu-smi
    smi = subprocess.Popen(["xpu-smi", "dump", *xpu_args], stdout=subprocess.PIPE)

    try:
        with smi.stdout as smi_output:
            # Skip to the header
            try:
                while not (header := next(smi_output)).startswith(b"Timestamp"):
                    pass
            except StopIteration:
                curses.endwin()
                print("XPU-SMI Error:", header.decode().rstrip("\n"))
                print("XPU-SMI Args:", *smi.args)
                exit(smi.poll())

            header = parse_line(header)

            # Loop through the output
            for line in iter(smi_output.readline, b''):
                items = parse_line(line)

                rows = tuple(zip(header, items))

                # Get the max character length of the header and items
                header_len = max(len(i) for i, j in rows if j != "")
                item_len = max(len(i) for i in items)

                # Start drawing the table
                stdscr.clear()
                stdscr.addstr(0, 0, gen_hr(header_len, item_len, "ARC-REACTOR"), colour2)

                y_adjust = 1
                num = 0

                for key, value in rows:
                    if value == "" or value == "N/A":  # Skip empty values
                        continue

                    if num == 2:  # Draw a horizontal rule after the first two lines (Timestamp and Device)
                        stdscr.addstr(num + y_adjust, 0, gen_hr(header_len, item_len, "METRIC", "VALUE" if item_len > 5 else ""), colour2)
                        y_adjust += 1

                    # Draw the values in the table
                    stdscr.addstr(num + y_adjust, 0, '|', colour2)
                    stdscr.addstr(num + y_adjust, 2, key)
                    stdscr.addstr(num + y_adjust, header_len + 3, '|', colour2)
                    stdscr.addstr(num + y_adjust, header_len + 5, value)
                    stdscr.addstr(num + y_adjust, header_len + 6 + item_len, '|', colour2)
                    num += 1

                # Draw the footer
                stdscr.addstr(num + y_adjust, 0, gen_hr(header_len, item_len, "BY-STEVE-TECH"), colour2)
                stdscr.refresh()

    except KeyboardInterrupt:
        # Gracefully exit
        smi.send_signal(subprocess.signal.SIGINT)

    smi.wait()
    curses.endwin()

if __name__ == "__main__":
    parser = argparse.ArgumentParser()
    parser.add_argument('-d', '--device', help='The device IDs or PCI BDF addresses to query.')
    parser.add_argument('-m', '--metrics', help='Metrics type to collect raw data, options. Separated by the comma. Default: All.', default=",".join(str(i) for i in range(0,36)))
    parser.add_argument('-i', help='The interval (in seconds) to dump the device statistics to screen. Default value: 1 second.')
    parser.add_argument('-n', help='Number of the device statistics dump to screen. The dump will never be ended if this parameter is not specified.')
    parser.add_argument('-c', help='Enable Colourful output.', action='store_true')
    args = parser.parse_args()

    curses.wrapper(main)