functional models

.gitignore

@@ -31,7 +31,9 @@ run_dir/
 
 # Archived catalog-generation logic (kept locally, not checked in)
 scripts/_catalog_gen.py
-scripts/__pycache__/
+
+# Python
+__pycache__
 
 # Include images of  waveforms in Brave New World bug readme
 !tests/fpga-debugging/axis-async-fifo-c4/*.png

Cargo.toml

@@ -1,6 +1,6 @@
 [workspace]
 resolver = "3"
-members = ["bi", "interp", "protocols", "cli", "graph-interp"]
+members = ["bi", "functional-models", "interp", "protocols", "cli", "graph-interp"]
 
 
 [workspace.package]
@@ -13,6 +13,7 @@ license = "MIT"
 
 [workspace.dependencies]
 protocols = { path = "protocols" }
+functional = { path = "functional-models" }
 baa = { version = "0.19.3", features = ["rand1"] }
 patronus = { git="https://github.com/Nikil-Shyamsunder/patronus#" }
 rand = "0.10"

functional-models/Cargo.toml

@@ -0,0 +1,18 @@
+[package]
+name = "functional"
+version = "0.1.0"
+description = "load functional models generated by our python library"
+edition.workspace = true
+rust-version.workspace = true
+repository.workspace = true
+license.workspace = true
+
+[dependencies]
+protocols.workspace = true
+clap-verbosity-flag = "3.0.4"
+clap.workspace = true
+baa.workspace = true
+patronus.workspace = true
+rustc-hash.workspace = true
+serde = { version = "1.0.229", features = ["derive"] }
+serde_json = "1.0.151"

functional-models/README.md

@@ -0,0 +1,4 @@
+# Functional Models
+
+Define your functional model in `main.py`, export to JSON and
+then load it using the rust library.

functional-models/fun.py

@@ -0,0 +1,134 @@
+import json
+from dataclasses import dataclass, field
+from typing import Optional, Tuple
+
+from pypatronus import (
+    TransitionSystem,
+    State,
+    BitVec,
+    ExprRef,
+    BitVecVal,
+    If,
+    Interpreter,
+)
+
+
+@dataclass
+class Method:
+    name: str
+    inputs: list[ExprRef] = field(default_factory=list)
+    outputs: list[Tuple[str, ExprRef]] = field(default_factory=list)
+    nexts: Optional[list[ExprRef]] = None
+    # indicates whether the method can be executed based on the current model state
+    guard: Optional[ExprRef] = None
+
+
+@dataclass
+class FunctionalModel:
+    name: str
+    methods: list[Method] = field(default_factory=list)
+    states: list[ExprRef] = field(default_factory=list)
+
+
+class Sim:
+    def __init__(self, model: FunctionalModel):
+        self.sys = _build_sys(model)
+        self.model = model
+        self.sim = Interpreter(self.sys)
+        for idx, m in enumerate(model.methods):
+            # note: python lambdas capture the context instead of the value of idx be default which is why
+            #       we need the nested lambdas!
+            setattr(
+                self,
+                m.name,
+                (
+                    lambda ii: (
+                        lambda *args, **kwargs: self._exec_method(ii, *args, **kwargs)
+                    )
+                )(idx),
+            )
+
+    def _exec_method(self, idx: int, *args, **kwargs):
+        assert len(kwargs) == 0, "TODO: support keyword args"
+        method = self.model.methods[idx]
+        inputs = list(args)
+        assert len(inputs) == len(method.inputs), (
+            f"Wrong number of inputs {len(inputs)} != {len(method.inputs)}"
+        )
+
+        assert False, f"TODO: exec {method.name} {inputs}"
+
+
+def verify_model(m: FunctionalModel):
+    for method in m.methods:
+        allowed_symbols = set(m.states) | set(method.inputs)
+
+        for out_name, out_expr in method.outputs:
+            unallowed = out_expr.symbols() - allowed_symbols
+            assert len(unallowed) == 0, (
+                f"Output {out_name}={out_expr} uses symbols that are neither inputs nor state: {unallowed}"
+            )
+
+        if method.nexts is not None:
+            assert len(method.nexts) == len(m.states), (
+                f"[{method.name}] {len(method.nexts)} next state assignments, but model has {len(m.states)} states."
+            )
+            for state, next in zip(m.states, method.nexts):
+                assert state.sort() == next.sort(), (
+                    f"[{method.name}] {state} : {state.sort()} = {next} : {next.sort()}"
+                )
+                unallowed = next.symbols() - allowed_symbols
+                assert len(unallowed) == 0, (
+                    f"[{method.name}] State update {state}={next} uses symbols that are neither inputs nor state: {unallowed}"
+                )
+
+        # check guard
+        if method.guard is not None:
+            allowed_symbols = set(m.states)
+            unallowed = method.guard.symbols() - allowed_symbols
+            assert len(unallowed) == 0, (
+                f"Guard {method.guard} uses symbols that are not state: {unallowed}"
+            )
+
+
+def _build_sys(m: FunctionalModel) -> TransitionSystem:
+    verify_model(m)
+    sys = TransitionSystem(name=m.name)
+    next_states = list(m.states)
+    for t in m.methods:
+        commit_signal = BitVec(f"{t.name}_commit", 1)
+        sys.add_input(commit_signal)
+        guard_signal = BitVecVal(1, 1) if t.guard is None else t.guard
+        sys.add_output(f"{t.name}_guard", guard_signal)
+        input_map = {}
+        for inp in t.inputs:
+            renamed = BitVec(f"{t.name}_in_{inp.name()}", inp.width())
+            input_map[inp] = renamed
+            sys.add_input(renamed)
+        for out_name, out_expr in t.outputs:
+            out_expr = out_expr.replace(input_map)
+            sys.add_output(f"{t.name}_out_{out_name}", out_expr)
+        if t.nexts is not None:
+            for idx, next in enumerate(t.nexts):
+                expr = next.replace(input_map)
+                next_states[idx] = If(commit_signal, expr, next_states[idx])
+    assert len(next_states) == len(m.states)
+    sys.states = [
+        State(sym.name(), next=next) for (sym, next) in zip(m.states, next_states)
+    ]
+    return sys
+
+
+def serialize(m: FunctionalModel, filename):
+    sys = _build_sys(m)
+
+    info = {
+        "name": m.name,
+        "methods": [t.name for t in m.methods],
+        "states": [s.name() for s in m.states],
+    }
+
+    print(sys)
+
+    with open(filename, "w") as f:
+        json.dump({"info": info, "sys": sys.to_btor2_str()}, f)

functional-models/main.py

@@ -0,0 +1,125 @@
+# Copyright 2026 Cornell University
+# released under MIT License
+# author: Kevin Laeufer <laeufer@cornell.edu>
+
+from pypatronus import BitVec, SignExt, ZeroExt, Slice, Update, If, Array, BitVecVal
+from fun import FunctionalModel, Method, serialize, Sim
+
+
+def picorv32_pcpi_mul():
+    """
+    https://github.com/ekiwi/paso/blob/ad2bf83f420ca704ff0e76e7a583791a0e80a545/benchmarks/src/benchmarks/picorv32/PicoRV32Spec.scala#L8
+    """
+    m = FunctionalModel(name="picorv32_pcpi_mul")
+    rs1, rs2 = BitVec("rs1_data", 32), BitVec("rs2_data", 32)
+    m.methods = [
+        Method("pcpi_mul", [rs1, rs2], [("rd_data", rs1 * rs2)]),
+        Method(
+            "pcpi_mulh",
+            [rs1, rs2],
+            [("rd_data", Slice(63, 32, SignExt(32, rs1) * SignExt(32, rs2)))],
+        ),
+        Method(
+            "pcpi_mulhu",
+            [rs1, rs2],
+            [("rd_data", Slice(63, 32, ZeroExt(32, rs1) * ZeroExt(32, rs2)))],
+        ),
+        Method(
+            "pcpi_mulhsu",
+            [rs1, rs2],
+            [("rd_data", Slice(63, 32, SignExt(32, rs1) * ZeroExt(32, rs2)))],
+        ),
+        Method("pcpi_mul_reset"),
+        Method("idle"),
+    ]
+    return m
+
+
+def fifo(data_width: int, num_elements: int, push_pop: bool = False):
+    """https://github.com/ekiwi/paso/blob/ad2bf83f420ca704ff0e76e7a583791a0e80a545/benchmarks/src/benchmarks/fifo/FifoSpec.scala"""
+    counter_width = 12
+    assert num_elements < ((1 << (counter_width - 1)) - 1)
+    mem = Array("mem", counter_width, data_width)
+    count = BitVec("count", counter_width)
+    read = BitVec("read", counter_width)
+    m = FunctionalModel(name="fifo", states=[mem, count, read])
+    num_elements_bv = BitVecVal(num_elements, counter_width)
+    full = count.equals(num_elements_bv)
+    zero = BitVecVal(0, counter_width)
+    empty = count.equals(zero)
+    input = BitVec("input", data_width)
+
+    non_wrap = count + read
+    write_adr = If(non_wrap < num_elements_bv, non_wrap, non_wrap - num_elements_bv)
+    read_plus_one = read + BitVecVal(1, counter_width)
+    read_incr = If(
+        read_plus_one.equals(num_elements_bv),
+        zero,
+        read_plus_one,
+    )
+
+    m.methods = [
+        Method(
+            "push",
+            [input],
+            [],
+            [
+                Update(mem, write_adr, input),  # mem
+                count + BitVecVal(1, counter_width),  # count
+                read,  # read
+            ],
+            ~full,
+        ),
+        Method(
+            "pop",
+            [],
+            [("output", mem[read])],
+            [
+                mem,  # mem
+                count - BitVecVal(1, counter_width),  # count
+                read_incr,  # read
+            ],
+            ~empty,
+        ),
+        Method("reset", nexts=[mem, zero, zero]),
+        Method("idle"),
+    ]
+    if push_pop:
+        m.methods.append(
+            Method(
+                "push_pop",
+                [input],
+                [("output", mem[read])],
+                [
+                    Update(mem, write_adr, input),  # mem
+                    count,  # count
+                    read_incr,  # read
+                ],
+            ),
+        )
+    return m
+
+
+def test_fifo(m: FunctionalModel, num_elements: int, push_pop: bool = False):
+    sim = Sim(m)
+    # sim.push(123)
+    # assert sim.pop() == 123
+    pass  # TODO: implement simulator for testing
+
+
+def main():
+    serialize(picorv32_pcpi_mul(), "picorv32_pcpi_mul.json")
+    params = [
+        {"data_width": 32, "num_elements": 8},
+        {"data_width": 32, "num_elements": 16},
+        {"data_width": 32, "num_elements": 128},
+    ]
+    for p in params:
+        file_name = "fifo_" + "_".join(f"{k}={v}" for k, v in p.items()) + ".json"
+        m = fifo(**p)
+        test_fifo(m, num_elements=p["num_elements"])
+        serialize(m, file_name)
+
+
+if __name__ == "__main__":
+    main()

functional-models/pyproject.toml

@@ -0,0 +1,9 @@
+[project]
+name = "functional-models"
+version = "0.1.0"
+description = "Add your description here"
+readme = "README.md"
+requires-python = ">=3.12"
+dependencies = [
+    "pypatronus==0.39.6",
+]