choice.py

"""
choice.py -- is a gain below naive ever a choice of raises?

QUESTION. A delay set F (every unit's delays alive at latency at most
T, brute's window 0..L*+3) at T below naive with no RAISE-FREE point
(every unit at or below its floor) is a GAIN NEEDING A RAISE. glue.py
found 101 at Y4, Y5 and X6 (X6 at three cells), each with one least
raised set over the whole set. Is any gain a CHOICE: two minimal
raised sets buying the same drop?

KEYING. Per point p, D(p) the units below their floors, R(p) those
above. F's points grouped by D(p) exactly; a CHOICE at D when the
minimal elements of {R(p) : D(p) = D} number two or more. A set counts
once if it holds a choice at any D. Beside it the SET-LEVEL count
(minimal R over all of F, glue.py's), which also reads two gains each
forced as a choice: printed, not the kill.

ARMS. HAND: h = (z, z), c = (z, h), a chain x1 = (z, z), x2 = (z, x1)
[, x3 = (z, x2)], r = (c, x_last), 5 and 6 units, a tree, at the 441
cells. X6: minclose.six() at the 441 cells, sharded by cell index mod
4, rehearsed at 10 cells first.

BY HAND (frozen before code). v_r rises in each inset; c's inset rises
in L_c and, through h's inset, in L_h. Where x's path is longer than
c's, c has slack, so either raise alone fits; if either alone lifts r's
leftover over 0 at L*_r - 1, the drop {r} has minimal raised sets {h}
and {c}. A unit's two inputs cannot both be in slack while it is
critical, so a raise OR must be two units on one input's side.
TRANSPLANT: a raise lifts the inset where the room does not bind.

PREDICTIONS. HAND gains with a choice at some cell (weakly); X6 at 441
cells holds a choice (weakly); every choice found is two ancestors on
one input's side (moderately).

KILL (frozen). "KILL gains below naive with a choice of raises: N
(gains needing a raise M)" over HAND and X6 together, N per drop;
fires if N == 0, UNREADABLE if M < 10.

CONTROLS. K1 synthetic, floors (2, 2, 2): {(3, 2, 1), (2, 3, 1)} a
choice at D = {2}; {(3, 2, 1), (3, 3, 1)} forced; {(3, 2, 1),
(2, 3, 2)} no choice per drop (D differ) and a choice set-level. K2 X6
at the three cells reproduces glue.py's 94 gains needing a raise, all
forced set-level. K3 a plant: each K2 set with a point added, its
forced raise moved to another unit at its floor, same D, read back as
a choice.

FINDINGS (copied from the runs below). K1 passes; K2 94 gains needing
a raise at the three cells, 94 forced set-level (glue.py's X6 count);
K3 94 of 94 moved raises read back as a choice (by construction, it
tests the grouping on real sets). K4 (added at the audit: a found
choice's key was a tuple, which json cannot dump, so --shard would have
crashed on the first choice) passes.
HAND: below-naive sets 26 (hand2) and 33 (hand3), gains needing a
raise 0 at both (hand2's chain matches h, c, so c has no slack there;
hand3 is the shape the argument wants): the shape gains, always with a point that raises
nothing, so its drop is bought by drops alone.
X6 at the 441 cells: below-naive sets 33,500, gains needing a raise
102, choice per drop 0, set-level choice 0.
KILL gains below naive with a choice of raises: 0 (gains needing a
raise 102). FIRES; both predictions of a choice were wrong, and the
third (a choice is two ancestors on one side) is unread.
Added after the kill was read: the 102 by cell, (6, 2, 4) 72,
(35, 8, 27) 20, (4, 2, 2) 2, and 8 off the three cells, (9, 2, 7) 2,
(18, 3, 15) 2, (36, 5, 31) 2, (10, 5, 5) 1, (6, 3, 3) 1; by naive - T,
99 at 1 and 3 at 2 (each first printed raises one unit).
The first at (10, 5, 5): h, a = (z, h), b = (z, a), c = (z, b),
(h, a), (c, (h, a)) at T = 5 against 6, floors (2, 1, 1, 1, 1, 1), the
unit (h, a), read only by the sink, raised. So over X6 at the 441
cells a gain below naive is decided by drops alone or by a forced
raise, at 8 cells. Keyed per exact drop set, a choice layered over a
forced set, or two raised sets buying one unit's drop beside
different further drops, is unread.

RUN RECORD. Controls 3/3 (before K4), 5.8 s, peak commit 19.2 MB. HAND 1.8 s.
X6 rehearsed at 10 cells, 19.0 s (2 gains needing a raise); then four
shards by cell mod 4, 191.5 to 203.6 s each, peak commit 20.3 MB;
rerun with the by-cell print, 192.2 to 200.5 s, the same counts. In
code/: four shards, 192.5 to 202.8 s, every X6 count above reproduced.
"""
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 minclose import CELLS, cell_named, six  # noqa: E402

CHECKS = []
HAND = (
    ("hand2", ((None, None), (None, 0), (None, None), (None, 2), (1, 3))),
    ("hand3", ((None, None), (None, 0), (None, None), (None, 2), (None, 3),
               (1, 4))),
)


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 minimal(sets_):
    sets_ = set(sets_)
    return sorted((r for r in sets_ if not any(q < r for q in sets_)),
                  key=lambda r: sorted(r))


def read(P, floors):
    """(raise_free, per-drop choices {D: minimal Rs} with two or more,
    set-level minimal Rs)."""
    by = {}
    allr = set()
    free = False
    for p in P:
        D = frozenset(u for u, (x, f) in enumerate(zip(p, floors)) if x < f)
        R = frozenset(u for u, (x, f) in enumerate(zip(p, floors)) if x > f)
        free |= not R
        by.setdefault(D, set()).add(R)
        allr.add(R)
    ch = {D: minimal(Rs) for D, Rs in by.items() if len(minimal(Rs)) > 1}
    return free, ch, minimal(allr)


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):
        P = median.sets(d, nv, rows, T)["FD"]
        out[(tag, "below")] += 1
        free, ch, mins = read(P, d.Ls)
        if free:
            continue
        out[(tag, "M")] += 1
        # added after the kill was read: where they sit, and the first
        out[(tag, "M", graphs.name(cell))] += 1
        out[(tag, "M gap", nv - T)] += 1
        if len(first.setdefault("M " + tag, [])) < 12:
            first["M " + tag].append((graphs.name(cell), units, T, nv, d.Ls,
                                      [sorted(r) for r in mins]))
        if len(mins) > 1:
            out[(tag, "set-level choice")] += 1
        if ch:
            out[(tag, "N")] += 1
            out[(tag, "N", graphs.name(cell))] += 1
            if len(first.setdefault(tag, [])) < 6:
                first[tag].append((graphs.name(cell), units, T, nv, d.Ls,
                                   {str(sorted(D)): [sorted(r) for r in Rs]
                                    for D, Rs in ch.items()}))


def controls():
    print("K1..K4  the instrument")
    f = (2, 2, 2)
    a = read({(3, 2, 1), (2, 3, 1)}, f)
    b = read({(3, 2, 1), (3, 3, 1)}, f)
    c = read({(3, 2, 1), (2, 3, 2)}, f)
    check("K1 synthetic: a choice, forced, set-level only",
          (not a[0]) and set(a[1]) == {frozenset({2})}
          and not b[1] and len(b[2]) == 1
          and not c[1] and len(c[2]) == 2,
          f"{a[1]}, {b[1]}/{b[2]}, {c[1]}/{c[2]}")
    M = forced = 0
    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):
                P = median.sets(d, nv, rows, T)["FD"]
                free, ch, mins = read(P, d.Ls)
                if free:
                    continue
                M += 1
                forced += len(mins) == 1
                # K3: move the forced raise of one point to a unit at
                # its floor outside D, same D
                p = next(iter(sorted(P)))
                Dp = {u for u in range(d.n) if p[u] < d.Ls[u]}
                Rp = [u for u in range(d.n) if p[u] > d.Ls[u]]
                vs = [v for v in range(d.n) if p[v] == d.Ls[v]]
                if not vs:
                    continue
                q = list(p)
                for u in Rp:
                    q[u] = d.Ls[u]
                q[vs[0]] += 1
                q = tuple(q)
                if q in P:
                    continue
                planted += 1
                _, ch2, _ = read(P | {q}, d.Ls)
                caught += frozenset(Dp) in ch2
    check("K2 X6 at the three cells: 94 gains needing a raise, all "
          "forced set-level", M == 94 and forced == 94, f"{M}, {forced}")
    check("K3 a moved raise read back as a choice", planted >= 10
          and caught == planted, f"{caught} of {planted}")
    # K4 (added at the audit): a shard's record with a choice in it
    # survives dump and merge
    import tempfile
    out, first = Counter(), {}
    d = dags.Dag(cell_named("(6, 2, 4)"), six()[0])
    out[("X6", "N")] += 1
    first["X6"] = [("(6, 2, 4)", six()[0], 5, 6, d.Ls,
                    {str(sorted(D)): [sorted(r) for r in Rs]
                     for D, Rs in a[1].items()})]
    path = os.path.join(tempfile.mkdtemp(), "k4.json")
    try:
        dump(path, out, first, 0.0, 1)
        back, bf = merge([path])
        ok = back[("X6", "N")] == 1 and len(bf["X6"]) == 1
    except TypeError as e:
        ok = False
        print(f"    K4: {e}")
    check("K4 a record holding a choice dumps and merges", ok)


def run(arm, shard, ncells):
    cells_ = list(graphs.cells())
    if ncells:
        step = max(1, len(cells_) // ncells)
        cells_ = cells_[::step][:ncells]
    out, first = Counter(), {}
    t0 = time.time()
    nrun = 0
    for i, cell in enumerate(cells_):
        if shard is not None and i % 4 != shard:
            continue
        nrun += 1
        if arm == "hand":
            for tag, units in HAND:
                sweep_one(tag, cell, units, out, first)
        else:
            for units in six():
                sweep_one("X6", cell, units, out, first)
    return out, first, time.time() - t0, nrun


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']} cells, "
              f"{rec['secs']:.1f} s")
    return out, first


def report(out, first):
    tags = sorted({k[0] for k in out})
    for tag in tags:
        print(f"  {tag}: below-naive sets {out[(tag, 'below')]}, gains "
              f"needing a raise {out[(tag, 'M')]}, choice per drop "
              f"{out[(tag, 'N')]}, set-level choice "
              f"{out[(tag, 'set-level choice')]}, skipped "
              f"{out[(tag, 'skipped')]}")
        cells_ = {k[2]: v for k, v in out.items()
                  if k[0] == tag and k[1] == "N" and len(k) == 3}
        if cells_:
            print(f"    choices by cell {cells_}")
        ms = {k[2]: v for k, v in out.items()
              if k[0] == tag and k[1] == "M" and len(k) == 3}
        gp = {k[2]: v for k, v in out.items()
              if k[0] == tag and k[1] == "M gap"}
        if ms:
            print(f"    gains needing a raise by cell {dict(sorted(ms.items()))}"
                  f"; by naive - T {dict(sorted(gp.items()))}")
        seen = set()
        for x in first.get("M " + tag, []):
            if x[0] not in seen:
                seen.add(x[0])
                print(f"    first needing a raise at {x[0]}: {x[1:]}")
        for x in first.get(tag, [])[:6]:
            print(f"    choice {x}")
    N = sum(out[(t, "N")] for t in tags)
    M = sum(out[(t, "M")] for t in tags)
    print(f"KILL gains below naive with a choice of raises: {N} (gains "
          f"needing a raise {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":
        arm, shard, ncells, path = argv[1], int(argv[2]), int(argv[3]), \
            argv[4]
        out, first, secs, n = run(arm, None if shard < 0 else shard, ncells)
        dump(path, out, first, secs, n)
        print(f"{arm} 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:]))