move baseline into class, support for pattern list import, fault list export, avoid resolving circuits twice.

Makefile

@@ -0,0 +1,2 @@
+atpg:
+	quaigh atpg --output c6288.test tests/c6288.bench

README.md

@@ -36,6 +36,31 @@ uv run jupyter lab
 Choose `uv-env` as kernel in Jupyter Lab.
 
 
+## Performance Statistics
+
+Numbers are given in `gfs/s` = `gates * faults * patterns / second`.
+
+Stuck-at fault sim on 1024 patterns, baseline:
+
+| OS, CPU, RAM | `tests/c6288.bench` | `polito-itc99-b15-sky130` |
+|--------------|---------------------|---------------------------|
+| MacOS, M3 Max, 32GiB (Stefan)  | 1.52G | 1.97G |
+| WSL, i7-14650HX, 15GiB (Zhang) | 1.68G | 2.03G |
 
 
+## Notes
 
+```
+Common Fault Classes:
+DT - Detected
+DS - Detected By Simulation
+DI - Detected By Implication
+PT - Possibly detected
+UD - Undetectable
+UB - Undetectable - blocked
+UR - Undetectable - redundant
+AU - ATPG untestable
+AN - ATPG untestable - not detected
+ND - Not detected
+NO - Not detected - not observed
+```

main.py

@@ -6,22 +6,22 @@ import time
 
 import numpy as np
 
-from kyupy import verilog, bench, log, logic, batchrange
+from kyupy import verilog, bench, log, logic, batchrange, atalanta, stil
 from kyupy.techlib import techlib_by_name, KYUPY
-from kyupy.logic_sim import LogicSim2V
 
 from fsim.static import LineRoles, FaultSet
+from fsim.baseline import SAFSimSimple
+
 
 def main():
 
     parser = argparse.ArgumentParser(description='A basic stuck-at fault simulator.')
     parser.add_argument('-t', '--tlib', default='SKY130', help=f'Techlib for verilog circuit. Default: SKY130, available: {sorted(techlib_by_name.keys())}.')
-    parser.add_argument('-p', '--patterns', default=1024, help='Number of random patterns to simulate. Default: 1024.')
+    parser.add_argument('-p', '--patterns', default=1024, help='Pattern file or number of random patterns to simulate. Default: 1024.')
+    parser.add_argument('-o', '--output', default=None, help='')
     parser.add_argument('--seed', type=int, default=42, help='Random seed for reproducibility. Default: 42.')
     parser.add_argument('circuit', help='Gate-level verilog, bench, or nix package to import. See available packages: "nix flake show github:s-holst/benchmark-circuits".')
     args = parser.parse_args()
-    args.patterns = int(args.patterns)
-    args.tlib = techlib_by_name[args.tlib]
 
     if not (circuit_path := Path(args.circuit)).exists(): # fallback to published nix package.
         nix_cmd = f"nix build github:s-holst/benchmark-circuits#{args.circuit} --print-out-paths --no-link"
@@ -30,75 +30,71 @@ def main():
 
     log.info(f'loading {circuit_path} ...')
     if circuit_path.name.endswith('.bench'):
-        c = bench.load(circuit_path)
         args.tlib = techlib_by_name['KYUPY']
+        c = bench.load(circuit_path)
     else:
+        args.tlib = techlib_by_name[args.tlib]
         c = verilog.load(circuit_path, tlib=args.tlib)
+
     stats = {k.replace('__',''): v for k, v in c.stats(args.tlib).items() if k.startswith('__') or k.endswith('put')}
     log.info(f'circuit {stats=}')
 
     lr = LineRoles(c, args.tlib)
     log.info(f'line role stats={lr.stats}')
 
-    fs = FaultSet(c, args.tlib)
+    c_resolved = c.copy()
+    c_resolved.resolve_tlib_cells(args.tlib)
+
+    fs = FaultSet(c, args.tlib, c_resolved)
     log.info(f'fault sites: {len(fs.fault_sites)}')
+    log.info(f'uncollapsed stuck-at fault count: {len(fs.saf_set)}')
     log.info(f'collapsed stuck-at fault count: {len(fs.saf_equiv_classes)}')
 
-    c.resolve_tlib_cells(args.tlib)
-
     ffr_stems = []
-    for stem, _ in c.fanout_free_regions(KYUPY):
+    for stem, _ in c_resolved.fanout_free_regions(KYUPY):
         if len(stem.outs) > 0 and stem.outs[0] is not None:
             ffr_stems.append(stem.outs[0])
     ffr_stems = np.array(ffr_stems, dtype=np.uint32)
 
     log.info(f'FFR count: {len(ffr_stems)}')
 
-    sim = LogicSim2V(c, sims=min(args.patterns, 10240))
-
     rng = np.random.default_rng(args.seed)
-    patterns = rng.choice(
-                [logic.ZERO, logic.ONE],
-                size=(sim.s_len, args.patterns),
-            ).astype(np.uint8)
 
-    golden = np.zeros_like(patterns)
+    if not (tests_path := Path(args.patterns)).exists(): # fallback to random patterns.
-    log.info(f'{sim=}')
+        patterns = rng.choice(
+                    [logic.ZERO, logic.ONE],
+                    size=(len(c_resolved.s_nodes(KYUPY)), int(args.patterns)),
+                ).astype(np.uint8)
+    else:
+        log.info(f'loading {tests_path} ...')
+        if tests_path.name.endswith('.stil'):
+            patterns = stil.load(tests_path).tests(c_resolved)
+        else:
+            patterns = atalanta.load(tests_path).tests(c_resolved)
 
-    sim.simulate(patterns, golden)
+    saf_collapsed = np.array(list(fs.saf_equiv_classes.keys()), dtype=np.uint32)
-    log.info(f'golden sim finished.')
+    rng.shuffle(saf_collapsed)
 
-    syndrome = np.zeros_like(patterns)
+    safsim = SAFSimSimple(c_resolved, min(patterns.shape[1], 10240))
+    log.info(f'{safsim.sim=}')
 
-    injection_faults = np.array(list(fs.saf_equiv_classes.keys()), dtype=np.uint32)
+    fclasses = safsim.classify_faults(saf_collapsed, patterns)
-    rng.shuffle(injection_faults)
 
-    undetected = set()
+    log.info(f'fsim time: {safsim.sim_time:.2f}s')
-    detected = set()
+    sim_performance = stats['comb'] * len(saf_collapsed) * patterns.shape[1] / safsim.sim_time
-
-    start_time = time.perf_counter()
-
-    with log.progress() as p:
-        for fidx, fault in enumerate(injection_faults):
-            fault_site = fault//2
-            fault_polarity = fault&1
-            p.update((fidx+1) / len(injection_faults), f'd:{len(detected)} u:{len(undetected)}')
-            for bo, bs in batchrange(patterns.shape[1], sim.sims):
-                sim.s_assign[:, :bs] = patterns[:, bo:bo+bs]
-                sim.s_to_c()
-                sim.c_prop(fault_line=fault_site, fault_model=fault_polarity)
-                sim.c_to_s()
-                syndrome[:, bo:bo+bs] = sim.s_result[:,:bs]
-            if np.allclose(golden, syndrome):
-                undetected.add(fault)
-            else:
-                detected.add(fault)
-
-    sim_time = time.perf_counter() - start_time
-    log.info(f'fsim time: {sim_time:.2f}s')
-    sim_performance = stats['comb'] * len(injection_faults) * patterns.shape[1] / sim_time
     log.info(f'fsim performance: {sim_performance:.2e} gfp/s')
-    log.info(f'detected by simulation: {len(detected)}/{len(injection_faults)}  -  {len(detected)/len(injection_faults)*100:.2f}%')
+    log.info(f'detected by simulation (collapsed): {len(fclasses["DS"])}/{len(saf_collapsed)}  -  {len(fclasses["DS"])/len(saf_collapsed)*100:.2f}%')
+
+    if args.output is not None:
+        out_path = Path(args.output)
+        with open(out_path, 'w') as f:
+            log.info(f'Writing {out_path.absolute()} ...')
+            for flt_rep in fclasses["NO"]:
+                for flt in fs.saf_equiv_classes[flt_rep]:
+                    f.write(f'{fs.fault_type_str(flt)}\tNO\t{fs.fault_site_str(c, args.tlib, flt)}\n')
+            for flt_rep in fclasses["DS"]:
+                for flt in fs.saf_equiv_classes[flt_rep]:
+                    f.write(f'{fs.fault_type_str(flt)}\tDS\t{fs.fault_site_str(c, args.tlib, flt)}\n')
 
 if __name__ == "__main__":
     main()

src/fsim/baseline.py

@@ -0,0 +1,47 @@
+import time
+from collections.abc import Collection
+
+import numpy as np
+
+from kyupy import log, batchrange
+from kyupy.circuit import Circuit
+from kyupy.logic_sim import LogicSim2V
+
+class SAFSimSimple:
+
+    def __init__(self, circuit_resolved: Circuit, batch_size: int):
+        self.sim = LogicSim2V(circuit_resolved, sims=batch_size)
+        self.sim_time = 0
+        pass
+
+    def classify_faults(self, faults: Collection[int], patterns: np.ndarray):
+
+        golden = np.zeros_like(patterns)
+        self.sim.simulate(patterns, golden)
+        log.info(f'golden sim finished.')
+
+        syndrome = np.zeros_like(patterns)
+
+        fclass_NO = set()
+        fclass_DS = set()
+
+        start_time = time.perf_counter()
+
+        with log.progress() as p:
+            for fidx, fault in enumerate(faults):
+                fault_site = fault//2
+                fault_polarity = fault&1
+                p.update((fidx+1) / len(faults), f'DS:{len(fclass_DS)} NO:{len(fclass_NO)}')
+                for bo, bs in batchrange(patterns.shape[1], self.sim.sims):
+                    self.sim.s_assign[:, :bs] = patterns[:, bo:bo+bs]
+                    self.sim.s_to_c()
+                    self.sim.c_prop(fault_line=fault_site, fault_model=fault_polarity)
+                    self.sim.c_to_s()
+                    syndrome[:, bo:bo+bs] = self.sim.s_result[:,:bs]
+                if np.allclose(golden, syndrome):
+                    fclass_NO.add(fault)
+                else:
+                    fclass_DS.add(fault)
+
+        self.sim_time += time.perf_counter() - start_time
+        return {'DS': fclass_DS, 'NO': fclass_NO}

src/fsim/static.py

@@ -90,7 +90,7 @@ class FaultSet:
     Stuck-at faults are collapsed using basic controlling-value theory of simple gates.
     """
 
-    def __init__(self, circuit: Circuit, tlib: TechLib):
+    def __init__(self, circuit: Circuit, tlib: TechLib, circuit_resolved: Circuit):
         lr = LineRoles(circuit, tlib)
 
         self.fault_sites = {l.index for l in circuit.lines
@@ -117,22 +117,22 @@ class FaultSet:
         representative f always contains at least f (:math:`f \in` saf_equiv_classes[f]).
         """
 
-        def collect_equivalent_faults(circuit: Circuit, fault: int) -> set:
+        def collect_equivalent_faults(circuit_resolved: Circuit, fault: int) -> set:
             site = fault//2
             polarity = fault&1
-            driver = circuit.lines[site].driver
+            driver = circuit_resolved.lines[site].driver
             # Stop collapsing at fanout stems, primary inputs, and cells whose
             # output faults have no equivalent single-input fault (XOR/XNOR, MUX, the two-term
             # AND-OR/OR-AND cells, and sequential elements).
             if len(driver.outs) > 1 or len(driver.ins) == 0 or driver.kind in ('XOR2', 'XOR3', 'XOR4', 'XNOR2', 'XNOR3', 'XNOR4', 'MUX21', 'AO22', 'AOI22', 'OA22', 'OAI22', 'DFF', 'LATCH'): return {fault}
-            if driver.kind == 'INV1': return {fault} | collect_equivalent_faults(circuit, driver.ins[0].index*2 + (1-polarity))
+            if driver.kind == 'INV1': return {fault} | collect_equivalent_faults(circuit_resolved, driver.ins[0].index*2 + (1-polarity))
-            if driver.kind in ('BUF1', '__fork__'): return {fault} | collect_equivalent_faults(circuit, driver.ins[0].index*2 + (polarity))
+            if driver.kind in ('BUF1', '__fork__'): return {fault} | collect_equivalent_faults(circuit_resolved, driver.ins[0].index*2 + (polarity))
 
             def collapse(*faults: tuple) -> set:
                 """Collapse the output fault onto the given (input_line, stuck_value) pairs."""
                 equiv = {fault}
                 for site, value in faults:
-                    equiv |= collect_equivalent_faults(circuit, site.index*2 + value)
+                    equiv |= collect_equivalent_faults(circuit_resolved, site.index*2 + value)
                 return equiv
 
             # Simple gates: driving any input to its controlling value forces the output.
@@ -151,13 +151,13 @@ class FaultSet:
 
             # AND-OR / OR-AND complex gates: only the "single" (un-paired) terms force the output.
             #   AO21  = (i0&i1) | i2        : out s-a-1 == i2 s-a-1
-            #   AOI21 = ~AO21              : out s-a-0 == i2 s-a-1
+            #   AOI21 = ~AO21               : out s-a-0 == i2 s-a-1
             #   OA21  = (i0|i1) & i2        : out s-a-0 == i2 s-a-0
-            #   OAI21 = ~OA21              : out s-a-1 == i2 s-a-0
+            #   OAI21 = ~OA21               : out s-a-1 == i2 s-a-0
             #   AO211 = (i0&i1) | i2 | i3   : out s-a-1 == i2 s-a-1, i3 s-a-1
-            #   AOI211= ~AO211             : out s-a-0 == i2 s-a-1, i3 s-a-1
+            #   AOI211= ~AO211              : out s-a-0 == i2 s-a-1, i3 s-a-1
             #   OA211 = (i0|i1) & i2 & i3   : out s-a-0 == i2 s-a-0, i3 s-a-0
-            #   OAI211= ~OA211             : out s-a-1 == i2 s-a-0, i3 s-a-0
+            #   OAI211= ~OA211              : out s-a-1 == i2 s-a-0, i3 s-a-0
             if driver.kind == 'AO21':   return collapse((driver.ins[2], 1)) if polarity == 1 else {fault}
             if driver.kind == 'AOI21':  return collapse((driver.ins[2], 1)) if polarity == 0 else {fault}
             if driver.kind == 'OA21':   return collapse((driver.ins[2], 0)) if polarity == 0 else {fault}
@@ -167,11 +167,9 @@ class FaultSet:
             if driver.kind == 'OA211':  return collapse((driver.ins[2], 0), (driver.ins[3], 0)) if polarity == 0 else {fault}
             if driver.kind == 'OAI211': return collapse((driver.ins[2], 0), (driver.ins[3], 0)) if polarity == 1 else {fault}
 
-            raise ValueError(f'Unknown node kind {driver.kind}. Circuit should have been resolved to techlib.KYUPY?')
+            raise ValueError(f'Unknown node kind {driver.kind}. circuit_resolved must be resolved to techlib.KYUPY with resolve_tlib_cells().')
 
         remaining = self.saf_set.copy()
-        circuit_resolved = circuit.copy()
-        circuit_resolved.resolve_tlib_cells(tlib)
         for n in circuit.reversed_topological_order(tlib):
             for il in n.ins.without_nones():
                 for polarity in (0, 1):
@@ -181,13 +179,16 @@ class FaultSet:
                         remaining.difference_update(equiv)
                         self.saf_equiv_classes[repr] = equiv
 
-        self.circuit = circuit
+    @staticmethod
-        self.tlib = tlib
+    def fault_site_str(circuit: Circuit, tlib: TechLib, fault: int) -> str:
-
+        site = circuit.lines[fault//2]
-    def fault_str(self, fault: int) -> str:
-        site = self.circuit.lines[fault//2]
-        polarity = '@1' if fault&1 else '@0'
         if site.driver.kind == '__fork__':
-            return f'{site.reader.name}/{self.tlib.pin_name(site.reader.kind, site.reader_pin)}{polarity}'
+            return f'{site.reader.name}/{tlib.pin_name(site.reader.kind, site.reader_pin)}'
+        elif site.driver.kind == 'input':
+            return f'{site.driver.name}'
         else:
-            return f'{site.driver.name}/{self.tlib.pin_name(site.driver.kind, site.driver_pin, output=True)}{polarity}'
+            return f'{site.driver.name}/{tlib.pin_name(site.driver.kind, site.driver_pin, output=True)}'
+
+    @staticmethod
+    def fault_type_str(fault: int) -> str:
+        return 'sa1' if fault&1 else 'sa0'

tests/test_fault_set.py

@@ -4,15 +4,15 @@ from fsim.static import FaultSet
 
 def test_trivial_inv():
     c = bench.parse('input(i) output(o) o=INV(i)')
-    fs = FaultSet(c, KYUPY)
+    fs = FaultSet(c, KYUPY, c)
     assert len(fs.saf_set) == 4
     assert len(fs.saf_equiv_classes) == 2
     oline = c.forks['o'].ins[0]
     iline = oline.driver.ins[0]
-    assert fs.fault_str(oline.index*2) == 'o/o@0'
+    assert fs.fault_site_str(c, KYUPY, oline.index*2) == 'o/o'
-    assert fs.fault_str(oline.index*2+1) == 'o/o@1'
+    assert fs.fault_site_str(c, KYUPY, oline.index*2+1) == 'o/o'
-    assert fs.fault_str(iline.index*2) == 'o/i0@0'
+    assert fs.fault_site_str(c, KYUPY, iline.index*2) == 'o/i0'
-    assert fs.fault_str(iline.index*2+1) == 'o/i0@1'
+    assert fs.fault_site_str(c, KYUPY, iline.index*2+1) == 'o/i0'
     assert oline.index*2 in fs.saf_equiv_classes
     assert iline.index*2+1 in fs.saf_equiv_classes[oline.index*2]
     assert oline.index*2+1 in fs.saf_equiv_classes
@@ -22,7 +22,7 @@ def test_and():
     c = bench.parse('input(i0, i1) output(o) o=AND(i0, i1)')
     for n in c.nodes:
         print(n)
-    fs = FaultSet(c, KYUPY)
+    fs = FaultSet(c, KYUPY, c)
     assert len(fs.saf_set) == 6
     assert len(fs.saf_equiv_classes) == 4
     oline = c.forks['o'].ins[0]
@@ -39,7 +39,7 @@ def test_and():
 def _class_sizes(src):
     """Returns (#equiv class of output s-a-0, #equiv class of output s-a-1)."""
     c = bench.parse(src)
-    fs = FaultSet(c, KYUPY)
+    fs = FaultSet(c, KYUPY, c)
     oline = c.forks['o'].ins[0]
     return (len(fs.saf_equiv_classes.get(oline.index*2, set())),
             len(fs.saf_equiv_classes.get(oline.index*2+1, set())))
@@ -106,7 +106,7 @@ def test_s27():
         G12 = NOR(G1, G7)
         G13 = NOR(G2, G12)
     ''')
-    fs = FaultSet(c, KYUPY)
+    fs = FaultSet(c, KYUPY, c)
     assert len(fs.saf_set) == 52
     assert len(fs.saf_equiv_classes) == 32
     g11_line = c.cells['G11'].outs[0]