File size: 6,114 Bytes
2a64ad4 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 |
#!/usr/bin/env python3
"""
Commitment Conservation Harness CLI
Runs the operational harness via a single command so users don't need to
navigate internal modules. Outputs JSON/CSV receipts.
"""
import os
import json
import argparse
from datetime import datetime
# Force non-GUI plotting backend (prevents macOS blocking)
os.environ.setdefault("MPLBACKEND", "Agg")
def _now_iso() -> str:
return datetime.utcnow().replace(microsecond=0).isoformat() + "Z"
def simple_extraction(text: str, quiet: bool = False, as_json: bool = False) -> int:
"""Simple commitment extraction (default mode)."""
try:
from src.extraction import extract_hard_commitments
import spacy
except ImportError as e:
print(f"Import error: {e}", file=__import__('sys').stderr)
return 1
try:
nlp = spacy.load("en_core_web_sm")
except OSError:
print("Error: spaCy model 'en_core_web_sm' not found.", file=__import__('sys').stderr)
print("Install with: python -m spacy download en_core_web_sm", file=__import__('sys').stderr)
return 1
commitments = extract_hard_commitments(text, nlp)
if as_json:
import json
print(json.dumps({"input": text, "commitments": sorted(list(commitments))}, indent=2))
elif quiet:
for c in sorted(commitments):
print(c)
else:
print(f"Extracted {len(commitments)} commitment(s) from: \"{text[:60]}{'...' if len(text) > 60 else ''}\"")
if commitments:
for i, c in enumerate(sorted(commitments), 1):
print(f" {i}. {c}")
else:
print(" (none)")
return 0
def main() -> int:
import sys
# Check if this is a 'run' subcommand or simple extraction
if len(sys.argv) > 1 and sys.argv[1] == "run":
# Experimental mode
return run_experiment()
else:
# Simple extraction mode
return run_simple_extraction()
def run_simple_extraction() -> int:
"""Simple extraction CLI."""
import sys
p = argparse.ArgumentParser(
prog="commitment-harness",
description="Extract commitments from text.",
epilog="For full experiments, use: python analyze.py run {compression|recursion|full}"
)
p.add_argument("text", help="Text to analyze")
p.add_argument("--quiet", "-q", action="store_true", help="Output only commitments (no headers)")
p.add_argument("--json", action="store_true", help="Output as JSON")
args = p.parse_args()
return simple_extraction(args.text, quiet=args.quiet, as_json=args.json)
def run_experiment() -> int:
"""Experimental harness CLI."""
import sys
p = argparse.ArgumentParser(
prog="commitment-harness run",
description="Run commitment conservation experiments and export receipts."
)
sub = p.add_subparsers(dest="experiment", required=True)
# compression experiment
pc = sub.add_parser("compression", help="Run compression sweep on a signal.")
pc.add_argument("--signal", required=True, help="Input signal text.")
pc.add_argument("--out", default="outputs/compression_receipt.json", help="Output receipt path (json).")
# recursion experiment
pr = sub.add_parser("recursion", help="Run recursion test on a signal.")
pr.add_argument("--signal", required=True, help="Input signal text.")
pr.add_argument("--depth", type=int, default=8, help="Recursion depth.")
pr.add_argument("--enforced", action="store_true", help="Use enforcement mode.")
pr.add_argument("--out", default="outputs/recursion_receipt.json", help="Output receipt path (json).")
# full pipeline
pf = sub.add_parser("full", help="Run the deterministic pipeline (if available).")
pf.add_argument("--out", default="outputs/full_receipt.json", help="Output receipt path (json).")
# Remove 'run' from argv so argparse sees the subcommand correctly
sys.argv.pop(1)
args = p.parse_args()
os.makedirs(os.path.dirname(args.out), exist_ok=True)
# Import from your harness implementation
try:
from src.deterministic_pipeline import compression_sweep, recursion_test, deterministic_pipeline
except ImportError:
# Fallback to test_harness if deterministic_pipeline doesn't exist
try:
from src.test_harness import compression_sweep, recursion_test
deterministic_pipeline = None
except Exception as e:
raise SystemExit(f"Import error: {e}\n(Verify you run this from the harness/ directory.)")
receipt = {
"timestamp_utc": _now_iso(),
"experiment": args.experiment,
"python": {
"mpl_backend": os.environ.get("MPLBACKEND"),
},
}
if args.experiment == "compression":
sigma_vals, fid_vals = compression_sweep(args.signal)
receipt.update({
"input_signal": args.signal,
"n": len(fid_vals),
"sigma_values": sigma_vals,
"fidelities": fid_vals,
})
elif args.experiment == "recursion":
deltas = recursion_test(args.signal, depth=args.depth, enforced=args.enforced) if hasattr(recursion_test, '__code__') and 'enforced' in recursion_test.__code__.co_varnames else recursion_test(args.signal, depth=args.depth)
receipt.update({
"input_signal": args.signal,
"depth": args.depth,
"enforced": args.enforced if hasattr(recursion_test, '__code__') and 'enforced' in recursion_test.__code__.co_varnames else False,
"deltas": deltas,
})
elif args.experiment == "full":
if deterministic_pipeline is None:
raise SystemExit("deterministic_pipeline not available. (Missing src/deterministic_pipeline.py import.)")
result = deterministic_pipeline()
receipt.update({"result": result})
with open(args.out, "w", encoding="utf-8") as f:
json.dump(receipt, f, indent=2, ensure_ascii=False)
print(f"✓ Wrote receipt: {args.out}")
return 0
if __name__ == "__main__":
raise SystemExit(main())
|