raises.py

Checks: LATENCY: Trees of units property; rule; observation.

Cited on: LATENCY.

To run it: save raises.py, with chains.py, dags.py, graphs.py, manystream.py, median.py, minclose.py, poles.py, redundant.py, sublattice.py beside it; needs numpy.

"""
raises.py -- do a graph's delays carry a clause of three raises?

QUESTION. median.py's FD sets (every unit's delays alive at latency at
most T) at each (class, cell, T) of Y5 and X6 that are not
median-closed. Which clauses over threshold literals on three units do
they force, and does any carry a RAISE on each of its three units?

DESIGN (frozen before code). Within the set's box hull a
clause on units (i, j, k) is violated on a sub-box, each side a proper
sub-interval of the unit's hull range; the set forces it iff its
projection on (i, j, k) misses the box. The PRIME clauses on three
units are the maximal empty boxes with three proper sides. A side
[s, t] of unit u, hull [lo, hi], floor f = L*_u, carries [L_u <= s-1]
if s > lo (CAP) and [L_u >= t+1] if t < hi (RAISE if t+1 > f, else
DROP-GUARD); PURE if s = lo.

PREDICTIONS. At most two RAISE units per prime 3-clause, the third a
DROP-GUARD (moderately); CAPs appear (weakly); no pure three-RAISE
clause (moderately).

KILL (frozen). "KILL prime 3-unit clauses with a RAISE on each unit:
N (sets M, prime 3-unit clauses K)" over every non-median-closed FD
set of Y5 and X6; fires if N == 0, UNREADABLE if K < 10.

CONTROLS. C1 the box finder against a brute clause enumeration on 40
random subsets of {0..3}^3 and {0..2}^4. C2 the least witness at
(6, 2, 4), T = 6: exactly [L_h >= 3] or [L_c >= 3] or [L_r >= 2].
C3 every median-closed FD set of Y4 and Y5: no prime 3-unit clause.
C4 a planted [L_a >= f+1] or [L_b >= f+1] or [L_c >= f+1] cut from
median-closed Y5 FD sets, hulls kept and no two-unit clause implying
it, read back each time as a prime 3-unit clause of three RAISEs whose
box holds the plant: by construction the plant box is a maximal empty
one, so C4 re-tests C1 on real sets and the R tag on real floors.

FINDINGS (copied from the run below). C1 40 sets, 0 off; C2 the
witness's FD is {(2, 2, 2), (2, 3, 1), (3, 2, 1)}, its one prime
clause sides h [2, 2], c [2, 2], r [1, 1], kinds R, R, D; C3 3,745
median-closed sets, 0 clauses; C4 984 of 984 plants read back.
KILL prime 3-unit clauses with a RAISE on each unit: 0 (sets 990,
prime 3-unit clauses 3063). FIRES: no three-raise clause at Y5 or X6.
Patterns (per unit C cap, D drop-guard, R raise), X6 then Y5:
  C.C.C. 1251, 317; D.R.R. 886, 85; C.D.R. 320, 22; C.C.D. 93, 64;
  C.C.R. 12, 0; D.D.R. 9, 0; C.D.D. 2, 0; C.R.R. 2, 0.
Median-closed sets (2,036 X6, 2,868 Y5): none. Added after the kill
was read, units on one directed path: C.C.C. 1251 of 1251, 317 of
317; D.R.R. 733 of 886, 75 of 85; C.C.D. 24 of 93, 0 of 64; D.D.R.
6 of 9; C.C.R. 3 of 12; C.D.R. none. Added at the audit, C.C.C. with
every cap [L_u <= s - 1] at or above its floor (s - 1 >= L*_u): 162,
6; the other 1,400 hold a cap below a floor, a literal that demands a
drop.
So the predictions split: no pure three-raise clause (held); at most
two raises per clause (held, the kill); "the third a drop-guard"
(wrong: half the clauses are three CAPs, every one's units on a
directed path, read, not checked, as the path's latency budget). The
forced clauses in delays come in two families: three caps on a path,
1,400 of them with a literal that demands a drop and 168 capping
raises only, and raises
guarded by a drop (r below its floor wants h or c above theirs), with
mixed CAP-DROP-RAISE ones off paths.

RUN RECORD. main: 4/4 checks, 56.7 s, peak commit 21.5 MB; with the
path print, 56.3 s; with the floor print, 57.0 s, 21.3 MB. In code/: 4/4 checks, 57.6 s,
21.4 MB, every count above reproduced.
"""
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 median  # noqa: E402
import sublattice as sl  # noqa: E402
from minclose import cell_named  # 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 boxes3(Q):
    """Maximal empty boxes with three proper sides of a set Q of
    3-tuples, within its hull: a list of ((s0, t0), (s1, t1), (s2, t2))
    in Q's own values."""
    X = np.array(sorted(Q), dtype=np.int64)
    lo, hi = X.min(0), X.max(0)
    n = hi - lo + 1
    if (n < 2).any():
        return []
    G = np.zeros(tuple(n), dtype=np.int64)
    G[tuple((X - lo).T)] = 1
    S = np.zeros(tuple(n + 1), dtype=np.int64)
    S[1:, 1:, 1:] = G.cumsum(0).cumsum(1).cumsum(2)
    a = [np.arange(m) for m in n]
    s0, t0, s1, t1, s2, t2 = np.ix_(a[0], a[0], a[1], a[1], a[2], a[2])

    def cnt(s0, t0, s1, t1, s2, t2):
        return (S[t0 + 1, t1 + 1, t2 + 1] - S[s0, t1 + 1, t2 + 1]
                - S[t0 + 1, s1, t2 + 1] - S[t0 + 1, t1 + 1, s2]
                + S[s0, s1, t2 + 1] + S[s0, t1 + 1, s2]
                + S[t0 + 1, s1, s2] - S[s0, s1, s2])

    ok = (s0 <= t0) & (s1 <= t1) & (s2 <= t2)
    ok &= ~((s0 == 0) & (t0 == n[0] - 1)) & ~((s1 == 0) & (t1 == n[1] - 1))
    ok &= ~((s2 == 0) & (t2 == n[2] - 1))
    sc = [np.clip(x, 0, None) for x in (s0, t0, s1, t1, s2, t2)]
    ok &= cnt(*sc) == 0
    side = [(s0, t0, 0), (s1, t1, 1), (s2, t2, 2)]
    for s, t, ax in side:
        for grow in ("down", "up"):
            g = [s0, t0, s1, t1, s2, t2]
            if grow == "down":
                can = s > 0
                g[2 * ax] = np.clip(s - 1, 0, None)
            else:
                can = t < n[ax] - 1
                g[2 * ax + 1] = np.clip(t + 1, None, n[ax] - 1)
            g = [np.clip(x, 0, None) for x in g]
            ok &= ~(can & (cnt(*g) == 0))
    out = []
    for idx in zip(*np.nonzero(ok)):
        out.append(tuple((int(idx[2 * k] + lo[k]), int(idx[2 * k + 1] + lo[k]))
                         for k in range(3)))
    return out


def brute3(P):
    """Prime clauses on exactly three units of P, by literals: per unit
    an interval-complement, implication by every point, primality by
    every one-step strengthening. Keyed (units, sides) as boxes3's."""
    P = sorted(set(P))
    m = len(P[0])
    lo = [min(p[u] for p in P) for u in range(m)]
    hi = [max(p[u] for p in P) for u in range(m)]
    out = set()
    for U in itertools.combinations(range(m), 3):
        opts = []
        for u in U:
            o = [(s, t) for s in range(lo[u], hi[u] + 1)
                 for t in range(s, hi[u] + 1)
                 if not (s == lo[u] and t == hi[u])]
            opts.append(o)
        for sides in itertools.product(*opts):
            def implied(sd):
                return not any(all(sd[k][0] <= p[U[k]] <= sd[k][1]
                                   for k in range(3)) for p in P)
            if not implied(sides):
                continue
            prime = True
            for k in range(3):
                s, t = sides[k]
                for g in ((s - 1, t), (s, t + 1)):
                    if g[0] < lo[U[k]] or g[1] > hi[U[k]]:
                        continue
                    sd = list(sides)
                    sd[k] = g
                    if implied(sd):
                        prime = False
            if prime:
                out.add((U, sides))
    return out


def clauses(P):
    """Every prime 3-unit clause of P: (units, sides)."""
    P = sorted(set(P))
    m = len(P[0])
    out = []
    for U in itertools.combinations(range(m), 3):
        Q = {tuple(p[u] for u in U) for p in P}
        for b in boxes3(Q):
            out.append((U, b))
    return out


def kinds(P, floors, cl):
    """Per unit of a clause: its literal kinds, e.g. 'R', 'CD'."""
    U, sides = cl
    tags = []
    for u, (s, t) in zip(U, sides):
        lo = min(p[u] for p in P)
        hi = max(p[u] for p in P)
        k = ""
        if s > lo:
            k += "C"
        if t < hi:
            k += "R" if t + 1 > floors[u] else "D"
        tags.append(k)
    return tags


def controls():
    print("C1..C4  the instrument")
    off = n = 0
    for seed in range(40):
        rg = random.Random(seed)
        dim, top = (3, 3) if seed % 2 else (4, 2)
        box = list(itertools.product(range(top + 1), repeat=dim))
        P = [p for p in box if rg.random() < 0.6]
        if len(P) < 3:
            continue
        n += 1
        a = set((U, tuple(b)) for U, b in clauses(P))
        b = brute3(P)
        off += a != b
    check("C1 boxes3 = brute on random subsets of {0..3}^3, {0..2}^4",
          n >= 30 and off == 0, f"{n} sets, {off} off")
    w = ((None, None), (None, 0), (0, 1))
    d = dags.Dag(cell_named("(6, 2, 4)"), w)
    rows = sl.alive_all(d, d.naive())
    P = median.sets(d, d.naive(), rows, 6)["FD"]
    cl = clauses(P)
    print(f"    least witness FD at T = 6: {sorted(P)}")
    print(f"    its prime 3-unit clauses: "
          f"{[(c, kinds(P, d.Ls, c)) for c in cl]}")
    lo = tuple(min(p[u] for p in P) for u in range(3))
    want = [((0, 1, 2), ((lo[0], 2), (lo[1], 2), (lo[2], 1)))]
    check("C2 the least witness: exactly [L_h >= 3] or [L_c >= 3] or "
          "[L_r >= 2], kinds R, R, D", cl == want
          and kinds(P, d.Ls, cl[0]) == ["R", "R", "D"], f"floors {d.Ls}")
    closed_n = closed_cl = 0
    planted = caught = 0
    pops = {t: (c, cs) for t, c, cs in median.populations()}
    for tag in ("Y4", "Y5"):
        cls, cells_ = pops[tag]
        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)
                for T in range(least, nv + 1):
                    P = median.sets(d, nv, rows, T)["FD"]
                    if len(P) < 3 or not median.fast(P)[0]:
                        continue
                    closed_n += 1
                    closed_cl += len(clauses(P))
                    if tag == "Y5":
                        r = plant(P, d.Ls)
                        if r is not None:
                            planted += 1
                            caught += r
    check("C3 median-closed FD sets of Y4, Y5: no prime 3-unit clause",
          closed_n > 0 and closed_cl == 0,
          f"{closed_n} sets, {closed_cl} clauses")
    check("C4 a planted three-RAISE cut read back", planted >= 10
          and caught == planted, f"{caught} of {planted}")


def plant(P, floors):
    """Cut P by a three-RAISE clause on its first eligible triple;
    None if no triple is eligible, else whether it is read back."""
    P = set(P)
    m = len(next(iter(P)))
    lo = [min(p[u] for p in P) for u in range(m)]
    hi = [max(p[u] for p in P) for u in range(m)]
    for U in itertools.combinations(range(m), 3):
        if not all(lo[u] <= floors[u] < hi[u] for u in U):
            continue
        cut = {p for p in P if any(p[u] > floors[u] for u in U)}
        if cut == P or not cut:
            continue
        if any(min(p[u] for p in cut) != lo[u]
               or max(p[u] for p in cut) != hi[u] for u in range(m)):
            continue
        # no two-unit clause on U implies the plant
        if any(not any(all(p[U[k]] <= floors[U[k]] for k in (0, 1, 2)
                           if k != skip) for p in cut) for skip in range(3)):
            continue
        for cl in clauses(cut):
            if cl[0] == U and all(s == lo[u] and t >= floors[u]
                                  for u, (s, t) in zip(U, cl[1])) \
                    and kinds(cut, floors, cl) == ["R", "R", "R"]:
                return True
        return False
    return None


def on_path(d, U):
    """Do the units U lie on one directed path of d (added after the
    kill was read)?"""
    below = [set() for _ in range(d.n)]
    for j, u in enumerate(d.units):
        for x in u:
            if x is not None:
                below[j] |= below[x] | {x}
    a, b, c = sorted(U)
    return a in below[b] and b in below[c]


def sweep(tag, cls, cells_):
    t0 = time.time()
    n = Counter()
    first = {}
    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)
            for T in range(least, nv + 1):
                P = median.sets(d, nv, rows, T)["FD"]
                if len(P) < 3:
                    continue
                arm = "open" if median.fast(P)[0] else "fail"
                n[(arm, "sets")] += 1
                for cl in clauses(P):
                    tg = kinds(P, d.Ls, cl)
                    pat = "".join(sorted(t.ljust(2, ".") for t in tg))
                    n[(arm, "clauses")] += 1
                    n[(arm, "pattern", pat)] += 1
                    three = all("R" in t for t in tg)
                    pure = all(t == "R" for t in tg)
                    n[(arm, "three")] += three
                    n[(arm, "pure")] += pure
                    n[(arm, "on a path", pat)] += on_path(d, cl[0])
                    if pat == "C.C.C.":
                        # added at the audit: [L_u <= s - 1] below the
                        # floor demands a drop, at or above it caps a
                        # raise
                        n[(arm, "C.C.C. caps at or above floor")] += all(
                            s - 1 >= d.Ls[u] for u, (s, t) in zip(*cl))
                    if three:
                        n[("cell", graphs.name(cell))] += 1
                        first.setdefault(arm, (graphs.name(cell), units,
                                               T, cl, tg))
                    first.setdefault((arm, pat), (graphs.name(cell), units,
                                                  T, cl))
    print(f"\n{tag}  {len(cls)} classes at {len(cells_)} cells, "
          f"{time.time() - t0:.1f} s")
    for arm in ("fail", "open"):
        print(f"    {arm}: sets {n[(arm, 'sets')]}, prime 3-unit clauses "
              f"{n[(arm, 'clauses')]}, a RAISE on each unit "
              f"{n[(arm, 'three')]}, pure {n[(arm, 'pure')]}")
        pats = {k[2]: v for k, v in sorted(n.items())
                if k[0] == arm and k[1] == "pattern"}
        print(f"      patterns {pats}")
        if pats:
            print(f"      C.C.C. with every cap at or above its floor "
                  f"{n[(arm, 'C.C.C. caps at or above floor')]}")
            print("      of them on one directed path " + str(
                {k[2]: v for k, v in sorted(n.items())
                 if k[0] == arm and k[1] == "on a path"}))
    print(f"    three-RAISE by cell: "
          f"{ {k[1]: v for k, v in sorted(n.items()) if k[0] == 'cell'} }")
    for k in sorted(first, key=str):
        print(f"    first {k}: {first[k]}")
    return n


def main():
    t0 = time.time()
    controls()
    if not all(CHECKS):
        print("\nCONTROL FAILED: nothing below is read.")
        return 1
    tot = Counter()
    for tag, cls, cells_ in median.populations():
        if tag == "Y4":
            continue
        tot += sweep(tag, cls, cells_)
    print(f"\nKILL prime 3-unit clauses with a RAISE on each unit: "
          f"{tot[('fail', 'three')]} (sets {tot[('fail', 'sets')]}, prime "
          f"3-unit clauses {tot[('fail', 'clauses')]})"
          f"{'  UNREADABLE' if tot[('fail', 'clauses')] < 10 else ''}")
    print(f"\n{sum(CHECKS)}/{len(CHECKS)} checks, {time.time() - t0:.1f} s")
    return 0


if __name__ == "__main__":
    sys.exit(main())