splits.py

"""
splits.py -- where sharing costs, does the unfolding split a unit's
copies with one raised above its floor?

QUESTION. dags.py: over the shared classes of 2..5 units at the 34
cells where S or Q gains, sharing costs (BRUTE > UNF(3)) only where
the unfolding is raise-priced. Finer: does every least assignment of
the unfolding raise a copy of a unit whose copies it splits?

THE DESIGN (frozen before code). The FOREST of a DAG: every sink's
unfolding as one Dag, each unit a COPY of a DAG unit (its origin). In
an assignment of it a unit is SPLIT where its copies sit at two
delays, by a RAISE where one copy sits above its floor (copies share
tau, so one floor). Property: in a pair that costs every least forest
assignment splits (unsplit, read once per unit it is a DAG assignment
of latency UNF(3) in brute's window). A HEAD (a unit over two fresh
streams) reads inset 0, so no copy of it sits below its floor and any
split of a head is a raise split: the kill can fire only on an
assignment that splits no head, counted as its population.

CONTROLS (read first). WC the (6, 2, 4) witness h, c = (z, h),
r = (h, c): least 5, h's copy read by c at 2, the one read by r at 3
or 4 (as frozen "3 only"; the hand answer missed 4, since r at 1 reads
c at 4 and h at 4 arrives no later, and was refrozen in rehearsal
before the kill was read). WA the forest's brute = UNF(3), WB every
least forest assignment splits, at every pair that costs; WD each pair
read once (its planted break, X5's rows at (6, 2, 4) and (35, 8, 27)
read twice, prints 130 rows, 66 distinct, and fails).

KILL (frozen): "least, no raise split" > 0 over every pair that costs
among dags.py's X5 (classes of 2..5 units at the 34 cells; X4's 8 cost
pairs, at (6, 2, 4) and (35, 8, 27), lie among them). Predicted 0, in
part in-sample: the two (4, 2, 2) pairs were read before the freeze.

FINDINGS (copied from the run's prints). WC green; WD 66 rows, 66
distinct; WA 66 pairs, 0 off; WB 1,607 assignments, each split. The
kill does not fire: no least assignment lacks a raise split. Its
population is 2: all but 2 of the 1,607 split a head, a raise by
construction. A second print, frozen before its run, lists each least
assignment that splits no head (pair, delays, split origins, those
split by a raise), predicted both at one pair, the split unit the one
non-head unit read twice, its raised copy the one its deeper reader
reads. It prints both at (4, 2, 2), at two pairs, units
((None, None), (None, None), (0, 1), (0, 2), (2, 3)) and
((None, None), (None, 0), (None, 1), (0, 2), (2, 3)); at each the split
unit is 2, the one non-head unit read twice, its copy read by
r = (2, 3) at delay 2 over its floor 1, the copy read by c = (0, 2) at
1. The prediction missed on "one pair" and held on the rest. Both end
as the (6, 2, 4) witness h, c = (z, h), r = (h, c) does, c = (0, 2)
and r = (2, 3) reading the split unit h = 2, r's copy raised; but h is
(0, 1) over two heads at one pair and (None, 1) over the chain unit
(None, 0) at the other, and c's other input is head 0, in h's cone,
not a fresh stream. So every least assignment of the 66 pairs, in
brute's window, splits some unit by a raise: a head at 1,605, the one
non-head unit read twice at the other 2. WD guards only against rows
read twice, not a drift of the cells.

RUN RECORD. main: 61.5 s wall, peak commit 8.1 MB (an earlier run
without the second print: 60.4 s, X5 54.6 s, W 2.8 s). The kill and
the head-split counts are prints, read at the run, not checks.
"""
import sys
import time
from collections import Counter

import chains
import dags
import graphs
from dags import Dag, brute, check, name, unfold_least


def forest(d):
    """Every sink's unfolding as one Dag, and each copy's origin in d."""
    units, origin = [], []

    def walk(j):
        ins = tuple(None if x is None else walk(x) for x in d.units[j])
        units.append(ins)
        origin.append(j)
        return len(units) - 1

    for s in d.sinks():
        walk(s)
    return Dag(d.cell, units), origin


def least_all(d, lat):
    """Every assignment in brute's window, alive, of latency <= lat."""
    left, arr, Ls = [None] * d.n, [0] * d.n, [0] * d.n

    def go(j):
        if j == d.n:
            yield list(Ls)
            return
        for L in range(0, d.Ls[j] + 4):
            a = d.arrival(j, L, arr)
            if a > lat:
                break
            r = d.leftover(j, L, left)
            if r is None:
                continue
            left[j], arr[j], Ls[j] = r, a, L
            yield from go(j + 1)
        left[j], arr[j] = None, 0

    yield from go(0)


def splits(f, origin, Ls):
    """The split origins of one forest assignment, and those split by a
    raise (a copy above its floor)."""
    at = {}
    for k, j in enumerate(origin):
        at.setdefault(j, []).append(k)
    split = [j for j, ks in at.items() if len({Ls[k] for k in ks}) > 1]
    up = [j for j in split if any(Ls[k] > f.Ls[k] for k in at[j])]
    return split, up


def raise_splits(rows):
    """Every least forest assignment of each pair: counts, and WA's
    misses."""
    n, off, ex, deep = Counter(), 0, [], []
    for cell, units in rows:
        d = Dag(cell, units)
        heads = {j for j, u in enumerate(d.units) if u == (None, None)}
        f, origin = forest(d)
        lat = brute(f)[0]
        off += lat != unfold_least(d)
        for Ls in least_all(f, lat):
            split, up = splits(f, origin, Ls)
            n["least"] += 1
            n["least, no split"] += not split
            n["least, no head split"] += not (set(split) & heads)
            n["least, no raise split"] += not up
            if not up and len(ex) < 4:
                ex.append((name(cell), units, Ls))
            if not set(split) & heads:
                deep.append((name(cell), units, Ls, split, up))
        n["pairs"] += 1
    return n, off, ex, deep


def witness_splits():
    """WC: (6, 2, 4)'s h, c = (z, h), r = (h, c), each least forest
    assignment's delays for h's copies (read by c, by r)."""
    d = Dag(chains.ms.Cell(6, 2, 4), [(None, None), (None, 0), (0, 1)])
    f, origin = forest(d)
    lat = brute(f)[0]
    reader = {x: origin[m] for m, u in enumerate(f.units)
              for x in u if x is not None}
    hs = [k for k, j in enumerate(origin) if j == 0]
    return lat, sorted({tuple(sorted((reader[k], Ls[k]) for k in hs))
                        for Ls in least_all(f, lat)})


def main():
    t0 = time.time()
    got = witness_splits()
    check("WC (6, 2, 4) witness: least 5, h's copies at 2 (read by c), "
          "3 or 4 (by r)", got == (5, [((1, 2), (2, 3)), ((1, 2), (2, 4))]),
          f"{got}")
    if not all(dags.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 name(c) in ("(6, 2, 4)", "(35, 8, 27)")
          or gain(c, "S") or gain(c, "Q")]
    rows = dags.sweep("X5", 5, sq, extra=True)
    keys = [(name(c), u) for c, u in rows]
    check("WD each pair that costs read once", len(set(keys)) == len(keys),
          f"{len(keys)} rows, {len(set(keys))} distinct")
    t1 = time.time()
    n, off, ex, deep = raise_splits(rows)
    print(f"\nW  least forest assignments of the {n['pairs']} pairs that "
          f"cost, {time.time() - t1:.1f} s")
    check("WA forest brute = UNF(3) at every pair", off == 0,
          f"{n['pairs']} pairs, {off} off")
    check("WB every least forest assignment splits",
          n["least, no split"] == 0 and n["least"] > 0,
          f"{n['least']} assignments")
    if not all(dags.CHECKS):
        print("\nCONTROL FAILED: the kill is not read.")
        return 1
    for k in ("least", "least, no head split", "least, no raise split"):
        print(f"    {k:24s} {n[k]}")
    for e in ex:
        print(f"    KILL {e}")
    for cell, units, Ls, split, up in deep:
        print(f"    no head split: {cell} units {units} delays {Ls} "
              f"split {split} by a raise {up}")
    print(f"\n{time.time() - t0:.1f} s")
    return 0


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