compose.py

"""
compose.py -- does any graph carry a gate off the point E's reach?

QUESTION. schaefer.py found one gate at (4, 2, 2): on h, 1 = (z, h),
2 = (z, h), 3 = (1, 2), 4 = (1, 3), 5 = (1, 4), 6 = (2, 5) at T = 7
(naive 8) the raise-free set reads d6 = NAND(d1, d2). Pricing below
naive would be NP-hard if such gates composed. Do they, and where can a
composed gate be read?

BY HAND (frozen before code). The two composites below read no gate
at B: c1 (gate B's input a := 6, b = (z, h) on A's head) has B's floors
2, 1, 0, 0, 1 and naive 10, its b side 4 slack at every T >= 6; c2 (B
over two gates' outputs) has B's floors all 0, naive 8. (Frozen as
"read nothing"; K3 prints c1 hard off E's reach, so only the gate
claim stands.) At
(4, 2, 2) a unit has floor >= 1 iff lam > 1/4 at inset 0; a deep unit's
tau is a product of its leaves', so a unit over two deep inputs sits at
floor 0, never drops and hands about w - 1 whatever its inputs did.
LEMMA E (property). Let some unit read a head, every reader of every
head be a (z, h) unit,
and the cell such that (z, h) is alive at L*_zh - 1 on its head's floor
leftover (at (4, 2, 2) exactly: h hands 2/9, and (z, h) at L = 1 needs
2/9). Then at every T >= naive - 1 the point E, every (z, h) unit at
L* - 1 and every other unit at L*, lies in RF. A path starts at a head;
its next unit reads the head, so is a (z, h) unit, and a (z, h) unit
reads no unit but a head: every path of two or more units holds exactly
one and loses exactly 1. Units at floor are alive whatever their
insets. So, typed by kind, every relation read at T >= naive - 1 holds
E's projection, and an instance glued from such graphs (each part's
paths are its own, schaefer.py's D1) holds E at the union's naive - 1:
the gate above is PROTECTED, and composition must deliver a gate read
off E's reach (below naive - 1, or where the hypotheses fail).

DESIGN. RF by a box walk (each unit 0..L*, alive, latency <= T),
checked against drops.cut of median.sets. Over X6 at the 441 cells and
X7 at the four (pairwise.jobs), each (graph, T) with RF non-empty is
PROTECTED where the lemma's hypotheses hold and T = naive - 1, else
OFF. A HARD TRIPLE is three units whose drop relation (a drop by 2
folded) lies in none of Schaefer's six classes alone (schaefer.py's
classifier, untyped: the stronger reading). A unit whose drop is
constant in RF is skipped: its triples are a binary relation times a
constant, bijunctive.

PREDICTIONS. No hard triple off E's reach (moderately: at naive - 2 a
critical path needs two drops, and the gate's graph at T = 6 is empty).
E misses 0 (strongly, a proof). The cell condition holds at few cells
(weakly).

KILL (frozen). "KILL hard triples off E's reach: N (RF sets off it M,
hard triples on it G)"; fires if N == 0 (the only gates at this scale
are protected), UNREADABLE if G == 0 or M < 10.

CONTROLS. K0 the box walk equals cut of median.sets, X6 at (4, 2, 2),
every T. K1 the gate's graph at T = 7: E in RF; the plant, E with unit 1
back at its floor, is not. K2 the census reads (1, 2, 6) hard there,
protected; the plant, RF with E removed, is read by the E check as a
miss. K3 c1 and c2 printed: floors, naive, RF per T, hard triples
touching B's units.

FINDINGS (copied from the runs below). K0 0 off of 876 (graph, T); K1
E (2, 1, 1, 1, 1, 1, 1) in RF (4 points), the plant out; K2 hard
[(1, 2, 6)], E miss 0, the plant's 1. K3 c1: T = 9 RF 38, E in, no hard
triple on B; T = 8 RF 18, E out, hard (1, 6, 8) and (2, 6, 8), on
(1, 6, 8) {001, 010, 100, 101, 110}; T = 7 RF 2. c2: T = 7 RF 16, E in,
none on B. The census: the cell condition holds at 17 of the 441 cells,
(4, 2, 2) among them; E misses 0 of 3,912 protected sets, none of them
empty. Hard triples
off E's reach 169: (11, 3, 8) 64, (6, 2, 4) 64, (8, 2, 6) 31, (4, 2, 2)
10; in 160 sets at naive - 1 (on graphs where a hypothesis of the
lemma fails: the cell condition, or a head feeding a non-(z, h) unit)
and 2
at naive - 2, none lower; on it 22, all at (4, 2, 2). The printed
examples off it (three per offset per shard, the first hard triple
each; relations are not counted by kind) read at naive - 1 three
consecutive units of one path, {001, 010, 100, 101}: the path drops at
least once and no two neighbours drop; at (4, 2, 2), T = 6, naive 8,
on h, 1 = (z, h),
2 = 3 = (h, 1), 4 = (2, 3), 5 = (1, 4), 6 = (1, 5), the triple (1, 4, 6)
reads exactly two of three.
KILL hard triples off E's reach: 169 (RF sets off it 49632, hard
triples on it 22). MISSES, against the moderate prediction; E as
proved; the condition at 17 cells, as weakly predicted.
Added after the kill was read (the count printed by read_set): in all
169, and all 22 protected, one unit's cone holds the other two. So over
X6 at the 441 cells and X7 at the four, every hard triple read (191
incidences of (graph, T, triple)) is a unit and two of its ancestors,
none three units that share apart (schaefer.py's D1: a shared unit
brings its cone): at this scale a gate is a unit over its own
ancestors, and composing one is extending its graph, as c1 does. There
c1's three (z, h) units 1, 2, 7 read 8 of 8 patterns at T = 8 and 9
(K3): the hard relation stays on a unit and its cone.

RUN RECORD. Controls 3/3, 1.9 s, peak commit 19.2 MB. Rehearsed at one
cell per population, 12.2 s. Four shards, 41.8 to 43.5 s each, peak
commit 31.0 MB (run four times as prints were added; every count
reproduced). Rerun from code/: controls 3/3, 2.0 s, 20.1 MB.
"""
import itertools
import json
import os
import sys
import time
from collections import Counter

os.environ.setdefault("OPENBLAS_NUM_THREADS", "1")

import dags  # noqa: E402
import graphs  # noqa: E402
import median  # noqa: E402
import sublattice as sl  # noqa: E402
from drops import cut  # noqa: E402
from minclose import cell_named  # noqa: E402
from pairwise import jobs, seven  # noqa: E402
from schaefer import NAND1, classes_of  # noqa: E402

CHECKS = []
C1 = list(NAND1) + [(None, 0), (6, 7), (6, 8), (6, 9), (7, 10)]
C2 = (list(NAND1) + [(None, 0), (None, 0), (7, 8), (7, 9), (7, 10), (8, 11)]
      + [(6, 12), (6, 13), (6, 14), (12, 15)])


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 rf_rows(d, cap):
    """Every point of the raise-free box (each unit 0..L*) alive with
    latency <= cap: a list of (latency, delays)."""
    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)))
            return
        for L in range(0, d.Ls[j] + 1):
            a = d.arrival(j, L, arr)
            if a > cap:
                continue
            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 at(rows, T):
    return {p for lat, p in rows if lat <= T}


def cond(cell):
    """(z, h) alive at L*_zh - 1 on its head's floor leftover."""
    try:
        d = dags.Dag(cell, [(None, None), (None, 0)])
    except AssertionError:
        return False
    if d.Ls[1] < 1:
        return False
    return d.evaluate([d.Ls[0], d.Ls[1] - 1])[1]


def heads_feed_zh(d):
    for h, (a, b) in enumerate(d.units):
        if (a, b) != (None, None):
            continue
        for j in d.readers[h]:
            if tuple(d.units[j]) not in ((None, h), (h, None)):
                return False
    return True


def e_point(d):
    zh = {j for j, u in enumerate(d.units)
          if None in u and any(x is not None and d.units[x] == (None, None)
                               for x in u)}
    return tuple(L - 1 if j in zh else L for j, L in enumerate(d.Ls))


def cone(d, j):
    """The units below j (its ancestors)."""
    out = set()
    for x in d.units[j]:
        if x is not None:
            out |= {x} | cone(d, x)
    return out


def hard_triples(P, floors):
    D = {tuple(int(x < f) for x, f in zip(p, floors)) for p in P}
    vary = [u for u in range(len(floors)) if len({x[u] for x in D}) == 2]
    out = []
    for U in itertools.combinations(vary, 3):
        R = {tuple(x[u] for u in U) for x in D}
        if (0, 0, 0) in R or (1, 1, 1) in R:
            continue
        if not classes_of(R):
            out.append(U)
    return out


def controls():
    print("K0..K3  the instrument")
    cell = cell_named("(4, 2, 2)")
    bad = sets = 0
    for units in dags.classes(6):
        if len(units) != 6:
            continue
        try:
            d = dags.Dag(cell, units)
        except AssertionError:
            continue
        nv = d.naive()
        rows = sl.alive_all(d, nv)
        mine = rf_rows(d, nv)
        for T in range(min(r[0] for r in rows), nv):
            want = cut(median.sets(d, nv, rows, T)["FD"], d.Ls)[0]
            got = at(mine, T)
            sets += 1
            bad += want != got
    check("K0 the box walk = cut of median.sets, X6 at (4, 2, 2), every T",
          bad == 0, f"{bad} off of {sets} (graph, T)")
    dA = dags.Dag(cell, NAND1)
    RF = at(rf_rows(dA, 7), 7)
    E = e_point(dA)
    plant = tuple(dA.Ls[1] if j == 1 else x for j, x in enumerate(E))
    check("K1 gate A at T = 7: E in RF; E with unit 1 at floor is not",
          E in RF and plant not in RF and heads_feed_zh(dA) and cond(cell),
          f"E {E}, RF {len(RF)}")
    H = hard_triples(RF, dA.Ls)
    o1, o2 = Counter(), Counter()
    read_set(dA, RF, 7, 8, True, o1, [])
    read_set(dA, RF - {E}, 7, 8, True, o2, [])
    check("K2 the census reads (1, 2, 6) hard at T = 7, protected; RF less "
          "E reads an E miss", H == [(1, 2, 6)] and o1["hard on"] == 1
          and o1["E miss"] == 0 and o2["E miss"] == 1,
          f"hard {H}; {dict(o1)}; plant E miss {o2['E miss']}")
    for nm, units, first in (("c1", C1, 7), ("c2", C2, 13)):
        d = dags.Dag(cell, units)
        nv = d.naive()
        rows = rf_rows(d, nv - 1)
        per = []
        for T in range(nv - 1, -1, -1):
            P = at(rows, T)
            if not P:
                break
            touch = [U for U in hard_triples(P, d.Ls)
                     if any(u >= first for u in U)]
            D = {tuple(int(x < f) for x, f in zip(p, d.Ls)) for p in P}
            rel = sorted({tuple(x[u] for u in touch[0]) for x in D}) \
                if touch else ""
            zh = [j for j, x in enumerate(e_point(d)) if x < d.Ls[j]]
            per.append(f"T={T} RF {len(P)} E {e_point(d) in P} hard on B "
                       f"{touch} {rel}; (z, h) units {zh} read "
                       f"{len({tuple(x[u] for u in zh) for x in D})} of "
                       f"{2 ** len(zh)} patterns")
        print(f"  K3 {nm}: L* {d.Ls}, naive {nv}; " + "; ".join(per))


def sweep_one(cell, units, ok_cell, out, first):
    try:
        d = dags.Dag(cell, units)
    except AssertionError:
        out["skipped"] += 1
        return
    nv = d.naive()
    rows = rf_rows(d, nv - 1)
    prot = ok_cell and heads_feed_zh(d)
    out["protected, RF empty at naive - 1"] += prot and not at(rows, nv - 1)
    for T in range(nv - 1, -1, -1):
        P = at(rows, T)
        if not P:
            break
        read_set(d, P, T, nv, prot, out, first)


def read_set(d, P, T, nv, prot, out, first):
    on = prot and T == nv - 1
    arm = "on" if on else "off"
    out[f"M {arm}"] += 1
    if on:
        out["E miss"] += e_point(d) not in P
    H = hard_triples(P, d.Ls)
    out[f"hard {arm}"] += len(H)
    out[f"hard {arm} at {graphs.name(d.cell)}"] += len(H)
    out[f"hard {arm}, one unit's cone holds the other two"] += sum(
        any(set(U) - {u} <= cone(d, u) for u in U) for U in H)
    out[f"sets with hard {arm} at naive{T - nv}"] += bool(H)
    if H and not on:
        tag = f"naive{T - nv}"
        if sum(f[0] == tag for f in first) < 3:
            D = {tuple(int(x < f) for x, f in zip(p, d.Ls)) for p in P}
            rel = sorted({tuple(x[u] for u in H[0]) for x in D})
            first.append([tag, graphs.name(d.cell), d.units, T, nv, d.Ls,
                          H[:4], rel])


def run(shard, ncells):
    out, first = Counter(), []
    conds = {}
    t0 = time.time()
    n = 0
    for tag, cell, cls in jobs(shard, ncells, seven()):
        if tag not in ("X6", "X7"):
            continue
        name = graphs.name(cell)
        if name not in conds:
            conds[name] = cond(cell)
        n += 1
        for units in cls:
            sweep_one(cell, units, conds[name], out, first)
    return out, first, conds, time.time() - t0, n


def dump(path, out, first, conds, secs, n):
    with open(path, "w", encoding="utf-8", newline="\n") as fh:
        json.dump({"out": dict(out), "first": first, "conds": conds,
                   "secs": secs, "cells": n}, fh)


def merge(paths):
    out, first, conds = Counter(), [], {}
    for p in paths:
        with open(p, encoding="utf-8") as fh:
            rec = json.load(fh)
        out.update(rec["out"])
        first += rec["first"]
        conds.update(rec["conds"])
        print(f"  {os.path.basename(p)}: {rec['cells']} (pop, cell) jobs, "
              f"{rec['secs']:.1f} s")
    return out, first, conds


def report(out, first, conds):
    held = sorted(k for k, v in conds.items() if v)
    print(f"\n  cell condition holds at {len(held)} of {len(conds)} cells "
          f"read: {held[:12]}")
    for k in ("E miss", "protected, RF empty at naive - 1"):
        print(f"  {k}: {out[k]}")
    for k in sorted(out):
        if out[k]:
            print(f"  {k}: {out[k]}")
    for f in first:
        print(f"    off: {f}")
    N, M, G = out["hard off"], out["M off"], out["hard on"]
    tail = "  UNREADABLE" if G == 0 or M < 10 else ""
    print(f"KILL hard triples off E's reach: {N} (RF sets off it {M}, hard "
          f"triples on it {G}){tail}")


def main(argv):
    t0 = time.time()
    if argv and argv[0] == "--merge":
        report(*merge(argv[1:]))
        return 0
    if argv and argv[0] == "--shard":
        shard, ncells, path = int(argv[1]), int(argv[2]), argv[3]
        out, first, conds, secs, n = run(None if shard < 0 else shard,
                                         ncells)
        dump(path, out, first, conds, secs, n)
        print(f"shard {shard}: {secs:.1f} s")
        return 0
    controls()
    if not all(CHECKS):
        print("\nCONTROL FAILED: nothing below is read.")
        return 1
    print(f"\n{sum(CHECKS)}/{len(CHECKS)} checks, {time.time() - t0:.1f} s")
    return 0


if __name__ == "__main__":
    sys.exit(main(sys.argv[1:]))