feat: add per-generator NIST health tests and dead-ring monitor, onod3000 output test (FPGA v4)

This commit is contained in:
randogoth 2026-10-25 13:36:55 +02:00
parent 585e755c66
commit 67bd7eec58
9 changed files with 436 additions and 28 deletions

View file

@ -5,7 +5,8 @@ Subcommands:
freq measure ring frequencies in every generator
rings estimate each ring's relative predictability at k=1 (sizing input)
sweep capture sources at increasing decimation k and fit the accumulated jitter
output check every generator's spatial XOR and processed output for bias and correlation
output run the onod3000 test suite on every generator's processed output (256 KB each)
health show the health monitors (NIST repetition count / adaptive proportion, dead rings)
pairs capture the same ring index in two generators simultaneously and test for correlation
Entropy per bit is estimated from conditional block entropies and converted to a
@ -21,6 +22,7 @@ deviation grows with sqrt(k). On synthetic captures it overestimates by a factor
import argparse
import math
import subprocess
import sys
import time
from itertools import combinations
@ -39,6 +41,8 @@ REG_SOURCE_A, REG_DECIM_HIGH, REG_DECIM_LOW = 0x08, 0x09, 0x0A
REG_CONTROL, REG_STATUS, REG_SOURCE_B, REG_MODE = 0x0B, 0x0C, 0x0D, 0x0E
REG_RING_SELECT, REG_FREQ_COUNT3 = 0x11, 0x12
REG_GEN_ENABLE, REG_TEMPORAL, REG_FREQ_GEN = 0x16, 0x17, 0x18
REG_HEALTHY, REG_HEALTH_CTRL, REG_RCT_CUTOFF, REG_APT_HIGH, REG_APT_LOW, REG_MAX_DEAD = 0x20, 0x21, 0x22, 0x23, 0x24, 0x25
REG_GEN_HEALTH = 0x30 # + 4g: flags, RCT events, APT events, dead rings
CONTROL_START_FREQ = 0x04
STATUS_FREQ_VALID = 0x08
@ -72,8 +76,8 @@ class Board:
while self.serial.readline():
pass
self.serial.timeout = 200
if not identity.startswith("STRX v3"):
sys.exit(f"unexpected board identity: {identity!r} (needs the v3 FPGA image)")
if not identity.startswith("STRX v4"):
sys.exit(f"unexpected board identity: {identity!r} (needs the v4 FPGA image)")
self.generators = self.read(REG_GENERATORS)
self.rings = self.read(REG_RINGS)
print(f"board: {identity}, {self.generators} generators x {self.rings} rings")
@ -272,18 +276,45 @@ def run_sweep(board: Board | None, args) -> None:
def run_output(board: Board | None, args) -> None:
# Longer contexts are undersampled at 256 Kbit and read falsely low on random data.
order = min(args.order, 8)
temporal = board.read(REG_TEMPORAL) if board else args.temporal
generators = board.generators if board else 4
print(f"temporal XOR = {temporal}; output captured with decimation {temporal}")
print("ones should be within about +-0.003 of 0.5 (3 sigma for 256 Kbit)\n")
print_header()
if not args.onod.exists():
sys.exit(f"{args.onod} not found; build it with `devbox run onod-build`")
print(f"temporal XOR = {temporal}; {args.bursts} bursts of 32 KB per generator\n")
for g in range(generators):
for source, k in [(SOURCE_SPATIAL + g, 1), (SOURCE_OUTPUT + g, temporal)]:
data = cached_capture(board, args.out / f"output_s{source:02x}_k{k}.bin", source, k)
if data is not None:
print_row(source_name(source, 0), k, analyze(to_bits(data), order))
source = SOURCE_OUTPUT + g
bursts = [cached_capture(board, args.out / f"output_g{g}_t{temporal}_b{b}.bin", source, temporal)
for b in range(args.bursts)]
if any(b is None for b in bursts):
continue
path = args.out / f"output_g{g}_t{temporal}.bin"
path.write_bytes(b"".join(bursts))
report = subprocess.run([str(args.onod), "-f", str(path)], capture_output=True, text=True).stdout
lines = report.splitlines()
if args.verbose:
print(f"generator {g}:\n{report}")
else:
summary = next((l for l in lines if "tests passed" in l), "no summary")
failed = [l.split()[0] for l in lines if l.rstrip().endswith("❌")]
print(f"generator {g}: {summary}" + (f" failed: {', '.join(failed)}" if failed else ""))
def run_health(board: Board | None, args) -> None:
if board is None:
sys.exit("health needs the board")
if args.clear:
board.write(REG_HEALTH_CTRL, 1)
apt = board.read(REG_APT_HIGH) << 8 | board.read(REG_APT_LOW)
print(f"cutoffs: repetition {board.read(REG_RCT_CUTOFF)}, adaptive proportion {apt}/1024, "
f"dead rings tolerated {board.read(REG_MAX_DEAD)}")
healthy = board.read(REG_HEALTHY)
print(f"{'gen':>3} {'healthy':>7} {'active':>6} {'swept':>5} {'RCT':>4} {'APT':>4} {'dead':>4}")
for g in range(board.generators):
flags = board.read(REG_GEN_HEALTH + 4 * g)
rct, apt_events, dead = (board.read(REG_GEN_HEALTH + 4 * g + i) for i in (1, 2, 3))
yes = lambda bit: "yes" if flags >> bit & 1 else "no"
print(f"{g:>3} {('yes' if healthy >> g & 1 else 'NO'):>7} {yes(3):>6} {yes(4):>5} "
f"{rct:>4} {apt_events:>4} {dead:>4}")
def run_pairs(board: Board | None, args) -> None:
@ -332,6 +363,11 @@ def main() -> None:
default=[1, 2, 4, 8, 16, 32, 64, 128, 256, 512, 1024, 2048, 4096])
output = sub.add_parser("output")
output.add_argument("--temporal", type=int, default=5, help="for --analyze-only")
output.add_argument("--bursts", type=int, default=8, help="32 KB captures per generator")
output.add_argument("--onod", type=Path, default=Path(".build/onod3000/release/onod3000"))
output.add_argument("--verbose", action="store_true", help="print the full onod3000 report")
health = sub.add_parser("health")
health.add_argument("--clear", action="store_true", help="clear failures and event counters first")
pairs = sub.add_parser("pairs")
pairs.add_argument("--rings", type=int, nargs="+", default=list(range(0, 90, 9)))
pairs.add_argument("-k", type=int, default=64, help="decimation")
@ -340,7 +376,7 @@ def main() -> None:
args.out.mkdir(parents=True, exist_ok=True)
board = None if args.analyze_only else Board(args.port)
handlers = {"freq": run_freq, "rings": run_rings, "sweep": run_sweep,
"output": run_output, "pairs": run_pairs}
"output": run_output, "health": run_health, "pairs": run_pairs}
handlers[args.command](board, args)