sublattice.py

"""
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())