median.py
Checks: LATENCY: Trees of units property; rule; observation.
Cited on: LATENCY.
To run it: save median.py, with chains.py, dags.py, graphs.py, manystream.py, minclose.py, poles.py, redundant.py, sublattice.py beside it; needs numpy.
"""
median.py -- is a graph's feasible set closed under the componentwise
median?
QUESTION. sublattice.py's sets at each (class, cell, T), T from brute's
least to naive: D and A (the shared units' delays and arrivals) and,
unprojected, FD and F (every unit's delays, every unit's arrivals).
Populations Y4 (dags.classes(4), the 441 cells), Y5 (exactly 5 units,
the 34 SQ cells), X6 (exactly 6 units, (6, 2, 4), (35, 8, 27),
(4, 2, 2)). Is each set closed under med(p, q, r)?
BY HAND (frozen before code). A clause of two threshold literals is
median-closed: a literal holds at the median iff it holds at two of the
three points, and each point meets one of the clause's two literals.
Projection keeps median-closure, so FD median-closed gives D. The min
witness's A at T = 8 (minclose.py S3) is its box less three points,
each one clause's only violation, so median-closed, and
median-testable: med((2, 6), (3, 5), (4, 7)) = (3, 6).
PREDICTIONS. D and A median-closed everywhere (moderately); FD too
(weakly); F none.
KILL (frozen). "KILL D, A not median-closed on shared units: N
(median-testable M)" over Y4, Y5, X6; fires if N > 0, UNREADABLE if
M < 10. Testable: some triple whose median is none of the three.
CONTROLS. M1 synthetic: the three-point cross caught; {0, 1, 2}^3 cut
by a 3-clause caught; {0..3}^3 cut by 30 random two-threshold clauses
passes at 20 seeds, each testable (as coded, a cut leaving under 8
points is skipped, so a seed may apply fewer than 30). M2 a real
break: every median-closed, testable D and A set of X6 at (6, 2, 4)
with one new median that is a member deleted is caught.
M3 the fast detector equals the triple loop on every Y4 set at
(6, 2, 4) and the X6 sets there under 60 points. HAND the min witness's
D set at T = 8.
FINDINGS (copied from the run below). M1 green (20 closed, 20
testable), M2 286 of 286 deleted medians caught (first run 375 of 375,
89 of them vacuous: the median not a member, or the set already not
closed; M2 now deletes members of closed sets only and counts the
vacuous, 0), M3 2,802 sets, 0 off.
HAND: the min witness at T = 8, D (h', v) = {(2, 1), (2, 2), (2, 3),
(3, 1), (3, 2), (3, 3), (4, 1), (4, 2), (4, 3), (5, 1)} and A both
median-closed, as predicted; FD not (|FD| 37), F is.
Median-testable D, A sets: Y4 0, 0 (no Y4 set reads the kill); Y5 234,
118; X6 573, 267. Not median-closed:
D: Y5 6, X6 108, all above shared; A: X6 1, above shared.
By cell, D: X6 (6, 2, 4) 88, (35, 8, 27) 10, (4, 2, 2) 10; Y5
(6, 2, 4) 4, (9, 4, 5) 1, (9, 5, 4) 1. A: X6 (6, 2, 4) 1.
FD: Y4 16, Y5 159, X6 831 (heads-only kinds included); F: Y5 1,
X6 46.
KILL D, A not median-closed on shared units: 115 (median-testable
1192). FIRES (predicted 0).
Y5's first D failure, at (6, 2, 4): h, c = (z, h), r = (h, c), with
(z, r) and (c, r) reading on, T = 8, (L_h, L_c, L_r) = (2, 2, 2),
(2, 3, 1), (3, 2, 1) alive, their median (2, 2, 1) not (X6's first
fails on the same triple): the least witness (h, c, r alone) read on
by two units, r at 1 alive only behind h or c raised. In delays that is the
three-literal clause [L_r >= 2] or [L_h >= 3] or [L_c >= 3], which no
majority operation preserves; FD's first failure (Y4) is the same
triple on the bare witness at T = 6. The one A failure: X6, (6, 2, 4),
h, u1 = (z, h), u2 = (h, u1), u3 = (z, u2), (z, u3), (u2, u3), T = 10,
(a_h, a_u2, a_u3) = (2, 7, 8), (3, 6, 8), (3, 7, 9) alive, (3, 7, 8)
not: u3 at delay 1 behind u2 arriving at 7 lives only with h at 2.
So the graph's feasible set is not median-closed in delays at five
cells ((6, 2, 4), (35, 8, 27), (4, 2, 2), (9, 4, 5), (9, 5, 4)) nor in
arrivals at (6, 2, 4); over the cells each population runs, arrivals
nearly are (1 failure of 385 testable).
RUN RECORD. main: 3/3 checks, 35.3 s, peak 21.1 MB.
"""
import itertools
import os
import random
import sys
import time
from collections import Counter
os.environ.setdefault("OPENBLAS_NUM_THREADS", "1")
import numpy as np # noqa: E402
import dags # noqa: E402
import graphs # noqa: E402
import sublattice as sl # noqa: E402
from minclose import CELLS, MINW, cell_named, six # noqa: E402
CHECKS = []
def check(name, ok, detail=""):
CHECKS.append(bool(ok))
tail = f" ({detail})" if detail else ""
print(f" {'PASS' if ok else 'FAIL'} {name}{tail}")
def med(p, q, r):
return tuple(sorted(t)[1] for t in zip(p, q, r))
def slow(P):
"""(median-closed, testable, a missing median's triple or None)."""
P = set(P)
ok, test, wit = True, False, None
for p, q, r in itertools.combinations(sorted(P), 3):
m = med(p, q, r)
if m not in (p, q, r):
test = True
if m not in P:
ok = False
wit = wit or (p, q, r)
return ok, test, wit
def fast(P):
"""(median-closed, testable, a missing median's triple or None):
med(p, q, r) is r clamped into [min(p, q), max(p, q)], so per pair
one clamp of every point."""
P = sorted(set(P))
n = len(P)
if n < 3:
return True, False, None
X = np.array(P, dtype=np.int64)
lo, span = X.min(0), X.max(0) - X.min(0) + 1
w = np.cumprod(np.concatenate(([1], span[:-1])))
code = np.sort((X - lo) @ w)
ok, test, wit = True, False, None
for i in range(n):
for j in range(i + 1, n):
a = np.minimum(X[i], X[j])
b = np.maximum(X[i], X[j])
M = np.clip(X, a, b)
new = ~((M == X[i]).all(1) | (M == X[j]).all(1)
| (M == X).all(1))
if not new.any():
continue
test = True
c = (M[new] - lo) @ w
k = np.searchsorted(code, c)
k[k == n] = 0
miss = code[k] != c
if miss.any():
r = np.flatnonzero(new)[np.argmax(miss)]
return False, test, (P[i], P[j], P[r])
return ok, test, wit
def sets(d, nv, rows, T):
S = sl.shared_units(d)
sel = [r for r in rows if r[0] <= T]
return {"D": {tuple(r[1][j] for j in S) for r in sel},
"A": {tuple(r[2][j] for j in S) for r in sel},
"FD": {r[1] for r in sel},
"F": {r[2] for r in sel}}
def controls():
print("M1..M3 the instrument")
got = {}
got["cross"] = fast({(1, 1, 2), (1, 2, 1), (2, 1, 1)})[:2]
box3 = set(itertools.product(range(3), repeat=3))
got["3-clause"] = fast({p for p in box3 if max(p) >= 2})[:2]
rnd = Counter()
for seed in range(20):
rg = random.Random(seed)
P = set(itertools.product(range(4), repeat=3))
for _ in range(30):
i, j = rg.sample(range(3), 2)
a, b = rg.randint(0, 3), rg.randint(0, 3)
si, sj = rg.choice((1, -1)), rg.choice((1, -1))
lit = lambda p, k, t, s: (p[k] >= t) if s > 0 else (p[k] <= t)
Q = {p for p in P if lit(p, i, a, si) or lit(p, j, b, sj)}
if len(Q) >= 8:
P = Q
c = fast(P)
rnd["closed"] += c[0]
rnd["testable"] += c[1]
check("M1 cross and 3-clause caught; 20 two-threshold sets closed, "
"testable", got == {"cross": (False, True),
"3-clause": (False, True)}
and rnd["closed"] == 20 and rnd["testable"] == 20,
f"{got}, {dict(rnd)}")
cell = cell_named("(6, 2, 4)")
planted = caught = vacuous = 0
cmp_off = cmp_n = 0
t0 = time.time()
for units in six():
d = dags.Dag(cell, units)
nv = d.naive()
rows = sl.alive_all(d, nv)
least = min(r[0] for r in rows)
for T in range(least, nv + 1):
ss = sets(d, nv, rows, T)
for arm in ("D", "A"):
P = ss[arm]
if len(P) < 60:
cmp_n += 1
cmp_off += fast(P)[:2] != slow(P)[:2]
if not fast(P)[0]:
continue
m = None
for p, q, r in itertools.combinations(sorted(P), 3):
x = med(p, q, r)
if x not in (p, q, r) and x in P:
m = x
break
if m is None:
continue
planted += 1
vacuous += m not in P or not fast(P)[0]
caught += not fast(P - {m})[0]
for units in dags.classes(4):
d = dags.Dag(cell, units)
nv = d.naive()
rows = sl.alive_all(d, nv)
least = min(r[0] for r in rows)
for T in range(least, nv + 1):
for arm, P in sets(d, nv, rows, T).items():
if len(P) < 60:
cmp_n += 1
cmp_off += fast(P)[:2] != slow(P)[:2]
check("M2 a deleted new median caught in every median-closed, "
"testable X6 D, A set at (6, 2, 4), each deletion a member",
planted > 0 and caught == planted and vacuous == 0,
f"{caught} of {planted}, {vacuous} vacuous")
check("M3 fast = slow on Y4 at (6, 2, 4) and X6 D, A there, under "
"60 points", cmp_n > 0 and cmp_off == 0,
f"{cmp_n} sets, {cmp_off} off, {time.time() - t0:.1f} s")
def hand():
print("\nHAND the min witness at (6, 2, 4), T = 8")
d = dags.Dag(cell_named("(6, 2, 4)"), MINW)
nv = d.naive()
rows = sl.alive_all(d, nv)
ss = sets(d, nv, rows, 8)
for arm in ("D", "A"):
print(f" {arm} (h', v): {sorted(ss[arm])} "
f"closure {fast(ss[arm])}")
print(f" |FD| {len(ss['FD'])}, |F| {len(ss['F'])}: "
f"FD {fast(ss['FD'])[:2]}, F {fast(ss['F'])[:2]}")
def sweep(tag, cls, cells_, arms):
t0 = time.time()
n = Counter()
first = {}
big = Counter()
for cell in cells_:
for units in cls:
d = dags.Dag(cell, units)
nv = d.naive()
rows = sl.alive_all(d, nv)
least = min(r[0] for r in rows)
k = sl.kind(d, sl.shared_units(d))
for T in range(least, nv + 1):
ss = sets(d, nv, rows, T)
n[(k, "triples")] += 1
for arm in arms:
P = ss[arm]
big[arm] = max(big[arm], len(P))
ok, test, wit = fast(P)
n[(k, f"{arm} testable")] += test
if not ok:
n[(k, f"{arm} not median-closed")] += 1
n[("cell", graphs.name(cell), arm)] += 1
first.setdefault((k, arm), (
graphs.name(cell), units, T, wit))
print(f"\n{tag} {len(cls)} classes at {len(cells_)} cells, "
f"{time.time() - t0:.1f} s, largest sets {dict(big)}")
kinds = ("heads only", "shared non-head", "shared above shared")
for k in kinds:
print(f" {k:20s} triples {n[(k, 'triples')]}; testable " + str(
{a: n[(k, f'{a} testable')] for a in arms}))
for k in kinds:
print(f" {k:20s} not median-closed " + str(
{a: n[(k, f'{a} not median-closed')] for a in arms}))
print(" not median-closed by cell: " + str(
{x[1:]: v for x, v in sorted(n.items()) if x[0] == "cell"}))
for key in sorted(first):
print(f" first {key}: {first[key]}")
return n
def populations():
"""(tag, classes, cells): X6 at CELLS, Y4 at the 441, Y5 at the 34
where S or Q gains (with (6, 2, 4) and (35, 8, 27))."""
cs = list(graphs.cells())
gain = lambda c, k: (lambda p: p[1] < p[0])(graphs.price(
c, graphs.SHAPES[k]))
sq = [c for c in cs if graphs.name(c) in ("(6, 2, 4)", "(35, 8, 27)")
or gain(c, "S") or gain(c, "Q")]
return (("X6", six(), [cell_named(nm) for nm in CELLS]),
("Y4", dags.classes(4), cs),
("Y5", [u for u in dags.classes(5) if len(u) == 5], sq))
def main(arms=("D", "A", "FD", "F")):
t0 = time.time()
controls()
if not all(CHECKS):
print("\nCONTROL FAILED: nothing below is read.")
return 1
hand()
tot = Counter()
for tag, cls, cells_ in populations():
tot += sweep(tag, cls, cells_, arms)
kinds = ("heads only", "shared non-head", "shared above shared")
M = sum(tot[(k, f"{a} testable")] for k in kinds for a in ("D", "A"))
N = sum(tot[(k, f"{a} not median-closed")] for k in kinds
for a in ("D", "A"))
for a in arms:
print(f"{a}: testable {sum(tot[(k, f'{a} testable')] for k in kinds)}"
f", not median-closed "
f"{sum(tot[(k, f'{a} not median-closed')] for k in kinds)}")
print(f"\nKILL D, A not median-closed on shared units: {N} "
f"(median-testable {M}){' UNREADABLE' if M < 10 else ''}")
print(f"\n{sum(CHECKS)}/{len(CHECKS)} checks, {time.time() - t0:.1f} s")
return 0
if __name__ == "__main__":
sys.exit(main())