"""
schaefer.py -- is a (z, h) triple ever a hard relation?

QUESTION. Below naive, is pricing a graph of units at one cell
NP-hard? A union of graphs sharing units has, at every T, F (and RF,
the raise-free part) exactly the points whose projection on each graph
lies in that graph's own (D1 below), so an instance is a CSP: gadget
graphs, their private units projected out, the shared units the
variables. A variable that can sit in any slot of any gadget has one
cone everywhere: the VARIABLE KIND is (z, h), h one shared head (a
head never drops: its floor is the least delay alive on fresh inputs).
On drop indicators (below floor or not) the CSP over a set Γ of
relations is NP-complete iff Γ lies inside none of Schaefer's six
tractable classes (0-valid, 1-valid, Horn, dual-Horn, bijunctive,
affine; Schaefer 1978, known, its text not opened), read on the whole
set. Γ is what one (cell, T) realizes over every graph there, on (z, h)
singles, pairs and triples.

BY HAND (frozen before code). D1: a shared unit is its inputs, so it
brings its cone; every path ends at a unit of some graph G_i and every
unit on it is an ancestor of that unit, so the path lies in G_i; each
unit's arrival and leftover read only its cone. So gluing by shared
units is the conjunction of the copies, at every T. A NAND of three is
Horn and 0-valid, an OR of three dual-Horn and 1-valid: each alone is
tractable, together monotone 3-SAT. All relations 0-valid means all
variables at floor extends everywhere. TRANSPLANTs: (a) RF only;
pricing reads F, where a raise can rescue a drop, so F's sets are read
beside; (b) a drop by two is folded into "dropped"; (c) the
(4, 2, 2) NAND's third literal is its sink, not a (z, h) unit.

PREDICTIONS. Every (z, h) relation in RF at 7 units is 0-valid
(moderately: private units absorb), so no Γ escapes and the kill
fires; F's sets escape neither (weakly).

KILL (frozen). "KILL (cell, T) whose (z, h) relations escape
Schaefer's six classes: N ((cell, T) read M, RF sets with three (z, h)
units K)" over X6 at the 441 cells and X7 at the four cells; fires if
N == 0 (the census offers no hard gadget set at this scale),
UNREADABLE if K < 10. Beside: the same over F; per (cell, T)
the classes Γ sits in; the triple relations not 0-valid.

CONTROLS. C0 by D1: two copies of the first (4, 2, 2) NAND graph
glued on h and 1, F and RF at T = 6, 7, 8 equal to the join of the
copies' own sets; a plant, copy B mapped onto the wrong units (its
2' onto 2), reads a join off the composite's. (Frozen with glue.join
as the reader and a deleted point as the plant; the deletion is caught
by construction once the join matches, and glue.join joins a set's own
projections, not the copies' own sets: replaced by glued, its plant
on the map, on review.) C1 the
classifier: NAND3 {0-valid, Horn}; OR3 {1-valid, dual-Horn}; 1-in-3
none; x + y + z + w = 1 mod 2 {affine}; the set {NAND3, OR3} none; a
planted 1-in-3 with (0, 0, 0) added read back as 0-valid. (Frozen with
x + y + z = 1 as the affine-only case; by hand before code it holds
(1, 1, 1), so it is 1-valid too: replaced by the four-variable parity.)
C2 the RF of pairwise.py's first fire projected on (1, 2, 6) reads
NAND3 exactly. (Frozen so; the first run read {001, 011, 101, 110},
and by hand the frozen answer forgot the paths: at T = 7 the paths
h, 1, 3, 4, 5, 6 and h, 2, 3, 4, 5, 6 floor at 8, and with h, 3, 4, 5
never dropping they force d1 or d6 and d2 or d6, beside not all three.
So the relation is d6 = NAND(d1, d2), a gate; C2 now asks that, and
that h, 3, 4, 5 sit at their floors in every point.)

FINDINGS (copied from the runs below). C0 F and RF equal the join at
T = 6, 7, 8 (T = 7: F 20, RF 8; T = 8: F 568, RF 25; T = 6 empty), the
plant read; C1 NAND3 [0-valid, Horn], OR3 [1-valid, dual-Horn], 1-in-3
none, parity 4 [affine], the pair none, the plant [0-valid, Horn,
bijunctive]; C2 {001, 011, 101, 110}, h, 3, 4, 5 at floor, classes
none: the gate d6 = NAND(d1, d2) escapes all six alone.
RF: (cell, T) read 807; Γ in {0-valid, Horn, dual-Horn, bijunctive,
affine} 721, {1-valid, Horn, dual-Horn, bijunctive, affine} 85,
{1-valid, Horn, dual-Horn, bijunctive} 1; triple relations not
0-valid 108 (summed over (cell, T), not distinct). F: read 808; 718,
84, all six 4, {1-valid, dual-Horn, bijunctive} 1, {1-valid, Horn,
dual-Horn, bijunctive} 1; not 0-valid 108.
KILL (cell, T) whose (z, h) relations escape Schaefer's six classes: 0
((cell, T) read 807, RF sets with three (z, h) units 1424). FIRES, as
predicted; F escapes neither, as weakly predicted. Every RF Γ lies in
Horn, dual-Horn and bijunctive at once, and in 0-valid or 1-valid;
every F Γ in 0-valid or 1-valid (one, {1-valid, dual-Horn,
bijunctive}, not Horn): so over X6 at the 441 cells and X7 at the
four, relations of arity at most 3, each gadget read below its own
naive, a (z, h) variable layer carries no hard gadget set. Unread:
groups of four or more (z, h) units on one head as one relation, and
gadgets read at or above their own naive. The prediction's reason
was half wrong: 86 of 807 are not 0-valid (the (z, h) units forced
down), yet then all-dropped extends on indicators (a drop by one in
one graph and by two in another unread). What C2 found instead is a
gate on non-interchangeable units, the sink's drop a NAND of the two (z, h)
units' drops: hardness, if it is there, is a question of whether such
gates compose.

RUN RECORD. Controls 5/5, 0.1 s, peak commit 19.4 MB. Rehearsed at
one cell per population, shard 0, 13.4 s. Four shards, 199.3 to
207.9 s, peak commit 31.0 MB. Rerun from code/: controls 5/5,
0.2 s, 18.4 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

SIX = ("0-valid", "1-valid", "Horn", "dual-Horn", "bijunctive", "affine")
NAND1 = ((None, None), (None, 0), (None, 0), (1, 2), (1, 3), (1, 4),
         (2, 5))
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 classes_of(R):
    """Schaefer's classes a non-empty Boolean relation lies in."""
    R = set(R)
    k = len(next(iter(R)))
    op2 = lambda f: all(tuple(f(x, y) for x, y in zip(a, b)) in R
                        for a in R for b in R)
    op3 = lambda f: all(tuple(f(x, y, z) for x, y, z in zip(a, b, c)) in R
                        for a in R for b in R for c in R)
    out = set()
    if (0,) * k in R:
        out.add("0-valid")
    if (1,) * k in R:
        out.add("1-valid")
    if op2(lambda x, y: x & y):
        out.add("Horn")
    if op2(lambda x, y: x | y):
        out.add("dual-Horn")
    if op3(lambda x, y, z: (x & y) | (y & z) | (x & z)):
        out.add("bijunctive")
    if op3(lambda x, y, z: x ^ y ^ z):
        out.add("affine")
    return out


def set_classes(G):
    out = set(SIX)
    for R in G:
        out &= classes_of(R)
    return out


def zh(d):
    """{h: the (z, h) units reading head h}."""
    heads = [j for j, u in enumerate(d.units) if tuple(u) == (None, None)]
    return {h: [j for j, u in enumerate(d.units)
                if tuple(u) in ((None, h), (h, None))] for h in heads}


def relations(P, floors, groups):
    """Every drop relation of P on (z, h) singles, pairs and triples."""
    out = set()
    for V in groups.values():
        for r in (1, 2, 3):
            for U in itertools.combinations(V, r):
                out.add(frozenset(tuple(int(p[u] < floors[u]) for u in U)
                                  for p in P))
    return out


def sets_at(d, T, rows=None):
    nv = d.naive()
    rows = rows if rows is not None else sl.alive_all(d, max(T, nv))
    F = median.sets(d, nv, rows, T)["FD"]
    return F, cut(F, d.Ls)[0] if F else set()


def glued(PA, A, B, n):
    """The join of two copies' sets, each copy's own set PA, copy A on
    composite units A and copy B on B (same unit order)."""
    out = set()
    for a in PA:
        for b in PA:
            p = [None] * n
            ok = True
            for U, v in ((A, a), (B, b)):
                for u, x in zip(U, v):
                    ok &= p[u] is None or p[u] == x
                    p[u] = x
            if ok:
                out.add(tuple(p))
    return out


def controls():
    print("C0..C2  the instrument")
    cell = cell_named("(4, 2, 2)")
    comp = list(NAND1) + [(None, 0), (1, 7), (1, 8), (1, 9), (7, 10)]
    A = list(range(7))
    B = [0, 1, 7, 8, 9, 10, 11]
    dA, dG = dags.Dag(cell, NAND1), dags.Dag(cell, comp)
    ok, detail = True, []
    for T in (6, 7, 8):
        FA, RA = sets_at(dA, T)
        FG, RG = sets_at(dG, T)
        for arm, PA, PG in (("F", FA, FG), ("RF", RA, RG)):
            J = glued(PA, A, B, dG.n)
            ok &= J == PG
            detail.append(f"T={T} {arm} {len(PG)}/{len(J)}")
    FA, _ = sets_at(dA, 7)
    FG, _ = sets_at(dG, 7)
    wrong = [0, 1, 2, 8, 9, 10, 11]
    ok_plant = glued(FA, A, wrong, dG.n) != FG
    check("C0 glued on h and 1: F, RF = the copies' join, T = 6, 7, 8",
          ok, ", ".join(detail))
    check("C0 plant: copy B mapped onto 2 reads a join off the composite's",
          ok_plant)
    NAND3 = {p for p in itertools.product((0, 1), repeat=3) if sum(p) < 3}
    OR3 = {p for p in itertools.product((0, 1), repeat=3) if sum(p) > 0}
    ONE3 = {(1, 0, 0), (0, 1, 0), (0, 0, 1)}
    PAR4 = {p for p in itertools.product((0, 1), repeat=4) if sum(p) % 2}
    hand = ((NAND3, {"0-valid", "Horn"}), (OR3, {"1-valid", "dual-Horn"}),
            (ONE3, set()), (PAR4, {"affine"}))
    good = all(classes_of(R) == want for R, want in hand)
    good &= set_classes([NAND3, OR3]) == set()
    planted = classes_of(ONE3 | {(0, 0, 0)})
    check("C1 the classifier on NAND3, OR3, 1-in-3, parity 4, the pair",
          good, " ".join(f"{sorted(classes_of(R))}" for R, _ in hand))
    check("C1 plant: 1-in-3 with (0, 0, 0) reads 0-valid",
          "0-valid" in planted, f"{sorted(planted)}")
    _, RF = sets_at(dA, 7)
    R = {tuple(int(p[u] < dA.Ls[u]) for u in (1, 2, 6)) for p in RF}
    still = all(p[u] == dA.Ls[u] for p in RF for u in (0, 3, 4, 5))
    gate = {(a, b, 1 - (a & b)) for a in (0, 1) for b in (0, 1)}
    check("C2 the first fire's RF on (1, 2, 6): h, 3, 4, 5 never drop, "
          "so d6 = NAND(d1, d2)", still and R == gate,
          f"{sorted(R)}, h 3 4 5 at floor {still}, "
          f"classes {sorted(classes_of(R))}")


def sweep_one(cell, units, gam, out):
    try:
        d = dags.Dag(cell, units)
    except AssertionError:
        out["skipped"] += 1
        return
    groups = {h: V for h, V in zh(d).items() if V}
    if not groups:
        return
    three = any(len(V) >= 3 for V in groups.values())
    nv = d.naive()
    rows = sl.alive_all(d, nv)
    least = min(r[0] for r in rows)
    name = graphs.name(cell)
    for T in range(least, nv):
        F = median.sets(d, nv, rows, T)["FD"]
        if not F:
            continue
        RF = cut(F, d.Ls)[0]
        key = f"{name}|{T}"
        g = gam.setdefault(key, {"RF": set(), "F": set()})
        g["F"] |= relations(F, d.Ls, groups)
        if RF:
            g["RF"] |= relations(RF, d.Ls, groups)
            out["K"] += three


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


def dump(path, gam, out, secs, n):
    enc = {k: {a: [sorted(R) for R in v[a]] for a in v}
           for k, v in gam.items()}
    with open(path, "w", encoding="utf-8", newline="\n") as fh:
        json.dump({"gam": enc, "out": dict(out), "secs": secs,
                   "cells": n}, fh)


def merge(paths):
    gam, out = {}, Counter()
    for p in paths:
        with open(p, encoding="utf-8") as fh:
            rec = json.load(fh)
        for k, v in rec["gam"].items():
            g = gam.setdefault(k, {"RF": set(), "F": set()})
            for a in v:
                g[a] |= {frozenset(tuple(t) for t in R) for R in v[a]}
        out.update(rec["out"])
        print(f"  {os.path.basename(p)}: {rec['cells']} (pop, cell) jobs, "
              f"{rec['secs']:.1f} s")
    return gam, out


def report(gam, out):
    for arm in ("RF", "F"):
        keys = [k for k in gam if gam[k][arm]]
        by = Counter()
        esc, not0 = [], 0
        for k in sorted(keys):
            G = gam[k][arm]
            sc = set_classes(G)
            by[tuple(c for c in SIX if c in sc)] += 1
            not0 += sum(1 for R in G if len(next(iter(R))) == 3
                        and (0, 0, 0) not in R)
            if not sc:
                esc.append(k)
        print(f"\n  {arm}: (cell, T) read {len(keys)}; by classes G sits "
              f"in {dict(by)}; triple relations not 0-valid {not0}")
        for k in esc[:5]:
            print(f"    escapes: {k}, {len(gam[k][arm])} relations")
        tail = "  UNREADABLE" if out["K"] < 10 else ""
        label = "" if arm == "RF" else " (F, beside)"
        print(f"KILL (cell, T) whose (z, h) relations escape Schaefer's six "
              f"classes{label}: {len(esc)} ((cell, T) read {len(keys)}, RF "
              f"sets with three (z, h) units {out['K']}){tail}")
    print(f"  skipped {out['skipped']}")


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]
        gam, out, secs, n = run(None if shard < 0 else shard, ncells)
        dump(path, gam, out, 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:]))