""" sublattice.py -- is a graph's feasible set a lattice on its shared units? QUESTION. ors.py: at (6, 2, 4) a cone reading pinned heads has, at each latency T, a sublattice of head settings. Over dags.py's shared classes (2..5 units at the 34 SQ cells, 2..4 at the 441), at each T from brute's least to naive: F_T, the assignments in brute's window (L_j in 0..L*_j + 3) with every unit alive and latency <= T, projected onto the SHARED units (two or more readers), the other units existentially optimized. Is F_T closed under componentwise max (min)? If max-closed (in some coordinates), the greatest feasible setting is found by propagation and the graph is priced in polynomial time. COORDINATES. D, the shared units' delays; A, their arrivals. A head (fresh inputs only) has arrival = delay, so A = D wherever only heads are shared. BY HAND (frozen before code). Only heads shared: the other units form cones rooted at sinks over fresh streams and heads, F_T the intersection of their head tables, a head read twice in one cone the diagonal of a two-mark table; by ors.py each is a sublattice at (6, 2, 4) within its sizes, so closed both ways there (elsewhere a TRANSPLANT). A shared unit above a shared unit: in D the path sum L_h + L_g <= c holds at (2, 3) and (3, 2), not at (3, 3), so D can fail by it alone. In A the latency bound is unary and g's own relations have the form a_g >= phi(a_h), max-closed whatever phi; a reader of h and g spends budget as a_g rises while gaining inset, so A can fail through the projection. KILL (frozen): "A not max-closed" > 0 over the (class, cell, T) with a shared non-head; predicted 0, weakly. D predicted closed where only heads are shared, "D not max-closed" > 0 where a shared unit sits above another. Frozen after reading the kill: F arm, the arrivals of EVERY unit (no projection), "F not min-closed" > 0, predicted to fire weakly (an internal unit's delay trades against its reader's along one path); and where A fails max-closure, its cells; then the testable count and A's testable cells. Y5 reads exactly 5 units, so no triple is counted twice. CONTROLS. C1 the enumerator: its least latency = dags.brute's, and its alive set at latency <= naive = an itertools.product filter over the full window, every class of 2..4 units at (6, 2, 4) and (8, 3, 5), and of 5 at (6, 2, 4). C2 the detector on synthetic sets. C3 the hand witness h, c = (z, h), r = (h, c) at (6, 2, 4), T = 6: F_6 = {2, 3} on h's delay. FINDINGS (copied from the run below). C1 148 (class, cell) pairs, 0 off; C2, C3 green. Y4 (17 classes of 2..4 units, 441 cells) and Y5 (the 114 of exactly 5, 34 SQ cells): 13,900 (class, cell, T) triples. Added after reading the kill: a projection can fail closure only where two of its points are incomparable (TESTABLE), and A's testable cells. Heads only (5,569 + 2,442; testable 12 + 69 in both arms, Y4's at 12 cells none of them (6, 2, 4), Y5's at 28): closed under min and max at every triple, as predicted at (6, 2, 4), observed at the others. A shared non-head, none above another (941 + 700; testable 0 + 26, one at each of 26 cells): closed both ways, both arms. A shared unit above another (1,407 + 2,841): D fails max-closure at every testable triple (92 of 92, 1,105 of 1,105), min at 6 + 4, the first met at (4, 2, 2), h, c = (z, h), (z, c), (h, c), T = 5, delays (3, 1) and (2, 2) alive, (3, 2) not: the path sum, as predicted. A is testable at 0 + 70 and fails max at 4, all at (6, 2, 4): KILL FIRES (predicted 0). The first met: h, c = (z, h), r = (h, c), (z, r) twice, T = 8, arrivals (h, r) = (2, 7) and (3, 6) alive, (3, 7) not. A is min-closed at every triple (testable 12 + 165). F, every unit's arrival: not max-closed at 1 + 10, not min-closed at 0 + 9 (testable 254 + 1,867), as predicted. So a greatest-element pass in arrivals is dead at (6, 2, 4); a least one is not refuted, on 177 testable triples. RUN RECORD. main: 3/3 checks, 32.8 s, peak commit 9.4 MB. """ import itertools import sys import time from collections import Counter import chains import dags import graphs 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 alive_all(d, cap): """Every assignment in brute's window, every unit alive, latency <= cap: a list of (latency, delays, arrivals).""" out = [] left, arr, Ls = [None] * d.n, [0] * d.n, [0] * d.n def go(j): if j == d.n: out.append((max(arr), tuple(Ls), tuple(arr))) return for L in range(0, d.Ls[j] + 4): a = d.arrival(j, L, arr) if a > cap: break r = d.leftover(j, L, left) if r is None: continue left[j], arr[j], Ls[j] = r, a, L go(j + 1) left[j], arr[j] = None, 0 go(0) return out def alive_product(d, cap): out = [] for Ls in itertools.product(*[range(0, x + 4) for x in d.Ls]): lat, ok = d.evaluate(list(Ls)) if ok and lat <= cap: arr = [0] * d.n for j in range(d.n): arr[j] = d.arrival(j, Ls[j], arr) out.append((lat, tuple(Ls), tuple(arr))) return out def closed(P): """(min-closed, max-closed, a max witness (p, q) or None, testable: some pair incomparable, so closure can fail at all).""" lo, hi, wit, inc = True, True, None, False P = set(P) for p in P: for q in P: if not (all(a <= b for a, b in zip(p, q)) or all(a >= b for a, b in zip(p, q))): inc = True if tuple(map(min, p, q)) not in P: lo = False if tuple(map(max, p, q)) not in P: if hi: wit = (p, q) hi = False return lo, hi, wit, inc def shared_units(d): return [j for j in range(d.n) if len(d.readers[j]) >= 2] def is_head(d, j): return d.units[j] == (None, None) def above(d, g, h): """g reads h, directly or through other units.""" stack = [x for x in d.units[g] if x is not None] seen = set() while stack: x = stack.pop() if x == h: return True if x in seen: continue seen.add(x) stack += [y for y in d.units[x] if y is not None] return False def kind(d, S): if all(is_head(d, j) for j in S): return "heads only" if any(above(d, g, h) for g in S for h in S if g != h): return "shared above shared" return "shared non-head" def read_pair(cell, units): """[(T, kind, D closure, A closure)] for T = least..naive.""" d = dags.Dag(cell, units) nv = d.naive() rows = alive_all(d, nv) least = min(r[0] for r in rows) S = shared_units(d) k = kind(d, S) out = [] for T in range(least, nv + 1): PD = {tuple(r[1][j] for j in S) for r in rows if r[0] <= T} PA = {tuple(r[2][j] for j in S) for r in rows if r[0] <= T} PF = {r[2] for r in rows if r[0] <= T} out.append((T, k, closed(PD), closed(PA), closed(PF))) return d, least, out def controls(): print("C1..C3 the instrument") off, cnt = 0, 0 pop = [(c, u) for c in (chains.ms.Cell(6, 2, 4), chains.ms.Cell(8, 3, 5)) for u in dags.classes(4)] pop += [(chains.ms.Cell(6, 2, 4), u) for u in dags.classes(5) if len(u) == 5] for cell, units in pop: d = dags.Dag(cell, units) nv = d.naive() a = alive_all(d, nv) b = alive_product(d, nv) cnt += 1 off += sorted(a) != sorted(b) off += min(r[0] for r in a) != dags.brute(d)[0] check("C1 enumerator = product filter, least = brute (2..4 units at " "two cells, 5 at (6, 2, 4))", off == 0, f"{cnt} (class, cell) pairs, {off} off") syn = {"cross": {(1, 2), (2, 1)}, "square": {(1, 2), (2, 1), (2, 2), (1, 1)}, "chain3": {(1, 1), (2, 2), (1, 2)}} got = {k: closed(v)[:2] for k, v in syn.items()} check("C2 detector: cross neither, square both, chain3 both", got == { "cross": (False, False), "square": (True, True), "chain3": (True, True)}, f"{got}") cell = chains.ms.Cell(6, 2, 4) w = ((None, None), (None, 0), (0, 1)) d = dags.Dag(cell, w) rows = alive_all(d, 6) F = sorted({(r[1][0],) for r in rows if r[0] <= 6}) check("C3 the hand witness at T = 6: h's delays {2, 3}", F == [(2,), (3,)], f"{F}, shared {shared_units(d)}") def sweep(tag, nmin, nmax, cells_): t0 = time.time() cls = [u for u in dags.classes(nmax) if len(u) >= nmin] n = Counter() first = {} for cell in cells_: for units in cls: d, least, out = read_pair(cell, units) for T, k, cd, ca, cf in out: n[(k, "triples")] += 1 if not ca[1]: n[(k, f"A max fails at {graphs.name(cell)}")] += 1 for arm, c in (("D", cd), ("A", ca), ("F", cf)): n[(k, f"{arm} testable")] += c[3] if c[3] and arm == "A": n[(k, f"A testable at {graphs.name(cell)}")] += 1 if not c[1]: n[(k, f"{arm} not max-closed")] += 1 key = (k, arm, "max") if key not in first: first[key] = (graphs.name(cell), units, T, shared_units(d), c[2]) if not c[0]: n[(k, f"{arm} not min-closed")] += 1 key = (k, arm, "min") if key not in first: first[key] = (graphs.name(cell), units, T, shared_units(d)) print(f"\n{tag} {len(cls)} shared classes of {nmin}..{nmax} units at " f"{len(cells_)} cells, {time.time() - t0:.1f} s") for k in ("heads only", "shared non-head", "shared above shared"): row = {s: n[(k, s)] for s in ("triples", "D testable", "A testable", "F testable", "D not max-closed", "D not min-closed", "A not max-closed", "A not min-closed", "F not max-closed", "F not min-closed")} print(f" {k:20s} {row}") where = {s: v for (kk, s), v in n.items() if kk == k and (s.startswith("A max fails") or s.startswith("A testable at"))} if where: print(f" {'':20s} {where}") for key in sorted(first): print(f" first {key}: {first[key]}") return n def main(): t0 = time.time() controls() if not all(CHECKS): print("\nCONTROL FAILED: nothing below is read.") return 1 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")] print(f"SQ cells: {len(sq)}") n4 = sweep("Y4", 2, 4, cs) n5 = sweep("Y5", 5, 5, sq) kill = sum(n[(k, "A not max-closed")] for n in (n4, n5) for k in ("shared non-head", "shared above shared")) print(f"\nKILL A not max-closed over shared non-heads: {kill}") print(f"\n{sum(CHECKS)}/{len(CHECKS)} checks, {time.time() - t0:.1f} s") return 0 if __name__ == "__main__": sys.exit(main())