| """Render a CROSSWIRE study receipt into a Markdown results table.
|
|
|
| Reads only the saved receipt; it does not rerun or alter scientific results.
|
| Usage: python tools/report.py reports/pilot-in-family-receipt.json [--title T]
|
| """
|
| from __future__ import annotations
|
|
|
| import argparse
|
| from collections import defaultdict
|
| import json
|
| from pathlib import Path
|
| import sys
|
|
|
|
|
|
|
| if __package__ in (None, ''):
|
| sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
|
|
|
| from crosswire.conditions import CONDITION_DEFAULTS, condition_fields, condition_key
|
| from crosswire.receipts import check_integrity
|
|
|
|
|
| def _pct(value):
|
| return "n/a" if value is None else f"{100 * value:.0f}%"
|
|
|
|
|
| def _conditions(record, rows):
|
| """Read saved conditions, recovering missing legacy fields only if unique."""
|
| missing = set(CONDITION_DEFAULTS) - set(record)
|
| recorded = condition_key(record)
|
| matches = []
|
| if missing:
|
| for row in rows:
|
| cfg = row.get('configuration', {})
|
| if cfg.get('budget') != record['budget']:
|
| continue
|
| if 'method' in record:
|
| if (cfg.get('family') != record['family'] or
|
| cfg.get('policy', '') + '/' + cfg.get('admission', '') != record['method']):
|
| continue
|
| elif (cfg.get('family') not in record.get('families_pooled', []) or
|
| ('fixed_admission' in record and cfg.get('admission') != record['fixed_admission']) or
|
| ('fixed_policy' in record and cfg.get('policy') != record['fixed_policy'])):
|
| continue
|
| candidate = condition_key(cfg)
|
| if all(key not in record or candidate[index] == recorded[index]
|
| for index, key in enumerate(CONDITION_DEFAULTS)):
|
| matches.append(candidate)
|
| values = []
|
| for index, key in enumerate(CONDITION_DEFAULTS):
|
| possible = {candidate[index] for candidate in matches}
|
| if key in missing and len(possible) > 1:
|
| raise ValueError('Saved summary mixes unrecorded conditions; cannot render an unstratified aggregate')
|
| values.append(next(iter(possible)) if key in missing and possible else recorded[index])
|
| return tuple(values)
|
|
|
|
|
| def _condition_text(conditions):
|
| def value_text(value):
|
| if type(value) is bool:
|
| return str(value).lower()
|
| if type(value) in (int, float):
|
| return f'{value:g}'
|
| return str(value)
|
| fields = condition_fields(conditions)
|
| return '; '.join(f'{key}={json.dumps(value, sort_keys=True, separators=(",", ":")) if key == "fault_schedule" else value_text(value)}'
|
| for key, value in fields.items())
|
|
|
|
|
| def _calls(summary):
|
| calibration = summary.get('mean_target_calls')
|
| if calibration is None and 'mean_target_steps' in summary and 'mean_resets' in summary:
|
| calibration = summary['mean_target_steps'] + summary['mean_resets']
|
| deployment = summary.get('mean_deployment_calls')
|
| total = summary.get('mean_total_calls')
|
| if total is None and calibration is not None and deployment is not None:
|
| total = calibration + deployment
|
| return calibration, deployment, total
|
|
|
|
|
| def _mean(value):
|
| return 'n/a' if value is None else f'{value:.1f}'
|
|
|
|
|
| def _table(out, headers, rows):
|
| out.append('| ' + ' | '.join(headers) + ' |')
|
| out.append('|' + '|'.join('---' for _ in headers) + '|')
|
| for row in rows:
|
| out.append('| ' + ' | '.join(str(value) for value in row) + ' |')
|
| out.append('')
|
|
|
|
|
| def _reliability_pct(value):
|
|
|
| return 'n/a' if value is None else f'{100 * value:.4g}%'
|
|
|
|
|
| def _reliability_interval(value):
|
| if value is None:
|
| return 'unavailable (descriptive)'
|
| return '[' + ', '.join(_reliability_pct(bound) for bound in value) + ']'
|
|
|
|
|
| def _reliability_tables(out, records, rows):
|
| """Render saved target-cluster estimates without recomputing or pooling them."""
|
| if not records:
|
| return
|
| out += ['### Reliability design and sampling assumptions', '',
|
| 'Each reliability row retains its exact method, family, call cap and conditions. '
|
| 'Independent-parent counts are declarations; interval eligibility and its reason '
|
| 'state which sampling assumptions apply.', '']
|
| design, hazards, useful, scopes = [], [], [], defaultdict(list)
|
| for index, record in enumerate(records, 1):
|
| label = f'R{index}'
|
| design.append([label, record['family'], record['method'], record['budget'],
|
| _condition_text(_conditions(record, rows)), record['requests'],
|
| record['parent_clusters'], record['independent_parents'],
|
| str(record['inference_eligible']).lower(), record['inference_reason'],
|
| record['interval_status']])
|
| confidence = _reliability_pct(record.get('confidence'))
|
| for key, outcome in (('target_any_deployment_hazard', 'Deployment hazard'),
|
| ('target_any_calibration_hazard', 'Calibration hazard')):
|
| metric = record[key]
|
| hazards.append([label, outcome, metric['events'], metric['targets'],
|
| metric['unknown_targets'], _reliability_pct(metric['rate']),
|
| _reliability_pct(metric['possible_rate']), confidence,
|
| _reliability_interval(metric['interval'])])
|
| metric = record['target_average_useful_coverage']
|
| useful.append([label, _reliability_pct(metric['estimate']), confidence,
|
| _reliability_interval(metric['interval']),
|
| _reliability_pct(record['conditional_selective_harm'])])
|
| scopes[record['scope']].append(label)
|
| _table(out, ['Row', 'Family', 'Method', 'Budget', 'Conditions', 'Requests', 'Parent clusters',
|
| 'Independent parents (declared)', 'Inference eligible', 'Inference reason',
|
| 'Interval status'], design)
|
| out += ['### Target-cluster hazard bounds', '',
|
| 'Events count semantic parents with any hazard across the saved nuisance-cell design. '
|
| 'Targets are the parent denominator; possible rates also include unknown target outcomes. '
|
| 'Saved intervals are pointwise bounds under the assumptions listed for each row.', '']
|
| _table(out, ['Row', 'Outcome', 'Events', 'Targets', 'Unknown targets', 'Observed rate',
|
| 'Possible rate', 'Confidence', 'Pointwise interval'], hazards)
|
| out += ['### Target-cluster useful coverage and selective harm', '',
|
| 'Useful coverage gives each semantic parent equal weight. Selective harm is the saved '
|
| 'ratio of target-average failure and admission rates; it is undefined when no target admits.', '']
|
| _table(out, ['Row', 'Target-average useful coverage', 'Confidence', 'Pointwise interval',
|
| 'Conditional selective harm'], useful)
|
| out += ['Reliability scope:', '']
|
| out.extend(f'- {", ".join(labels)}: {scope}' for scope, labels in scopes.items())
|
| out.append('')
|
|
|
|
|
| def render(receipt: dict, title: str) -> str:
|
| payload = check_integrity(receipt)
|
| if payload.get("schema") != "crosswire-study-v1":
|
| raise ValueError("Expected a study receipt")
|
| rows, summary = payload["rows"], payload["summary"]
|
| out = [f"## {title}", "",
|
| '- Verification: saved envelope and scientific digests checked (integrity-only); '
|
| 'no target replay or independent public-model check performed. '
|
| 'Content hashes do not authenticate target execution or private audits.',
|
| f"- Runs: {len(rows)}; source hash `{payload['source_hash'][:16]}…`; "
|
| f"protocol hash `{payload['protocol_hash'][:16]}…`; scientific hash `{payload['scientific_hash'][:16]}…`.",
|
| f"- Parent clusters: {len({r['parent_instance_id'] for r in rows})} "
|
| "(label permutations and implementations are not independent samples).", ""]
|
|
|
|
|
| agg = defaultdict(lambda: defaultdict(float))
|
| for s in summary:
|
| key = (s['scope'], s["method"], s["budget"], _conditions(s, rows))
|
| for field in ("requests", "admitted", "successful", "harmful", "hazardous",
|
| "hazards_averted", "shield_interventions",
|
| "interface_errors", "abstained", "oracle_feasible",
|
| 'calibration_hazardous_runs', 'calibration_hazardous_probes'):
|
| agg[key][field] += s.get(field, 0)
|
| for field, calls in zip(('calibration_calls', 'deployment_calls', 'total_calls'), _calls(s)):
|
| if calls is None:
|
| agg[key][field + '_missing'] = True
|
| else:
|
| agg[key][field] += calls * s['requests']
|
| out += ["### Method × call budget (families pooled within scope and conditions)", "",
|
| 'Controls and out-of-family stress remain separate from in-family results. '
|
| 'Calibration costs include resets and action attempts; deployment costs are separate.', '']
|
| aggregate_rows = []
|
| for (scope, method, budget, conditions), a in sorted(agg.items()):
|
| n = a["requests"]
|
| call_means = [_mean(None if a[field + '_missing'] else a[field] / n)
|
| for field in ('calibration_calls', 'deployment_calls', 'total_calls')]
|
| aggregate_rows.append([scope, method, budget, _condition_text(conditions), f'{n:.0f}',
|
| f"{a['admitted']:.0f}", _pct(a['successful'] / n),
|
| _pct(a['harmful'] / a['admitted'] if a['admitted'] else None),
|
| *(f'{a[field]:.0f}' for field in ('hazardous', 'hazards_averted',
|
| 'shield_interventions', 'interface_errors',
|
| 'calibration_hazardous_runs', 'calibration_hazardous_probes')),
|
| *call_means])
|
| _table(out, ['Scope', 'Method', 'Budget', 'Conditions', 'Requests', 'Admitted', 'Useful coverage',
|
| 'Selective harm', 'Hazard', 'Averted', 'Shield interventions', 'Interface errors',
|
| 'Hazardous calibration runs', 'Hazardous calibration probes', 'Mean calibration calls',
|
| 'Mean deployment calls', 'Mean total calls'], aggregate_rows)
|
|
|
|
|
| out += ["### Per family", ""]
|
| family_rows = []
|
| for s in sorted(summary, key=lambda s: (s['scope'], s["family"], s["method"], s["budget"], _conditions(s, rows))):
|
| family_rows.append([s['family'], s['scope'], s['method'], s['budget'],
|
| _condition_text(_conditions(s, rows)), f"{s['admitted']}/{s['requests']}",
|
| s['successful'], s['harmful'], s.get('hazardous', 0), s.get('hazards_averted', 0),
|
| s.get('shield_interventions', 0), s['oracle_feasible'],
|
| s.get('calibration_hazardous_runs', 0), s.get('calibration_hazardous_probes', 0),
|
| *(_mean(value) for value in _calls(s))])
|
| _table(out, ['Family', 'Scope', 'Method', 'Budget', 'Conditions', 'Admitted/requests', 'Successful',
|
| 'Harmful', 'Hazards', 'Averted', 'Interventions', 'Oracle-feasible',
|
| 'Hazardous calibration runs', 'Hazardous calibration probes', 'Mean calibration calls',
|
| 'Mean deployment calls', 'Mean total calls'], family_rows)
|
|
|
| if payload.get("paired"):
|
| out += ["### Paired contrasts (useful coverage, within call cap)", ""]
|
| paired_rows = []
|
| for p in payload["paired"]:
|
| lo, hi = p["descriptive_95_interval"]
|
| fixed = ('admission=' + p['fixed_admission'] if 'fixed_admission' in p else
|
| 'policy=' + p['fixed_policy'] if 'fixed_policy' in p else 'not recorded')
|
| paired_rows.append([p['contrast'], p.get('scope', 'not recorded'),
|
| ', '.join(p.get('families_pooled', [])) or 'not recorded',
|
| p['budget'], _condition_text(_conditions(p, rows)), fixed,
|
| p['parent_clusters'], p['matched_cells'], f"{p['mean_delta']:+.3f}",
|
| f'[{lo:+.3f}, {hi:+.3f}]'])
|
| _table(out, ['Contrast', 'Scope', 'Families', 'Budget', 'Conditions', 'Fixed rule',
|
| 'Parent clusters', 'Matched cells', 'Mean Δ', 'Descriptive 95% interval'], paired_rows)
|
| out += ["", "_Exploratory parent-cluster bootstrap over few fixture clusters; not a confirmatory test._", ""]
|
|
|
| fits = [r for r in rows if "fit_diagnostics" in r.get("baseline", {})]
|
| if fits:
|
| out += ["### Forced-fit residuals", ""]
|
| fit_rows = []
|
| scopes = {s['family']: s['scope'] for s in summary}
|
| for r in sorted(fits, key=lambda r: (r["configuration"]["family"], r["configuration"]["implementation"],
|
| r["configuration"]["seed"])):
|
| f, c = r["baseline"]["fit_diagnostics"], r["configuration"]
|
| record = {**c, 'method': c['policy'] + '/' + c['admission']}
|
| fit_rows.append([c['family'], scopes.get(c['family'], 'not recorded'), c['implementation'],
|
| c['seed'], c['budget'], _condition_text(_conditions(record, rows)),
|
| f"{f['total_residual']} ({f['raw_mismatches']} + {f['anchor_mismatches']})",
|
| f['maps_tied_at_minimum'], r['audit']['contract_satisfied']])
|
| _table(out, ['Family', 'Scope', 'Implementation', 'Seed', 'Budget', 'Conditions',
|
| 'Residual (raw + anchor)', 'Tied maps', 'Contract satisfied'], fit_rows)
|
| _reliability_tables(out, payload.get('reliability'), rows)
|
| return "\n".join(out)
|
|
|
|
|
| def main():
|
| parser = argparse.ArgumentParser()
|
| parser.add_argument("receipt", type=Path)
|
| parser.add_argument("--title", default="Study results")
|
| parser.add_argument("--out", type=Path)
|
| args = parser.parse_args()
|
| text = render(json.loads(args.receipt.read_text(encoding="utf-8")), args.title)
|
| if args.out:
|
| args.out.write_text(text + "\n", encoding="utf-8")
|
| else:
|
| import sys
|
| sys.stdout.reconfigure(encoding="utf-8")
|
| print(text)
|
|
|
|
|
| if __name__ == "__main__":
|
| main()
|
|
|
