import and validate vcd with kyupy

.gitignore

@@ -1,5 +1,7 @@
 .claude
 .venv
-results
+result
 *.vvp
-*.vcd
+*.vcd
+*.npy
+*.npz

Makefile

@@ -1,22 +1,44 @@
-JPEG_CORE_VVP := jpeg_core_tb.vvp
+
-INFILE        ?= test.jpg
+#
-OUTFILE       ?= output.ppm
+# jpeg_core config
-VVP_FLAGS     ?=
+#
+
+JPEG_CORE_VVP       := jpeg_core_tb.vvp
+JPEG_CORE_INFILE    ?= test.jpg
+JPEG_CORE_OUTFILE   ?= output.ppm
+JPEG_CORE_VVP_FLAGS ?=
 
 JPEG_CORE_DIR    := core_jpeg/src_v
 JPEG_CORE_SRCS   := $(wildcard $(JPEG_CORE_DIR)/*.v)
 JPEG_CORE_TB     := jpeg_core_tb.v
 
-.PHONY: all sim clean
+#
+# General targets
+#
+
+.PHONY: all jpeg_rtl_sim clean
 
 all: $(JPEG_CORE_VVP)
 
+clean:
+	rm -f $(JPEG_CORE_VVP) $(JPEG_CORE_OUTFILE) *.vcd
+
+#
+# jpeg_core targets
+#
+
 $(JPEG_CORE_VVP): $(JPEG_CORE_TB) $(JPEG_CORE_SRCS)
 	iverilog -g2005 -Wall -o $@ $^
 
-# Run simulation
+# Run RTL simulation
-sim: $(JPEG_CORE_VVP)
+jpeg_rtl_sim: $(JPEG_CORE_VVP)
-	vvp $< $(VVP_FLAGS) +infile=$(INFILE) +outfile=$(OUTFILE)
+	vvp $< $(JPEG_CORE_VVP_FLAGS) +infile=$(JPEG_CORE_INFILE) +outfile=$(JPEG_CORE_OUTFILE)
+
+#
+# picorv32 targets
+#
+
+picorv32_rtl_sim:
+	make -C picorv32 test_vcd
+	cp picorv32/testbench.vcd picorv32.vcd
 
-clean:
-	rm -f $(JPEG_CORE_VVP) $(OUTFILE) *.vcd

picorv32_vcd_import.py

@@ -0,0 +1,158 @@
+#!/usr/bin/env -S uv run
+
+import re
+import argparse
+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 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 propagate_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()
+    parser.add_argument('dut', type=str, help='verilog file with unit under test (uut)')
+    parser.add_argument('vcd', type=str, help='vcd file to process')
+    args = parser.parse_args()
+
+    c = verilog.load(args.dut, tlib=SKY130)
+    log.info(f'circuit: {c}')
+    c.resolve_tlib_cells(tlib=SKY130)
+    s_locs = {} #{n.name: l for l, n in enumerate(c.s_nodes)}
+    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_picorv(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):
+        clk_val = values[clk_loc]
+        return clk_val == 0 and time < 100e6
+
+    sim = LogicSim4V(c, sims=1024)
+    log.info(f'sim: {sim}')
+
+    v = vcd.load(args.vcd, var_locs=var_locs_picorv, 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
+    np.save('testbench_orig.npy', merged)
+    for i in range(30):
+        merged = propagate_and_check(sim, merged, node2name)
+        log.info(f'PropCycle[{i}] total bits unassigned: {(merged == logic.UNASSIGNED).sum()} unknown: {(merged == logic.UNKNOWN).sum()}')
+
+    np.save('testbench_prop.npy', merged)
+
+
+if __name__ == "__main__":
+    main()