Reporting & reproducibility

profile_packed_linear

profile_packed_linear(
    *,
    m: int = 64,
    n: int = 4096,
    k: int = 4096,
    reps: int = 20,
    warmup: int = 5,
    compare_baselines: bool = True,
) -> ProfileBreakdown

Break down pack_weight / pack_act / gemm / scale vs torch FP32 / INT8-WO.

Source code in bnn/profile.py

def profile_packed_linear(
    *,
    m: int = 64,
    n: int = 4096,
    k: int = 4096,
    reps: int = 20,
    warmup: int = 5,
    compare_baselines: bool = True,
) -> ProfileBreakdown:
    """Break down pack_weight / pack_act / gemm / scale vs torch FP32 / INT8-WO."""
    torch.manual_seed(0)
    lin = nn.Linear(n, k, bias=True)
    x = torch.randn(m, n)
    # Weight pack (once)
    for _ in range(max(warmup, 1)):
        packed_mod = PackedBinaryXNORLinear(lin.weight.data, lin.bias.data)
    # timed pack
    t0 = time.perf_counter()
    for _ in range(reps):
        PackedBinaryXNORLinear(lin.weight.data, lin.bias.data)
    pack_w_ms = (time.perf_counter() - t0) / reps * 1e3

    packed_mod = PackedBinaryXNORLinear(lin.weight.data, lin.bias.data)
    x_np = x.detach().float().cpu().numpy()
    # Act pack
    for _ in range(warmup):
        _pack_activations_fast(x_np, n)
    t0 = time.perf_counter()
    for _ in range(reps):
        xp = _pack_activations_fast(x_np, n)
    pack_a_ms = (time.perf_counter() - t0) / reps * 1e3

    xp = _pack_activations_fast(x_np, n)
    gemm_fn = (
        binary_gemm_native_prepacked
        if packed_mod.uses_native
        else binary_gemm_numpy_prepacked
    )
    for _ in range(warmup):
        y = gemm_fn(xp, packed_mod._wp_np, n)
        assert y is not None
    t0 = time.perf_counter()
    for _ in range(reps):
        y = gemm_fn(xp, packed_mod._wp_np, n)
        assert y is not None
    gemm_ms = (time.perf_counter() - t0) / reps * 1e3

    y = gemm_fn(xp, packed_mod._wp_np, n)
    assert y is not None
    for _ in range(warmup):
        yy = y * packed_mod._alpha_np
        if packed_mod._bias_np is not None:
            yy = yy + packed_mod._bias_np
    t0 = time.perf_counter()
    for _ in range(reps):
        yy = y * packed_mod._alpha_np
        if packed_mod._bias_np is not None:
            yy = yy + packed_mod._bias_np
    scale_ms = (time.perf_counter() - t0) / reps * 1e3

    # e2e forward
    for _ in range(warmup):
        packed_mod(x)
    t0 = time.perf_counter()
    for _ in range(reps):
        packed_mod(x)
    e2e_ms = (time.perf_counter() - t0) / reps * 1e3

    # FP32 baseline
    for _ in range(warmup):
        lin(x)
    t0 = time.perf_counter()
    for _ in range(reps):
        lin(x)
    fp_ms = (time.perf_counter() - t0) / reps * 1e3

    int8_ms = 0.0
    if compare_baselines:
        int8_ms = _time_int8_weight_only(x, lin, reps=reps, warmup=warmup)

    overhead = (e2e_ms - gemm_ms) / max(gemm_ms, 1e-9)
    baselines = {
        "torch_fp32_ms": float(fp_ms),
        "torch_int8_weight_only_ms": float(int8_ms),
        "packed_e2e_ms": float(e2e_ms),
        "packed_gemm_ms": float(gemm_ms),
    }
    return ProfileBreakdown(
        m=m,
        n=n,
        k=k,
        reps=reps,
        pack_weight_ms=pack_w_ms,
        pack_act_ms=pack_a_ms,
        gemm_ms=gemm_ms,
        scale_bias_ms=scale_ms,
        e2e_forward_ms=e2e_ms,
        torch_fp32_ms=fp_ms,
        native=bool(native_kernel_available() and packed_mod.uses_native),
        overhead_vs_gemm=float(overhead),
        speedup_vs_fp32=float(fp_ms / max(e2e_ms, 1e-9)),
        torch_int8_wo_ms=float(int8_ms),
        speedup_vs_int8_wo=float(int8_ms / max(e2e_ms, 1e-9)) if int8_ms > 0 else 0.0,
        baselines=baselines,
    )

ProfileBreakdown dataclass

Source code in bnn/profile.py

@dataclass
class ProfileBreakdown:
    m: int
    n: int
    k: int
    reps: int
    pack_weight_ms: float
    pack_act_ms: float
    gemm_ms: float
    scale_bias_ms: float
    e2e_forward_ms: float
    torch_fp32_ms: float
    native: bool
    overhead_vs_gemm: float  # (e2e - gemm) / gemm
    speedup_vs_fp32: float
    torch_int8_wo_ms: float = 0.0
    speedup_vs_int8_wo: float = 0.0
    baselines: dict[str, float] = field(default_factory=dict)

    def to_dict(self) -> dict:
        return asdict(self)

check_soft_budgets

check_soft_budgets(
    breakdown: ProfileBreakdown | dict[str, Any],
) -> list[str]

Return soft-budget violations (empty ⇒ within CI ceilings).

Callers decide severity: bnn eval-suite warns unless --strict-budgets; focused pytest may assert empty violations on the tiny smoke shape so CI still catches catastrophic regressions. Never mutates golden floors.

Source code in bnn/profile.py

def check_soft_budgets(breakdown: ProfileBreakdown | dict[str, Any]) -> list[str]:
    """Return soft-budget violations (empty ⇒ within CI ceilings).

    Callers decide severity: ``bnn eval-suite`` warns unless ``--strict-budgets``;
    focused pytest may assert empty violations on the tiny smoke shape so CI
    still catches catastrophic regressions. Never mutates golden floors.
    """
    if isinstance(breakdown, ProfileBreakdown):
        d = breakdown.to_dict()
    else:
        d = dict(breakdown)
    key = (int(d["m"]), int(d["n"]), int(d["k"]))
    ceilings = SOFT_BUDGETS_MS.get(key)
    if ceilings is None:
        return []
    violations: list[str] = []
    for metric, ceiling in ceilings.items():
        val = float(d.get(metric, 0.0) or 0.0)
        if val > ceiling:
            violations.append(f"{metric}={val:.3f}ms exceeds soft budget {ceiling}ms @ {key}")
    return violations

check_committed_bench_soft_floors

check_committed_bench_soft_floors(
    bench: dict[str, Any],
    *,
    floor_fraction: float = SOFT_SPEEDUP_FLOOR_FRACTION,
) -> list[str]

Soft-check committed results/benchmark.json for corruption + thread curves.

floor_fraction is an absolute minimum on speedup_compute_vs_numpy_fp32 (default SOFT_SPEEDUP_FLOOR_FRACTION), not a fraction of a historical headline. It only rejects nonsense rows (e.g. speedup 0). Also requires thread_scaling lists with ≥2 points.

Source code in bnn/profile.py

def check_committed_bench_soft_floors(
    bench: dict[str, Any],
    *,
    floor_fraction: float = SOFT_SPEEDUP_FLOOR_FRACTION,
) -> list[str]:
    """Soft-check committed ``results/benchmark.json`` for corruption + thread curves.

    ``floor_fraction`` is an **absolute** minimum on
    ``speedup_compute_vs_numpy_fp32`` (default ``SOFT_SPEEDUP_FLOOR_FRACTION``),
    not a fraction of a historical headline. It only rejects nonsense rows
    (e.g. speedup 0). Also requires ``thread_scaling`` lists with ≥2 points.
    """
    rows = bench.get("results") or bench.get("rows") or bench.get("benchmarks") or []
    violations: list[str] = []
    for r in rows:
        if not isinstance(r, dict):
            continue
        s = r.get("speedup_compute_vs_numpy_fp32")
        if not isinstance(s, (int, float)):
            continue
        # Absolute corruption floor — not relative to a prior machine run.
        if float(s) < floor_fraction:
            sh = r.get("shape") or {}
            violations.append(
                f"shape {sh}: speedup_compute_vs_numpy_fp32={s} "
                f"below soft absolute floor {floor_fraction}"
            )
        scaling = r.get("thread_scaling")
        if scaling is not None and not isinstance(scaling, list):
            violations.append(f"shape {r.get('shape')}: thread_scaling must be a list")
        elif isinstance(scaling, list) and len(scaling) < 2:
            violations.append(
                f"shape {r.get('shape')}: thread_scaling needs ≥2 points (W13.T04)"
            )
    return violations

SOFT_BUDGETS_MS module-attribute

SOFT_BUDGETS_MS: dict[
    tuple[int, int, int], dict[str, float]
] = {
    (8, 256, 256): {
        "gemm_ms": 25.0,
        "e2e_forward_ms": 40.0,
        "torch_fp32_ms": 40.0,
    },
    (64, 512, 512): {
        "gemm_ms": 80.0,
        "e2e_forward_ms": 120.0,
        "torch_fp32_ms": 120.0,
    },
}

memory_report

memory_report(model: Module) -> MemoryReport

Per-layer resident vs theoretical footprint for model.

Only Linear/Conv-shaped modules (packed or not) are tracked as layers; everything else — embeddings, norms, biases on other modules — is summed into other_* so the totals still reconcile with the real model size.

Source code in bnn/memory.py

def memory_report(model: nn.Module) -> MemoryReport:
    """Per-layer resident vs theoretical footprint for ``model``.

    Only Linear/Conv-shaped modules (packed or not) are tracked as layers;
    everything else — embeddings, norms, biases on other modules — is summed
    into ``other_*`` so the totals still reconcile with the real model size.
    """
    report = MemoryReport()
    tracked: set[int] = set()

    for name, mod in model.named_modules():
        if not isinstance(mod, (*_PACKED_TYPES, nn.Linear, nn.Conv2d)):
            continue
        packed = isinstance(mod, _PACKED_TYPES)
        resident = _module_bytes(mod)
        fp32 = _fp32_equivalent(mod)
        # isinstance against the literal tuple, so the packed types narrow:
        # nn.Module.__getattr__ is typed Tensor | Module, and neither a
        # hasattr guard nor a bool flag lets a type checker resolve the call.
        if isinstance(
            mod,
            (
                PackedBinaryXNORLinear,
                TernaryWeightOnlyLinear,
                BinaryWeightOnlyDequantLinear,
                PackedBinaryConv2d,
            ),
        ):
            theoretical = int(mod.packed_weight_bytes())
        else:
            theoretical = resident
        report.layers.append(
            LayerFootprint(
                name=name or "<root>",
                kind=type(mod).__name__,
                packed=packed,
                resident_bytes=resident,
                theoretical_bytes=theoretical,
                fp32_equivalent_bytes=fp32,
            )
        )
        tracked.update(id(p) for p in mod.parameters(recurse=False))
        tracked.update(id(b) for b in mod.buffers(recurse=False))

    report.other_param_bytes = int(
        sum(p.numel() * p.element_size() for p in model.parameters() if id(p) not in tracked)
    )
    report.other_buffer_bytes = int(
        sum(b.numel() * b.element_size() for b in model.buffers() if id(b) not in tracked)
    )
    return report

MemoryReport dataclass

Whole-model footprint, split into packed and unpacked contributions.

Source code in bnn/memory.py

@dataclass
class MemoryReport:
    """Whole-model footprint, split into packed and unpacked contributions."""

    layers: list[LayerFootprint] = field(default_factory=list)
    other_param_bytes: int = 0
    other_buffer_bytes: int = 0

    @property
    def packed_layers(self) -> list[LayerFootprint]:
        return [x for x in self.layers if x.packed]

    def totals(self) -> dict[str, int]:
        resident = sum(x.resident_bytes for x in self.layers)
        theoretical = sum(x.theoretical_bytes for x in self.layers)
        fp32 = sum(x.fp32_equivalent_bytes for x in self.layers)
        return {
            "tracked_resident_bytes": resident,
            "tracked_theoretical_bytes": theoretical,
            "tracked_fp32_equivalent_bytes": fp32,
            "other_param_bytes": self.other_param_bytes,
            "other_buffer_bytes": self.other_buffer_bytes,
            "model_resident_bytes": resident + self.other_param_bytes + self.other_buffer_bytes,
        }

    def to_dict(self) -> dict[str, Any]:
        t = self.totals()
        fp32 = t["tracked_fp32_equivalent_bytes"]
        # Whole-model ratio including the FP parts that were deliberately not
        # wrapped (attention, norms, embeddings) — the honest end-to-end number.
        whole_fp32 = fp32 + t["other_param_bytes"] + t["other_buffer_bytes"]
        return {
            "schema": "bnn_memory_report_v1",
            **t,
            "tracked_resident_compression": fp32 / max(t["tracked_resident_bytes"], 1),
            "tracked_theoretical_compression": fp32 / max(t["tracked_theoretical_bytes"], 1),
            "whole_model_resident_compression": (
                whole_fp32 / max(t["model_resident_bytes"], 1)
            ),
            "packed_layer_count": len(self.packed_layers),
            "layer_count": len(self.layers),
            "layers": [x.to_dict() for x in self.layers],
            "thesis_note": (
                "resident_* is measured from real buffers; theoretical_* is the "
                "encoding's pack ratio. Neither is a latency claim — use "
                "`bnn profile` / `bnn bench` for wall-clock."
            ),
        }

LayerFootprint dataclass

Bytes attributable to one module.

Source code in bnn/memory.py

@dataclass
class LayerFootprint:
    """Bytes attributable to one module."""

    name: str
    kind: str
    packed: bool
    resident_bytes: int
    theoretical_bytes: int
    fp32_equivalent_bytes: int

    @property
    def resident_compression(self) -> float:
        """What you actually save in RAM today."""
        return self.fp32_equivalent_bytes / max(self.resident_bytes, 1)

    @property
    def theoretical_compression(self) -> float:
        """What the encoding allows — not necessarily what is stored."""
        return self.fp32_equivalent_bytes / max(self.theoretical_bytes, 1)

    def to_dict(self) -> dict[str, Any]:
        d = asdict(self)
        d["resident_compression"] = self.resident_compression
        d["theoretical_compression"] = self.theoretical_compression
        return d

resident_compression property

resident_compression: float

What you actually save in RAM today.

theoretical_compression property

theoretical_compression: float

What the encoding allows — not necessarily what is stored.

forward_transient_bytes

forward_transient_bytes(
    batch: int, in_features: int, out_features: int
) -> dict[str, float]

Transient bytes a packed Linear forward allocates, by stage.

Useful for sizing edge deployments: the weight saving is permanent, but a forward still needs packed activations and an FP32 output.

Source code in bnn/memory.py

def forward_transient_bytes(
    batch: int, in_features: int, out_features: int
) -> dict[str, float]:
    """Transient bytes a packed Linear forward allocates, by stage.

    Useful for sizing edge deployments: the weight saving is permanent, but a
    forward still needs packed activations and an FP32 output.
    """
    words = (in_features + 63) // 64
    packed_act = batch * words * 8
    output = batch * out_features * 4
    fp32_act = batch * in_features * 4
    return {
        "packed_activation_bytes": int(packed_act),
        "output_bytes": int(output),
        "fp32_activation_bytes": int(fp32_act),
        "total_transient_bytes": int(packed_act + output),
        "activation_pack_compression": float(fp32_act) / max(packed_act, 1),
    }

render_summary

render_summary(results_dir: Path | None = None) -> str

Source code in bnn/eval_report.py

def render_summary(results_dir: Path | None = None) -> str:
    # Resolve locally rather than rebinding the module-level RESULTS: mutating
    # the global leaked the caller's directory into every later call in the
    # process, so a no-arg render_summary() after a scoped one silently read
    # the wrong (often deleted) directory.
    base = Path(results_dir) if results_dir is not None else RESULTS

    bench_raw = _load("benchmark.json", base)
    train_raw = _load("train_results.json", base)
    cifar = _load("cifar10_proxy.json", base) or {}
    wrap = _load("wrap_demo.json", base) or {}
    energy = _load("energy_bound.json", base) or {}
    fgsm = _load("robustness_fgsm.json", base) or {}
    card = machine_card()

    lines = [
        "# Results summary (this workspace)",
        "",
        f"_Regenerated: {card['generated_utc']}_",
        f"_Machine: {card['platform']} | torch {card['torch']} | CUDA={card['cuda']}_",
        "",
        "## Kernel (CPU packed XNOR)",
        "",
    ]

    bench = bench_raw if isinstance(bench_raw, dict) else {}
    rows = bench.get("rows") or bench.get("results") or []
    if isinstance(bench.get("benchmarks"), list):
        rows = bench["benchmarks"]
    if rows:
        lines += [
            "| Shape | S vs NumPy FP32 | S vs Torch FP32 | Err |",
            "|-------|----------------:|----------------:|----:|",
        ]
        for r in rows:
            sh = r.get("shape")
            if isinstance(sh, dict):
                shape = f"{sh.get('batch')}×{sh.get('in_features')}×{sh.get('out_features')}"
            else:
                shape = sh or f"{r.get('batch')}×{r.get('n')}×{r.get('m')}"
            s_np = (
                r.get("speedup_compute_vs_numpy_fp32")
                or r.get("speedup_vs_numpy_fp32")
                or r.get("speedup_numpy")
            )
            s_t = (
                r.get("speedup_compute_vs_torch_fp32")
                or r.get("speedup_vs_torch_fp32")
                or r.get("speedup_torch")
            )
            err = r.get("max_abs_error_vs_fp32") or r.get("max_err") or r.get("err") or 0
            s_np_s = f"{s_np:.2f}" if isinstance(s_np, (int, float)) else "—"
            s_t_s = f"{s_t:.2f}" if isinstance(s_t, (int, float)) else "—"
            lines.append(f"| {shape} | {s_np_s} | {s_t_s} | {err} |")
        comp = None
        if rows and isinstance(rows[0].get("theoretical"), dict):
            comp = rows[0]["theoretical"].get("weight_compression")
        comp = bench.get("compression") or bench.get("weight_compression") or comp
        if comp:
            lines.append("")
            lines.append(f"Compression: **{comp}×**. Source: `benchmark.json`.")
    else:
        lines.append("_No benchmark.json rows found — run `bnn bench`._")

    lines += ["", "## MNIST", ""]
    if isinstance(train_raw, list):
        models = train_raw
    elif isinstance(train_raw, dict):
        models = train_raw.get("results") or train_raw.get("models") or []
    else:
        models = []
    if models:
        lines += ["| Model | Acc % |", "|-------|------:|"]
        for m in models:
            name = m.get("model") or m.get("name")
            acc = m.get("test_acc") or m.get("acc")
            lines.append(f"| {name} | {acc} |")
        lines.append("")
        lines.append("Source: `train_results.json`.")
    else:
        lines.append("_No train_results.json — run `bnn train`._")

    image = _load("image_cifar.json", base) or cifar
    audio = _load("audio_synth.json", base) or {}

    lines += ["", "## Image (CIFAR-10 Bi-Real)", ""]
    if image:
        cres = image.get("results") or []
        fp = next((r for r in cres if "fp32" in str(r.get("model", "")).lower()), None)
        bn = next(
            (
                r
                for r in cres
                if "binary" in str(r.get("model", "")).lower()
                and "vit" not in str(r.get("model", "")).lower()
            ),
            None,
        )
        if fp and bn:
            lines.append(f"- FP32 CNN: **{fp['test_acc']:.2f}%**")
            lines.append(f"- Binary Bi-Real: **{bn['test_acc']:.2f}%**")
            gap = image.get("acc_gap_pp_fp_vs_binary_cnn", image.get("acc_gap_pp"))
            if gap is None:
                gap = fp["test_acc"] - bn["test_acc"]
            lines.append(f"- Gap: **{gap:.2f} pp**")
        src = "image_cifar.json" if (_load("image_cifar.json", base)) else "cifar10_proxy.json"
        lines.append(f"Source: `{src}`. Tutorial: `docs/tutorials/04_image_cifar.md`.")
    else:
        lines.append("_No image results — run `bnn train-image`._")

    lines += ["", "## Audio (synthetic tones)", ""]
    if audio:
        ares = audio.get("results") or []
        fp = next((r for r in ares if "fp32" in str(r.get("model", "")).lower()), None)
        bn = next((r for r in ares if "binary" in str(r.get("model", "")).lower()), None)
        if fp and bn:
            lines.append(f"- FP32 CNN: **{fp['test_acc']:.2f}%**")
            lines.append(f"- Binary CNN: **{bn['test_acc']:.2f}%**")
            lines.append(f"- Gap: **{audio.get('acc_gap_pp', fp['test_acc']-bn['test_acc']):.2f} pp**")
        lines.append(
            "Source: `audio_synth.json`. "
            "**Not production ASR** — INT8 Whisper/ORT for real speech. "
            "Tutorial: `docs/tutorials/05_audio.md`."
        )
    else:
        lines.append("_No audio_synth.json — run `bnn train-audio`._")

    lines += ["", "## Wrap / energy / robustness", ""]
    if wrap:
        e2e_fp = wrap.get("e2e_latency_ms_fp")
        e2e_w = wrap.get("e2e_latency_ms_wrapped")
        e2e_s = wrap.get("e2e_speedup")
        comp = wrap.get("weight_compression_replaced_layers") or wrap.get("compression")
        cos = wrap.get("output_cosine_vs_fp")
        gemm = (wrap.get("layer_microbench") or {}).get("speedup_gemm_only_vs_torch_linear")
        lines.append(
            f"- Wrap e2e latency: FP **{e2e_fp:.2f}** ms → wrapped **{e2e_w:.2f}** ms "
            f"(e2e **{e2e_s:.2f}×**)"
            if isinstance(e2e_fp, (int, float)) and isinstance(e2e_w, (int, float))
            else f"- Wrap e2e: {e2e_fp} → {e2e_w} ms"
        )
        if comp is not None:
            lines.append(f"- Weight compression (replaced layers): **{comp}×** (exact bit-pack)")
        if gemm is not None:
            lines.append(f"- Layer gemm_only vs torch Linear: **{gemm:.2f}×** (kernel ROI)")
        if cos is not None:
            lines.append(
                f"- Output cosine vs FP: **{cos:.3f}** "
                "(low without QAT is expected — not a transparent wrap)"
            )
    if energy:
        er = energy.get("energy_reduction_latency_only_same_power")
        if isinstance(er, (int, float)):
            lines.append(
                f"- Energy (latency-only, same power proxy): **{er:.2f}×** (`energy_bound.json`)"
            )
        else:
            lines.append(f"- Energy latency-only reduction: **{er}×** (`energy_bound.json`)")
    if fgsm:
        for r in fgsm.get("results", []):
            lines.append(
                f"- FGSM {r.get('model')}: clean {r.get('clean_acc')}% → "
                f"{r.get('fgsm_acc')}% (drop {r.get('drop_pp')} pp)"
            )

    lines += [
        "",
        "## Honesty / dual reporting",
        "",
        "| Quantity | Meaning | Do not claim as |",
        "|----------|---------|-----------------|",
        "| Weight compression **32×** | Bit-pack bytes | e2e latency |",
        "| Theoretical word reduction ~64× | XNOR-popcount ops | wall-clock |",
        "| Kernel speedup (bench) | Prepacked GEMM vs NumPy/Torch FP | full-model FPS |",
        "| E2E wrap speedup | Whole forward | quality-preserving wrap |",
        "",
        r"Amdahl: \(S_{e2e}=\frac{1}{(1-f)+f/S_{kernel}}\). "
        "Fake `sign()`+torch Linear is often **slower** than FP32 on GPU.",
        "",
        "Repro gates: `tests/golden_floors.json` · `bnn repro` · "
        "[`REPRODUCIBILITY.md`](../REPRODUCIBILITY.md).",
        "",
        "More: `docs/19_GAP_CLOSURE_REPORT.md`, `docs/28_IMAGE_AUDIO_COMPLETION.md`, "
        "`docs/29_FINAL_COMPLETION.md`, `docs/31_QUALITY_UPGRADE.md`.",
        "",
    ]
    return "\n".join(lines)

write_summary

write_summary(
    out: Path | None = None, results_dir: Path | None = None
) -> Path

Source code in bnn/eval_report.py

def write_summary(out: Path | None = None, results_dir: Path | None = None) -> Path:
    text = render_summary(results_dir)
    out = Path(out) if out else (results_dir or ROOT / "results") / "SUMMARY.md"
    out.parent.mkdir(parents=True, exist_ok=True)
    out.write_text(text, encoding="utf-8")
    return out

machine_card

machine_card() -> dict[str, Any]

Source code in bnn/eval_report.py

def machine_card() -> dict[str, Any]:
    import torch

    return {
        "platform": platform.platform(),
        "processor": platform.processor(),
        "python": platform.python_version(),
        "torch": torch.__version__,
        "cuda": torch.cuda.is_available(),
        "generated_utc": datetime.now(UTC).isoformat(),
    }

pareto

Pareto report: accuracy / compression / latency / energy-proxy (W7.T03).

Dual-metric rule

compression is theoretical pack ratio. latency_ms / energy_proxy are wall-clock / estimate — never conflate with 32× theory.

ParetoPoint dataclass

One optimiser / baseline configuration on the fair protocol.

Source code in bnn/eval/pareto.py

@dataclass
class ParetoPoint:
    """One optimiser / baseline configuration on the fair protocol."""

    name: str
    accuracy: float | None
    compression: float
    latency_ms: float | None
    energy_proxy: float | None = None
    notes: str = ""
    # Honesty tags
    accuracy_metric: str = "cosine_or_top1"  # document which
    compression_is_theory: bool = True
    latency_is_wall_clock: bool = True
    extra: dict[str, Any] = field(default_factory=dict)

    def to_dict(self) -> dict[str, Any]:
        d = asdict(self)
        extra = d.pop("extra", {}) or {}
        d.update(extra)
        return d

build_pareto_report

build_pareto_report(
    points: list[ParetoPoint] | list[dict[str, Any]],
    *,
    protocol: str = "docs/FAIR_EVAL_PROTOCOL.md",
    bench_shapes_ref: str = "docs/BENCH_SHAPES.md",
    warmup: int | None = None,
    threads: int | None = None,
    meta: dict[str, Any] | None = None,
) -> dict[str, Any]

Build a versioned Pareto JSON payload.

Source code in bnn/eval/pareto.py

def build_pareto_report(
    points: list[ParetoPoint] | list[dict[str, Any]],
    *,
    protocol: str = "docs/FAIR_EVAL_PROTOCOL.md",
    bench_shapes_ref: str = "docs/BENCH_SHAPES.md",
    warmup: int | None = None,
    threads: int | None = None,
    meta: dict[str, Any] | None = None,
) -> dict[str, Any]:
    """Build a versioned Pareto JSON payload."""
    serialised: list[dict[str, Any]] = []
    for p in points:
        if isinstance(p, ParetoPoint):
            serialised.append(p.to_dict())
        else:
            serialised.append(dict(p))

    out: dict[str, Any] = {
        "schema": PARETO_SCHEMA_ID,
        "schema_version": PARETO_SCHEMA_VERSION,
        "protocol": protocol,
        "bench_shapes_ref": bench_shapes_ref,
        "thesis_note": (
            "Dual-metric: compression is theory (pack ratio); latency_ms and "
            "energy_proxy are wall-clock / estimate. Never claim GPU 32× from sign()/STE."
        ),
        "warmup": warmup,
        "threads": threads,
        "machine": _machine_meta(),
        "points": serialised,
    }
    if meta:
        out["meta"] = meta
    return out

validate_pareto_report

validate_pareto_report(
    payload: dict[str, Any],
) -> list[str]

Return validation errors (empty ⇒ OK).

Source code in bnn/eval/pareto.py

def validate_pareto_report(payload: dict[str, Any]) -> list[str]:
    """Return validation errors (empty ⇒ OK)."""
    errors: list[str] = []
    if not isinstance(payload, dict):
        return ["report must be a dict"]
    if payload.get("schema") != PARETO_SCHEMA_ID:
        errors.append(f"schema must be {PARETO_SCHEMA_ID!r}")
    if payload.get("schema_version") != PARETO_SCHEMA_VERSION:
        errors.append(f"schema_version must be {PARETO_SCHEMA_VERSION}")
    points = payload.get("points")
    if not isinstance(points, list) or not points:
        errors.append("points must be a non-empty list")
        return errors
    for i, pt in enumerate(points):
        if not isinstance(pt, dict):
            errors.append(f"points[{i}] must be a dict")
            continue
        for key in REQUIRED_POINT_KEYS:
            if key not in pt:
                errors.append(f"points[{i}] missing {key}")
        # Soft honesty: very high compression + latency labeled as speedup elsewhere
        note = str(payload.get("thesis_note") or "")
        if "dual-metric" not in note.lower() and "theory" not in note.lower():
            errors.append("thesis_note must mention theory vs wall-clock / dual-metric")
            break
    return errors

demo_points

demo_points() -> list[ParetoPoint]

Tiny synthetic points for CI / schema smoke (not golden floors).

Source code in bnn/eval/pareto.py

def demo_points() -> list[ParetoPoint]:
    """Tiny synthetic points for CI / schema smoke (not golden floors)."""
    return [
        ParetoPoint(
            name="fp32_baseline",
            accuracy=1.0,
            compression=1.0,
            latency_ms=10.0,
            energy_proxy=1.0,
            notes="Reference FP32; compression=1 means no pack",
            accuracy_metric="relative_ref",
        ),
        ParetoPoint(
            name="binary_xnor_packed",
            accuracy=0.92,
            compression=32.0,
            latency_ms=4.0,
            energy_proxy=0.4,
            notes="Illustrative dual-metric point — not a published golden",
            accuracy_metric="cosine",
        ),
        ParetoPoint(
            name="ternary_weight_only",
            accuracy=0.97,
            compression=16.0,
            latency_ms=7.0,
            energy_proxy=0.7,
            notes="Size win; GEMM still FP — honesty",
            accuracy_metric="cosine",
        ),
    ]

set_repro_seed

set_repro_seed(
    seed: int = 0,
    *,
    deterministic: bool = True,
    force_cpu: bool = True,
) -> dict[str, Any]

Seed Python / NumPy / Torch; optionally enable deterministic algorithms.

Returns a small status dict (useful for logging in result JSON).

Source code in bnn/determinism.py

def set_repro_seed(
    seed: int = 0,
    *,
    deterministic: bool = True,
    force_cpu: bool = True,
) -> dict[str, Any]:
    """Seed Python / NumPy / Torch; optionally enable deterministic algorithms.

    Returns a small status dict (useful for logging in result JSON).
    """
    import numpy as np
    import torch

    random.seed(seed)
    np.random.seed(seed)  # noqa: NPY002 — seeding the legacy global RNG is the point
    torch.manual_seed(seed)
    if torch.cuda.is_available():
        torch.cuda.manual_seed_all(seed)

    status: dict[str, Any] = {
        "seed": seed,
        "force_cpu": force_cpu,
        "deterministic_requested": deterministic,
        "cuda_available": torch.cuda.is_available(),
        "notes": [],
    }

    if force_cpu:
        os.environ.setdefault("CUDA_VISIBLE_DEVICES", "")
        status["device_policy"] = "cpu"
    else:
        status["device_policy"] = "auto"

    if deterministic:
        # Prefer reproducibility over throughput for golden / smoke trains.
        torch.backends.cudnn.deterministic = True
        torch.backends.cudnn.benchmark = False
        try:
            torch.use_deterministic_algorithms(True, warn_only=True)
            status["torch_deterministic"] = True
        except (TypeError, RuntimeError) as exc:
            # Older torch: warn_only may be unavailable.
            try:
                torch.use_deterministic_algorithms(True)
                status["torch_deterministic"] = True
            except RuntimeError:
                status["torch_deterministic"] = False
                status["notes"].append(f"deterministic_algorithms unavailable: {exc}")
        # Avoid nondeterministic CPU reduction paths when possible.
        os.environ.setdefault("CUBLAS_WORKSPACE_CONFIG", ":4096:8")
    else:
        status["torch_deterministic"] = False

    status["notes"].append(
        "Some ops (certain CUDA kernels, rare Conv paths) remain nondeterministic; "
        "golden repro uses CPU + tolerance gates, not bit-identical floats."
    )
    return status

pack_linear_weight

pack_linear_weight(weight: Tensor) -> dict[str, Any]

Pack ±1 signs of a Linear weight into uint64 + scale alpha.

Source code in bnn/export.py

def pack_linear_weight(weight: torch.Tensor) -> dict[str, Any]:
    """Pack ±1 signs of a Linear weight into uint64 + scale alpha."""
    w = weight.detach().float().cpu()
    alpha = float(w.abs().mean().clamp(min=1e-4).item())
    pm1 = binary_sign(w).numpy().astype(np.float32)
    packed, n = pack_binary_pm1(pm1, axis=1)
    return {
        "packed": packed,
        "n": int(n),
        "out_features": int(w.shape[0]),
        "in_features": int(w.shape[1]),
        "alpha": alpha,
        "fp32_bytes": int(w.numel() * 4),
        "packed_bytes": int(packed.nbytes),
        "compression": (w.numel() * 4) / max(packed.nbytes, 1),
    }