# -*- coding: utf-8 -*-
"""
TPC cross-residence corpus runner (TPC-PY side), increment 9
=============================================================
Runs TPC-PY against every published vector corpus. Exit 0 iff all cases
pass. A residence "reproduces the corpus byte for byte" (PMP-TPM TPC)
exactly when this passes behind that residence. Standard library only;
Windows/Linux:
python run_corpus.py # every vectors/*.json
python run_corpus.py vectors/x.json # one corpus
Two corpus forms are read, told apart by the case keys (the corpora are
never rewritten; the runner adapts):
BASE FORM (increments 1-8, hold): {"meta": ..., "cases": [{"id", "fn",
"input", "expect"}]}; the result is canon(dispatch(fn, input)), an
exception rendered "ERROR: <msg>".
RUN_CASE FORM (increment 9: falcon, biblio, cluster, hunk, identity,
sponsor, beta, score, probe, volunteer, fingerprint; increment 10:
lint, module; the wiki-base hunk vectors, hunk_wiki, answered by
tpc_hunk): {"corpus": ...,
"cases": [{"name", "fn", "args", "expected"}]}; the result is the
module's run_case(fn, args), the module named by the corpus file
(<name>_corpus.json -> tpc_<name>); tpc_cluster's run_case takes the
whole case. run_case renders its own refusals "ERROR:<msg>".
"""
import glob
import json
import os
import platform
import sys
import unicodedata
import tpc_strike as S
import tpc_digest as G
import tpc_sr9 as C
import tpc_zone as Z
import tpc_yardstick as Y
import tpc_challenge as X
import tpc_units as U
import tpc_tests as K
import tpc_hold as HD
import tpc_form as F
import tpc_eom as EO
# increment 9 (run_case form); imported in dependency order
import tpc_beta as BT
import tpc_sponsor as SP
import tpc_falcon as FA
import tpc_biblio as BI
import tpc_cluster as CL
import tpc_hunk as HK
import tpc_identity as ID
import tpc_probe as PR
import tpc_volunteer as VO
import tpc_score as SO
import tpc_fingerprint as FP
# increment 10 (run_case form): the prompt lint, D182.1 F5
import tpc_lint as LI
# increment 10 (run_case form): environment modules, D178.1 (P2-04; tpm_module_spec 0.2.0)
import tpc_module as MO
# corpus file stem -> the module whose run_case answers it
RUN_CASE_MODULES = {
"falcon": FA, "biblio": BI, "cluster": CL, "hunk": HK, "hunk_wiki": HK, "identity": ID,
"sponsor": SP, "beta": BT, "score": SO, "probe": PR, "volunteer": VO,
"fingerprint": FP, "lint": LI, "module": MO,
}
def dispatch(fn, inp):
# the shared oracle itself: canon() of a value
if fn == "canon":
return inp["x"]
# strike-rule arithmetic
if fn == "a_of_r":
return S.a_of_r(inp["rid"], inp["records"], inp.get("tfk_assumptions", []))
if fn == "d_of_r":
return S.d_of_r(inp["rid"], inp["records"], inp.get("tfk_assumptions", []))
if fn == "sr8_scale_factor":
return S.sr8_scale_factor(inp["measurements"])
if fn == "mm_contradicts":
return S.mm_contradicts(inp["x1"], inp["u1"], inp["x2"], inp["u2"], inp["k"], inp.get("S", 1.0))
if fn == "lm_contradicts":
return S.lm_contradicts(inp["x_l"], inp["x_m"], inp["u_m"], inp["k"], inp.get("S", 1.0))
if fn == "sr5_order":
return S.sr5_order(inp["d1"], inp["d2"])
if fn == "sr6_independence_count":
return S.sr6_independence_count(inp["sets"], inp.get("same_model"))
if fn == "sr6_audit_opens":
return S.sr6_audit_opens(inp["n_ind"], inp["n_x"])
# the interim SR3 and SR8 scope (tpc_strike 0.2.0, proposed; D180.3)
if fn == "sr3_scope":
return S.sr3_scope(inp["tfk"], inp["quantity"])
# digests
if fn == "digest_fields":
return G.digest_fields(inp["fields"])
if fn == "combine_digests":
return G.combine_digests(inp["hex_digests"])
if fn == "d_tfk":
return G.d_tfk(inp["d_tpk"], inp["d_trm"], inp["d_tms"])
# SR9 C-test harness
if fn == "sr9_c1":
return C.sr9_c1(inp["entry"], inp["suite"])
if fn == "sr9_c2":
return C.sr9_c2(inp["entry"], inp["suite"])
if fn == "sr9_c4":
return C.sr9_c4(inp["entry"], inp["suite"])
if fn == "sr9_harness":
return C.sr9_harness(inp["entry"], inp["suite"])
if fn == "tms_params_ok":
return C.tms_params_ok(inp["params"])
# TFK zone schemas (increment 4)
if fn == "normalize_dimension":
r = Z.normalize_dimension(inp["s"])
return "null" if r is None else r
if fn == "canon_decimal":
r = Z.canon_decimal(inp["s"])
return "null" if r is None else r
if fn == "canonical_term":
return Z.canonical_term(inp["t"])
if fn == "is_canonical_dimension":
return Z.is_canonical_dimension(inp["s"])
if fn == "canon_dimension":
return Z.canon_dimension(inp["exps"])
if fn == "t1_well_formed":
return Z.t1_well_formed(inp["tpk"])
if fn == "trm_well_formed":
return Z.trm_well_formed(inp["trm"], inp["tpk"])
if fn == "tms_well_formed":
return Z.tms_well_formed(inp["tms"])
if fn == "zone_preimage":
return Z.zone_preimage(inp["zone"], inp["value"])
if fn == "d_tpk":
return Z.d_tpk(inp["tpk"])
if fn == "d_trm":
return Z.d_trm(inp["trm"])
if fn == "d_tms":
return Z.d_tms(inp["tms"])
if fn == "d_tfk_of":
return Z.d_tfk_of(inp["tfk"])
if fn == "classify_successor":
return Z.classify_successor(inp["a"], inp["b"])
# yardstick evaluator (increment 5)
if fn == "yardstick_key":
return Y.yardstick_key(inp["tms"], inp["name"], inp["rid"], inp["records"],
inp.get("tfk_assumptions", []))
if fn == "yardstick_checks":
return Y.yardstick_checks(inp["tms"], inp["name"], inp["suite"])
if fn == "yardstick_harness":
return Y.yardstick_harness(inp["tms"], inp["name"], inp["suite"])
# execution challenge (D101.1)
if fn == "challenge_preimage":
return X.challenge_preimage(inp["nonce"])
if fn == "challenge_response":
return X.challenge_response(inp["nonce"])
if fn == "challenge_verify":
return X.challenge_verify(inp["nonce"], inp["response"])
# units (increment 6; D103.1, D104.1, D105.1)
if fn == "unit_parse":
return U.unit_parse(inp["u"])
if fn == "unit_scale":
return U.unit_scale(inp["u"])
if fn == "unit_matches":
return U.unit_matches(inp["u"], inp["dimension"])
if fn == "normalize_measurement":
return U.normalize_measurement(inp["value"], inp["uncertainty"], inp["u"])
if fn == "mm_contradicts_units":
return U.mm_contradicts_units(inp["x1"], inp["u1"], inp["unit1"], inp["x2"], inp["u2"],
inp["unit2"], inp["k"], inp.get("S", 1.0))
if fn == "lm_contradicts_units":
return U.lm_contradicts_units(inp["x_l"], inp["unit_l"], inp["x_m"], inp["u_m"],
inp["unit_m"], inp["k"], inp.get("S", 1.0))
if fn == "sr8_scale_factor_units":
return U.sr8_scale_factor_units(inp["measurements"])
# kernel tests and isomorphism checks (increment 7; D106-D108 pause)
if fn == "t2_findings":
return K.t2_findings(inp["prop"])
if fn == "t2_atomicity":
return K.t2_atomicity(inp["tpk"])
if fn == "t3_consistency":
return K.t3_consistency(inp["tpk"], inp["tms"])
if fn == "t5_dof_yield":
return K.t5_dof_yield(inp["tpk"])
if fn == "kernel_tests":
return K.kernel_tests(inp["tpk"])
if fn == "ic_checks":
return K.ic_checks(inp["trm"], inp["tpk"])
if fn == "final_checks":
return K.final_checks(inp["tfk"])
# the hold-scope function (increment "hold scope"; D111.1, CO-21; 0.2.0: the
# bibliography list, D171 (j))
if fn == "hold_scope":
return HD.hold_scope(inp["history"], inp["lookback_passes"])
if fn == "held_closure":
return HD.held_closure(inp["records"], inp["seeds"])
if fn == "tpx_allocate":
return HD.tpx_allocate(inp["range"])
if fn == "hold_preimage":
return HD.hold_preimage(inp["cause"], inp["seeds"], inp["held"], inp["biblio"])
if fn == "hold_record":
return HD.hold_record(inp["hold_id"], inp["cause"], inp["seeds"], inp["held"], inp["biblio"])
if fn == "release_record":
return HD.release_record(inp["release_id"], inp["hold"], inp["reproduction"], inp["records"])
# the declared form (increment 8; D108.1 at Option 1 of every slot)
if fn == "parse_form":
return F.parse_form(inp["text"], inp["mode"])
if fn == "form_names":
return F.form_names(inp["text"], inp["mode"])
if fn == "form_dimension":
return F.form_dimension(inp["text"], inp["mode"], inp["dimensions"])
if fn == "form_totality":
return F.form_totality(inp["text"], inp["dimensions"])
if fn == "eval_form":
return F.eval_form(inp["text"], inp["mode"], inp["bindings"])
if fn == "names_check":
return F.names_check(inp["tpk"])
if fn == "form_coverage":
return F.form_coverage(inp["tpk"])
if fn == "t4_homogeneity":
return F.t4_homogeneity(inp["tpk"])
if fn == "t6_totality":
return F.t6_totality(inp["tpk"])
if fn == "t3_laws":
return F.t3_laws(inp["tpk"])
if fn == "ic4a_signature":
return F.ic4a_signature(inp["trm"], inp["tpk"])
# the D variable (tpc_form 0.2.0, unreleased; D177.1)
if fn == "d_variables":
return F.d_variables(inp["tpk"])
if fn == "d_variable_check":
return F.d_variable_check(inp["tpk"])
# the public parse tree (tpc_form 0.2.0, unreleased; S32)
if fn == "form_tree":
return F.form_tree(inp["text"], inp["mode"])
if fn == "grammar_witness_count":
return F.GRAMMAR_WITNESS_COUNT
# the structure check of the equations of motion (tpc_eom 0.1.0, proposed; D176.1, D177.1)
if fn == "eom_map":
return EO.eom_map(inp["tfk"])
if fn == "eom_findings":
return EO.eom_findings(inp["tfk"])
if fn == "eom_class":
return EO.eom_class(inp["tfk"])
if fn == "eom_register_test":
return EO.eom_register_test(inp["tfk"], inp["records"])
raise KeyError("unknown fn: %s" % fn)
def corpus_form(path, corpus):
"""BASE or RUN_CASE, read from the first case's keys; the module for a
RUN_CASE corpus from the file stem <name>_corpus.json."""
cases = corpus["cases"]
if not cases:
return "BASE", None
c0 = cases[0]
if "input" in c0 and "expect" in c0:
return "BASE", None
stem = os.path.basename(path)
if stem.endswith("_corpus.json"):
stem = stem[:-len("_corpus.json")]
if stem not in RUN_CASE_MODULES:
raise KeyError("no run_case module for corpus %s" % os.path.basename(path))
return "RUN_CASE", RUN_CASE_MODULES[stem]
def run_one(path):
with open(path, "r", encoding="utf-8") as f:
corpus = json.load(f)
form, mod = corpus_form(path, corpus)
passed, failures = 0, []
for c in corpus["cases"]:
if form == "BASE":
try:
got = S.canon(dispatch(c["fn"], c["input"]))
except Exception as e:
got = "ERROR: %s" % e
expect, cid = c["expect"], c["id"]
else:
try:
if mod is CL:
got = mod.run_case(c)
else:
got = mod.run_case(c["fn"], c["args"])
except Exception as e:
got = "EXCEPTION: %s: %s" % (type(e).__name__, e)
expect, cid = c["expected"], c["name"]
if got == expect:
passed += 1
else:
failures.append((cid, expect, got))
print(" %-22s %d/%d" % (os.path.basename(path), passed, len(corpus["cases"])),
"PASS" if not failures else "FAIL")
for cid, exp, got in failures:
print(" FAIL %s expected %r got %r" % (cid, exp, got))
return not failures
def main(argv):
here = os.path.dirname(os.path.abspath(__file__))
if len(argv) > 1:
paths = [argv[1]]
else:
paths = sorted(glob.glob(os.path.join(here, "vectors", "*.json")))
print("TPC-PY: strike %s, digest %s, sr9 %s, zone %s, yardstick %s, challenge %s, units %s, tests %s, hold %s, form %s, eom %s" % (
S.TPC_STRIKE_VERSION, G.TPC_DIGEST_VERSION, C.TPC_SR9_VERSION, Z.TPC_ZONE_VERSION,
Y.TPC_YARDSTICK_VERSION, X.TPC_CHALLENGE_VERSION, U.TPC_UNITS_VERSION, K.TPC_TESTS_VERSION,
HD.TPC_HOLD_VERSION, F.TPC_FORM_VERSION, EO.TPC_EOM_VERSION))
print(" falcon %s, biblio %s, cluster %s, hunk %s, identity %s, sponsor %s, beta %s, score %s, probe %s, volunteer %s, fingerprint %s, lint %s, module %s" % (
FA.VERSION, BI.VERSION, CL.VERSION, HK.VERSION, ID.VERSION, SP.VERSION, BT.VERSION,
SO.VERSION, PR.VERSION, VO.VERSION, FP.VERSION, LI.VERSION, MO.VERSION))
ok = True
for p in paths:
ok = run_one(p) and ok
print("RESULT:", "ALL PASS" if ok else "FAILURES")
# the self-test fields an agent declares before claiming EXECUTED (D101.1, 4b)
print("ENVIRONMENT: python %s, unicode %s" % (platform.python_version(), unicodedata.unidata_version))
return 0 if ok else 1
if __name__ == "__main__":
sys.exit(main(sys.argv))
