feat: add live statistics with per-trial snapshots and von Neumann option (FPGA v5)
This commit is contained in:
randogoth
parent
67bd7eec58
commit
5305d66102
8 changed files with 427 additions and 17 deletions
|
|
@ -7,6 +7,7 @@ Subcommands:
|
|||
sweep capture sources at increasing decimation k and fit the accumulated jitter
|
||||
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)
|
||||
stats follow the live statistics trial by trial, with rolling and cumulative windows
|
||||
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
|
||||
|
|
@ -43,6 +44,8 @@ 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
|
||||
REG_TRIAL_HIGH, REG_TRIAL_LOW, REG_STATS_CTRL = 0x26, 0x27, 0x28
|
||||
CHANNEL_NAMES = ["g0", "g1", "g2", "g3", "combined"]
|
||||
CONTROL_START_FREQ = 0x04
|
||||
STATUS_FREQ_VALID = 0x08
|
||||
|
||||
|
|
@ -76,8 +79,8 @@ class Board:
|
|||
while self.serial.readline():
|
||||
pass
|
||||
self.serial.timeout = 200
|
||||
if not identity.startswith("STRX v4"):
|
||||
sys.exit(f"unexpected board identity: {identity!r} (needs the v4 FPGA image)")
|
||||
if not identity.startswith("STRX v5"):
|
||||
sys.exit(f"unexpected board identity: {identity!r} (needs the v5 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")
|
||||
|
|
@ -123,6 +126,17 @@ class Board:
|
|||
return count * 256 / gate_seconds
|
||||
|
||||
|
||||
def read_snapshot(board: "Board") -> dict | None:
|
||||
"""Latest trial snapshot (layout in fpga/rtl/stats_block.v), or None if the read was torn."""
|
||||
raw = bytes.fromhex(board.command("stats"))
|
||||
word = lambda i: int.from_bytes(raw[i:i + 4], "big")
|
||||
if word(0) != word(len(raw) - 4):
|
||||
return None
|
||||
channels = [(word(8 + 12 * c), word(12 + 12 * c), word(16 + 12 * c)) for c in range((len(raw) - 12) // 12)]
|
||||
return {"sequence": word(0), "healthy": raw[4], "von_neumann": bool(raw[5] & 1),
|
||||
"trial_ms": raw[6] << 8 | raw[7], "channels": channels}
|
||||
|
||||
|
||||
def cached_capture(board: Board | None, path: Path, *args) -> bytes | None:
|
||||
if board:
|
||||
path.write_bytes(board.capture(*args))
|
||||
|
|
@ -317,6 +331,75 @@ def run_health(board: Board | None, args) -> None:
|
|||
f"{rct:>4} {apt_events:>4} {dead:>4}")
|
||||
|
||||
|
||||
def z_scores(bits: int, ones: int, agreements: int) -> tuple[float, float]:
|
||||
"""Bias z = (2*ones - n)/sqrt(n); lag-1 autocorrelation z = (2*agreements - n)/sqrt(n)."""
|
||||
if bits == 0:
|
||||
return float("nan"), float("nan")
|
||||
root = math.sqrt(bits)
|
||||
return (2 * ones - bits) / root, (2 * agreements - bits) / root
|
||||
|
||||
|
||||
def run_stats(board: Board | None, args) -> None:
|
||||
if board is None:
|
||||
sys.exit("stats needs the board")
|
||||
board.write(REG_TRIAL_HIGH, args.trial_ms >> 8)
|
||||
board.write(REG_TRIAL_LOW, args.trial_ms & 0xFF)
|
||||
board.write(REG_STATS_CTRL, (1 if args.von_neumann else 0) | 2) # bit 1 restarts trials
|
||||
# The snapshot still holds the last trial from before the restart until the first new
|
||||
# trial ends; wait for the sequence to start over.
|
||||
while (snapshot := read_snapshot(board)) is None or snapshot["sequence"] > 1:
|
||||
time.sleep(args.trial_ms / 4000)
|
||||
|
||||
count = len(CHANNEL_NAMES)
|
||||
totals = np.zeros((count, 3), dtype=np.int64)
|
||||
recent: list[np.ndarray] = []
|
||||
trial_z = [[] for _ in range(count)]
|
||||
trial_ac = [[] for _ in range(count)]
|
||||
last_sequence, missed, torn = -1, 0, 0
|
||||
|
||||
print(f"trials of {args.trial_ms} ms, von Neumann {'on' if args.von_neumann else 'off'}, "
|
||||
f"rolling window {args.rolling} trials")
|
||||
print(f"{'trial':>5} {'healthy':>7} " + " ".join(f"{n + ' z':>11}" for n in CHANNEL_NAMES)
|
||||
+ f" {'rolling':>8} {'walk':>9}")
|
||||
while len(trial_z[0]) < args.trials:
|
||||
snapshot = read_snapshot(board)
|
||||
if snapshot is None:
|
||||
torn += 1
|
||||
continue
|
||||
sequence = snapshot["sequence"]
|
||||
if sequence == last_sequence:
|
||||
time.sleep(args.trial_ms / 4000)
|
||||
continue
|
||||
if last_sequence >= 0 and sequence != last_sequence + 1:
|
||||
missed += sequence - last_sequence - 1
|
||||
last_sequence = sequence
|
||||
|
||||
counts = np.array(snapshot["channels"], dtype=np.int64)
|
||||
totals += counts
|
||||
recent = (recent + [counts])[-args.rolling:]
|
||||
rolling = np.sum(recent, axis=0)
|
||||
zs = []
|
||||
for c in range(count):
|
||||
z, ac = z_scores(*counts[c])
|
||||
trial_z[c].append(z)
|
||||
trial_ac[c].append(ac)
|
||||
zs.append(z)
|
||||
rolling_z, _ = z_scores(*rolling[count - 1])
|
||||
walk = 2 * totals[count - 1][1] - totals[count - 1][0]
|
||||
print(f"{sequence:>5} {snapshot['healthy']:>7x} " + " ".join(f"{z:>11.2f}" for z in zs)
|
||||
+ f" {rolling_z:>8.2f} {walk:>9d}")
|
||||
|
||||
print(f"\n{len(trial_z[0])} trials, {missed} missed, {torn} torn reads (retried)")
|
||||
print("baseline: trial z-scores should have mean about 0 and standard deviation about 1\n")
|
||||
print(f"{'channel':<9} {'bits':>12} {'cum. z':>7} {'trial z mean':>12} {'sd':>5} {'|z|>3':>5}"
|
||||
f" {'autocorr z mean':>15} {'sd':>5}")
|
||||
for c in range(count):
|
||||
z_total, _ = z_scores(*totals[c])
|
||||
z, ac = np.array(trial_z[c]), np.array(trial_ac[c])
|
||||
print(f"{CHANNEL_NAMES[c]:<9} {totals[c][0]:>12d} {z_total:>7.2f} {z.mean():>12.3f} {z.std():>5.2f}"
|
||||
f" {int(np.sum(np.abs(z) > 3)):>5d} {ac.mean():>15.3f} {ac.std():>5.2f}")
|
||||
|
||||
|
||||
def run_pairs(board: Board | None, args) -> None:
|
||||
generators = board.generators if board else 4
|
||||
rng = np.random.default_rng(0)
|
||||
|
|
@ -366,6 +449,11 @@ def main() -> None:
|
|||
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")
|
||||
stats = sub.add_parser("stats")
|
||||
stats.add_argument("--trials", type=int, default=100)
|
||||
stats.add_argument("--trial-ms", type=int, default=200)
|
||||
stats.add_argument("--rolling", type=int, default=10, help="trials in the rolling window")
|
||||
stats.add_argument("--von-neumann", action="store_true")
|
||||
health = sub.add_parser("health")
|
||||
health.add_argument("--clear", action="store_true", help="clear failures and event counters first")
|
||||
pairs = sub.add_parser("pairs")
|
||||
|
|
@ -376,7 +464,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, "health": run_health, "pairs": run_pairs}
|
||||
"output": run_output, "health": run_health, "stats": run_stats, "pairs": run_pairs}
|
||||
handlers[args.command](board, args)
|
||||
|
||||
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue