infield/picorv32_vcd_import.py

#!/usr/bin/env -S uv run

import re
import argparse
from pathlib import Path

import numpy as np
from PIL import Image

from kyupy import verilog, vcd, log, logic
from kyupy.techlib import SKY130
from kyupy.logic_sim import LogicSim4V


def path_for(circuit: str):
    import subprocess
    return Path(subprocess.check_output(
        ["nix", "build", "--print-out-paths", "--no-link",
         f"github:s-holst/benchmark-circuits#{circuit}" ],
        text=True,
    ).strip())


def verilog_nl_path_for(circuit: str):
    return next(path_for(circuit).glob("*/nl/*.nl.v"))


def check_conflicts(patterns1, patterns2, names):
    conflicts = (patterns1 != patterns2) & (patterns1 != logic.UNKNOWN) & (patterns1 != logic.UNASSIGNED) & (patterns2 != logic.UNKNOWN) & (patterns2 != logic.UNASSIGNED)
    if conflicts.sum() > 0:
        for pat_idx in range(patterns1.shape[1]):
            for c, p1, p2, n in zip(conflicts[:,pat_idx], patterns1[:,pat_idx], patterns2[:,pat_idx], names):
                if c:
                    log.error(f'incompatible logic values: pattern: {pat_idx} (p)pio: {n} {p1} != {p2}')
    assert conflicts.sum() == 0, "incompatible patterns"


def sim_and_check(sim: LogicSim4V, patterns: np.ndarray, node2name: dict):
    responses = np.full_like(patterns, logic.UNASSIGNED)
    sim.simulate(patterns, responses)
    #to_image(np.concatenate([patterns, responses]), 'pat_res.png')
    po_names = [node2name[sim.circuit.s_nodes[loc]] for loc in sim.po_s_locs]
    ppio_names = [node2name[sim.circuit.s_nodes[loc]] for loc in sim.ppio_s_locs]
    merged = np.full_like(patterns, logic.UNASSIGNED)
    merged[sim.pi_s_locs] = patterns[sim.pi_s_locs]  # copy primary inputs
    merged[sim.po_s_locs] = logic.mv_merge(patterns[sim.po_s_locs], responses[sim.po_s_locs])  # merge primary outputs
    merged[sim.ppio_s_locs,0] = patterns[sim.ppio_s_locs,0]  # copy pseudo-primary io for very first clock cycle (simulation does not give these)
    merged[sim.ppio_s_locs,1:] = logic.mv_merge(patterns[sim.ppio_s_locs,1:], responses[sim.ppio_s_locs,:-1])  # merge pseudo-primary io, shifted by one clock cycle (discard ppio of very last clock cycle from sim)
    check_conflicts(patterns[sim.po_s_locs], responses[sim.po_s_locs], po_names)
    check_conflicts(patterns[sim.ppio_s_locs,1:], responses[sim.ppio_s_locs,:-1], ppio_names)
    return merged


def to_image(data: np.ndarray, path: str):
    _palette = np.full(8, 128, dtype=np.uint8)
    _palette[0] = 0    # black
    _palette[3] = 255  # white
    pixels = _palette[data]
    Image.fromarray(pixels, mode='L').save(path)


def find_name(c, n):
    for nn in c.fanout([n], node_filter=lambda n: n.kind in ('input', 'BUF1', '__fork__', 'DFF')):
        if nn.name[0] == '_': continue
        if nn.name.startswith('net'): continue
        if nn.name.startswith('fanout'): continue
        if nn.name.startswith('hold'): continue
        if nn.name.startswith('output'): continue
        if nn.name.startswith('load_slew'): continue
        if nn.name.startswith('max_cap'): continue
        if nn.name.startswith('output'): continue
        if nn.name.startswith('wire'): continue
        if nn.name.startswith('ANTENNA'): continue
        return nn.name
    return n.name


def main():
    parser = argparse.ArgumentParser(description='Imports a VCD file from PicoRV32 simulation into kyupy, simulates a few cycles and save the patterns as npy file for later fault simulation.')
    parser.add_argument('vcd', type=str, help='input vcd file')
    parser.add_argument('npy', type=str, help='output pattern file in npy format')
    parser.add_argument('-c', type=int, default=16, help='number of clock cycles to re-simulate for filling unknowns')
    parser.add_argument('-n', type=int, default=0, help='number of patterns to convert (default 0=all)')
    args = parser.parse_args()

    c = verilog.load(verilog_nl_path_for("picorv32-sky130"), tlib=SKY130)
    log.info(f'circuit: {c}')
    c.resolve_tlib_cells(tlib=SKY130)
    s_locs = {}
    name2node = {}
    node2name = {}

    for l, n in enumerate(c.s_nodes):
        name = find_name(c, n)
        s_locs[name] = l
        name2node[name] = n
        node2name[n] = name

    clk_loc = next(s_locs[name] for name in ['blif_clk_net', 'clk'] if name in s_locs)

    s_found: list[str|None] = [None] * len(c.s_nodes)

    def var_assign(var: vcd.Var, name: str):
        if var.width == 1:
            s_found[s_locs[name]] = name
            return [s_locs[name]]
        else:
            locs = []
            for i in reversed(range(var.width)):
                if loc := s_locs.get(f'{name}[{i}]'):
                    s_found[loc] = f'{name}[{i}]'
                    locs.append(loc)
                else:
                    locs.append(-1)
            return locs

    def var_locs(var: vcd.Var):
        if var.scope.name in ['uut', 'picorv32_core', 'pcpi_mul', 'pcpi_div']:
            if var.reference.startswith('cpu_state'):  # FSM got re-coded during synthesis
                return []
            elif var.reference.startswith('mem_wordsize'):  # FSM got re-coded during synthesis
                return []
            elif var.reference.startswith('dbg_reg_x'):
                name = re.sub(r'dbg_reg_x([0-9]+)', r'cpuregs[\1]', var.reference)
            elif var.scope.name == 'pcpi_mul':
                name = 'genblk1.genblk1.pcpi_mul.' + var.reference
            elif var.scope.name == 'pcpi_div':
                name = 'genblk2.pcpi_div.' + var.reference
            else:
                name = var.reference

            if name in s_locs:
                return var_assign(var, name)
            elif var.width > 1:
                if ' ' not in name: return []
                name, indices = name.split()
                assert indices == f'[{var.width-1}:0]', f'unsupported indices: {var.reference}'
                return var_assign(var, name)
        return []

    def step_filter(time, values, var_map):
        if args.n > 0 and time is not None and time >= (args.n*10000):
            return False
        return values[clk_loc] == 0

    sim = LogicSim4V(c, sims=10240, c_reuse=True, strip_forks=True)
    log.info(f'sim: {sim}')

    v = vcd.load(args.vcd, var_locs=var_locs, step_filter=step_filter, s_len=sim.s_len)
    log.info(f'vcd: {v}')

    unknown_pi = 0
    unknown_ppio = 0
    for l, n in enumerate(sim.circuit.s_nodes):
        if n.kind == 'output': continue
        if 'cpu_state' in node2name[n] or 'mem_wordsize' in node2name[n]: continue  # re-coded FSMs
        if s_found[l] is None:
            log.warn(f'No VCD $var found for netlist port loc: {l} name: {node2name[n]} node: {n}')
            if n.kind == 'input':
                unknown_pi += 1
            else:
                unknown_ppio += 1
    log.info(f'Unassigned PI: {unknown_pi}/{len(sim.pi_s_locs)} PPIO: {unknown_ppio}/{len(sim.ppio_s_locs)}')
    log.info(f'Total bits unassigned: {(v.data == logic.UNASSIGNED).sum()} unknown: {(v.data == logic.UNKNOWN).sum()}')

    merged = v.data
    for i in range(args.c):
        merged = sim_and_check(sim, merged, node2name)
        log.info(f'Cycle[{i}] total bits unassigned: {(merged == logic.UNASSIGNED).sum()} unknown: {(merged == logic.UNKNOWN).sum()}')

    log.info(f'Writing {args.npy}')
    np.save(args.npy, merged)


if __name__ == "__main__":
    main()