Jump to content

The Principia Machina/TPC-PY/check draft.py

From The Invariance Itself
# -*- coding: utf-8 -*-
"""
check_draft.py --- check a draft TFK on your own machine with TPC-PY
====================================================================
Advisory. TPX of record decides at the new-UWC gate; this tool runs
the same TPC functions ahead of it so an author sees every refusal before
submitting. Standard library only; Python 3.8+; written for Windows and
Linux. Version 0.6.0 was run on Linux under Python 3.8.20, 3.10, 3.11,
3.12 and 3.13 over the 27 examples in examples/, the 86 research drafts
of the case table and the 32 cases of the fixtures (cd_fixtures/, beside
tpc/ in the build tree: S54's and F-HF2-1's), the output the same in each
(increment 12, 2026-09-29); the Windows run is owed.

    python3 check_draft.py [--catalogue CATALOGUE.json] my_draft.json
                                              (Windows: py check_draft.py ...)

The draft is one JSON object with up to three zones and, where the author
chose ENVIRONMENT MODULES in the interview, the claims:
    {"TPK": {...}, "TrM": {...}, "TMS": {...}, "MODULES_USED": [...]}
MODULES_USED is a list of claims, each {"ID", "VERSION", "DIGEST",
"BIND", "VALUES", "ANSWERS"}: TPS 5.22.8's MODULES USED (ID, VERSION,
digest, bindings, values and answers), in tpm_module_spec 0.2.0 5.1's
form; the TPK then holds the author's own entries, never an expansion
(TPS 5.22.7; AC-4.51). The catalogue is one JSON object {"LINES": [...],
"MODULES": [...]}: the MODULE DECLARATIONS in chain order and the text
of every version they declare (tpm_module_spec 5.3 and 5.2), as the
modules page publishes them. Missing zones are reported and skipped.
What it does, in the order TPX does it (TPS 5.5 and 5.22; tpm_zone_schemas
0.4.0 with schema 0.5.0; tpm_units 0.2.1; tpm_kernel_tests 0.3.0;
tpm_increment_8 0.1.0, the declared form; D176.1 and D177.1, the ROLE
fields, the TPK PARAMETERS part with D_VARIABLES, the TMS switch
STRUCTURE_CHECK and the D variable; D178.1, the ENVIRONMENT MODULES):
    1. reads the file as UTF-8 and refuses what is not RFC 8259 JSON
       (NaN, Infinity and -Infinity included), duplicate object keys (two
       readers could otherwise disagree on which value counts) and nesting
       deeper than MAX_DEPTH levels, before any TPC function runs; reads
       the catalogue the same way, when one is given;
    2. reads MODULES_USED, when the draft has it (S54; TPC tpc_module):
       with a catalogue, each claim's status as the check page shows it
       (OFFERED; SUPERSEDED, with whether the OFFERED version's expansion
       on this draft is the SAME or DIFFERS; RETIRED; WITHDRAWN, with its
       date, its reason and the question the declaration form asks,
       TPS 5.22.14), then a preview of the expansion TPX makes (D(TPK),
       the terms composed into each law, the entries written, the terms
       dropped along a FIXed DOF), and every check below reads the
       expanded draft; without a catalogue, only the claims' shape, and
       the checks read the author's own entries. The interview answer is
       not read: TPX records each round from its own state (TPS 5.22.4).
       A claim refused makes the exit status 1, and the checks go on over
       the author's own entries; the digests are then not printed, being
       of a draft the certificate would not carry;
    3. runs T1 on the TPK (units included: every UNIT parses and has the
       DIMENSION of what it measures; every EQUATION and CONDITION parses
       and names exactly its REFS; every LAW's ROLE is in its grammar and
       EOM(<NAME>) names a DOF; a DOF's ROLE is INPUT; D_VARIABLES is ONE
       or SEVERAL; every D variable is a DOF that no CONSTRAINT targets, and
       one only under ONE) and, if it passes, the kernel tests T2
       (the syntactic part), T5, T4 (over every LAW with an EQUATION) and
       T6 (over every BOUNDARY with a CONDITION); a law or boundary without
       a form is listed as an inspection, never silently passed;
    4. runs the structural check on the TrM against that TPK (a SIGNATURE's
       FORM and IMAGE_FORM included) and, if both pass their structure, the
       isomorphism checks IC1-IC4 and CODOMAIN, IC4 (a) syntactically where a
       signature carries its forms, holding any untied DOF as a named gap;
    5. runs the structural check on the TMS, then each yardstick of a
       well-formed TMS through the SR9 harness (C1, C2, C4) on the
       synthetic record suite; a yardstick with a FIELD term reads the
       field table, which this tool builds as TPS 2.4.8 and
       tpm_record_functions 0.2.0 section 10 have the caller build it:
       tpc_record field_values over each record of the suite, an L
       record's U_C read as 0, the suite's two M records measured at 50 ms
       (R3) and 1 mm (R6), as tpc/vectors/yardstick_corpus.json's field
       cases take them (F-HF2-1);
    6. if everything above passes, the final pass: T3 over the constraints
       and over the laws with an EQUATION (the ground fragment at depth 1),
       and IC2's disposition under the TMS's IC2_POLICY;
    7. if the final pass admits the draft, prints the structure check of
       its equations of motion (D176.1; tpc_eom): the EOM map (each DOF
       with its label and the law the matching gives it, the index and
       initial values of the determined part, the closed and invariant
       laws), the findings each with its question for the declaring
       authority, and the class the candidate certificate's STRUCTURE line
       would carry. The map and the findings are shown whatever the TMS's
       STRUCTURE_CHECK says (S26); the class line reads the switch, and
       under OFF the gate lists nothing for the declaring authority. A
       finding refuses nothing (D176.3), so it does not change the exit
       status;
    8. if the final pass admits the draft, prints D(TPK), D(TrM), D(TMS)
       and D(TFK) (of the expanded draft where claims were read against a
       catalogue), and lists the checks that stay inspections for the
       declaring authority.
Every problem is printed. Exit status 0 when all three zones are present
and every check passes (a named gap under IC2_POLICY GAP passes; a claimed
version's status refuses nothing); 1 when a zone is missing, a check
refuses or MODULES_USED is refused; 2 when the draft or the catalogue
cannot be read or the draft cannot be checked to the end, on a usage
error, and when the reader closes the output early (the run then ends
quietly). No input ends in a traceback. The last line names the TPC-PY modules by their own versions: a
TPC-PY version exists only when the declaring authority declares one (TPS
5.11.9).
"""
import errno
import json
import os
import sys

HERE = os.path.dirname(os.path.abspath(__file__))
sys.path.insert(0, HERE)

import tpc_zone as Z          # noqa: E402
import tpc_yardstick as Y     # noqa: E402
import tpc_units as U         # noqa: E402
import tpc_tests as K         # noqa: E402
import tpc_form as F          # noqa: E402
import tpc_eom as E           # noqa: E402
import tpc_module as MO       # noqa: E402
import tpc_record as RC       # noqa: E402

CHECK_DRAFT_VERSION = "0.6.0"     # proposed MINOR: MODULES_USED read against a catalogue (S54); the field table (F-HF2-1, increment 12), the same bump (S95)

# The synthetic suite's measurements (F-HF2-1): VALUE, U_C and UNIT of its M records, from
# which tpc_record field_values gives each record's U_C in SI; every L record reads 0.
SUITE_MEASURED = {"R3": ("2", "50", "ms"), "R6": ("3", "1", "mm")}


def _field_table(suite):
    """{rid: {"U_C": canonical decimal}} over the synthetic suite, by tpc_record field_values
    (TPS 2.4.8; tpm_record_functions 0.2.0 section 10). A record field_values refuses has no
    row, and a yardstick reading it is refused with the evaluator's own reason."""
    fields = {}
    for rid in sorted(suite["records"]):
        rec = {"CLASS": "RESULT", "REGIME": suite["records"][rid].get("regime")}
        if rid in SUITE_MEASURED:
            rec["VALUE"], rec["U_C"], rec["UNIT"] = SUITE_MEASURED[rid]
        fv = RC.field_values(rec)
        if fv[0] == "OK":
            fields[rid] = fv[1]
    return fields

# The deepest draft the zone schemas admit nests 7 containers (the draft,
# TMS, YARDSTICKS, a yardstick, LIMITING, a limiting case, KEY); the
# examples reach 6. The loader refuses more before the JSON parser or any
# TPC function recurses into it.
MAX_DEPTH = 64

# What no mechanical body reads (tpm_kernel_tests 0.2.0 sections 2, 12;
# tpm_increment_8 0.1.0). The per-entry inspections a draft owes because an
# entry carries no form are printed beside the checks, from the draft itself.
OWED_INSPECTIONS = [
    "T2 the semantic part: each ASSUMPTION states one claim (guide 3.6)",
    "T3 over the assumptions, and over the laws beyond the ground fragment at depth 1, to depth S_MAX (guide 3.6)",
    "T4, T6 and T3 over every law or boundary without a declared form, as listed beside the kernel tests (D108.1)",
    "IC4 (a) that each SIGNATURE image holds in the environment, beyond the syntactic reading, and each borne law is expressible (guide 4.6)",
    "IC5 falsifier invariance under the TRANSFORMATIONS (guide 4.6)",
    "equivalence of environment quantities under the TRANSFORMATIONS (names are taken as written)",
    "every STATEMENT, PROPOSITION, PREDICATE and IMAGE names only what its REFS or OVER list, and agrees with its declared form where one exists",
]


class DuplicateKey(ValueError):
    pass


def _no_duplicates(pairs):
    d = {}
    for k, v in pairs:
        if k in d:
            raise DuplicateKey("duplicate key " + Z._show(k))
        d[k] = v
    return d


class NonJsonConstant(ValueError):
    pass


def _no_constant(token):
    """json's parse_constant hook: RFC 8259 has no NaN, Infinity or
    -Infinity, which Python's json module would otherwise accept."""
    raise NonJsonConstant(token)


def _where(text, i):
    """The position as json reports it: line, column, char."""
    return "line %d column %d (char %d)" % (text.count("\n", 0, i) + 1, i - text.rfind("\n", 0, i), i)


def _scan(text):
    """One linear pass over the raw text, outside strings: [the index of
    the first bracket opened deeper than MAX_DEPTH, the index of the first
    N or I]. Either is None when absent. Outside strings in text JSON has
    read, an N or an I can only begin NaN or Infinity."""
    depth, in_str, esc, const = 0, False, False, None
    for i, ch in enumerate(text):
        if in_str:
            if esc:
                esc = False
            elif ch == "\\":
                esc = True
            elif ch == '"':
                in_str = False
        elif ch == '"':
            in_str = True
        elif ch == "[" or ch == "{":
            depth += 1
            if depth > MAX_DEPTH:
                return [i, const]
        elif ch == "]" or ch == "}":
            depth -= 1
        elif const is None and (ch == "N" or ch == "I"):
            const = i - 1 if i > 0 and text[i - 1] == "-" else i
    return [None, const]


def _load_json(path):
    """The file's one JSON value, or raises ValueError with a readable
    reason. utf-8-sig accepts the byte-order mark some Windows editors
    write. The nesting bound is checked on the raw text, so the parser
    never recurses past it; only the text before the first too-deep
    bracket is parsed, and a problem found there is reported first, as the
    file's first."""
    with open(path, "r", encoding="utf-8-sig") as f:
        text = f.read()
    deep, const = _scan(text)
    try:
        draft = json.loads(text if deep is None else text[:deep],
                           object_pairs_hook=_no_duplicates, parse_constant=_no_constant)
    except NonJsonConstant as e:
        raise ValueError("%s is not JSON (RFC 8259 has no NaN, Infinity or -Infinity)%s"
                         % (e, "" if const is None else ": " + _where(text, const)))
    except json.JSONDecodeError as e:
        if deep is None or e.pos < deep:
            raise
    except RecursionError:
        raise ValueError("nesting deeper than the checker reads")
    else:
        if deep is not None:        # a whole value, then a bracket: what json calls extra data
            raise ValueError("Extra data: " + _where(text, deep))
    if deep is not None:
        raise ValueError("nesting deeper than the checker reads (more than %d levels): %s"
                         % (MAX_DEPTH, _where(text, deep)))
    return draft


def load_draft(path):
    """The draft as a dict, or raises ValueError with a readable reason."""
    draft = _load_json(path)
    if not isinstance(draft, dict):
        raise ValueError("the draft must be one JSON object")
    return draft


def load_catalogue(path):
    """The catalogue as a dict with exactly LINES and MODULES, or raises
    ValueError; their contents are tpc_module's to check (5.2, 5.3)."""
    cat = _load_json(path)
    if not isinstance(cat, dict):
        raise ValueError("the catalogue must be one JSON object with LINES and MODULES")
    for k in sorted(cat):
        if k not in ("LINES", "MODULES"):
            raise ValueError("unknown field " + Z._show(k))
    for k in ("LINES", "MODULES"):
        if k not in cat:
            raise ValueError("missing field " + k)
    return cat


def _count(n, word):
    return "%d %s%s" % (n, word, "" if n == 1 else "s")


def _result(out, label, passed, probs):
    if passed == "NO FORM":
        out("%s NO FORM (by inspection)" % label)
        return
    out("%s %s" % (label, "PASS" if passed else "REFUSED, " + _count(len(probs), "problem")))
    for p in probs:
        out("       " + p)


def _structure(draft, out):
    """The structure check of the equations of motion, after the final pass
    (D176.1; S26): the map and the findings whatever the switch says, then
    the class, which reads the switch. Information: nothing here refuses."""
    m = E.eom_map(draft)
    if m[0] != "OK":
        out("STRUCTURE not computed: " + m[1])
        return
    r = dict((row[0], row[1:]) for row in m[1])
    dv = r["D_VARIABLES"][0]
    out("STRUCTURE map (shown whatever STRUCTURE_CHECK says): analysis " + r["ANALYSIS"][0]
        + ("; D variable" + ("s " if len(dv) != 1 else " ") + ", ".join(dv) if dv else "; no D variable"))
    for dof, label, law in r["MAP"][0]:
        out("     %s: %s%s" % (dof, label, " (LAW " + law + ")" if law else ""))
    if r["INDEX"][0] == "NONE":
        out("     index and initial values: not computed")
    else:
        out("     index %s, initial values %s (the determined part; information)" % (r["INDEX"][0], r["INITIAL_VALUES"][0]))
    if r["CLOSED"][0]:
        out("     closed: " + ", ".join("%s (%s)" % (law, kind) for law, kind in r["CLOSED"][0]))
    if r["INVARIANT"][0]:
        out("     invariant, outside the matching: " + ", ".join(r["INVARIANT"][0]))
    f = E.eom_findings(draft)
    if f[0] != "OK":
        out("STRUCTURE findings not computed: " + f[1])
        return
    c = E.eom_class(draft)
    off = c[0] == "STRUCTURE" and len(c) > 2 and c[2] == ["CHECK OFF"]
    head = "STRUCTURE findings" + (" (information: under STRUCTURE_CHECK OFF the gate lists none)" if off else "")
    out(head + ": " + (str(len(f[1])) if f[1] else "none"))
    for fd in f[1]:
        out("       " + fd[0] + ": " + (fd[-1] + " (D variables " + ", ".join(fd[1]) + ")" if fd[0] == "PDE" else fd[-1]))
    if c[0] != "STRUCTURE":
        out("STRUCTURE class not computed: " + c[1])
    elif len(c) == 2:
        out("STRUCTURE " + c[1])
    else:
        out("STRUCTURE " + c[1] + ": " + ", ".join(c[2]))


OWN_ENTRIES = "     the checks below read your own entries, not the expansion TPX checks at the gate"
NOT_PRINTED = ("D(TPK), D(TrM), D(TMS) and D(TFK) not printed: your own entries were checked,"
               " not the expansion the certificate would carry")


def _claim_row(row):
    """One row of draft_module_status's CLAIMS as the check page words it
    (TPS 5.22.14)."""
    head = "     %s %s: " % (row[0], row[1])
    if row[2] == "SUPERSEDED" and len(row) == 4:
        return head + "SUPERSEDED (no version of %s is OFFERED)" % row[0]
    if row[2] == "SUPERSEDED":
        return head + ("SUPERSEDED by %s (the OFFERED version's expansion on this draft: %s)"
                       % (row[3], row[4]))
    if row[2] == "RETIRED":
        return head + "RETIRED on " + row[3]
    if row[2] == "WITHDRAWN":
        return head + "WITHDRAWN on %s for: %s" % (row[3], row[4])
    return head + row[2]


def _modules_used(draft, catalogue, out):
    """S54: the claims read against the catalogue, when both are given.
    Returns [refused?, the draft the checks read, digests printable?]."""
    own = dict((k, v) for k, v in draft.items() if k != "MODULES_USED")
    given = "" if catalogue is None else "; the catalogue given is not used"
    if "MODULES_USED" not in draft:
        if catalogue is not None:
            out("MODULES USED none claimed" + given)
        return [False, own, True]
    claims = draft["MODULES_USED"]
    if claims == []:
        out("MODULES USED none claimed" + given)
        return [False, own, True]
    if catalogue is None:
        try:
            MO._claims(claims)
        except MO._Refused as e:
            out("MODULES USED REFUSED: " + str(e))
            out(OWN_ENTRIES)
            return [True, own, False]
        out("MODULES USED %s, well-formed; not read against a catalogue (give --catalogue FILE:"
            " the MODULE DECLARATIONS and their texts)" % _count(len(claims), "claim"))
        out(OWN_ENTRIES)
        return [False, own, False]
    lines, texts = catalogue["LINES"], catalogue["MODULES"]
    st = MO.draft_module_status(own, claims, texts, lines)
    if st[0] != "OK":
        out("MODULES USED REFUSED: " + st[1])
        out(OWN_ENTRIES)
        return [True, own, False]
    offered = MO.catalogue_state(lines)[2][1]
    x = MO._render_expand(own, claims, texts)
    out("MODULES USED %s, read against a catalogue of %s"
        % (_count(len(claims), "claim"), "%d MODULE DECLARATION%s" % (len(lines), "" if len(lines) == 1 else "S")))
    out("     offered: " + (", ".join(i + " " + v for i, v, _ in offered) if offered else "none"))
    for row in st[1][1]:
        out(_claim_row(row))
    for q in st[2][1]:
        out("     question for the declaring authority: " + q)
    out("     expansion (a preview; TPX expands at the gate): D(TPK) " + x[1][1])
    for law, eq in x[2][1]:
        out("       composed into LAW %s: %s" % (law, eq))
    for kind, name, src in x[3][1]:
        out("       emitted %s %s (%s)" % (kind, name, ", ".join(src)))
    for cv, q in x[4][1]:
        out("       dropped along %s: the terms of %s (FIXED:%s answered YES)" % (q, cv, q))
    out("     the checks below read the expanded draft")
    return [False, MO.module_expand(own, claims, texts)[1], True]


def check(draft, out, catalogue=None):
    ok = all(z in draft for z in ("TPK", "TrM", "TMS"))
    for k in sorted(draft):
        if k not in ("TPK", "TrM", "TMS", "MODULES_USED"):
            out("DRAFT: unknown zone " + Z._show(k))
            ok = False
    draft0 = draft
    refused, draft, digests = _modules_used(draft, catalogue, out)
    tpk, trm, tms = draft.get("TPK"), draft.get("TrM"), draft.get("TMS")
    t1_ok = kt_ok = trm_ok = ic_ok = tms_ok = False
    # the Prime Kernel: T1, then T2 and T5
    if "TPK" not in draft:
        out("TPK  MISSING (not checked)")
    else:
        kt = K.kernel_tests(tpk)
        t1_ok = kt[1][0][1] is True
        _result(out, "TPK ", t1_ok, kt[1][0][2])
        if t1_ok:
            for name, passed, probs in kt[1][1:]:
                _result(out, "     " + name, passed if passed == "NO FORM" else passed is True, probs)
            for i in kt[2]:
                out("     inspection " + i)
        kt_ok = kt[0] is True
        ok = ok and kt_ok
    # the Translation Manifold: structure, then IC1-IC4 and CODOMAIN
    if "TrM" not in draft:
        out("TrM  MISSING (not checked)")
    else:
        r = Z.trm_well_formed(trm, draft.get("TPK", {}))
        trm_ok = r[0]
        _result(out, "TrM ", trm_ok, r[1])
        ok = ok and trm_ok
        if trm_ok and not t1_ok:
            out("     IC1-IC4 not run: the TPK " + ("does not pass T1" if "TPK" in draft else "is missing"))
        elif trm_ok:
            ic = K.ic_checks(trm, tpk)
            if ic[0] == "REJECT":
                _result(out, "     IC1-IC4", False, ic[1])
            elif ic[0] == "GAP":
                out("     IC1-IC4 PASS with " + _count(len(ic[2]), "untied DOF")
                    + ", held for the final pass")
            elif ic[0] == "PASS":
                out("     IC1-IC4 PASS")
            else:
                out("     IC1-IC4 not run: " + "; ".join(ic[1]))
            if ic[0] == "REJECT" and ic[2]:
                out("     and " + _count(len(ic[2]), "untied DOF") + ", held for the final pass")
            for g in (ic[2] if ic[0] in ("REJECT", "GAP") else []):
                out("       " + g)
            if ic[0] != "NOT_RUN":
                for u in F.ic4a_signature(trm, tpk)[2]:
                    out("     inspection " + u)
                for sg in trm["HEADER"].get("SIGNATURE", []):
                    if "FORM" not in sg:
                        out("     inspection SIGNATURE:" + sg["NAME"] + ": no FORM; IC4 (a) by inspection")
            ic_ok = ic[0] in ("PASS", "GAP")
            ok = ok and ic_ok
    # the Measure Set: structure, then each yardstick through SR9
    if "TMS" not in draft:
        out("TMS  MISSING (not checked)")
    else:
        r = Z.tms_well_formed(tms)
        tms_ok = r[0]
        _result(out, "TMS ", tms_ok, r[1])
        ok = ok and tms_ok
    if tms_ok:
        suite = json.load(open(os.path.join(HERE, "vectors", "sr9_corpus.json"), encoding="utf-8"))["cases"][0]["input"]["suite"]
        for y in tms.get("YARDSTICKS", []):
            ys = dict(suite, fields=_field_table(suite)) if "FIELD:" in json.dumps(y) else suite
            r = Y.yardstick_checks(tms, y["NAME"], ys)
            if r[0] != "OK":
                out("YARDSTICK %s: REFUSED, %s" % (y["NAME"], r[1]))
                ok = False
                continue
            c1, c2, c4 = r[1], r[2], r[3]
            line = "YARDSTICK %s: C1 %s, C2 %s, C4 %s" % (
                y["NAME"], "pass" if c1[0] else "FAIL", "pass" if c2[0] else "FAIL", "pass" if c4[0] else "FAIL")
            if c2[2]:
                line += "; exempt in a declared limiting DOMAIN: " + ", ".join(c2[2])
            out(line)
            for w in c1[1] + c2[1] + c4[1]:
                out("       witness " + w)
            ok = ok and c1[0] and c2[0] and c4[0]
    # the final pass: T3 and IC2's disposition
    if not all(z in draft for z in ("TPK", "TrM", "TMS")):
        out("FINAL not run: a zone is missing")
    elif not ok:
        out("FINAL not run: it runs once every check above passes")
    else:
        fin = K.final_checks(draft)
        if fin[0] == "PASS":
            out("FINAL T3 (constraints and laws) and IC2 PASS")
        elif fin[0] == "GAP":
            out("FINAL T3 (constraints and laws) and IC2 PASS, admitted with " + _count(len(fin[2]), "named gap")
                + " (IC2_POLICY GAP)")
            for g in fin[2]:
                out("       " + g)
        elif fin[0] == "REJECT":
            _result(out, "FINAL T3 (constraints and laws) and IC2", False, fin[1])
        else:
            out("FINAL not run: " + "; ".join(fin[1]))
        ok = fin[0] in ("PASS", "GAP")
        if ok:
            _structure(draft, out)
    if ok:
        out("")
        if digests:
            out("D(TPK) " + Z.d_tpk(tpk))
            out("D(TrM) " + Z.d_trm(trm))
            out("D(TMS) " + Z.d_tms(tms))
            out("D(TFK) " + Z.d_tfk_of(draft))
        else:
            out(NOT_PRINTED)
        out("")
        out("Inspections still owed to the declaring authority before declaring:")
        for i in OWED_INSPECTIONS:
            out("  - " + i)
    out("")
    out("Advisory result (TPC-PY: tpc_zone %s, tpc_units %s, tpc_tests %s, tpc_form %s, tpc_eom %s%s; schema %s). "
        "TPX decides at the gate." % (Z.TPC_ZONE_VERSION, U.TPC_UNITS_VERSION, K.TPC_TESTS_VERSION, F.TPC_FORM_VERSION,
                                      E.TPC_EOM_VERSION,
                                      "" if "MODULES_USED" not in draft0 else ", tpc_module " + MO.TPC_MODULE_VERSION,
                                      Z.SCHEMA_VERSION))
    return ok and not refused


USAGE = "usage: check_draft.py [--catalogue CATALOGUE.json] DRAFT.json"


def _run(argv):
    args = argv[1:]
    if len(args) == 3 and args[0] == "--catalogue":
        cat_path, draft_path = args[1], args[2]
    elif len(args) == 1 and args[0] != "--catalogue":
        cat_path, draft_path = None, args[0]
    else:
        print(USAGE)
        return 2
    try:
        draft = load_draft(draft_path)
    except (OSError, ValueError) as e:
        print("DRAFT: cannot be read: " + str(e))
        return 2
    except MemoryError:
        print("DRAFT: cannot be read: the file does not fit in memory")
        return 2
    catalogue = None
    if cat_path is not None:
        try:
            catalogue = load_catalogue(cat_path)
        except (OSError, ValueError) as e:
            print("CATALOGUE: cannot be read: " + str(e))
            return 2
        except MemoryError:
            print("CATALOGUE: cannot be read: the file does not fit in memory")
            return 2
    try:
        return 0 if check(draft, print, catalogue) else 1
    except RecursionError:
        print("DRAFT: not checked to the end: a value nests deeper than TPC-PY's checks read"
              " (RecursionError), such as a deeply nested EQUATION, CONDITION, FORM or IMAGE_FORM;"
              " TPX decides at the gate")
        return 2
    except MemoryError:
        print("DRAFT: not checked to the end: out of memory in the checks; TPX decides at the gate")
        return 2


def _closed(e):
    """A reader that closed the output: BrokenPipeError, or on Windows an
    OSError of errno EINVAL on the write (O-G1V-2)."""
    return isinstance(e, BrokenPipeError) or getattr(e, "errno", None) in (errno.EPIPE, errno.EINVAL)


def main(argv):
    """_run, with the output flushed here, so a reader that closes the pipe
    early ends the run quietly: stdout moved to the null device, so the
    interpreter's own flush at exit cannot raise again, and exit status 2
    (the draft was not checked to the end for that reader)."""
    if hasattr(sys.stdout, "reconfigure"):
        sys.stdout.reconfigure(errors="backslashreplace")
    try:
        status = _run(argv)
        sys.stdout.flush()
        return status
    except OSError as e:
        if not _closed(e):
            raise
        try:
            os.dup2(os.open(os.devnull, os.O_WRONLY), sys.stdout.fileno())
        except (OSError, ValueError):
            pass
        return 2


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