The Principia Machina/TPC-PY/tpc strike.py: Difference between revisions
Appearance
m 1 revision imported |
m 1 revision imported |
||
| (One intermediate revision by the same user not shown) | |||
| Line 31: | Line 31: | ||
lists as "[a,b,...]". Nothing here exposes an enforcement path or the | lists as "[a,b,...]". Nothing here exposes an enforcement path or the | ||
canary/probe suite. | canary/probe suite. | ||
0.3.0 (PROPOSED, increment 12; D196.1, RD-RS-29, RD-RS-33, RD-RS-36; | |||
tpm_record_functions 0.2.0 section 9): three functions added over the | |||
record field schema (tpm_record_schema 0.3.0): conditions_of(records, id), | |||
comparison_set(records, words, tfk, quantity) and sr8_groups(records, | |||
words, tfk, quantity). They read the register's keyed records and a WORDS | |||
map {id: status word} the caller computes with tpc_record status_words; | |||
this module imports nothing of tpc_record. Vectors: tpc/vectors/ | |||
conditions_corpus.json (gen_conditions_corpus.py), answered through | |||
tpc_record's run_case. tpc_units is imported on first use (tpc_units | |||
imports this module). sr3_scope is unchanged, kept for row 21. | |||
L-L contradiction (SR3) is not here: it needs a derivation/logic oracle and | L-L contradiction (SR3) is not here: it needs a derivation/logic oracle and | ||
| Line 38: | Line 49: | ||
import math | import math | ||
TPC_STRIKE_VERSION = "0. | TPC_STRIKE_VERSION = "0.3.0" # increment 12 (S95: one MINOR bump over the declared 0.2.0 across increments 11 and 12): conditions_of, comparison_set, sr8_groups (D196.1); 0.2.1 C-MS-5; 0.2.0 sr3_scope, D180.3 | ||
| Line 372: | Line 383: | ||
raise ValueError("sr6: N_x is a TMS parameter, at least 2") | raise ValueError("sr6: N_x is a TMS parameter, at least 2") | ||
return int(n_ind) >= int(n_x) | return int(n_ind) >= int(n_x) | ||
# -------------------------------------------------------------------------- | |||
# CONDITIONS (D196.1; tpm_record_functions 0.2.0 section 9): which pairs of | |||
# records SR3 compares and which groups SR8 scales, over the record field | |||
# schema. Refusals are tpc_record's: ["REFUSED", [[code, field, message]]]. | |||
# -------------------------------------------------------------------------- | |||
def _units(): | |||
"""tpc_units, imported on first use (it imports this module).""" | |||
import tpc_units | |||
return tpc_units | |||
def _cshow(v): | |||
"""A slot value in a message (tpm_record_schema 12.1): every code point | |||
of general category C, and U+2028 and U+2029, written <U+XXXX>.""" | |||
import unicodedata | |||
return "".join("<U+%04X>" % ord(c) if unicodedata.category(c)[0] == "C" or c in ("
", "
") | |||
else c for c in v) | |||
def _ccj(v): | |||
"""TPM's canonical JSON (tpm_record_schema 11.4), a double by canon.""" | |||
if v is True: | |||
return "true" | |||
if v is False: | |||
return "false" | |||
if v is None: | |||
return "null" | |||
if type(v) is int: | |||
return str(v) | |||
if type(v) is float: | |||
return canon(v) | |||
if type(v) is str: | |||
out = ['"'] | |||
for ch in v: | |||
if ch == '"': | |||
out.append('\\"') | |||
elif ch == "\\": | |||
out.append("\\\\") | |||
elif ord(ch) < 0x20: | |||
out.append("\\u%04x" % ord(ch)) | |||
else: | |||
out.append(ch) | |||
out.append('"') | |||
return "".join(out) | |||
if type(v) is list: | |||
return "[" + ",".join(_ccj(x) for x in v) + "]" | |||
if type(v) is dict: | |||
return "{" + ",".join(_ccj(k) + ":" + _ccj(v[k]) for k in sorted(v)) + "}" | |||
raise TypeError("canonical JSON: unsupported type %r" % type(v)) | |||
def _cone(code, field, msg): | |||
return ["REFUSED", [[code, field, msg]]] | |||
def _cbad(argument, value, what): | |||
return _cone("REC-50", "-", "%s %s is not %s" % (argument, _cshow(_ccj(value)), what)) | |||
def _cv(v): | |||
"""<v>: a string shown as given, any other value as its canonical JSON.""" | |||
return _cshow(v) if type(v) is str else _cshow(_ccj(v)) | |||
def _ckeyed(records): | |||
"""{id: record} of the keyed members, or None when records is not a | |||
register (a list whose members are objects with a string ID, no ID | |||
twice, or list-form records, which are ignored).""" | |||
if type(records) is not list: | |||
return None | |||
keyed = {} | |||
for r in records: | |||
if type(r) is dict: | |||
if type(r.get("ID")) is not str or r["ID"] in keyed: | |||
return None | |||
keyed[r["ID"]] = r | |||
elif type(r) is not list: | |||
return None | |||
return keyed | |||
def _cmembers(v): | |||
return [x for x in v if type(x) is str] if type(v) is list else [] | |||
def _cconditions(keyed, rid): | |||
"""[situations, at] of a RESULT or VERIFICATION rid, or a refusal.""" | |||
rec = keyed[rid] | |||
base = rec | |||
if rec.get("CLASS") == "VERIFICATION": # RD-RS-36 | |||
base = keyed.get(rec.get("TARGET")) if type(rec.get("TARGET")) is str else None | |||
situations = [] | |||
if base is not None: | |||
view = {} | |||
for k, r in keyed.items(): | |||
view[k] = {"assumptions": _cmembers(r.get("ASSUMPTIONS")), | |||
"cites": [c for c in _cmembers(r.get("CITES")) if c in keyed]} | |||
for a in assumption_closure(base["ID"], view): | |||
r = keyed.get(a) | |||
if r is not None and r.get("CLASS") == "ASSUMPTION" and "SITUATION" in r: | |||
situations.append(a) | |||
situations.sort() | |||
at = rec.get("AT", []) | |||
points = [] | |||
ok = type(at) is list | |||
U = _units() | |||
for q in (at if ok else []): | |||
if type(q) is not dict or type(q.get("DOF")) is not str: | |||
ok = False | |||
break | |||
n = U.normalize_measurement(q.get("VALUE"), "0", q.get("UNIT")) | |||
if n[0] != "OK": | |||
ok = False | |||
break | |||
points.append([q["DOF"], _zone().canon_decimal(canon(n[1])), n[4]]) | |||
if not ok: | |||
return _cbad("AT", at, "a list of points that normalize") | |||
points.sort() | |||
return ["OK", [situations, points]] | |||
def conditions_of(records, id): | |||
"""9.1: ["OK", [situations, at]] of a RESULT or a VERIFICATION of the | |||
register: the SITUATION assumptions of its closure (a VERIFICATION's | |||
TARGET's, RD-RS-36), sorted, and its AT, each point [DOF, v, kind] | |||
with v the canonical decimal of the coherent-SI value, sorted by DOF. | |||
REFUSED: REC-50, REC-52, REC-73.""" | |||
keyed = _ckeyed(records) | |||
if keyed is None: | |||
return _cbad("records", records, "a register") | |||
if type(id) is not str or id not in keyed: | |||
return _cone("REC-52", "-", "record %s is not in the register" % _cv(id)) | |||
if keyed[id].get("CLASS") not in ("RESULT", "VERIFICATION"): | |||
return _cone("REC-73", "-", "conditions_of reads a RESULT or a VERIFICATION") | |||
return _cconditions(keyed, id) | |||
def _cscope(records, words, tfk, quantity): | |||
"""(keyed, fixed) or (None, refusal) for comparison_set and sr8_groups.""" | |||
keyed = _ckeyed(records) | |||
if keyed is None: | |||
return None, _cbad("records", records, "a register") | |||
if type(words) is not dict or not all(type(v) is str for v in words.values()): | |||
return None, _cbad("words", words, "a map of status words") | |||
if type(quantity) is not str: | |||
return None, _cone("REC-74", "-", "quantity %s is not an entry of the TrM" % _cv(quantity)) | |||
sc = sr3_scope(tfk, quantity) | |||
if type(sc) is list: | |||
if sc[1].startswith("QUANTITY "): | |||
return None, _cone("REC-74", "-", "quantity %s is not an entry of the TrM" % _cv(quantity)) | |||
return None, _cone("REC-39", "-", "the TFK is not readable") | |||
return keyed, sc == SR3_MECHANICAL | |||
def _cmembers_of(keyed, words, quantity, with_l): | |||
"""The standing RESULTs naming QUANTITY quantity: REGIME M, and with | |||
with_l REGIME L carrying VALUE; sorted in code point order.""" | |||
out = [] | |||
for rid in sorted(keyed): | |||
r = keyed[rid] | |||
if r.get("CLASS") != "RESULT" or r.get("QUANTITY") != quantity: | |||
continue | |||
if words.get(rid, "NOT GRADED") not in ("PROVISIONAL", "ESTABLISHED"): | |||
continue | |||
if r.get("REGIME") == "M" or (with_l and r.get("REGIME") == "L" and "VALUE" in r): | |||
out.append(rid) | |||
return out | |||
def comparison_set(records, words, tfk, quantity): | |||
"""9.2: ["OK", [[id1, id2, mode], ...]], the pairs SR3 compares among | |||
the standing RESULTs of quantity (M, and L carrying VALUE; never two L | |||
records) whose CONDITIONS are equal: mode CONVENTION (equal but for a | |||
point's KIND), TEST (the entry's DOF FIXed, or a non-empty AT) or | |||
INSPECT (row 21). REFUSED: REC-39, REC-50, REC-74.""" | |||
keyed, fixed = _cscope(records, words, tfk, quantity) | |||
if keyed is None: | |||
return fixed | |||
ids = _cmembers_of(keyed, words, quantity, True) | |||
cond = {} | |||
for rid in ids: | |||
c = _cconditions(keyed, rid) | |||
if c[0] != "OK": | |||
return c | |||
cond[rid] = c[1] | |||
out = [] | |||
for i in range(len(ids)): | |||
for j in range(i + 1, len(ids)): | |||
a, b = ids[i], ids[j] | |||
if keyed[a].get("REGIME") == "L" and keyed[b].get("REGIME") == "L": | |||
continue | |||
ca, cb = cond[a], cond[b] | |||
if ca[0] != cb[0] or [q[:2] for q in ca[1]] != [q[:2] for q in cb[1]]: | |||
continue | |||
if ca[1] != cb[1]: | |||
mode = "CONVENTION" | |||
elif fixed or ca[1]: | |||
mode = "TEST" | |||
else: | |||
mode = "INSPECT" | |||
out.append([a, b, mode]) | |||
return ["OK", out] | |||
def sr8_groups(records, words, tfk, quantity): | |||
"""9.3: ["OK", [[mode, [ids]], ...]], the standing M RESULTs of quantity | |||
grouped by equal CONDITIONS (a convention splits a group), ids sorted, | |||
groups by their first id; mode TEST when the entry's DOF is FIXed or | |||
the group's points are non-empty, else INSPECT. REFUSED: REC-39, | |||
REC-50, REC-74.""" | |||
keyed, fixed = _cscope(records, words, tfk, quantity) | |||
if keyed is None: | |||
return fixed | |||
groups = [] | |||
for rid in _cmembers_of(keyed, words, quantity, False): | |||
c = _cconditions(keyed, rid) | |||
if c[0] != "OK": | |||
return c | |||
for g in groups: | |||
if g[0] == c[1]: | |||
g[1].append(rid) | |||
break | |||
else: | |||
groups.append([c[1], [rid]]) | |||
return ["OK", [["TEST" if fixed or g[0][1] else "INSPECT", g[1]] for g in groups]] | |||
Latest revision as of 02:11, 30 September 2026
# -*- coding: utf-8 -*-
"""
TPC-PY --- strike-rule arithmetic (TPC, step 2 first increment)
===============================================================
The Principia Calculator, published advisory residence (TPC-PY).
Pure functions, Python standard library only, Python 3.7+.
Runs unmodified on Windows and Linux (no OS calls, no third-party imports).
Scope of this module: the fully-specified strike-rule arithmetic named in
PMP-TPM STRIKE RULE / TPS 4.4 --- SR4 distance d(r), SR8 uncertainty
scaling, the SR3 mechanical contradiction tests (M-M, L-M), SR5 ordering,
and the SR6 independence count. Digests (need the TPS 2.5 canonicalization),
the SR9 C-test harness, the isomorphism bodies and the kernel tests T1-T6
are later increments and are NOT in this file.
0.2.0 (PROPOSED, unreleased working label; D180.3, taken 2026-09-26): one
function added, sr3_scope(tfk, quantity), the interim scope of the SR3 M-M
and L-M tests and of SR8's scale factor (TPS 4.4.13 and 4.2.13 as staged
for revision 6, provisional numbers). Vectors: tpc/vectors/
sr3_scope_corpus.json (gen_sr3_scope_corpus.py). No earlier function
changed. sr3_scope reads with tpc_zone's helpers, imported on first use
(so sr3_scope needs Python 3.8+, as tpc_zone does; the rest stays 3.7+).
Residence model (D96.1): this file is the published, runnable residence.
The residence of record is TPC-PHP inside TPX, whose PUBLISHED form is
pseudocode (tpc_php/tpc_strike.pseudo.php). Both residences MUST reproduce
the published vector corpus (tpc/vectors/strike_corpus.json) byte for byte;
the corpus, compared through canon(), is the shared oracle. "Byte-identical"
is defined by canon(): booleans as "true"/"false", integers as base-10,
floats via the C printf format "%.12g" (identical in Python and PHP), and
lists as "[a,b,...]". Nothing here exposes an enforcement path or the
canary/probe suite.
0.3.0 (PROPOSED, increment 12; D196.1, RD-RS-29, RD-RS-33, RD-RS-36;
tpm_record_functions 0.2.0 section 9): three functions added over the
record field schema (tpm_record_schema 0.3.0): conditions_of(records, id),
comparison_set(records, words, tfk, quantity) and sr8_groups(records,
words, tfk, quantity). They read the register's keyed records and a WORDS
map {id: status word} the caller computes with tpc_record status_words;
this module imports nothing of tpc_record. Vectors: tpc/vectors/
conditions_corpus.json (gen_conditions_corpus.py), answered through
tpc_record's run_case. tpc_units is imported on first use (tpc_units
imports this module). sr3_scope is unchanged, kept for row 21.
L-L contradiction (SR3) is not here: it needs a derivation/logic oracle and
is an INSPECTION check, not a mechanical TPC function.
"""
import math
TPC_STRIKE_VERSION = "0.3.0" # increment 12 (S95: one MINOR bump over the declared 0.2.0 across increments 11 and 12): conditions_of, comparison_set, sr8_groups (D196.1); 0.2.1 C-MS-5; 0.2.0 sr3_scope, D180.3
# --------------------------------------------------------------------------
# Canonical serialization --- the byte-identity contract across residences.
# --------------------------------------------------------------------------
def _canon_float(x):
"""C "%.12g" of a double, derived from the digits of "%.11e": one
rounding to 12 significant digits, then fixed notation for exponents
-4..11 and exponent notation (sign, at least two digits) otherwise,
trailing zeros removed. Written out so that TPC-PHP can mirror it."""
if x != x: # NaN guard: never a valid TPC output
raise ValueError("canon: NaN is not a permitted result")
if x in (float("inf"), float("-inf")):
return "inf" if x > 0 else "-inf"
if x == 0:
return "0" # 0.0 and -0.0 alike
mant, exp = ("%.11e" % abs(x)).split("e")
digits = mant.replace(".", "") # twelve significant digits
e = int(exp)
if -4 <= e < 12:
if e >= 0:
ip, fp = digits[:e + 1], digits[e + 1:]
else:
ip, fp = "0", "0" * (-e - 1) + digits
fp = fp.rstrip("0")
out = ip + ("." + fp if fp else "")
else:
rest = digits[1:].rstrip("0")
out = digits[0] + ("." + rest if rest else "") + "e" + ("-" if e < 0 else "+") + "%02d" % abs(e)
return ("-" if x < 0 else "") + out
def canon(value):
"""Canonical string form of a result, identical across residences.
bool -> "true" / "false" (checked before int: bool subclasses int)
int -> base-10
float -> C printf "%.12g", built by _canon_float from the twelve
significant digits of "%.11e" so that both residences produce
it byte for byte (PHP's own sprintf('%.12g') writes exponents
differently: "2.0e-9" where C writes "2e-09"); -0.0 -> "0";
infinities -> "inf" / "-inf"
str -> itself
list/tuple -> "[e1,e2,...]" of canon(e)
"""
if isinstance(value, bool):
return "true" if value else "false"
if isinstance(value, int):
return str(value)
if isinstance(value, float):
return _canon_float(value)
if isinstance(value, str):
return value
if isinstance(value, (list, tuple)):
return "[" + ",".join(canon(v) for v in value) + "]"
raise TypeError("canon: unsupported type %r" % type(value))
# --------------------------------------------------------------------------
# SR4 --- distance d(r) = (a(r), s(r)), lexicographic, lower is closer.
# --------------------------------------------------------------------------
def assumption_closure(rid, records):
"""The set of assumption identifiers in the transitive closure of r:
r's own ASSUMPTIONS plus those of every record it cites, transitively.
Cited-record cycles are tolerated (a visited set guards them).
'records' maps record id -> {"assumptions":[aid...], "cites":[rid...]}.
"""
seen_records = set()
acc = set()
stack = [rid]
while stack:
cur = stack.pop()
if cur in seen_records:
continue
seen_records.add(cur)
rec = records[cur]
acc.update(rec.get("assumptions", []))
stack.extend(rec.get("cites", []))
return acc
def a_of_r(rid, records, tfk_assumptions=()):
"""SR4 a(r): count of DISTINCT assumption identifiers in r's closure,
EXCLUDING the TFK's own assumptions. Counted over identifiers (D41)."""
tfk = set(tfk_assumptions)
return len(assumption_closure(rid, records) - tfk)
def d_of_r(rid, records, tfk_assumptions=()):
"""SR4 d(r) = (a(r), s(r)). s(r) is r's own derivation-step count under
the UWC's STEP CONVENTION, carried on the record as 'steps'."""
a = a_of_r(rid, records, tfk_assumptions)
s = int(records[rid]["steps"])
return (a, s)
def d_compare(d1, d2):
"""Lexicographic compare of two d() tuples. -1 if d1 closer, 1 if d2
closer, 0 if equal (SR4; used by SR5)."""
if d1 < d2:
return -1
if d1 > d2:
return 1
return 0
# --------------------------------------------------------------------------
# SR8 --- uncertainty discipline / Birge scale factor.
# --------------------------------------------------------------------------
def sr8_scale_factor(measurements):
"""SR8 scale factor S for N>=2 standing M records of ONE quantity,
each {"x": value, "u": u_c > 0}, already normalized to one UNIT.
Weighted mean over the included set, reduced chi-square, and, per the
PDG Birge rule the spec cites, S computed over records with
u_i < 3*sqrt(N)*mean(u). S = sqrt(reduced) when reduced > 1, else 1.0.
Returns 1.0 for N < 2. Raises on a non-positive uncertainty.
"""
n = len(measurements)
if n < 2:
return 1.0
us = [float(m["u"]) for m in measurements]
xs = [float(m["x"]) for m in measurements]
if any(u <= 0.0 for u in us):
raise ValueError("sr8: every M record needs u_c > 0")
# The operations below follow TPC-PHP, the residence of record, one for
# one and left to right: explicit accumulation (Python 3.12's sum() adds
# floats with compensation, PHP does not), x * w with w = 1/(u*u), and
# d * d rather than d ** 2 (pow need not round as a product does).
u_mean = 0.0
for u in us:
u_mean = u_mean + u
u_mean = u_mean / n
cutoff = 3.0 * math.sqrt(n) * u_mean
incl = [i for i in range(n) if us[i] < cutoff]
if len(incl) < 2:
return 1.0
wsum, xw = 0.0, 0.0
for i in incl:
w = 1.0 / (us[i] * us[i])
wsum = wsum + w
xw = xw + xs[i] * w
xbar = xw / wsum
chi2 = 0.0
for i in incl:
d = xs[i] - xbar
chi2 = chi2 + (d * d) / (us[i] * us[i])
reduced = chi2 / (len(incl) - 1)
return math.sqrt(reduced) if reduced > 1.0 else 1.0
# --------------------------------------------------------------------------
# SR3 --- mechanical contradiction tests (M-M, L-M). k is a TMS parameter.
# --------------------------------------------------------------------------
def mm_contradicts(x1, u1, x2, u2, k, S=1.0):
"""SR3 M-M: |x1 - x2| > k * sqrt((S u1)^2 + (S u2)^2), after SR8 scaling.
Caller guarantees same QUANTITY and one UNIT (TPX refuses cross-entry)."""
su1 = S * float(u1)
su2 = S * float(u2)
return abs(float(x1) - float(x2)) > float(k) * math.sqrt(su1 * su1 + su2 * su2)
def lm_contradicts(x_l, x_m, u_m, k, S=1.0):
"""SR3 L-M: the M value fails the M-M test against the L record's derived
value. The L result carries no uncertainty, so the bound is k*S*u_m."""
return abs(float(x_m) - float(x_l)) > float(k) * S * float(u_m)
# --------------------------------------------------------------------------
# SR3 and SR8 for trajectories, INTERIM (D180.3): which QUANTITY is tested.
# --------------------------------------------------------------------------
SR3_MECHANICAL = "MECHANICAL"
SR3_INSPECTION = "INSPECTION"
def _zone():
"""tpc_zone, imported on first use (its helpers read the TFK's shape)."""
import tpc_zone
return tpc_zone
def _sr3_field(tag, e, f):
"""None when e[f] is a string; else the problem, in trm_well_formed's
and t1_well_formed's words."""
if f not in e:
return tag + ": missing field " + f
if not isinstance(e[f], str):
return tag + ": field " + f + " has the wrong type"
return None
def sr3_scope(tfk, quantity):
"""The interim scope of an M record's QUANTITY under SR3 and SR8, under
D180.3 until the record field schema (C-16) is written: TPX runs the SR3
M-M and L-M tests and SR8's scale factor only over quantities scoped
MECHANICAL, and lists the rest for row-21 inspection (TPS 4.4.13 and
4.2.13 as staged, provisional numbers). When C-16 lands, a CONDITIONS
field on the records replaces this interim (T-C3).
tfk: a TFK {"TPK", "TrM", "TMS"} in the parsed shape tpc_zone validates
(the caller has validated it). quantity: the QUANTITY an M record names,
a TrM entry NAME. Returns
"MECHANICAL" the TrM entry of that NAME exists and its DOF is the
TARGET of a TPK CONSTRAINT whose ACTION is FIX;
"INSPECTION" the entry exists and no FIX CONSTRAINT targets its DOF
(a REMOVE constraint does not count; the independent
variable is an ordinary DOF here);
["REFUSED", reason] no TrM entry has that NAME, or the input is not of
the shape read. Never raised; a reason that quotes input
escapes disallowed code points (zone schemas 6.3).
Matching is exact: NAMEs are ASCII identifiers, and a QUANTITY that is
not one is refused without being echoed.
WHAT IS READ, in this order, the first problem refusing: the QUANTITY;
the TFK; the TrM and its ENTRIES, each entry's NAME and DOF (every
entry, before the answer); the QUANTITY resolved (none: refused; two
entries of that NAME: refused as ambiguous); the TPK and its CONSTRAINT
kind (absent: none), each constraint's TARGET and ACTION (every
constraint). Nothing else is read or checked: not the TMS, not the other
fields or kinds, not the values T1, the TrM schema check, the IC checks
and T3 own (a TARGET that is no DOF, an ACTION other than FIX or REMOVE,
an entry on a REMOVEd DOF, a DOF both FIXed and REMOVEd). A FIX counts
only as ACTION "FIX" exactly with TARGET equal to the entry's DOF."""
Z = _zone()
def refuse(msg):
return ["REFUSED", Z._show(msg)]
if not Z._is_ident(quantity):
return refuse("QUANTITY is not an identifier")
if not Z._obj(tfk):
return refuse("TFK is not an object")
trm = tfk.get("TrM")
if not Z._obj(trm):
return refuse("TrM: zone is not an object")
if "ENTRIES" not in trm:
return refuse("TrM: missing part ENTRIES")
ents = trm["ENTRIES"]
if not Z._list(ents):
return refuse("TrM ENTRIES is not a list")
found = []
for e in ents:
if not Z._obj(e):
return refuse("TrM ENTRIES: entry is not an object")
tag = "ENTRY:" + Z._name(e)
for f in ("NAME", "DOF"):
p = _sr3_field(tag, e, f)
if p is not None:
return refuse(p)
if e["NAME"] == quantity:
found.append(e["DOF"])
if not found:
return refuse("QUANTITY " + quantity + " names no TrM entry")
if len(found) > 1:
return refuse("TrM: duplicate entry name " + quantity)
dof = found[0]
tpk = tfk.get("TPK")
if not Z._obj(tpk):
return refuse("TPK: zone is not an object")
cons = tpk.get("CONSTRAINT", [])
if not Z._list(cons):
return refuse("TPK kind CONSTRAINT is not a list")
fixed = False
for cn in cons:
if not Z._obj(cn):
return refuse("TPK kind CONSTRAINT: entry is not an object")
tag = "CONSTRAINT:" + Z._name(cn)
for f in ("TARGET", "ACTION"):
p = _sr3_field(tag, cn, f)
if p is not None:
return refuse(p)
if cn["ACTION"] == "FIX" and cn["TARGET"] == dof:
fixed = True
return SR3_MECHANICAL if fixed else SR3_INSPECTION
# --------------------------------------------------------------------------
# SR5 --- ordering of two contradicting standing records.
# --------------------------------------------------------------------------
def sr5_order(d1, d2):
"""SR5: of two contradicting standing records, the smaller d stands and
the other is struck; equal d opens an audit block. d1, d2 are d() tuples.
Returns "r1", "r2", or "AUDIT"."""
c = d_compare(tuple(d1), tuple(d2))
if c < 0:
return "r1"
if c > 0:
return "r2"
return "AUDIT"
# --------------------------------------------------------------------------
# SR6 --- independence count of the contradictions of a standing record r.
# --------------------------------------------------------------------------
_SR6_MAX_INPUTS = 24 # exact set-packing guard; realistic contradiction sets are tiny
def sr6_independence_count(extra_assumption_sets, same_model_flags=None):
"""SR6 n_ind(r): the largest sub-family of the contradictions whose
extra-assumption sets are PAIRWISE DISJOINT (as identifier sets).
A corroboration marked SAME-MODEL is excluded from the input first (D42).
extra_assumption_sets: list of iterables of assumption identifiers.
same_model_flags: optional list of bool, same length; True drops that one.
"""
sets = [frozenset(s) for s in extra_assumption_sets]
if same_model_flags is not None:
sets = [s for s, sm in zip(sets, same_model_flags) if not sm]
if len(sets) > _SR6_MAX_INPUTS:
raise ValueError("sr6: input exceeds exact packing guard (%d)" % _SR6_MAX_INPUTS)
best = [0]
def bt(i, used, count):
if count + (len(sets) - i) <= best[0]:
return
if i == len(sets):
if count > best[0]:
best[0] = count
return
bt(i + 1, used, count) # skip set i
if used.isdisjoint(sets[i]): # take set i
bt(i + 1, used | sets[i], count + 1)
bt(0, frozenset(), 0)
return best[0]
def sr6_audit_opens(n_ind, n_x):
"""SR6: an audit block opens on r when n_ind(r) reaches N_x (>= 2).
Reaching N_x never strikes r by itself; it opens the block."""
if int(n_x) < 2:
raise ValueError("sr6: N_x is a TMS parameter, at least 2")
return int(n_ind) >= int(n_x)
# --------------------------------------------------------------------------
# CONDITIONS (D196.1; tpm_record_functions 0.2.0 section 9): which pairs of
# records SR3 compares and which groups SR8 scales, over the record field
# schema. Refusals are tpc_record's: ["REFUSED", [[code, field, message]]].
# --------------------------------------------------------------------------
def _units():
"""tpc_units, imported on first use (it imports this module)."""
import tpc_units
return tpc_units
def _cshow(v):
"""A slot value in a message (tpm_record_schema 12.1): every code point
of general category C, and U+2028 and U+2029, written <U+XXXX>."""
import unicodedata
return "".join("<U+%04X>" % ord(c) if unicodedata.category(c)[0] == "C" or c in ("
", "
")
else c for c in v)
def _ccj(v):
"""TPM's canonical JSON (tpm_record_schema 11.4), a double by canon."""
if v is True:
return "true"
if v is False:
return "false"
if v is None:
return "null"
if type(v) is int:
return str(v)
if type(v) is float:
return canon(v)
if type(v) is str:
out = ['"']
for ch in v:
if ch == '"':
out.append('\\"')
elif ch == "\\":
out.append("\\\\")
elif ord(ch) < 0x20:
out.append("\\u%04x" % ord(ch))
else:
out.append(ch)
out.append('"')
return "".join(out)
if type(v) is list:
return "[" + ",".join(_ccj(x) for x in v) + "]"
if type(v) is dict:
return "{" + ",".join(_ccj(k) + ":" + _ccj(v[k]) for k in sorted(v)) + "}"
raise TypeError("canonical JSON: unsupported type %r" % type(v))
def _cone(code, field, msg):
return ["REFUSED", [[code, field, msg]]]
def _cbad(argument, value, what):
return _cone("REC-50", "-", "%s %s is not %s" % (argument, _cshow(_ccj(value)), what))
def _cv(v):
"""<v>: a string shown as given, any other value as its canonical JSON."""
return _cshow(v) if type(v) is str else _cshow(_ccj(v))
def _ckeyed(records):
"""{id: record} of the keyed members, or None when records is not a
register (a list whose members are objects with a string ID, no ID
twice, or list-form records, which are ignored)."""
if type(records) is not list:
return None
keyed = {}
for r in records:
if type(r) is dict:
if type(r.get("ID")) is not str or r["ID"] in keyed:
return None
keyed[r["ID"]] = r
elif type(r) is not list:
return None
return keyed
def _cmembers(v):
return [x for x in v if type(x) is str] if type(v) is list else []
def _cconditions(keyed, rid):
"""[situations, at] of a RESULT or VERIFICATION rid, or a refusal."""
rec = keyed[rid]
base = rec
if rec.get("CLASS") == "VERIFICATION": # RD-RS-36
base = keyed.get(rec.get("TARGET")) if type(rec.get("TARGET")) is str else None
situations = []
if base is not None:
view = {}
for k, r in keyed.items():
view[k] = {"assumptions": _cmembers(r.get("ASSUMPTIONS")),
"cites": [c for c in _cmembers(r.get("CITES")) if c in keyed]}
for a in assumption_closure(base["ID"], view):
r = keyed.get(a)
if r is not None and r.get("CLASS") == "ASSUMPTION" and "SITUATION" in r:
situations.append(a)
situations.sort()
at = rec.get("AT", [])
points = []
ok = type(at) is list
U = _units()
for q in (at if ok else []):
if type(q) is not dict or type(q.get("DOF")) is not str:
ok = False
break
n = U.normalize_measurement(q.get("VALUE"), "0", q.get("UNIT"))
if n[0] != "OK":
ok = False
break
points.append([q["DOF"], _zone().canon_decimal(canon(n[1])), n[4]])
if not ok:
return _cbad("AT", at, "a list of points that normalize")
points.sort()
return ["OK", [situations, points]]
def conditions_of(records, id):
"""9.1: ["OK", [situations, at]] of a RESULT or a VERIFICATION of the
register: the SITUATION assumptions of its closure (a VERIFICATION's
TARGET's, RD-RS-36), sorted, and its AT, each point [DOF, v, kind]
with v the canonical decimal of the coherent-SI value, sorted by DOF.
REFUSED: REC-50, REC-52, REC-73."""
keyed = _ckeyed(records)
if keyed is None:
return _cbad("records", records, "a register")
if type(id) is not str or id not in keyed:
return _cone("REC-52", "-", "record %s is not in the register" % _cv(id))
if keyed[id].get("CLASS") not in ("RESULT", "VERIFICATION"):
return _cone("REC-73", "-", "conditions_of reads a RESULT or a VERIFICATION")
return _cconditions(keyed, id)
def _cscope(records, words, tfk, quantity):
"""(keyed, fixed) or (None, refusal) for comparison_set and sr8_groups."""
keyed = _ckeyed(records)
if keyed is None:
return None, _cbad("records", records, "a register")
if type(words) is not dict or not all(type(v) is str for v in words.values()):
return None, _cbad("words", words, "a map of status words")
if type(quantity) is not str:
return None, _cone("REC-74", "-", "quantity %s is not an entry of the TrM" % _cv(quantity))
sc = sr3_scope(tfk, quantity)
if type(sc) is list:
if sc[1].startswith("QUANTITY "):
return None, _cone("REC-74", "-", "quantity %s is not an entry of the TrM" % _cv(quantity))
return None, _cone("REC-39", "-", "the TFK is not readable")
return keyed, sc == SR3_MECHANICAL
def _cmembers_of(keyed, words, quantity, with_l):
"""The standing RESULTs naming QUANTITY quantity: REGIME M, and with
with_l REGIME L carrying VALUE; sorted in code point order."""
out = []
for rid in sorted(keyed):
r = keyed[rid]
if r.get("CLASS") != "RESULT" or r.get("QUANTITY") != quantity:
continue
if words.get(rid, "NOT GRADED") not in ("PROVISIONAL", "ESTABLISHED"):
continue
if r.get("REGIME") == "M" or (with_l and r.get("REGIME") == "L" and "VALUE" in r):
out.append(rid)
return out
def comparison_set(records, words, tfk, quantity):
"""9.2: ["OK", [[id1, id2, mode], ...]], the pairs SR3 compares among
the standing RESULTs of quantity (M, and L carrying VALUE; never two L
records) whose CONDITIONS are equal: mode CONVENTION (equal but for a
point's KIND), TEST (the entry's DOF FIXed, or a non-empty AT) or
INSPECT (row 21). REFUSED: REC-39, REC-50, REC-74."""
keyed, fixed = _cscope(records, words, tfk, quantity)
if keyed is None:
return fixed
ids = _cmembers_of(keyed, words, quantity, True)
cond = {}
for rid in ids:
c = _cconditions(keyed, rid)
if c[0] != "OK":
return c
cond[rid] = c[1]
out = []
for i in range(len(ids)):
for j in range(i + 1, len(ids)):
a, b = ids[i], ids[j]
if keyed[a].get("REGIME") == "L" and keyed[b].get("REGIME") == "L":
continue
ca, cb = cond[a], cond[b]
if ca[0] != cb[0] or [q[:2] for q in ca[1]] != [q[:2] for q in cb[1]]:
continue
if ca[1] != cb[1]:
mode = "CONVENTION"
elif fixed or ca[1]:
mode = "TEST"
else:
mode = "INSPECT"
out.append([a, b, mode])
return ["OK", out]
def sr8_groups(records, words, tfk, quantity):
"""9.3: ["OK", [[mode, [ids]], ...]], the standing M RESULTs of quantity
grouped by equal CONDITIONS (a convention splits a group), ids sorted,
groups by their first id; mode TEST when the entry's DOF is FIXed or
the group's points are non-empty, else INSPECT. REFUSED: REC-39,
REC-50, REC-74."""
keyed, fixed = _cscope(records, words, tfk, quantity)
if keyed is None:
return fixed
groups = []
for rid in _cmembers_of(keyed, words, quantity, False):
c = _cconditions(keyed, rid)
if c[0] != "OK":
return c
for g in groups:
if g[0] == c[1]:
g[1].append(rid)
break
else:
groups.append([c[1], [rid]])
return ["OK", [["TEST" if fixed or g[0][1] else "INSPECT", g[1]] for g in groups]]