retroSoC/scripts/run_npu_p6_netlist.py

#!/usr/bin/env python3
"""Run IHP130 NPU block synthesis/STA and KWS/VWW netlist workloads."""

from __future__ import annotations

import argparse
import hashlib
import json
import os
import re
import subprocess
import sys
from pathlib import Path


ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT / "scripts"))
sys.path.insert(0, str(ROOT / "tests"))

from test_npu_rtl_p4 import _operator_cases  # noqa: E402


SRAM_MACRO = "RM_IHPSG13_1P_1024x32_c2_bm_bist"
DIRECTED_CASES = ("conv_kws_10x4", "dw33_s2_c7", "rej_stride3", "arith_acc_overflow")
REJECTED = re.compile(r"FAILED|FATAL|assertion failed|%Error|SIM_TEST_FAIL|SIM_TEST_TIMEOUT")
SPECIFY_BLOCK = re.compile(r"(?ms)^\s*specify\b.*?^\s*endspecify\s*$")
DELAYED_SIGNAL = re.compile(r"\bdelayed_([A-Za-z][A-Za-z0-9_]*)\b")


def identity(path: Path) -> dict[str, object]:
    data = path.read_bytes()
    return {"path": str(path.resolve()), "bytes": len(data), "sha256": hashlib.sha256(data).hexdigest()}


def run(command: list[str], *, log: Path, timeout: int | None = None) -> None:
    completed = subprocess.run(
        command,
        cwd=ROOT,
        text=True,
        stdout=subprocess.PIPE,
        stderr=subprocess.STDOUT,
        timeout=timeout,
        check=False,
        env={**os.environ, "CCACHE_DISABLE": "1"},
    )
    log.parent.mkdir(parents=True, exist_ok=True)
    log.write_text(completed.stdout, encoding="utf-8")
    if completed.returncode != 0:
        raise RuntimeError(f"command failed ({completed.returncode}): {' '.join(command)}\n{completed.stdout[-4000:]}")


def write_functional_stdcell(source: Path, output: Path) -> None:
    text = source.read_text(encoding="utf-8")
    transformed, specify_count = SPECIFY_BLOCK.subn("", text)
    transformed = DELAYED_SIGNAL.sub(r"\1", transformed)
    transformed, notifier_count = re.subn(
        r"(?m)^\s*reg notifier;\s*$", "\twire notifier = 1'b0;", transformed
    )
    if specify_count == 0 or notifier_count == 0 or "delayed_" in transformed:
        raise ValueError("IHP standard-cell functional transformation was incomplete")
    output.write_text(transformed, encoding="utf-8")


def plusarg_value(arguments: list[str], name: str) -> str:
    prefix = f"+{name}="
    matches = [argument[len(prefix) :] for argument in arguments if argument.startswith(prefix)]
    if len(matches) != 1:
        raise ValueError(f"directed netlist case has invalid {name} plusarg")
    return matches[0]


def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--output-dir", type=Path, required=True)
    parser.add_argument("--timeout-seconds", type=int, default=7200)
    parser.add_argument("--jobs", type=int, default=4)
    parser.add_argument(
        "--reuse-physical",
        action="store_true",
        help="reuse an existing isolated-npu.json after validating it (debug only)",
    )
    args = parser.parse_args()
    if args.jobs <= 0:
        parser.error("jobs must be positive")
    output = args.output_dir.resolve()
    physical = output / "physical"
    evidence_path = physical / "isolated-npu.json"
    if not args.reuse_physical:
        run(
            [
                sys.executable,
                str(ROOT / "scripts/run_npu_p1_flow.py"),
                "--build-root",
                str(physical),
                "--top",
                "npu_block_top",
                "--evidence",
                str(evidence_path),
            ],
            log=physical / "driver.log",
            timeout=args.timeout_seconds,
        )
    elif not evidence_path.is_file():
        raise ValueError("--reuse-physical requires an existing isolated-npu.json")
    physical_evidence = json.loads(evidence_path.read_text(encoding="utf-8"))
    top = physical_evidence["tops"]["npu_block_top"]
    if physical_evidence.get("verdict") != "PASS" or top["synth"]["sram_macro_count"] != 16:
        raise ValueError("NPU block physical evidence did not pass with exactly 16 SRAM macros")
    netlist = Path(top["synth"]["netlist"])
    transaction_netlist = netlist.with_name(netlist.name.replace("_yosys.v", "_yosys_debug.v"))
    if (
        not transaction_netlist.is_file()
        or transaction_netlist.read_text(encoding="utf-8", errors="replace").count(
            f"{SRAM_MACRO} u_mem"
        )
        != 16
    ):
        raise ValueError("technology-mapped transaction netlist does not contain 16 SRAM macros")
    pdk = ROOT / "physical/pdk/IHP-Open-PDK/ihp-sg13g2/libs.ref"
    stdcell_source = pdk / "sg13g2_stdcell/verilog/sg13g2_stdcell.v"
    stdcell_functional = output / "sg13g2_stdcell_functional.v"
    write_functional_stdcell(stdcell_source, stdcell_functional)
    compile_dir = output / "verilator-netlist"
    simulator = compile_dir / "npu_netlist_tb"
    compile_log = output / "compile.log"
    run(
        [
            "verilator",
            "--binary",
            "--timing",
            "--top-module",
            "npu_operator_tb",
            "-Wno-fatal",
            "--x-initial",
            "0",
            "--x-assign",
            "0",
            "-j",
            str(args.jobs),
            "-o",
            simulator.name,
            "--Mdir",
            str(compile_dir),
            "-I" + str(ROOT / "rtl/managed/clusterip/common/rtl/interface"),
            "-I" + str(ROOT / "rtl/ip/multimedia"),
            str(ROOT / "rtl/managed/clusterip/common/rtl/interface/axi4_if.sv"),
            str(pdk / "sg13g2_stdcell/verilog/sg13g2_udp.v"),
            str(stdcell_functional),
            str(ROOT / "tests/rtl/ihp130_npu_sram_model.sv"),
            str(transaction_netlist),
            str(ROOT / "tests/rtl/npu_operator_tb.sv"),
        ],
        log=compile_log,
        timeout=args.timeout_seconds,
    )
    artifact_dir = output / "artifacts"
    artifact_dir.mkdir(parents=True, exist_ok=True)
    generated = _operator_cases(artifact_dir, selected=set(DIRECTED_CASES))
    by_name = {plusarg_value(arguments, "NAME"): arguments for arguments in generated}
    if not set(DIRECTED_CASES) <= set(by_name):
        raise ValueError("P4 fixture generator omitted a required P6 netlist case")
    cases = []
    for name in DIRECTED_CASES:
        plusargs = by_name[name]
        log = output / f"{name}.log"
        command = [str(simulator), *plusargs]
        run(command, log=log, timeout=args.timeout_seconds)
        text = log.read_text(encoding="utf-8", errors="replace")
        if "NPU operator test passed" not in text or REJECTED.search(text):
            raise ValueError(f"{name} netlist workload did not produce a clean pass")
        artifacts = {
            label.lower(): identity(Path(plusarg_value(plusargs, label)))
            for label in ("IMG", "GOLD")
        }
        cases.append({
            "case": name,
            "status": "PASS",
            "command": command,
            "artifacts": artifacts,
            "log": identity(log),
        })
    report = {
        "schema": 1,
        "phase": "NPU-P6",
        "verification_ids": ["NPU-V017", "NPU-V018"],
        "git_revision": subprocess.check_output(
            ["git", "-C", str(ROOT), "rev-parse", "HEAD"], text=True
        ).strip(),
        "physical_reused": args.reuse_physical,
        "physical": identity(evidence_path),
        "netlist": identity(netlist),
        "transaction_netlist": identity(transaction_netlist),
        "stdcell_source": identity(stdcell_source),
        "stdcell_functional": identity(stdcell_functional),
        "simulator": identity(simulator),
        "compile_log": identity(compile_log),
        "sram_functional_model": identity(ROOT / "tests/rtl/ihp130_npu_sram_model.sv"),
        "sram_macro_count": 16,
        "simulation_policy": {
            "simulator": "Verilator",
            "state_model": "2-state",
            "x_initial": "0",
            "x_assign": "0",
            "timing_evidence": False,
        },
        "implementation": {
            name: identity(ROOT / name)
            for name in (
                "scripts/run_npu_p6_netlist.py",
                "scripts/run_npu_p1_flow.py",
                "physical/smoke/syn/yosys/npu_block_top.sv",
                "tests/rtl/npu_operator_tb.sv",
                "tests/rtl/ihp130_npu_sram_model.sv",
                "tests/test_npu_rtl_p4.py",
            )
        },
        "cases": cases,
        "summary": {"required_cases": 4, "passes": 4, "failures": 0, "skipped": 0},
        "verdict": "PASS",
    }
    report_path = output / "qualification-p6-netlist.json"
    report_path.write_text(json.dumps(report, indent=2, sort_keys=True) + "\n", encoding="utf-8")
    print(f"NPU-P6 netlist qualification: PASS ({report_path})")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())