troupe-standup/wizard-work/volt/q143-check-sweep/block_rows.py

#!/usr/bin/env python3
"""block_rows.py -- extract ONE signed block from check_entry_baseline.tsv.

WHY THIS EXISTS
  The baseline is one file holding 145 signed readings in about a dozen blocks,
  each opened by a `# row(s) added by hand (Q<n> ...` comment. CHECK-q159's
  register leg counted its block with an awk program that turned the count on at
  the Q159 marker and never turned it off, so six later sections' rows were
  inside the 27 it reported and the leg read `rows=27 red=2` against a block of
  21. That guards nothing: the leg reds when some *other* lane signs a row, and
  it would stay green if Q159's own block lost rows and gained them back from a
  neighbour. The count had no boundary, so it was not a count of anything.

  This is the boundary, in one place, executed by the guard and by its own
  selftest. It is not a second gate beside the awk -- the awk is gone.

WHAT A BLOCK IS
  The marker line, then every comment line under it, then the tab-separated
  rows. The block ENDS at the first blank line or the first comment line that
  appears after at least one row has been read; a later marker therefore starts
  a new block and cannot be counted into this one. Trailing comments with no new
  marker also end it, so a note appended under a block never joins it.

WHAT IT GRADES
  Rows: a line with at least two tab-separated fields whose second field is an
  integer exit code. Anything else inside the block is not a row -- and a block
  row whose code is not an integer is not guessed at: the run is ungradable
  (rc 2), because a code no reader can read is not a reading.

EXIT CODES (tools/ contract)
  0 the block was read; 1 reserved (never returned: a readable block is a
  reading, red or not); 2 could not grade -- file unreadable, marker absent,
  marker matched more than one line, or a row carried a non-integer code.

USAGE
  python3 block_rows.py --file F --marker 'rows added by hand (Q159'
  python3 block_rows.py --file F --marker 'row added by hand (Q164' --paths
  python3 block_rows.py --file F --marker '...' --red
  python3 block_rows.py --file F --marker '...' --json
  python3 block_rows.py --selftest

  The default summary is one line: `rows=<n> red=<m>`.
"""
import argparse
import json
import os
import sys
import tempfile

SUMMARY_KEYS = ("marker", "rows", "red", "paths", "red_paths")


class Ungradable(Exception):
    """The block cannot be read as a block."""


def read_block(path, marker):
    try:
        with open(path, "r", encoding="utf-8", errors="replace") as fh:
            lines = fh.read().splitlines()
    except OSError as exc:
        raise Ungradable("cannot read %s: %s" % (path, exc))

    hits = [i for i, line in enumerate(lines)
            if line.startswith("#") and marker in line]
    if not hits:
        raise Ungradable("marker not found: %r" % marker)
    if len(hits) > 1:
        raise Ungradable("marker matched %d lines (want 1): %r"
                         % (len(hits), marker))

    i = hits[0]
    j = i + 1
    # Leading comment lines belong to the block's prose, not to its rows.
    while j < len(lines) and (lines[j].startswith("#") or not lines[j].strip()):
        if not lines[j].strip():
            # A blank line before any row is prose spacing; it does not open
            # the block, but it also cannot be a row.
            if j + 1 < len(lines) and lines[j + 1].startswith("#"):
                j += 1
                continue
        j += 1
    rows = []
    while j < len(lines):
        line = lines[j]
        if not line.strip() or line.startswith("#"):
            break
        fields = line.split("\t")
        if len(fields) < 2:
            raise Ungradable("row without an exit-code field: %r" % line[:80])
        code = fields[1].strip()
        try:
            rc = int(code)
        except ValueError:
            raise Ungradable("non-integer exit code %r on row %r"
                             % (code, fields[0][:80]))
        rows.append((fields[0], rc))
        j += 1
    return rows


def as_dict(marker, rows):
    return {
        "marker": marker,
        "rows": len(rows),
        "red": sum(1 for _, rc in rows if rc != 0),
        "paths": [p for p, _ in rows],
        "red_paths": [p for p, rc in rows if rc != 0],
    }


# --- selftest ---------------------------------------------------------------

FIXTURE = (
    "# path\trc\tnote\n"
    "# rows added by hand (Q1, fixture): a two-row block.\n"
    "# a second prose line, still inside the block\n"
    "CHECK-a\t0\tfine\n"
    "CHECK-b\t1\tknown red\n"
    "\n"
    "# rows added by hand (Q2, fixture): a three-row block that must NOT count.\n"
    "CHECK-c\t0\tfine\n"
    "CHECK-d\t0\tfine\n"
    "CHECK-e\t1\tknown red\n"
)

NAIVE_TO_EOF = sum(
    1 for line in FIXTURE.splitlines()
    if line and not line.startswith("#") and "\t" in line
    and line.split("\t")[1].strip().isdigit()
)


def _write(tmp, name, text):
    path = os.path.join(tmp, name)
    with open(path, "w", encoding="utf-8") as fh:
        fh.write(text)
    return path


def selftest():
    cells = []

    def cell(name, ok, detail):
        cells.append((name, ok, detail))

    tmp = tempfile.mkdtemp(prefix="block_rows.")
    try:
        good = _write(tmp, "fixture.tsv", FIXTURE)

        rows = read_block(good, "rows added by hand (Q1")
        cell("boundary", len(rows) == 2,
             "Q1 block reads %d row(s), want 2" % len(rows))

        # The falsifiability cell: the defect this tool replaces counted to EOF.
        cell("defect-visible", NAIVE_TO_EOF == 5 and len(rows) == 2,
             "marker-to-EOF reads %d, block reads %d -- the boundary moves "
             "the number" % (NAIVE_TO_EOF, len(rows)))

        cell("red-count", sum(1 for _, rc in rows if rc != 0) == 1
             and [p for p, rc in rows if rc != 0] == ["CHECK-b"],
             "red rows: %r" % ([p for p, rc in rows if rc != 0],))

        try:
            read_block(good, "rows added by hand (Q9")
            cell("missing-marker", False, "no marker -> rc 0 (want rc 2)")
        except Ungradable:
            cell("missing-marker", True, "no marker -> ungradable")

        dup = _write(tmp, "dup.tsv", FIXTURE
                     + "# rows added by hand (Q1, again)\nCHECK-f\t0\tx\n")
        try:
            read_block(dup, "rows added by hand (Q1")
            cell("ambiguous-marker", False, "two markers -> read one (want rc 2)")
        except Ungradable:
            cell("ambiguous-marker", True, "two markers -> ungradable")

        bad = _write(tmp, "bad.tsv", "# rows added by hand (Q1 x)\n"
                                   "CHECK-g\tnot-a-code\tx\n")
        try:
            read_block(bad, "rows added by hand (Q1")
            cell("bad-code", False, "non-integer code -> read (want rc 2)")
        except Ungradable:
            cell("bad-code", True, "non-integer code -> ungradable")

        # A trailing comment with no new marker ends the block too.
        trail = _write(tmp, "trail.tsv", "# rows added by hand (Q1 t)\n"
                                       "CHECK-h\t0\tx\n"
                                       "# a trailing note, no marker\n"
                                       "CHECK-i\t1\tx\n")
        n = len(read_block(trail, "rows added by hand (Q1"))
        cell("trailing-note", n == 1, "trailing comment ends the block: %d row(s)"
             % n)

        parsed = as_dict("m", rows)
        cell("json-keys", all(k in parsed for k in SUMMARY_KEYS)
             and parsed["red_paths"] == ["CHECK-b"],
             "summary keys: %s" % ",".join(sorted(parsed)))
    finally:
        for name in os.listdir(tmp):
            os.unlink(os.path.join(tmp, name))
        os.rmdir(tmp)

    bad_cells = [c for c in cells if not c[1]]
    for name, ok, detail in cells:
        print("  %-16s %s  %s" % (name, "PASS" if ok else "FAIL", detail))
    print("block_rows selftest: %d cell(s), %d failed, 0 skipped"
          % (len(cells), len(bad_cells)))
    return 1 if bad_cells else 0


# --- main -------------------------------------------------------------------

def main(argv=None):
    ap = argparse.ArgumentParser(description="extract one signed baseline block")
    ap.add_argument("--file")
    ap.add_argument("--marker")
    ap.add_argument("--paths", action="store_true", help="print the block's paths")
    ap.add_argument("--red", action="store_true", help="print the non-zero paths")
    ap.add_argument("--json", action="store_true")
    ap.add_argument("--selftest", action="store_true")
    args = ap.parse_args(argv)

    if args.selftest:
        return selftest()
    if not args.file or not args.marker:
        ap.error("--file and --marker are required (or --selftest)")

    try:
        rows = read_block(args.file, args.marker)
    except Ungradable as exc:
        print("block_rows: cannot grade -- %s" % exc, file=sys.stderr)
        return 2

    doc = as_dict(args.marker, rows)
    if args.json:
        print(json.dumps(doc, sort_keys=True))
    elif args.paths:
        for path in doc["paths"]:
            print(path)
    elif args.red:
        for path in doc["red_paths"]:
            print(path)
    else:
        print("rows=%d red=%d" % (doc["rows"], doc["red"]))
    return 0


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