BFP Rauch: TPC-PY source tpc_probe.py (0.2.0), SHA-256 8b99a699b72469a7; conformed TPS 0.4.0; PROPOSED, not declared (TPC-PY 0.2.0)

2026-09-30T02:10:58Z

TPC-PY source tpc_probe.py (0.2.0), SHA-256 8b99a699b72469a7; conformed TPS 0.4.0; PROPOSED, not declared (TPC-PY 0.2.0)

← Older revision		Revision as of 02:10, 30 September 2026
Line 38:		Line 38:
	and normalized to one UNIT; TPX refuses a cross-entry M-M test"; the		and normalized to one UNIT; TPX refuses a cross-entry M-M test"; the
	user's Q12 answer "say half or double" (parrot speculation 12).		user's Q12 answer "say half or double" (parrot speculation 12).
	CP1 L: original and reproduction each {finding: {type: TEXT \| DECIMAL,		CP1 L: original and reproduction each {finding: {type: TEXT \| DECIMAL \|
	value}} in the ~~D99~~ canonical form, which this function checks ~~and~~		EQUATION, value}} in the canonical form, which this function checks
	never produces: TEXT is NFC with LF line endings; DECIMAL is the		and never produces: EQUATION is canonical when tpc_form parse_form(
			value, "equation") returns ["OK", value] (0.2.0; D195.3, RD-RS-13;
			tpm_record_functions 0.2.0 8.2); TEXT is NFC with LF line endings;
			DECIMAL is the
	zone serialization's canonical decimal ("0", or -?M(eE)? with M an		zone serialization's canonical decimal ("0", or -?M(eE)? with M an
	integer without leading or trailing zeros and E a non-zero integer;		integer without leading or trailing zeros and E a non-zero integer;
Line 161:		Line 164:
	from tpc_strike import mm_contradicts, sr8_scale_factor # noqa: E402		from tpc_strike import mm_contradicts, sr8_scale_factor # noqa: E402

	VERSION = "0.1.0"		VERSION = "0.2.0" # increment 12 (S95: one MINOR bump over the declared 0.1.0): CP1's EQUATION type
	TIME_MAX = 253402300799		TIME_MAX = 253402300799
	COUNT_MAX = 1000000000		COUNT_MAX = 1000000000
Line 297:		Line 300:
	raise ValueError("record")		raise ValueError("record")
	f = o["finding"]		f = o["finding"]
	if not _keys_are(f, ("type", "value")) or f["type"] not in ("TEXT", "DECIMAL"):		if not _keys_are(f, ("type", "value")) or f["type"] not in ("TEXT", "DECIMAL", "EQUATION"):
	raise ValueError("record")		raise ValueError("record")
	if type(f["value"]) is not str:		if type(f["value"]) is not str:
	raise ValueError("record")		raise ValueError("record")
	ok = _canon_text_ok(f["value"]) if f["type"] == "TEXT" else _canon_dec_ok(f["value"])		if f["type"] == "EQUATION":
			import tpc_form
			ok = tpc_form.parse_form(f["value"], "equation") == ["OK", f["value"]]
			else:
			ok = _canon_text_ok(f["value"]) if f["type"] == "TEXT" else _canon_dec_ok(f["value"])
	if not ok:		if not ok:
	raise ValueError("CANON")		raise ValueError("CANON")

BFP Rauch: TPC-PY source tpc_probe.py (0.1.0), SHA-256 b49228a3ae8d06ec; conformed TPS 0.3.0; PROPOSED, not declared (CO-8)

2026-09-27T15:43:18Z

TPC-PY source tpc_probe.py (0.1.0), SHA-256 b49228a3ae8d06ec; conformed TPS 0.3.0; PROPOSED, not declared (CO-8)

New page

"""tpc_probe 0.1.0 (DRAFT, pending the pause of landing round 8): the
probe machinery of D118-D120 as PMP-TPM 0.12.0 words it - the comparison
of a reproduction with its target, LOAD and EXPECTED LOSS, the PARROT
RATIO's epoch draw, the CANARY SCHEDULE's tier, burst and gap draws, the
probe draw with the RESET, the reset's end by state, and the
historical-view selection of a register revision.

Reference implementation, Python standard library only; imports the
tree's tpc_strike (SR3 mm_contradicts and SR8 sr8_scale_factor; reused,
never reimplemented) and tpc_beta's canon_float. DATA for landing round 8
of branch tpm_project_closeout; not a declared TPC-PY version and not TPX
code. Every function is a pure function of its arguments; every draw is a
pure function of recorded state and a supplied random value r (32
lowercase hex, as tpc_falcon's falcon_draw takes it), further uniform
values derived from r by the D97 netstring digest of [tag, "1", r,
counter] and never by a language RNG. Declared parameters and tuning
options are read from the params object under PMP-TPM 0.12.0's names
(PARROT RATIO, CANARY SCHEDULE, RESET RULE, CONSISTENCY SHARE, ORANGE
PASSES, HALF-LIFE); the TMS parameter k is an argument. Every refused
argument raises ValueError with a short fixed message. render_*() give
the canonical text the PHP residence (tpc_probe.php) reproduces byte for
byte; floats through canon_float. Ints are never bools.

uniform(tag, r, counter) -> float (the draw law, D62)
u = int(first 13 hex of SHA-256(netstring([tag, "1", r, counter]))) /
2^52: a double in [0, 1) exact in both residences.
udiscrete(lo, hi, u) = min(hi, lo + floor(u x (hi - lo + 1))): U{lo..hi}.

compare(kind, original, reproduction, standing, k, material_ratio)
-> dict (D119.1; SR3; SR8)
"The comparison, closing TPS 5.9.7 OWED C-04: L, equality of FINDING
after canonicalization (D99 forms), zero tolerance; M, SR3's test with
the UWC's k after SR8 scaling and unit normalization, a material u_c
difference noted on a MATCH (MATCH WITH DIFFERENCE NOTED); mechanical
over a declared form only, a prose FINDING INSPECTION until the record
field schema C-16 fixes a form"; SR3 "M-M, |x1 - x2| > k (u1^2 +
u2^2)^(1/2) after SR8 scaling, both records naming the same QUANTITY
and normalized to one UNIT; TPX refuses a cross-entry M-M test"; the
user's Q12 answer "say half or double" (parrot speculation 12).
CP1 L: original and reproduction each {finding: {type: TEXT | DECIMAL,
value}} in the D99 canonical form, which this function checks and
never produces: TEXT is NFC with LF line endings; DECIMAL is the
zone serialization's canonical decimal ("0", or -?M(eE)? with M an
integer without leading or trailing zeros and E a non-zero integer;
tpm_zone_schemas 5.6). A value not canonical is refused (CANON):
canonicalization is the serializer's, upstream. MATCH when the two
values are byte-equal, else NON-CORROBORATION carrying both.
CP2 M: each {x, u > 0, unit}; standing: the standing M records of the
QUANTITY as [{x, u}] (the original among them), over which SR8's S
is computed (tpc_strike.sr8_scale_factor; 1 below two records);
k > 0 the TMS parameter. A UNIT difference is refused (UNIT):
normalization to coherent SI is upstream (D103, tpc_units), so this
function compares one UNIT only (PD-SC-14). Contradiction by
tpc_strike.mm_contradicts(x1, u1, x2, u2, k, S); the SR3 statistic
|x1 - x2| / sqrt((S u1)^2 + (S u2)^2) is reported beside k.
NON-CORROBORATION carries both values, both u_c, S and the
statistic; else MATCH, and MATCH WITH DIFFERENCE NOTED when
max(u1, u2) / min(u1, u2) >= material_ratio (PD-SC-15: "say half
or double" read as the ratio 2, an argument, never a constant).

load(records, r) -> dict (D118.1; 17.7)
"LOAD the number of standing records whose citations or ASSUMPTIONS
closure reach r". records: {id: {cites: [ids], assumptions: [ids],
standing: bool}}; an identifier outside records is ignored (a TFK
identifier). LD1 A standing record s other than r counts when r is
in the transitive closure of s over cites and assumptions (through
any record, standing or not; PD-SC-16). The reaching ids are listed
in code-point order.

expected_loss(e_lo, load) -> float (D118.1; 17.7)
EL(r) = PD x LOAD, "PD one minus the author's evidence at its
pessimistic end", LGD 1: (1 - e_lo) x load.

ratio_step(state, r, params) -> dict (D120.1)
"PARROT RATIO, k parrots per canary, k : 1, k drawn per epoch of
U{50..100} work blocks from {1/2, 1/3, 1/5, 1/7} with weights 0.30 /
0.50 / 0.15 / 0.05"; RATIO EPOCH SCOPE decides whose state this is
(PER ACCOUNT: the account's; FLEET: the fleet's), the law the same.
state: {k: null or the value in force, remaining: blocks left in the
epoch}; params["PARROT RATIO"] = {set, weights, epoch: [lo, hi]}.
RS1 When k is null or remaining is 0 this block starts an epoch: its
length L = udiscrete(lo, hi, u0), k the first set member whose
cumulative weight exceeds u1 (the last member past every cumulant),
remaining = L - 1. Else remaining - 1. Tags TPC-RATIO-EPOCH and
TPC-RATIO-K, counter 0.

schedule_step(state, r, params) -> dict (D117.1)
"blocks 1-100, 1 in 5; blocks 101-1000, 1 in 10; blocks 1001 and
later, 1 in 20 by default, with bursts at 1 in 5 or 1 in 10 at even
odds, duration U{50..200} blocks, gap U{200..500} blocks from the end
of one burst to the start of the next, bursts after block 1000 only".
state: {blocks: landed under the certificate before this one, burst:
null or {rate, remaining}, gap: null or the blocks left}; params
["CANARY SCHEDULE"] = {tiers: [[from, to or null, rate], ...],
burst_rates, burst_duration: [lo, hi], burst_gap: [lo, hi]}.
SC1 b = blocks + 1 is this block's tenure number; its tier is the row
holding b; bursts exist only from the last row's first block.
SC2 In a burst: the burst's rate; remaining - 1; at 0 the burst ends
and a gap is drawn (tag TPC-SCHEDULE-GAP). Else a gap: the one in
force, or, on entering the burst regime, a fresh draw as if a
burst had just ended (PD-SC-17); gap - 1; at 0 a burst starts on
this block: its rate by even odds (TPC-SCHEDULE-RATE), its
duration (TPC-SCHEDULE-DURATION), this block the first; a
duration of 1 ends it here and draws the next gap (counter 1).
SC3 The result records tier, rate and burst for this block.

probe_draw(schedule, ratio_k, reset, has_own_record, r, params) -> dict
(D118.1; D120.1; RECORDING)
"The CANARY SCHEDULE read as the rate of probes, parrots inside the
budget"; "RESET RULE 1 in 5 per account ... whatever its tier or burst
state, the ratio unchanged"; "consistency parrots are CONSISTENCY
SHARE of parrots, drawn per parrot, skipped when the account has no
eligible own record". PD1 rate = RESET RULE when reset is in force,
else the schedule's; probe when u0 < 1 / rate (TPC-PROBE 0). PD2 A
probe is a parrot when u1 < k / (1 + k) (TPC-PROBE 1), else a CANARY.
PD3 A parrot is a CONSISTENCY PARROT when u2 < CONSISTENCY SHARE
(TPC-PROBE 2) and the account has an eligible own record, else a
CORROBORATION PARROT. The recording form carries probe, kind, tier,
burst, rate, ratio and reset.

probe_step(state, r, params, has_own_record) -> dict
ratio_step, schedule_step and probe_draw on one state {schedule,
ratio, reset}; the recorded probe form and the new state.

reset_update(reset, event, account_state, params) -> dict (D120.1)
"After a non-corroboration the reproducer's probe rate is 1 in 5 ...;
it ends by the account's state: YELLOW at the next canary pass,
ORANGE after ORANGE PASSES consecutive canary passes, SUSPENDED only
by the audit finding. A canary failure during a reset is CR6's case
...; a passed parrot ... does not end a reset". reset: {in_force,
passes}; event: NON-CORROBORATION | CANARY PASS | CANARY FAIL | PARROT
PASS | AUDIT FINDING; account_state: GREEN | YELLOW | ORANGE |
SUSPENDED. RU1 NON-CORROBORATION starts (or restarts) the reset. RU2
CANARY PASS counts; the reset ends at the first pass under GREEN or
YELLOW (PD-SC-18: GREEN, a state the rule does not name, ends as
YELLOW does), at ORANGE PASSES consecutive passes under ORANGE, never
under SUSPENDED. RU3 CANARY FAIL ends it as CR6's case. RU4 PARROT
PASS changes nothing. RU5 AUDIT FINDING ends it under SUSPENDED only.

historical_view(revisions, target) -> dict (D118.1)
"HISTORICAL VIEW: the source register served as it stood immediately
before the target record was allocated". revisions: [{rev, allocated:
[ids]}] in ascending rev order; target: a record id allocated in
exactly one revision. HV1 The view is the last revision before the
target's (null when the target's is the first) and the ids allocated
through it, in order.
"""

import hashlib
import math
import os
import re
import sys
import unicodedata

_here = os.path.dirname(os.path.abspath(__file__))
for _p in (_here,):
if _p not in sys.path:
sys.path.insert(0, _p)
from tpc_beta import canon_float # noqa: E402
from tpc_strike import mm_contradicts, sr8_scale_factor # noqa: E402

VERSION = "0.1.0"
TIME_MAX = 253402300799
COUNT_MAX = 1000000000
TWO52 = 4503599627370496.0
STATES = ("GREEN", "YELLOW", "ORANGE", "SUSPENDED")
RESET_EVENTS = ("NON-CORROBORATION", "CANARY PASS", "CANARY FAIL",
"PARROT PASS", "AUDIT FINDING")
HEX16 = "0123456789abcdef"
_CANON_DEC = re.compile(r"(0|-?[1-9]([0-9]*[1-9])?(e-?[1-9][0-9]*)?)")

# ---------------------------------------------------------------- common

def _esc(s):
out = []
for ch in s:
o = ord(ch)
if ch == "|" or ch == "\\":
out.append("\\" + ch)
elif o < 0x20 or o == 0x7F:
out.append("\\x%02x" % o)
else:
out.append(ch)
return "".join(out)

def _is_int(x, lo, hi):
return type(x) is int and lo <= x <= hi

def _is_name(x):
return type(x) is str and x != ""

def _keys_are(o, required, optional=()):
if type(o) is not dict:
return False
for k in required:
if k not in o:
return False
for k in o:
if k not in required and k not in optional:
return False
return True

def _num(x):
if type(x) is bool or not isinstance(x, (int, float)):
return None
x = float(x)
if not math.isfinite(x):
return None
return x

def _is_hex(s, n):
if type(s) is not str or len(s) != n:
return False
for ch in s:
if ch not in HEX16:
return False
return True

def _f(x):
return canon_float(x)

def _b(x):
return "true" if x else "false"

def _netstring(fields):
return b"".join(b"%d:%s" % (len(b), b)
for b in (f.encode("utf-8") for f in fields))

def _digest_fields(fields):
return hashlib.sha256(_netstring(fields)).hexdigest()

# --------------------------------------------------------------- uniform

def uniform(tag, r, counter):
if not _is_name(tag):
raise ValueError("tag")
if not _is_hex(r, 32):
raise ValueError("r")
if not _is_int(counter, 0, COUNT_MAX):
raise ValueError("counter")
h = _digest_fields([tag, "1", r, str(counter)])
return float(int(h[:13], 16)) / TWO52

def udiscrete(lo, hi, u):
n = hi - lo + 1
v = lo + int(math.floor(u * float(n)))
if v > hi:
v = hi
return v

def _pair(params, key, lo_min, hi_max, label):
if type(params) is not dict or key not in params:
raise ValueError("params:" + label)
v = params[key]
if type(v) is not list or len(v) != 2:
raise ValueError("params:" + label)
if not _is_int(v[0], lo_min, hi_max) or not _is_int(v[1], lo_min, hi_max):
raise ValueError("params:" + label)
if v[0] > v[1]:
raise ValueError("params:" + label)
return v[0], v[1]

# --------------------------------------------------------------- compare

def _canon_text_ok(s):
if type(s) is not str:
return False
if "\r" in s:
return False
for ch in s:
if 0xD800 <= ord(ch) <= 0xDFFF:
return False
return unicodedata.normalize("NFC", s) == s

def _canon_dec_ok(s):
return type(s) is str and _CANON_DEC.fullmatch(s) is not None

def _finding(o):
if not _keys_are(o, ("finding",)):
raise ValueError("record")
f = o["finding"]
if not _keys_are(f, ("type", "value")) or f["type"] not in ("TEXT", "DECIMAL"):
raise ValueError("record")
if type(f["value"]) is not str:
raise ValueError("record")
ok = _canon_text_ok(f["value"]) if f["type"] == "TEXT" else _canon_dec_ok(f["value"])
if not ok:
raise ValueError("CANON")
return f["type"], f["value"]

def _mrec(o):
if not _keys_are(o, ("x", "u", "unit")):
raise ValueError("record")
x = _num(o["x"])
u = _num(o["u"])
if x is None or u is None or u <= 0.0 or type(o["unit"]) is not str:
raise ValueError("record")
return x, u, o["unit"]

def compare(kind, original, reproduction, standing, k, material_ratio):
if kind == "L":
t1, v1 = _finding(original)
t2, v2 = _finding(reproduction)
if t1 != t2:
raise ValueError("form")
if v1 == v2:
return {"kind": "L", "outcome": "MATCH", "original": v1,
"reproduction": v2}
return {"kind": "L", "outcome": "NON-CORROBORATION", "original": v1,
"reproduction": v2}
if kind != "M":
raise ValueError("kind")
x1, u1, unit1 = _mrec(original)
x2, u2, unit2 = _mrec(reproduction)
if unit1 != unit2:
raise ValueError("UNIT")
kf = _num(k)
if kf is None or kf <= 0.0:
raise ValueError("k")
mr = _num(material_ratio)
if mr is None or mr < 1.0:
raise ValueError("material_ratio")
if type(standing) is not list:
raise ValueError("standing")
meas = []
for m in standing:
if not _keys_are(m, ("x", "u")):
raise ValueError("standing")
mx = _num(m["x"])
mu = _num(m["u"])
if mx is None or mu is None or mu <= 0.0:
raise ValueError("standing")
meas.append({"x": mx, "u": mu})
s = sr8_scale_factor(meas)
su1 = s * u1
su2 = s * u2
stat = abs(x1 - x2) / math.sqrt(su1 * su1 + su2 * su2)
contra = mm_contradicts(x1, u1, x2, u2, kf, s)
base = {"kind": "M", "x1": x1, "u1": u1, "x2": x2, "u2": u2, "S": s,
"k": kf, "statistic": stat, "unit": unit1}
if contra:
base["outcome"] = "NON-CORROBORATION"
return base
ratio = (u1 / u2) if u1 >= u2 else (u2 / u1)
base["ratio"] = ratio
base["outcome"] = "MATCH WITH DIFFERENCE NOTED" if ratio >= mr else "MATCH"
return base

def render_compare(d):
if d["kind"] == "L":
return "L|%s|%s|%s" % (d["outcome"], _esc(d["original"]),
_esc(d["reproduction"]))
out = "M|%s|%s|%s|%s|%s|%s|%s|%s|%s" % (
d["outcome"], _f(d["x1"]), _f(d["u1"]), _f(d["x2"]), _f(d["u2"]),
_f(d["S"]), _f(d["k"]), _f(d["statistic"]), _esc(d["unit"]))
if "ratio" in d:
out += "|" + _f(d["ratio"])
return out

# ------------------------------------------------------------------ load

def _graph(records):
if type(records) is not dict:
raise ValueError("records")
g = {}
for rid, rec in records.items():
if not _is_name(rid) or not _keys_are(rec, ("cites", "assumptions", "standing")):
raise ValueError("records")
if type(rec["cites"]) is not list or type(rec["assumptions"]) is not list:
raise ValueError("records")
if type(rec["standing"]) is not bool:
raise ValueError("records")
edges = []
for x in rec["cites"] + rec["assumptions"]:
if not _is_name(x):
raise ValueError("records")
edges.append(x)
g[rid] = (edges, rec["standing"])
return g

def _closure(g, start):
seen = set()
stack = [start]
while stack:
cur = stack.pop()
if cur in seen or cur not in g:
continue
seen.add(cur)
for e in g[cur][0]:
stack.append(e)
seen.discard(start)
return seen

def load(records, r):
g = _graph(records)
if not _is_name(r) or r not in g:
raise ValueError("r")
reach = []
for sid in sorted(g.keys()):
if sid == r or not g[sid][1]:
continue
if r in _closure(g, sid):
reach.append(sid)
return {"load": len(reach), "reach": reach}

def render_load(d):
return "%d|%s" % (d["load"], ",".join(_esc(x) for x in d["reach"]))

def expected_loss(e_lo, load_):
e = _num(e_lo)
if e is None or e < 0.0 or e > 1.0:
raise ValueError("e_lo")
if not _is_int(load_, 0, COUNT_MAX):
raise ValueError("load")
return (1.0 - e) * float(load_)

# ------------------------------------------------------------ ratio_step

def _ratio_params(params):
if type(params) is not dict or "PARROT RATIO" not in params:
raise ValueError("params:PARROT RATIO")
pr = params["PARROT RATIO"]
if not _keys_are(pr, ("set", "weights", "epoch")):
raise ValueError("params:PARROT RATIO")
st = pr["set"]
ws = pr["weights"]
if type(st) is not list or type(ws) is not list or not st or len(st) != len(ws):
raise ValueError("params:PARROT RATIO")
kset = []
wset = []
total = 0.0
for i in range(len(st)):
kv = _num(st[i])
wv = _num(ws[i])
if kv is None or kv <= 0.0 or wv is None or wv < 0.0:
raise ValueError("params:PARROT RATIO")
kset.append(kv)
wset.append(wv)
total = total + wv
if total <= 0.0 or abs(total - 1.0) > 1e-9:
raise ValueError("params:PARROT RATIO")
lo, hi = _pair(pr, "epoch", 1, COUNT_MAX, "PARROT RATIO")
return kset, wset, lo, hi

def ratio_step(state, r, params):
if not _keys_are(state, ("k", "remaining")):
raise ValueError("state")
if state["k"] is not None and (_num(state["k"]) is None or _num(state["k"]) <= 0.0):
raise ValueError("state")
if not _is_int(state["remaining"], 0, COUNT_MAX):
raise ValueError("state")
if not _is_hex(r, 32):
raise ValueError("r")
kset, wset, lo, hi = _ratio_params(params)
if state["k"] is None or state["remaining"] == 0:
u0 = uniform("TPC-RATIO-EPOCH", r, 0)
u1 = uniform("TPC-RATIO-K", r, 0)
length = udiscrete(lo, hi, u0)
cum = 0.0
k = kset[len(kset) - 1]
for i in range(len(kset)):
cum = cum + wset[i]
if u1 < cum:
k = kset[i]
break
return {"k": k, "remaining": length - 1, "drawn": True, "length": length}
return {"k": float(state["k"]), "remaining": state["remaining"] - 1,
"drawn": False, "length": None}

def render_ratio(d):
return "%s|%d|%s|%s" % (_f(d["k"]), d["remaining"], _b(d["drawn"]),
"-" if d["length"] is None else str(d["length"]))

# --------------------------------------------------------- schedule_step

def _schedule_params(params):
if type(params) is not dict or "CANARY SCHEDULE" not in params:
raise ValueError("params:CANARY SCHEDULE")
cs = params["CANARY SCHEDULE"]
if not _keys_are(cs, ("tiers", "burst_rates", "burst_duration", "burst_gap")):
raise ValueError("params:CANARY SCHEDULE")
tiers = cs["tiers"]
if type(tiers) is not list or not tiers:
raise ValueError("params:CANARY SCHEDULE")
rows = []
expect = 1
for i, t in enumerate(tiers):
if type(t) is not list or len(t) != 3:
raise ValueError("params:CANARY SCHEDULE")
frm, to, rate = t
last = i == len(tiers) - 1
if not _is_int(frm, 1, COUNT_MAX) or frm != expect or not _is_int(rate, 1, COUNT_MAX):
raise ValueError("params:CANARY SCHEDULE")
if last:
if to is not None:
raise ValueError("params:CANARY SCHEDULE")
else:
if not _is_int(to, frm, COUNT_MAX):
raise ValueError("params:CANARY SCHEDULE")
expect = to + 1
rows.append((frm, to, rate))
br = cs["burst_rates"]
if type(br) is not list or not br:
raise ValueError("params:CANARY SCHEDULE")
for x in br:
if not _is_int(x, 1, COUNT_MAX):
raise ValueError("params:CANARY SCHEDULE")
dlo, dhi = _pair(cs, "burst_duration", 1, COUNT_MAX, "CANARY SCHEDULE")
glo, ghi = _pair(cs, "burst_gap", 1, COUNT_MAX, "CANARY SCHEDULE")
return rows, br, (dlo, dhi), (glo, ghi)

def schedule_step(state, r, params):
if not _keys_are(state, ("blocks", "burst", "gap")):
raise ValueError("state")
if not _is_int(state["blocks"], 0, COUNT_MAX - 1):
raise ValueError("state")
burst = state["burst"]
if burst is not None:
if not _keys_are(burst, ("rate", "remaining")):
raise ValueError("state")
if not _is_int(burst["rate"], 1, COUNT_MAX) or not _is_int(burst["remaining"], 1, COUNT_MAX):
raise ValueError("state")
gap = state["gap"]
if gap is not None and not _is_int(gap, 1, COUNT_MAX):
raise ValueError("state")
if burst is not None and gap is not None:
raise ValueError("state")
if not _is_hex(r, 32):
raise ValueError("r")
rows, brates, dur, gp = _schedule_params(params)
b = state["blocks"] + 1
tier = 0
tier_rate = rows[len(rows) - 1][2]
for i in range(len(rows)):
frm, to, rate = rows[i]
if b >= frm and (to is None or b <= to):
tier = i + 1
tier_rate = rate
break
burst_from = rows[len(rows) - 1][0]
new_burst = None
new_gap = None
rate = tier_rate
in_burst = False
if b >= burst_from:
if burst is not None:
rate = burst["rate"]
in_burst = True
rem = burst["remaining"] - 1
if rem == 0:
new_gap = udiscrete(gp[0], gp[1], uniform("TPC-SCHEDULE-GAP", r, 0))
else:
new_burst = {"rate": burst["rate"], "remaining": rem}
else:
g = gap if gap is not None else udiscrete(gp[0], gp[1], uniform("TPC-SCHEDULE-GAP", r, 0))
g = g - 1
if g == 0:
idx = udiscrete(0, len(brates) - 1, uniform("TPC-SCHEDULE-RATE", r, 0))
length = udiscrete(dur[0], dur[1], uniform("TPC-SCHEDULE-DURATION", r, 0))
rate = brates[idx]
in_burst = True
if length - 1 == 0:
new_gap = udiscrete(gp[0], gp[1], uniform("TPC-SCHEDULE-GAP", r, 1))
else:
new_burst = {"rate": rate, "remaining": length - 1}
else:
new_gap = g
return {"blocks": b, "tier": tier, "rate": rate, "burst": in_burst,
"state": {"blocks": b, "burst": new_burst, "gap": new_gap}}

def render_schedule(d):
st = d["state"]
bs = "-" if st["burst"] is None else "%d/%d" % (st["burst"]["rate"], st["burst"]["remaining"])
gs = "-" if st["gap"] is None else str(st["gap"])
return "%d|%d|%d|%s|%s|%s" % (d["blocks"], d["tier"], d["rate"], _b(d["burst"]), bs, gs)

# ------------------------------------------------------------ probe_draw

def _reset(reset):
if not _keys_are(reset, ("in_force", "passes")):
raise ValueError("reset")
if type(reset["in_force"]) is not bool or not _is_int(reset["passes"], 0, COUNT_MAX):
raise ValueError("reset")

def probe_draw(schedule, ratio_k, reset, has_own_record, r, params):
if not _keys_are(schedule, ("tier", "rate", "burst")):
raise ValueError("schedule")
if (not _is_int(schedule["tier"], 1, COUNT_MAX) or not _is_int(schedule["rate"], 1, COUNT_MAX)
or type(schedule["burst"]) is not bool):
raise ValueError("schedule")
kf = _num(ratio_k)
if kf is None or kf <= 0.0:
raise ValueError("ratio")
_reset(reset)
if type(has_own_record) is not bool:
raise ValueError("has_own_record")
if not _is_hex(r, 32):
raise ValueError("r")
if type(params) is not dict or "RESET RULE" not in params or "CONSISTENCY SHARE" not in params:
raise ValueError("params:RESET RULE")
reset_rate = params["RESET RULE"]
if not _is_int(reset_rate, 1, COUNT_MAX):
raise ValueError("params:RESET RULE")
share = _num(params["CONSISTENCY SHARE"])
if share is None or share < 0.0 or share > 1.0:
raise ValueError("params:CONSISTENCY SHARE")
rate = reset_rate if reset["in_force"] else schedule["rate"]
u0 = uniform("TPC-PROBE", r, 0)
probe = u0 < 1.0 / float(rate)
kind = None
if probe:
u1 = uniform("TPC-PROBE", r, 1)
if u1 < kf / (1.0 + kf):
u2 = uniform("TPC-PROBE", r, 2)
if u2 < share and has_own_record:
kind = "CONSISTENCY PARROT"
else:
kind = "CORROBORATION PARROT"
else:
kind = "CANARY"
return {"probe": probe, "kind": kind, "tier": schedule["tier"],
"burst": schedule["burst"], "rate": rate, "ratio": kf,
"reset": reset["in_force"]}

def render_probe(d):
return "%s|%s|%d|%s|%d|%s|%s" % (_b(d["probe"]), "-" if d["kind"] is None else d["kind"],
d["tier"], _b(d["burst"]), d["rate"], _f(d["ratio"]),
_b(d["reset"]))

def probe_step(state, r, params, has_own_record):
if not _keys_are(state, ("schedule", "ratio", "reset")):
raise ValueError("state")
rs = ratio_step(state["ratio"], r, params)
sc = schedule_step(state["schedule"], r, params)
pd = probe_draw({"tier": sc["tier"], "rate": sc["rate"], "burst": sc["burst"]},
rs["k"], state["reset"], has_own_record, r, params)
return {"probe": pd, "ratio": rs, "schedule": sc,
"state": {"schedule": sc["state"], "ratio": {"k": rs["k"], "remaining": rs["remaining"]},
"reset": state["reset"]}}

def render_step(d):
return "\n".join([render_probe(d["probe"]), render_ratio(d["ratio"]),
render_schedule(d["schedule"])])

# ---------------------------------------------------------- reset_update

def reset_update(reset, event, account_state, params):
_reset(reset)
if event not in RESET_EVENTS:
raise ValueError("event")
if account_state not in STATES:
raise ValueError("account_state")
if type(params) is not dict or not _is_int(params.get("ORANGE PASSES"), 1, COUNT_MAX):
raise ValueError("params:ORANGE PASSES")
orange = params["ORANGE PASSES"]
if event == "NON-CORROBORATION":
return {"in_force": True, "passes": 0, "end": "-"}
if not reset["in_force"]:
return {"in_force": False, "passes": 0, "end": "-"}
if event == "CANARY FAIL":
return {"in_force": False, "passes": 0, "end": "CR6"}
if event == "PARROT PASS":
return {"in_force": True, "passes": reset["passes"], "end": "-"}
if event == "AUDIT FINDING":
if account_state == "SUSPENDED":
return {"in_force": False, "passes": 0, "end": "AUDIT"}
return {"in_force": True, "passes": reset["passes"], "end": "-"}
passes = reset["passes"] + 1
if account_state in ("GREEN", "YELLOW"):
return {"in_force": False, "passes": 0, "end": "PASS"}
if account_state == "ORANGE" and passes >= orange:
return {"in_force": False, "passes": 0, "end": "PASSES"}
return {"in_force": True, "passes": passes, "end": "-"}

def render_reset(d):
return "%s|%d|%s" % (_b(d["in_force"]), d["passes"], d["end"])

# ------------------------------------------------------- historical_view

def historical_view(revisions, target):
if type(revisions) is not list or not _is_name(target):
raise ValueError("revisions")
last = None
seen = set()
where = None
for rv in revisions:
if not _keys_are(rv, ("rev", "allocated")):
raise ValueError("revisions")
if not _is_int(rv["rev"], 0, COUNT_MAX) or (last is not None and rv["rev"] <= last):
raise ValueError("revisions")
if type(rv["allocated"]) is not list:
raise ValueError("revisions")
for x in rv["allocated"]:
if not _is_name(x) or x in seen:
raise ValueError("revisions")
seen.add(x)
if x == target:
where = rv["rev"]
last = rv["rev"]
if where is None:
raise ValueError("target")
view = None
ids = []
for rv in revisions:
if rv["rev"] < where:
view = rv["rev"]
ids.extend(rv["allocated"])
return {"rev": view, "ids": ids}

def render_view(d):
return "%s|%s" % ("-" if d["rev"] is None else str(d["rev"]),
",".join(_esc(x) for x in d["ids"]))

# ---------------------------------------------------------------- corpus

FNS = ("uniform", "compare", "load", "el", "ratio", "schedule", "probe",
"step", "reset", "view")

def run_case(fn, args):
"""One corpus case: the render, or "ERROR:" and the message."""
if fn not in FNS:
raise ValueError("fn")
try:
p = args.get("params")
if fn == "uniform":
return _f(uniform(args["tag"], args["r"], args["counter"]))
if fn == "compare":
return render_compare(compare(args["kind"], args["original"], args["reproduction"],
args.get("standing"), args.get("k"),
args.get("material_ratio")))
if fn == "load":
return render_load(load(args["records"], args["r"]))
if fn == "el":
return _f(expected_loss(args["e_lo"], args["load"]))
if fn == "ratio":
return render_ratio(ratio_step(args["state"], args["r"], p))
if fn == "schedule":
return render_schedule(schedule_step(args["state"], args["r"], p))
if fn == "probe":
return render_probe(probe_draw(args["schedule"], args["ratio"], args["reset"],
args["has_own_record"], args["r"], p))
if fn == "step":
return render_step(probe_step(args["state"], args["r"], p, args["has_own_record"]))
if fn == "reset":
return render_reset(reset_update(args["reset"], args["event"], args["account_state"], p))
return render_view(historical_view(args["revisions"], args["target"]))
except ValueError as ex:
return "ERROR:" + str(ex)

The Principia Machina/TPC-PY/tpc probe.py - Revision history

BFP Rauch: 1 revision imported

BFP Rauch: TPC-PY source tpc_probe.py (0.2.0), SHA-256 8b99a699b72469a7; conformed TPS 0.4.0; PROPOSED, not declared (TPC-PY 0.2.0)

BFP Rauch: 1 revision imported

BFP Rauch: TPC-PY source tpc_probe.py (0.1.0), SHA-256 b49228a3ae8d06ec; conformed TPS 0.3.0; PROPOSED, not declared (CO-8)

← Older revision	Revision as of 02:11, 30 September 2026
(No difference)

← Older revision	Revision as of 04:37, 28 September 2026
(No difference)