drops.py

"""
drops.py -- are drops alone median-closed?

QUESTION. Below naive (T from brute's least to naive - 1), each FD set
(every unit's delays alive at latency <= T, median.sets) is cut to its
RAISE-FREE part RF = {p : p_u <= L*_u every u}. Is RF median-closed?
Where RF is empty, the FORCED part {p : p_u <= L*_u off R0}, R0 the
least raised set (choice.py: one exists in every such set read), is
read the same way, apart.

BY HAND (frozen before code). Inside the box a unit at its floor is
alive whatever its inputs (naive is alive; a leftover rises in each
inset), so RF is cut by (a) the latency, a down-set: each path whose
floor sum passes T loses a drop on it, a clause of caps
[L_u <= L*_u - 1] over its droppable units, and (b) each drop's
aliveness, an up-set in its ancestors' delays. Three caps on one path
are not median-closed ((1,2,2), (2,1,2), (2,2,1) under floors 2; median
(2,2,2)), nor three up-literals. TRANSPLANT: a dropped ancestor
lowering a descendant's inset is read off the tree pass, not checked
on graphs with shared units.

PREDICTIONS. The kill FIRES (moderately), its breaking clauses C.C.C.
on one directed path (moderately); a D.D.D. break exists (weakly). The
forced parts median-closed (weakly), each forced unit one reader
(moderately).

KILL (frozen). "KILL raise-free parts not median-closed: N
(median-testable M)" over Y4, Y5 (median.populations) and X6 at the
441 cells; fires if N > 0, UNREADABLE if M < 10. Beside it, not the
kill: forced parts not closed / testable; per forced unit its readers,
its slack at the floors (naive minus the longest floor path through
it) and its least raise; on every RF set that fails, its prime 3-unit
clauses (raises.clauses, raises.kinds) by kind and on-path.

CONTROLS. K1 synthetic: F = {(1,2,2), (2,1,2), (2,2,1), (3,1,1)}
under floors (2,2,2): RF the first three, not closed, testable; the
forced part of {(3,1,2), (3,2,1)} is both, R0 = {0}. K2 X6 at the
three cells: 94 sets with RF empty (glue.py), each with a forced part.
K3 a plant: every closed, testable RF set of X6 at the three cells
with one new median that is a member deleted, read back not closed
(first frozen on Y5, which holds no testable RF set; moved, bar
planted >= 5, after that was read). K3 tests median.fast on real
RF sets, not the cut; K1 and K2 test the cut.

FINDINGS (copied from the runs below). K1 passes; K2 94 with RF
empty, 94 with a forced part; K3 8 of 8.
X6 at the 441 cells: below-naive sets 33,500, RF non-empty 33,398
(median-testable 68, not closed 8), RF empty 102 (each with a least
raise; forced parts testable 0). Y4: 420 below naive, every RF
non-empty, none testable. Y5: 2,107, RF empty 7, none testable.
KILL raise-free parts not median-closed: 8 (median-testable 68).
FIRES, as predicted. By cell: (11, 3, 8) 3, (6, 2, 4) 3, (4, 2, 2) 1,
(8, 2, 6) 1. Each of the 8 has one prime 3-unit clause, C.C.C. on one
directed path (8 of 8, as predicted); no D.D.D. break (the weak
prediction wrong). Forced units (readers, slack at the floors, least
raise), X6: (1, 1, 1) 4, (1, 2, 1) 38, (1, 4, 1) 8, (1, 6, 1) 1,
(2, 2, 1) 36, (2, 4, 1) 5, (3, 2, 1) 11; Y5: (1, 2, 1) 3, (2, 2, 1) 4.
So every forced unit's least raise is one, into a slack of at least
one, but 56
of the 110 forced units have two or three readers (the one-reader
prediction wrong; the six firsts printed before were one-reader).
The by-hand reason above argues the wrong way (a dropped ancestor
lowers an inset); the claim holds because graphs.floor is the least
delay alive at zero insets. Only 68 of the 33,398 RF sets are
median-testable, and the forced
parts never are, so their median-closure is unread.
Added after the kill was read (a scratch print of the 8 sets): 6, at
(6, 2, 4) and (11, 3, 8), end in one chain u, v = (z, u), x = (z, v)
at floors 2, T = naive - 1, RF the three single drops and {u, x}
together: the cap clause [L_u <= 1] or [L_v <= 1] or [L_x <= 1]
beside the conflicts not (u and v), not (v and x). At (8, 2, 6), v
reads u and a head, and RF adds {u, v}: the one conflict v and x. At
(4, 2, 2) the caps sit on u = (z, h), a = (y, (u, y)) with
y = (z, u), and the sink (u, a), the conflicts the sink beside
either. So below naive in this scope, drops alone
break two thresholds only by a path that must lose a drop at one of
three units, each able to take it alone, with a drop barred beside its
reader's.

RUN RECORD. Controls 3/3, 11.8 s, peak commit 20.4 MB (first run
failed K3 on Y5, 0 of 0, and K3 was moved). Rehearsed at 10 cells per
population, 20.8 s. Four shards, 166.4 to 177.1 s. Rerun from code/: controls
3/3, 11.8 s; four shards, 168.2 to 177.2 s, peak commit 20.2 MB,
every count above reproduced.
"""
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 raises  # noqa: E402
import sublattice as sl  # noqa: E402
from minclose import CELLS, 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 cut(P, floors):
    """(RF, R0, forced part): R0 and the forced part None where RF is
    not empty; R0 None where no least raised set exists."""
    up = lambda p: frozenset(u for u, (x, f) in enumerate(zip(p, floors))
                             if x > f)
    RF = {p for p in P if not up(p)}
    if RF:
        return RF, None, None
    Rs = {up(p) for p in P}
    R0 = frozenset.intersection(*Rs)
    if R0 not in Rs:
        return RF, None, None
    FP = {p for p in P if up(p) <= R0}
    return RF, R0, FP


def slack(d, nv):
    """naive minus the longest path at the floors through each unit."""
    arr = [0] * d.n
    for j in range(d.n):
        arr[j] = d.arrival(j, d.Ls[j], arr)
    tail = [0] * d.n
    for j in reversed(range(d.n)):
        tail[j] = max([d.Ls[r] + tail[r] for r in d.readers[j]] or [0])
    return [nv - arr[j] - tail[j] for j in range(d.n)]


def read_set(d, nv, P, T, tag, out, first):
    RF, R0, FP = cut(P, d.Ls)
    out[(tag, "below")] += 1
    if RF:
        out[(tag, "RF")] += 1
        ok, test, wit = median.fast(RF)
        out[(tag, "M")] += test
        if ok:
            return
        out[(tag, "N")] += 1
        out[(tag, "N", graphs.name(d.cell))] += 1
        pats = Counter()
        for cl in raises.clauses(RF):
            k = ".".join(sorted(x or "-" for x in raises.kinds(RF, d.Ls,
                                                                cl)))
            pats[(k, raises.on_path(d, cl[0]))] += 1
        for key, v in pats.items():
            out[(tag, "clause", key[0], key[1])] += v
        if len(first.setdefault("N " + tag, [])) < 6:
            first["N " + tag].append((graphs.name(d.cell), d.units, T, nv,
                                      d.Ls, wit, sorted(pats.items())))
        return
    out[(tag, "RF empty")] += 1
    if R0 is None:
        out[(tag, "no least raise")] += 1
        return
    ok, test, wit = median.fast(FP)
    out[(tag, "FP testable")] += test
    out[(tag, "FP not closed")] += not ok
    s = slack(d, nv)
    for u in R0:
        rise = min(p[u] for p in FP) - d.Ls[u]
        out[(tag, "forced", len(d.readers[u]), s[u], rise)] += 1
    if len(first.setdefault("F " + tag, [])) < 6:
        first["F " + tag].append((graphs.name(d.cell), d.units, T, nv, d.Ls,
                                  sorted(R0), len(FP), ok, wit))


def sweep_one(tag, cell, units, out, first):
    try:
        d = dags.Dag(cell, units)
    except AssertionError:
        out[(tag, "skipped")] += 1
        return
    nv = d.naive()
    rows = sl.alive_all(d, nv)
    least = min(r[0] for r in rows)
    for T in range(least, nv):
        read_set(d, nv, median.sets(d, nv, rows, T)["FD"], T, tag, out,
                 first)


def controls():
    print("K1..K3  the instrument")
    f = (2, 2, 2)
    RF, R0, FP = cut({(1, 2, 2), (2, 1, 2), (2, 2, 1), (3, 1, 1)}, f)
    a = (RF == {(1, 2, 2), (2, 1, 2), (2, 2, 1)}
         and median.fast(RF)[:2] == (False, True))
    RF2, R02, FP2 = cut({(3, 1, 2), (3, 2, 1)}, f)
    b = not RF2 and R02 == {0} and FP2 == {(3, 1, 2), (3, 2, 1)}
    check("K1 synthetic: the cross cut out and read open; R0 = {0}",
          a and b, f"{sorted(RF)}, {R02}, {FP2}")
    empty = withfp = 0
    for cell in [cell_named(nm) for nm in CELLS]:
        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):
                RF, R0, FP = cut(median.sets(d, nv, rows, T)["FD"], d.Ls)
                if not RF:
                    empty += 1
                    withfp += bool(FP)
    check("K2 X6 at the three cells: 94 with RF empty, each with a "
          "forced part", empty == 94 and withfp == 94, f"{empty}, {withfp}")
    planted = caught = 0
    for cell in [cell_named(nm) for nm in CELLS]:
        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):
                RF, _, _ = cut(median.sets(d, nv, rows, T)["FD"], d.Ls)
                if not RF or not all(median.fast(RF)[:2]):
                    continue
                m = None
                for p, q, r in itertools.combinations(sorted(RF), 3):
                    x = median.med(p, q, r)
                    if x not in (p, q, r) and x in RF:
                        m = x
                        break
                if m is None:
                    continue
                planted += 1
                caught += not median.fast(RF - {m})[0]
    check("K3 a deleted new median of a closed X6 RF set read open",
          planted >= 5 and caught == planted, f"{caught} of {planted}")


def jobs(shard, ncells):
    """(tag, cell, classes) for this shard: Y4 and X6 at the 441 cells,
    Y5 at the 34, each population's cells dealt by index mod 4."""
    X6 = ("X6", six(), list(graphs.cells()))
    pops = [X6] + [p for p in median.populations() if p[0] != "X6"]
    for tag, cls, cells_ in pops:
        if ncells:
            step = max(1, len(cells_) // ncells)
            cells_ = cells_[::step][:ncells]
        for i, cell in enumerate(cells_):
            if shard is None or i % 4 == shard:
                yield tag, cell, cls


def run(shard, ncells):
    out, first = Counter(), {}
    t0 = time.time()
    n = 0
    for tag, cell, cls in jobs(shard, ncells):
        n += 1
        for units in cls:
            sweep_one(tag, cell, units, out, first)
    return out, first, time.time() - t0, n


def dump(path, out, first, secs, n):
    rec = {"out": [[list(k), v] for k, v in out.items()],
           "first": first, "secs": secs, "cells": n}
    with open(path, "w", encoding="utf-8", newline="\n") as fh:
        json.dump(rec, fh, default=str)


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


def report(out, first):
    tags = sorted({k[0] for k in out})
    for tag in tags:
        g = lambda *k: out[(tag,) + k]
        print(f"\n  {tag}: below-naive sets {g('below')}, RF non-empty "
              f"{g('RF')} (testable {g('M')}, not closed {g('N')}); RF "
              f"empty {g('RF empty')} (no least raise "
              f"{g('no least raise')}; forced parts testable "
              f"{g('FP testable')}, not closed {g('FP not closed')}); "
              f"skipped {g('skipped')}")
        bc = {k[2]: v for k, v in out.items() if k[:2] == (tag, "N")
              and len(k) == 3}
        if bc:
            print(f"    RF not closed by cell {dict(sorted(bc.items()))}")
        cl = sorted(((k[2], k[3]), v) for k, v in out.items()
                    if k[:2] == (tag, "clause"))
        if cl:
            print("    prime 3-unit clauses of failing RF sets (kinds, "
                  "on one path): " + ", ".join(f"{a}{' path' if p else ''}"
                                               f" {v}" for (a, p), v in cl))
        fo = sorted((k[2:], v) for k, v in out.items()
                    if k[:2] == (tag, "forced"))
        if fo:
            print("    forced units (readers, slack, least raise): " +
                  ", ".join(f"{k} {v}" for k, v in fo))
        for x in first.get("N " + tag, [])[:3]:
            print(f"    first RF not closed: {x}")
        for x in first.get("F " + tag, [])[:2]:
            print(f"    first forced part: {x}")
    N = sum(out[(t, "N")] for t in tags)
    M = sum(out[(t, "M")] for t in tags)
    print(f"\nKILL raise-free parts not median-closed: {N} (median-testable"
          f" {M}){'  UNREADABLE' if M < 10 else ''}")


def main(argv):
    t0 = time.time()
    if argv and argv[0] == "--merge":
        out, first = merge(argv[1:])
        report(out, first)
        return 0
    if argv and argv[0] == "--shard":
        shard, ncells, path = int(argv[1]), int(argv[2]), argv[3]
        out, first, secs, n = run(None if shard < 0 else shard, ncells)
        dump(path, out, first, 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:]))