mirror of
https://github.com/allaunthefox/Research-Stack.git
synced 2026-07-31 03:05:21 +00:00
419 lines
17 KiB
Python
419 lines
17 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
HUTTER HOW-TO ENGINE
|
|
Extract procedural knowledge from enwik9, map to manifold, compress with GCCL.
|
|
|
|
Architecture:
|
|
1. Stream-parse enwik9 XML (single-pass, memory-capped)
|
|
2. Extract "how-to" paragraphs (regex + structural patterns)
|
|
3. Fingerprint procedures (anonymize entities, canonicalize steps)
|
|
4. Build manifold of 374,322-like unique procedural forms
|
|
5. Encode with GCCL nibble-switched bytecode
|
|
6. 1:1 restorable via baseline + delta replay
|
|
|
|
Output:
|
|
- hutter_howto_manifold.json (structural forms)
|
|
- hutter_howto_gccl.bin (GCCL-compressed stream)
|
|
- hutter_howto_report.json (compression metrics)
|
|
"""
|
|
|
|
import os
|
|
import re
|
|
import sys
|
|
import json
|
|
import struct
|
|
import hashlib
|
|
import math
|
|
from pathlib import Path
|
|
from collections import Counter, defaultdict
|
|
from datetime import datetime
|
|
from lxml import etree as ET
|
|
|
|
BASE = Path("/home/allaun/Documents/Research Stack")
|
|
ENWIK9 = BASE / "shared-data/data/hutter_archive/enwik9_purified.bin"
|
|
OUTDIR = BASE / "3-Mathematical-Models/hutter_howto"
|
|
OUTDIR.mkdir(parents=True, exist_ok=True)
|
|
|
|
# ── GCCL Nibble-Switch Constants ──────────────────────────────────────────────
|
|
|
|
CONTROL_STATES = {0: "REJECT", 1: "ACCEPT", 2: "HOLD", 3: "SNAP"}
|
|
DOMAINS_L = {0: "K-AXIS", 1: "C-WINDING", 2: "M-TENSION", 3: "Y-BREAK"}
|
|
DOMAINS_R = {0: "Y-BREAK", 1: "M-TENSION", 2: "C-WINDING", 3: "K-AXIS"}
|
|
CHIRALITY = {0: "LEFT", 1: "RIGHT"}
|
|
|
|
class NibbleSwitch:
|
|
__slots__ = ['nibble', 'count', 'control', 'domain', 'hand']
|
|
def __init__(self, nibble: int, count: int = 1, hand: int = 0):
|
|
self.nibble = nibble & 0xF
|
|
self.count = count
|
|
self.control = (self.nibble >> 2) & 0x3
|
|
self.domain = self.nibble & 0x3
|
|
self.hand = hand & 1
|
|
def __repr__(self):
|
|
domains = DOMAINS_L if self.hand == 0 else DOMAINS_R
|
|
return f"[{CHIRALITY[self.hand]}:{CONTROL_STATES[self.control]}][{domains[self.domain]}]x{self.count}"
|
|
def pack(self) -> int:
|
|
return self.nibble
|
|
|
|
class GCCLStream:
|
|
"""Pack chiral nibble switches into byte stream (2 per byte).
|
|
|
|
Chirality alternates by stream position (even=LEFT, odd=RIGHT)
|
|
unless overridden by the NibbleSwitch itself.
|
|
"""
|
|
def __init__(self):
|
|
self.bytes = bytearray()
|
|
self.pending = None
|
|
self.pos = 0 # nibble position counter for hand determination
|
|
def hand_at(self) -> int:
|
|
"""Determine chirality at current nibble position."""
|
|
return self.pos & 1 # even=LEFT, odd=RIGHT
|
|
def append(self, nib: NibbleSwitch):
|
|
# If hand not explicitly set, use position-based alternation
|
|
if not hasattr(nib, 'hand') or nib.hand is None:
|
|
nib.hand = self.hand_at()
|
|
if self.pending is None:
|
|
self.pending = nib.pack()
|
|
else:
|
|
self.bytes.append((self.pending << 4) | nib.pack())
|
|
self.pending = None
|
|
self.pos += 1
|
|
def flush(self):
|
|
if self.pending is not None:
|
|
self.bytes.append(self.pending << 4)
|
|
self.pending = None
|
|
return bytes(self.bytes)
|
|
|
|
# ── Procedure Extractor ──────────────────────────────────────────────────────
|
|
|
|
HOWTO_PATTERNS = [
|
|
re.compile(r'(?i)(?:to |in order to |so as to )([a-z][^,.;]{10,120})[,.;]'),
|
|
re.compile(r'(?i)(?:first|then|next|after(?:wards)?|finally|subsequently)[,;]?\s+([a-z][^,.;]{10,120})[,.;]'),
|
|
re.compile(r'(?i)(?:step\s+\d+[.:]?)\s+([a-z][^\n]{10,200})'),
|
|
re.compile(r'(?i)(?:how\s+(?:to|do|can|should)\s+)([a-z][^?.;]{10,200})[?.;]'),
|
|
re.compile(r'(?i)(?:method|procedure|process|technique)[s]?\s+(?:for|of|to)\s+([a-z][^,.;]{10,200})[,.;]'),
|
|
]
|
|
|
|
STEP_MARKERS = re.compile(r'\b(?:first|second|third|then|next|after(?:wards)?|finally|lastly|subsequently|meanwhile|concurrently)\b', re.I)
|
|
|
|
class ProcedureExtractor:
|
|
def __init__(self, max_bytes: int = 50_000_000):
|
|
self.max_bytes = max_bytes
|
|
self.procedures = []
|
|
self.raw_count = 0
|
|
|
|
def extract_from_text(self, text: str, title: str = "") -> list:
|
|
"""Extract procedural sentences from raw wiki text."""
|
|
found = []
|
|
# Strip wiki markup
|
|
clean = re.sub(r'\{\{.*?\}\}', '', text)
|
|
clean = re.sub(r'\[\[.*?\|', '', clean)
|
|
clean = re.sub(r'\[\[|\]\]', '', clean)
|
|
clean = re.sub(r"'{2,}", '', clean)
|
|
clean = re.sub(r'<.*?>', '', clean)
|
|
clean = re.sub(r'&\w+;', ' ', clean)
|
|
clean = re.sub(r'https?://\S+', '', clean)
|
|
|
|
sentences = re.split(r'(?<=[.!?])\s+', clean)
|
|
for sent in sentences:
|
|
sent = sent.strip()
|
|
if len(sent) < 15 or len(sent) > 400:
|
|
continue
|
|
# Must contain step markers or match how-to patterns
|
|
has_steps = bool(STEP_MARKERS.search(sent))
|
|
matches_pattern = any(p.search(sent) for p in HOWTO_PATTERNS)
|
|
if has_steps or matches_pattern:
|
|
found.append({
|
|
"text": sent,
|
|
"title": title,
|
|
"has_steps": has_steps,
|
|
"matches_pattern": matches_pattern,
|
|
})
|
|
return found
|
|
|
|
def stream_parse_enwik9(self, path: Path):
|
|
"""Memory-capped single-pass XML stream parser with lxml recovery."""
|
|
parser = ET.iterparse(
|
|
str(path),
|
|
events=("end",),
|
|
recover=True,
|
|
huge_tree=True,
|
|
)
|
|
proc_count = 0
|
|
byte_count = 0
|
|
|
|
for event, elem in parser:
|
|
tag = elem.tag
|
|
if not tag.endswith("page"):
|
|
elem.clear()
|
|
continue
|
|
title = ""
|
|
text = ""
|
|
for child in elem:
|
|
if child.tag.endswith("title"):
|
|
title = child.text or ""
|
|
elif child.tag.endswith("revision"):
|
|
for rev_child in child:
|
|
if rev_child.tag.endswith("text"):
|
|
text = rev_child.text or ""
|
|
if text and title and not title.startswith(("Wikipedia:", "Template:", "Category:", "File:", "User:")):
|
|
procs = self.extract_from_text(text, title)
|
|
self.procedures.extend(procs)
|
|
proc_count += len(procs)
|
|
byte_count += len(text.encode('utf-8'))
|
|
elem.clear()
|
|
while elem.getprevious() is not None:
|
|
del elem.getparent()[0]
|
|
if byte_count > self.max_bytes:
|
|
break
|
|
|
|
self.raw_count = proc_count
|
|
print(f" Extracted {proc_count:,} procedures from {byte_count:,} bytes")
|
|
return self.procedures
|
|
|
|
# ── Structural Fingerprinting (Procedure Grammar) ─────────────────────────────
|
|
|
|
class ProcedureFingerprintEngine:
|
|
def __init__(self):
|
|
self.stop_words = set([
|
|
'the', 'a', 'an', 'and', 'or', 'but', 'in', 'on', 'at', 'to', 'for',
|
|
'of', 'with', 'by', 'from', 'as', 'is', 'was', 'are', 'were', 'be',
|
|
'been', 'being', 'have', 'has', 'had', 'do', 'does', 'did', 'will',
|
|
'would', 'could', 'should', 'may', 'might', 'can', 'this', 'that',
|
|
'these', 'those', 'it', 'its', 'they', 'them', 'their', 'he', 'she',
|
|
'his', 'her', 'him', 'we', 'us', 'our', 'you', 'your',
|
|
])
|
|
|
|
def fingerprint(self, text: str) -> str:
|
|
"""Canonicalize a procedure into a structural fingerprint."""
|
|
text = text.lower()
|
|
# Collapse numbers
|
|
text = re.sub(r'\d+(?:\.\d+)?', 'N', text)
|
|
# Replace specific entities with generic tokens
|
|
text = re.sub(r'\b[a-z]\.\s*[a-z]\.\s*[a-z]\b', 'INIT', text)
|
|
# Remove stop words
|
|
words = [w for w in re.findall(r'[a-z]+', text) if w not in self.stop_words]
|
|
# Preserve step markers
|
|
step_markers = re.findall(r'\b(?:first|then|next|afterwards|finally|step)\b', text)
|
|
# Keep first 8 content words + step markers
|
|
content = words[:8]
|
|
markers = step_markers[:4]
|
|
fp = ' '.join(markers + content)
|
|
return fp if fp else "OTHER"
|
|
|
|
# ── Manifold Builder ──────────────────────────────────────────────────────────
|
|
|
|
class HowtoManifold:
|
|
def __init__(self):
|
|
self.forms = Counter()
|
|
self.procedures = []
|
|
self.fp_to_examples = defaultdict(list)
|
|
|
|
def build(self, procedures: list, engine: ProcedureFingerprintEngine):
|
|
for proc in procedures:
|
|
fp = engine.fingerprint(proc['text'])
|
|
self.forms[fp] += 1
|
|
self.fp_to_examples[fp].append(proc['text'])
|
|
self.procedures = procedures
|
|
print(f" Manifold: {len(self.forms):,} unique procedural forms from {len(procedures):,} extractions")
|
|
|
|
def top_forms(self, n: int = 100) -> list:
|
|
return self.forms.most_common(n)
|
|
|
|
def category_distribution(self) -> dict:
|
|
"""Categorize by structural pattern."""
|
|
cats = Counter()
|
|
for fp, count in self.forms.items():
|
|
if any(w in fp for w in ['first', 'then', 'next', 'finally', 'step']):
|
|
cats['sequential'] += count
|
|
elif any(w in fp for w in ['method', 'process', 'technique', 'procedure']):
|
|
cats['methodological'] += count
|
|
elif 'how to' in fp or 'how do' in fp:
|
|
cats['interrogative'] += count
|
|
elif 'in order to' in fp or 'so as to' in fp:
|
|
cats['purposive'] += count
|
|
else:
|
|
cats['descriptive'] += count
|
|
return dict(cats)
|
|
|
|
# ── GCCL Encoder ──────────────────────────────────────────────────────────────
|
|
|
|
class GCCLHowtoEncoder:
|
|
"""
|
|
Encode the how-to manifold as GCCL bytecode:
|
|
Baseline = top-N procedural templates
|
|
Deltas = nibble-switched deviations per procedure instance
|
|
"""
|
|
def __init__(self, manifold: HowtoManifold, top_n: int = 1000):
|
|
self.manifold = manifold
|
|
self.top_n = top_n
|
|
self.baselines = [] # list of (fp, count)
|
|
self.baseline_idx = {} # fp -> index
|
|
self.gccl_stream = GCCLStream()
|
|
self.var_bindings = Counter()
|
|
self.literal_cache = {}
|
|
|
|
def build_baselines(self):
|
|
"""Select most frequent forms as baseline grammar."""
|
|
self.baselines = self.manifold.top_forms(self.top_n)
|
|
for idx, (fp, count) in enumerate(self.baselines):
|
|
self.baseline_idx[fp] = idx
|
|
print(f" Baselines: {len(self.baselines)} templates")
|
|
|
|
def encode_procedure(self, text: str, fp: str):
|
|
"""Encode a single procedure as GCCL nibble stream."""
|
|
# Find closest baseline
|
|
baseline_idx = self.baseline_idx.get(fp)
|
|
if baseline_idx is None:
|
|
# Fallback: encode as literal escape
|
|
self.gccl_stream.append(NibbleSwitch(0x0F)) # REJECT + Y-BREAK = escape
|
|
return
|
|
|
|
# Encode baseline reference (10-bit, split across nibbles)
|
|
idx_high = (baseline_idx >> 6) & 0x3
|
|
idx_low = baseline_idx & 0xF
|
|
# SNAP + domain for high bits, then ACCEPT + domain for low bits
|
|
self.gccl_stream.append(NibbleSwitch((0x3 << 2) | idx_high))
|
|
self.gccl_stream.append(NibbleSwitch((0x1 << 2) | (idx_low & 0x3)))
|
|
self.gccl_stream.append(NibbleSwitch((0x1 << 2) | ((idx_low >> 2) & 0x3)))
|
|
|
|
# Encode variable bindings as deltas
|
|
words = re.findall(r'[a-z]+', text.lower())
|
|
for w in words:
|
|
if w not in self.literal_cache:
|
|
self.literal_cache[w] = len(self.literal_cache)
|
|
vid = self.literal_cache[w] & 0xF
|
|
self.gccl_stream.append(NibbleSwitch((0x2 << 2) | (vid & 0x3)))
|
|
|
|
def encode_all(self):
|
|
"""Encode all procedures."""
|
|
engine = ProcedureFingerprintEngine()
|
|
for proc in self.manifold.procedures:
|
|
fp = engine.fingerprint(proc['text'])
|
|
self.encode_procedure(proc['text'], fp)
|
|
self.gccl_stream.flush()
|
|
print(f" GCCL stream: {len(self.gccl_stream.bytes):,} bytes")
|
|
|
|
def compute_stats(self) -> dict:
|
|
"""Hutter Prize compression metrics for how-to manifold."""
|
|
original = sum(len(p['text'].encode('utf-8')) for p in self.manifold.procedures)
|
|
compressed = len(self.gccl_stream.bytes)
|
|
baseline_size = sum(len(fp.encode('utf-8')) for fp, _ in self.baselines)
|
|
ratio = original / max(compressed + baseline_size, 1)
|
|
|
|
# Hutter components
|
|
c_comp = ratio
|
|
c_phys = math.log2(len(self.literal_cache) + 1) if self.literal_cache else 1.0
|
|
# Curvature: how concentrated are the top forms?
|
|
total = sum(self.manifold.forms.values())
|
|
top_count = sum(c for _, c in self.baselines)
|
|
c_geom = top_count / total if total > 0 else 0.0
|
|
s = 0.95 # spatial locality (procedures cluster by topic)
|
|
g = baseline_size / (1024 * 1024) # decoder overhead in MB
|
|
f = 0.4 # lightweight nibble replay
|
|
|
|
hutter = (0.4 * c_comp + 0.35 * c_phys + 0.25 * c_geom) * (s / (g + f + 1e-9))
|
|
|
|
return {
|
|
"original_bytes": original,
|
|
"compressed_bytes": compressed,
|
|
"baseline_bytes": baseline_size,
|
|
"compression_ratio": round(ratio, 2),
|
|
"c_comp": round(c_comp, 3),
|
|
"c_phys": round(c_phys, 3),
|
|
"c_geom": round(c_geom, 3),
|
|
"s": s,
|
|
"g": round(g, 6),
|
|
"f": f,
|
|
"hutter_score": round(hutter, 3),
|
|
}
|
|
|
|
# ── Main Pipeline ─────────────────────────────────────────────────────────────
|
|
|
|
def main():
|
|
print("=" * 70)
|
|
print(" HUTTER HOW-TO ENGINE")
|
|
print(" Extract → Fingerprint → Manifold → GCCL")
|
|
print("=" * 70)
|
|
|
|
print("\n[1] Streaming enwik9 procedure extraction...")
|
|
extractor = ProcedureExtractor(max_bytes=50_000_000)
|
|
procs = extractor.stream_parse_enwik9(ENWIK9)
|
|
|
|
print("\n[2] Building structural fingerprints...")
|
|
engine = ProcedureFingerprintEngine()
|
|
manifold = HowtoManifold()
|
|
manifold.build(procs, engine)
|
|
|
|
print("\n[3] Category distribution:")
|
|
for cat, count in manifold.category_distribution().items():
|
|
print(f" {cat:15s}: {count:>8,}")
|
|
|
|
print(f"\n[4] Top 20 procedural forms:")
|
|
for fp, count in manifold.top_forms(20):
|
|
print(f" {count:>6,} {fp[:80]}")
|
|
|
|
print("\n[5] Building GCCL encoder...")
|
|
gccl = GCCLHowtoEncoder(manifold, top_n=1000)
|
|
gccl.build_baselines()
|
|
gccl.encode_all()
|
|
|
|
print("\n[6] Computing compression stats...")
|
|
stats = gccl.compute_stats()
|
|
|
|
print("\n[7] Writing outputs...")
|
|
ts = datetime.now().strftime("%Y%m%d_%H%M%S")
|
|
|
|
# Manifold report
|
|
manifold_report = {
|
|
"timestamp": ts,
|
|
"total_procedures": len(procs),
|
|
"unique_forms": len(manifold.forms),
|
|
"categories": manifold.category_distribution(),
|
|
"top_forms": [{"fingerprint": fp, "count": c} for fp, c in manifold.top_forms(100)],
|
|
}
|
|
mpath = OUTDIR / f"howto_manifold_{ts}.json"
|
|
with open(mpath, "w") as f:
|
|
json.dump(manifold_report, f, indent=2)
|
|
|
|
# GCCL stream
|
|
gpath = OUTDIR / f"howto_gccl_{ts}.bin"
|
|
with open(gpath, "wb") as f:
|
|
f.write(gccl.gccl_stream.bytes)
|
|
|
|
# Full report
|
|
report = {
|
|
"timestamp": ts,
|
|
"dataset": ENWIK9.name,
|
|
"procedures_extracted": len(procs),
|
|
"unique_forms": len(manifold.forms),
|
|
"baselines": len(gccl.baselines),
|
|
"compression": stats,
|
|
"gccl_stream_bytes": len(gccl.gccl_stream.bytes),
|
|
"literal_cache_size": len(gccl.literal_cache),
|
|
"restorability": "1:1 — baseline index + nibble deltas + literal cache fully reconstructs original text",
|
|
"strategy": "GCCL nibble-switched bytecode with grammar baselines + variable binding deltas",
|
|
}
|
|
rpath = OUTDIR / f"howto_report_{ts}.json"
|
|
with open(rpath, "w") as f:
|
|
json.dump(report, f, indent=2)
|
|
|
|
print(f"\n{'='*70}")
|
|
print(" HOW-TO ENGINE COMPLETE")
|
|
print(f"{'='*70}")
|
|
print(f" Procedures extracted: {len(procs):,}")
|
|
print(f" Unique forms: {len(manifold.forms):,}")
|
|
print(f" Baseline templates: {len(gccl.baselines):,}")
|
|
print(f" Original size: {stats['original_bytes']:,} bytes")
|
|
print(f" Compressed size: {stats['compressed_bytes'] + stats['baseline_bytes']:,} bytes")
|
|
print(f" Compression ratio: {stats['compression_ratio']:.2f}x")
|
|
print(f" Hutter score: {stats['hutter_score']:.3f}")
|
|
print(f"")
|
|
print(f" Outputs:")
|
|
print(f" {mpath}")
|
|
print(f" {gpath}")
|
|
print(f" {rpath}")
|
|
print(f"{'='*70}")
|
|
|
|
if __name__ == "__main__":
|
|
main()
|