randogoth/strix
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

feat: add two-path bias amplifier with random walk and majority vote stages (FPGA v6)

Browse source
This commit is contained in:
randogoth 2026-10-25 14:59:27 +02:00
parent 5305d66102
commit 8541cce59c
11 changed files with 741 additions and 58 deletions
Download patch file Download diff file Expand all files Collapse all files

88
tools/jitter.py
View file

@ -8,6 +8,7 @@ Subcommands:
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
amp configure the bias amplifier and follow its trials (baseline without intention)
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
@ -46,6 +47,8 @@ REG_HEALTHY, REG_HEALTH_CTRL, REG_RCT_CUTOFF, REG_APT_HIGH, REG_APT_LOW, REG_MAX
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"]
REG_AMP_CTRL, REG_AMP_SOURCES, REG_AMP_SUB_MS, REG_AMP_SUBS, REG_AMP_STAGE = 0x50, 0x51, 0x52, 0x53, 0x54
PATH_NAMES = ["bias", "autocorr"]
CONTROL_START_FREQ = 0x04
STATUS_FREQ_VALID = 0x08
@ -79,8 +82,8 @@ class Board:
while self.serial.readline():
pass
self.serial.timeout = 200
if not identity.startswith("STRX v5"):
sys.exit(f"unexpected board identity: {identity!r} (needs the v5 FPGA image)")
if not identity.startswith("STRX v6"):
sys.exit(f"unexpected board identity: {identity!r} (needs the v6 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")
@ -137,6 +140,22 @@ def read_snapshot(board: "Board") -> dict | None:
"trial_ms": raw[6] << 8 | raw[7], "channels": channels}
def read_amp_snapshot(board: "Board") -> dict | None:
"""Latest amplifier trial snapshot (layout in fpga/rtl/bias_amplifier.v), or None if torn."""
raw = bytes.fromhex(board.command("amp"))
word = lambda i: int.from_bytes(raw[i:i + 4], "big")
if word(0) != word(len(raw) - 4):
return None
paths = []
for p in range(2):
base = 8 + 16 * p
paths.append({"bits": word(base), "ones": word(base + 4),
"votes": raw[base + 8] << 8 | raw[base + 9], "voted": raw[base + 10] << 8 | raw[base + 11],
"result": raw[base + 13] & 1, "valid": bool(raw[base + 13] & 2)})
return {"sequence": word(0), "sources": raw[4] & 0x0F, "von_neumann": bool(raw[4] & 0x10),
"overflow": bool(raw[4] & 0x20), "subtrials": raw[5], "sub_ms": raw[6], "paths": paths}
def cached_capture(board: Board | None, path: Path, *args) -> bytes | None:
if board:
path.write_bytes(board.capture(*args))
@ -400,6 +419,60 @@ def run_stats(board: Board | None, args) -> None:
f" {int(np.sum(np.abs(z) > 3)):>5d} {ac.mean():>15.3f} {ac.std():>5.2f}")
def run_amp(board: Board | None, args) -> None:
if board is None:
sys.exit("amp needs the board")
if len(args.stages) != 3:
sys.exit("--stages needs three values")
mode = 1 if args.method == "vote" else 0
for s, param in enumerate(args.stages):
board.write(REG_AMP_STAGE + 3 * s, mode)
board.write(REG_AMP_STAGE + 3 * s + 1, param >> 8)
board.write(REG_AMP_STAGE + 3 * s + 2, param & 0xFF)
board.write(REG_AMP_SOURCES, args.sources)
board.write(REG_AMP_SUB_MS, args.sub_ms)
board.write(REG_AMP_SUBS, args.subtrials)
board.write(REG_AMP_CTRL, (1 if args.von_neumann else 0) | 2) # bit 1 restarts
while (snapshot := read_amp_snapshot(board)) is None or snapshot["sequence"] > 1:
time.sleep(args.sub_ms * args.subtrials / 4000)
total = math.prod(args.stages)
print(f"{args.method}, stages {' x '.join(map(str, args.stages))} (total {total}), sources {args.sources:#x}, "
f"{args.subtrials} x {args.sub_ms} ms sub-trials, von Neumann {'on' if args.von_neumann else 'off'}")
print(f"{'trial':>5} {'src':>3} " + " ".join(f"{n + ' bits':>13} {'z':>6} {'votes':>{args.subtrials}} {'res':>3}"
for n in PATH_NAMES))
results = [[] for _ in PATH_NAMES]
bit_z = [[] for _ in PATH_NAMES]
last_sequence = -1
while len(results[0]) < args.trials:
snapshot = read_amp_snapshot(board)
if snapshot is None or snapshot["sequence"] == last_sequence:
time.sleep(args.sub_ms * args.subtrials / 4000)
continue
last_sequence = snapshot["sequence"]
if snapshot["overflow"]:
print("warning: merge overflow reported")
cells = []
for p, path in enumerate(snapshot["paths"]):
z, _ = z_scores(path["bits"], path["ones"], 0)
votes = "".join(("1" if path["votes"] >> i & 1 else "0") if path["voted"] >> i & 1 else "."
for i in range(snapshot["subtrials"]))
result = str(path["result"]) if path["valid"] else "-"
if path["valid"]:
results[p].append(path["result"])
bit_z[p].append(z)
cells.append(f"{path['bits']:>13d} {z:>6.2f} {votes:>{args.subtrials}} {result:>3}")
print(f"{snapshot['sequence']:>5} {snapshot['sources']:>3x} " + " ".join(cells))
print("\nbaseline: amplified-bit z per trial should have sd about 1; trial results about 50% ones\n")
print(f"{'path':<9} {'trials':>6} {'ones':>5} {'rate':>6} {'z':>6} {'bit z mean':>10} {'sd':>5}")
for p, name in enumerate(PATH_NAMES):
n, ones = len(results[p]), sum(results[p])
z = (2 * ones - n) / math.sqrt(n) if n else float("nan")
bz = np.array(bit_z[p])
print(f"{name:<9} {n:>6} {ones:>5} {ones / max(n, 1):>6.3f} {z:>6.2f} {np.nanmean(bz):>10.3f} {np.nanstd(bz):>5.2f}")
def run_pairs(board: Board | None, args) -> None:
generators = board.generators if board else 4
rng = np.random.default_rng(0)
@ -454,6 +527,15 @@ def main() -> None:
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")
amp = sub.add_parser("amp")
amp.add_argument("--trials", type=int, default=50)
amp.add_argument("--method", choices=["walk", "vote"], default="walk")
amp.add_argument("--stages", type=int, nargs="+", default=[8, 8, 4],
help="bound (walk) or window (vote) of the three stages")
amp.add_argument("--sources", type=lambda s: int(s, 0), default=0xF, help="generator mask")
amp.add_argument("--sub-ms", type=int, default=40)
amp.add_argument("--subtrials", type=int, default=5)
amp.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")
@ -464,7 +546,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, "stats": run_stats, "pairs": run_pairs}
"output": run_output, "health": run_health, "stats": run_stats, "amp": run_amp, "pairs": run_pairs}
handlers[args.command](board, args)