LibreLane Dumpster Fire Local CI script

ci_local.py

# SPDX-License-Identifier: Unlicense
# Copyright (c) 2025 Mohamed Gaber
# Please don't actually use this.
import os
import math
import yaml
import curses
from time import sleep
from pathlib import Path
from concurrent.futures import ThreadPoolExecutor
import subprocess

__file_dir__ = Path(__file__).absolute().parent
lln_root = __file_dir__
TEST_SETS_FILE = lln_root / ".github" / "test_sets" / "test_sets.yml"

status = {}
assoc_data = {}

def run_test(tup):
    test_name, pdk, scl = tup
    global status
    status[tup] = "P"
    config_file, script = assoc_data[tup]
    design_dir = config_file.parent
    try:
        cmd = ["librelane", config_file]
        if script is not None:
            cmd = ["python3", script]
        cmd += [
            "--condensed",
            "--overwrite",
            "--run-tag", f"test-{test_name}-{pdk}-{scl}",
            "--pdk", pdk,
            "--scl", scl,
            "-j1",
            "-c", "DRT_THREADS=1",
        ]
        subprocess.check_call(cmd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
        status[tup] = "S"
    except subprocess.CalledProcessError as e:
        if e.returncode == 2:
            status[tup] = "F"
        else:
            status[tup] = "E"
    except Exception as e:
        with open(design_dir / f"runner_exception_{pdk}_{scl}.log", "w") as f:
            f.write(str(e))
        status[tup] = "E"


def run_tests(screen: curses.window):
    global status

    curses.start_color()
    curses.init_pair(1, curses.COLOR_WHITE, curses.COLOR_BLACK)
    curses.init_pair(2, curses.COLOR_YELLOW, curses.COLOR_BLACK)
    curses.init_pair(3, curses.COLOR_GREEN, curses.COLOR_BLACK)
    curses.init_pair(4, curses.COLOR_RED, curses.COLOR_BLACK)
    max_y, max_x = screen.getmaxyx()

    data_str = open(TEST_SETS_FILE).read()
    data = yaml.safe_load(data_str)
    test_set_data = filter(lambda e: e["name"] in ["fastest_test_set"], data)

    for test_set in list(test_set_data):
        pdk, scl = test_set["scl"].split("/")
        for design in test_set["designs"]:
            design_name = design
            test_name = design_name
            script = None
            config_filename = "config.json"
            if not isinstance(design, str):
                design_name = design["name"]
                test_name = design.get("test_name", design_name)
                config_filename = design.get("config_file", config_filename)
                script_filename = design.get("script")
                if script_filename:
                    script = lln_root / "test" / "designs" / design_name / script_filename
            config_file = lln_root / "test" / "designs" / design_name / config_filename
            assoc_data[(test_name, pdk, scl)] = (config_file, script)
            status[(test_name, pdk, scl)] = "N"

    max_test_width = max(len("/".join(key)) for key in status)
    test_cell_width = max_test_width + 3  # space indicator space
    tests_per_line = max_x // test_cell_width
    lines_for_tests = math.ceil(len(status) // tests_per_line)
    if max_y < lines_for_tests:
        screen.addstr("window too small")
        screen.refresh()
        screen.getkey()
        exit(-1)

    tpe = ThreadPoolExecutor(int(os.getenv("CI_LOCAL_JOBS", os.cpu_count())))
    futures = tpe.map(run_test, status)
    while True:
        x = 0
        y = 0
        screen.clear()
        try:
            for tup in status:
                value = status[tup]
                if x + test_cell_width >= max_x:
                    x = 0
                    y += 1
                screen.addstr(y, x, "/".join(tup), curses.color_pair(1))
                screen.addstr(
                    y,
                    x + test_cell_width - 2,
                    value,
                    curses.color_pair({"N": 1, "P": 2, "S": 3, "F": 4, "E": 5}[value]),
                )
                x += test_cell_width
            screen.refresh()
        except curses.error as e:
            screen.addstr(0, 0, str(e) + f"@{x, y}")
            screen.refresh()
        sleep(1)


curses.wrapper(run_tests)