files / problems / hilbert16-degree-8 / compute / q12 / search.py
#!/usr/bin/env python3
"""Bounded sign balls on the leftover q11-followup one-flip plan.
Does not stop on a candidate: every assigned ball is finished. A discovery
is only a candidate; this driver never claims algebraic realizability.
"""
import argparse
from collections import Counter
import fcntl
import json
from math import comb
from pathlib import Path
import subprocess
import time
from common import (HERE, LEFTOVER_PLAN_SHA, LEFTOVER_SEEDS_SHA, N_LEFTOVER,
atomic_json, baseline, canon, cert_plan, cert_seeds,
code_scheme, digest, leftover_plan, leftover_seeds,
source_hashes, task_tris, unpack, write_task)
def append_jsonl(path, rec):
path = Path(path)
path.parent.mkdir(parents=True, exist_ok=True)
line = json.dumps(rec, sort_keys=True) + '\n'
with open(path, 'a') as f:
fcntl.flock(f, fcntl.LOCK_EX)
f.write(line)
f.flush()
def parse_ids(args, nplan):
if args.ids:
text = Path(args.ids).read_text() if Path(args.ids).exists() and not args.ids[0].isdigit() else args.ids
ids = []
for part in text.replace(',', ' ').split():
if '-' in part:
a, b = map(int, part.split('-', 1))
ids.extend(range(a, b + 1))
else:
ids.append(int(part))
return ids
start = args.start
stop = args.stop
assert 0 <= start < stop <= nplan
return list(range(start, stop))
def main():
ap = argparse.ArgumentParser(description=__doc__)
ap.add_argument('--output', type=str, required=True)
ap.add_argument('--start', type=int, default=0)
ap.add_argument('--stop', type=int, default=N_LEFTOVER)
ap.add_argument('--ids', type=str, help='comma/space IDs or a file of IDs')
ap.add_argument('--certs', type=str,
help='search one-flip balls of certificates in this JSON')
ap.add_argument('--radius', type=int, default=3)
args = ap.parse_args()
assert args.radius == 3, 'this campaign records radius 3 only'
extra_known = set()
if args.certs:
cpath = Path(args.certs)
if not cpath.is_absolute():
cpath = HERE / cpath
rows, seedlist = (cert_plan(cpath, source_prefix='q12 addition: '),
cert_seeds(cpath, source_prefix='q12 addition: '))
domain = 'q12-certificate-flips'
extra_known = {canon(c['scheme']) for c in seedlist}
else:
rows, seedlist = leftover_plan(), leftover_seeds()
domain = 'q11-followup-leftover'
assert digest(rows) == LEFTOVER_PLAN_SHA
assert digest(seedlist) == LEFTOVER_SEEDS_SHA
provenance = dict(radius=args.radius, domain=domain,
plan_sha256=digest(rows), seeds_sha256=digest(seedlist),
sources=source_hashes())
path = Path(args.output)
if not path.is_absolute():
path = HERE / path
if args.start > len(rows):
args.start = len(rows)
if args.stop > len(rows):
args.stop = len(rows)
ids = parse_ids(args, len(rows))
for i in ids:
assert 0 <= i < len(rows)
if path.exists():
saved = json.loads(path.read_text())
assert saved['provenance'] == provenance, 'checkpoint source/input mismatch'
results = saved['tasks']
done = {r['id'] for r in results}
assert done <= set(ids)
else:
results, done = [], set()
work = HERE / 'work'
work.mkdir(exist_ok=True)
exe = work / 'ball'
if not exe.exists():
subprocess.run(['cc', '-O3', '-std=c11', '-Wall', '-Wextra',
str(HERE / 'ball.c'), '-o', str(exe)], check=True)
expected = sum(comb(45, k) for k in range(4))
known = baseline() | extra_known
start = time.monotonic()
cand_path = HERE / 'certs/candidates.jsonl'
for idx in ids:
if idx in done:
continue
row = rows[idx]
tris = task_tris(row, seedlist[row['seed']])
signs = unpack(seedlist[row['seed']])[2]
task = work / f'task-{path.stem}-{idx}.txt'
write_task(tris, signs, task, known=known | {'<>'})
p = subprocess.run([str(exe), str(task), str(args.radius)], check=True,
capture_output=True, text=True)
messages = [json.loads(line) for line in p.stdout.splitlines()]
summaries = [m for m in messages if m['kind'] == 'summary']
assert len(summaries) == 1
summary = summaries[0]
assert summary['complete'] and summary['evals'] == expected
counts = {code_scheme(code): n for code, n in summary['counts']}
assert len(counts) == len(summary['counts']) and sum(counts.values()) == expected
candidates = [m for m in messages if m['kind'] == 'witness']
novel = set(counts) - known - {'<>'}
assert {code_scheme(m['code']) for m in candidates} == novel
compact = [{k: m[k] for k in ('code', 'bits')} for m in candidates]
entry = dict(row, id=idx, complete=True, evals=expected, counts=counts,
candidates=compact, novel=sorted(novel))
results.append(entry)
results.sort(key=lambda r: r['id'])
total = Counter()
for r in results:
total.update(r['counts'])
saved = dict(provenance=provenance, planned_ids=ids,
complete=len(results) == len(ids), tasks=results,
evaluations=sum(r['evals'] for r in results),
counts=dict(total),
candidate_schemes=sorted(set(total) - known - {'<>'}))
atomic_json(path, saved)
task.unlink(missing_ok=True)
if novel:
rec = dict(id=idx, domain=domain, seed=row['seed'],
removed=row['removed'], added=row['added'],
triangulation_sha256=row['triangulation_sha256'],
schemes=sorted(novel), candidates=compact)
append_jsonl(cand_path, rec)
print(f'CANDIDATE task {idx} schemes {sorted(novel)}', flush=True)
if novel or (len(results) % 10 == 0) or len(results) == 1 or len(results) == len(ids):
print(f'{len(results)}/{len(ids)} this shard; {saved["evaluations"]} evaluations; '
f'{len(total)} schemes; {len(saved["candidate_schemes"])} candidates; '
f'{time.monotonic()-start:.1f}s', flush=True)
return 0 if len(results) == len(ids) else 2
if __name__ == '__main__':
raise SystemExit(main())