mirror of
https://github.com/allaunthefox/Research-Stack.git
synced 2026-07-31 03:05:21 +00:00
245 lines
9.3 KiB
Python
245 lines
9.3 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
Parallel Physics Equation Remapper — 4-6 concurrent workers
|
||
|
||
Leverages RTX 4070 SUPER: 12GB VRAM, currently 19% utilized.
|
||
llama3.1:8b (~4.9GB) can run 2-3 concurrent streams.
|
||
With 4 workers: ~4x speedup vs sequential.
|
||
|
||
Flow:
|
||
/dev/shm/physics_equations.db → read
|
||
4-worker ThreadPoolExecutor → LLM calls
|
||
/dev/shm/mapped.jsonl → write
|
||
flush to disk + compression + metaprobe
|
||
"""
|
||
|
||
import ctypes
|
||
import json
|
||
import os
|
||
import re
|
||
import sqlite3
|
||
import sys
|
||
import time
|
||
from concurrent.futures import ThreadPoolExecutor, as_completed
|
||
from dataclasses import dataclass, field
|
||
from pathlib import Path
|
||
from typing import Dict, List, Optional, Tuple
|
||
|
||
import requests
|
||
|
||
# ─── Config ─────────────────────────────────────────────────────────────────
|
||
|
||
OLLAMA_URL = "http://localhost:11434/api/generate"
|
||
MODEL = "llama3.1:8b"
|
||
DB_PATH = "/dev/shm/physics_equations.db"
|
||
SHM_OUT = "/dev/shm/physics_mapped_parallel.jsonl"
|
||
SHM_META = "/dev/shm/physics_metaprobe_parallel.jsonl"
|
||
DISK_DIR = Path("/home/allaun/Documents/Research Stack/3-Mathematical-Models")
|
||
OUTPUT_MD = DISK_DIR / "physics_eqs_mapped_parallel.md"
|
||
COMPRESSION_LOG = DISK_DIR / "physics_compression_parallel.log"
|
||
LIBMOIRE_PATH = "/tmp/libmoire.so"
|
||
MAX_WORKERS = 6 # local GPU: 6 concurrent streams
|
||
|
||
|
||
# ─── Moiré Decoder ──────────────────────────────────────────────────────────
|
||
|
||
class MoireDecoder:
|
||
def __init__(self):
|
||
self.lib = ctypes.CDLL(LIBMOIRE_PATH)
|
||
self.lib.moire_estimate_entropy.argtypes = [ctypes.POINTER(ctypes.c_uint8), ctypes.c_size_t]
|
||
self.lib.moire_estimate_entropy.restype = ctypes.c_double
|
||
|
||
def entropy(self, data: bytes) -> float:
|
||
src = (ctypes.c_uint8 * len(data))(*data)
|
||
return self.lib.moire_estimate_entropy(src, len(data))
|
||
|
||
def ratio(self, data: bytes) -> float:
|
||
ent = self.entropy(data)
|
||
return 8.0 / ent if ent > 0 else 1.0
|
||
|
||
|
||
# ─── LLM Single Equation Call ───────────────────────────────────────────────
|
||
|
||
PROMPT_TEMPLATE = """Map this physics equation to five symbols from:
|
||
Ω = Ψ [ B(θ) ⊗ C(n, α) ] ⊕ Δ(n, θ, α)
|
||
|
||
Equation: {title}
|
||
Domain: {domain}
|
||
Description: {description}
|
||
|
||
Definitions:
|
||
- Ω: Observable output / predicted quantity
|
||
- Ψ: Operator, mechanism, or underlying theory
|
||
- B: Conserved basis / fundamental component
|
||
- C: Dynamic context / variable parameter
|
||
- Δ: Residual error / noise / fundamental limit
|
||
|
||
Respond ONLY in valid JSON with exactly these five keys and short (≤15 words) values:
|
||
{{"Ω": "...", "Ψ": "...", "B": "...", "C": "...", "Δ": "..."}}
|
||
"""
|
||
|
||
|
||
def call_ollama(title: str, domain: str, description: str) -> Optional[Dict]:
|
||
prompt = PROMPT_TEMPLATE.format(title=title, domain=domain, description=description[:300])
|
||
try:
|
||
r = requests.post(
|
||
OLLAMA_URL,
|
||
json={"model": MODEL, "prompt": prompt, "stream": False,
|
||
"options": {"temperature": 0.1, "num_predict": 150}},
|
||
timeout=60,
|
||
)
|
||
r.raise_for_status()
|
||
content = r.json()["response"]
|
||
match = re.search(r'\{[^}]+\}', content)
|
||
if match:
|
||
return json.loads(match.group(0))
|
||
except Exception as e:
|
||
pass
|
||
return None
|
||
|
||
|
||
# ─── Data Loading ───────────────────────────────────────────────────────────
|
||
|
||
def load_equations() -> List[Tuple[int, str, str, str]]:
|
||
conn = sqlite3.connect(DB_PATH)
|
||
cursor = conn.cursor()
|
||
cursor.execute("SELECT eq_number, title, domain_id, significance FROM equations ORDER BY eq_number")
|
||
rows = cursor.fetchall()
|
||
cursor.execute("SELECT id, name FROM domains")
|
||
domains = {str(r[0]): r[1] for r in cursor.fetchall()}
|
||
conn.close()
|
||
return [(eq_num, title, domains.get(str(did), "Unknown"), desc or "")
|
||
for eq_num, title, did, desc in rows]
|
||
|
||
|
||
# ─── Progress Tracking ──────────────────────────────────────────────────────
|
||
|
||
class ProgressTracker:
|
||
def __init__(self, moire: MoireDecoder, total: int):
|
||
self.moire = moire
|
||
self.total = total
|
||
self.success = 0
|
||
self.fail = 0
|
||
self.domain_stats: Dict[str, int] = {}
|
||
self.entropies: List[float] = []
|
||
self.latencies: List[float] = []
|
||
|
||
def record(self, eq_num: int, title: str, domain: str, mapping: Dict, latency_ms: float):
|
||
self.success += 1
|
||
self.domain_stats[domain] = self.domain_stats.get(domain, 0) + 1
|
||
self.latencies.append(latency_ms)
|
||
# Entropy of this mapping
|
||
mapping_text = json.dumps(mapping, sort_keys=True).encode()
|
||
self.entropies.append(self.moire.entropy(mapping_text))
|
||
|
||
def record_fail(self):
|
||
self.fail += 1
|
||
|
||
def summary(self) -> str:
|
||
lines = ["=== Results ===", ""]
|
||
lines.append(f"Total: {self.total} | Success: {self.success} | Fail: {self.fail}")
|
||
if self.latencies:
|
||
avg_lat = sum(self.latencies) / len(self.latencies)
|
||
lines.append(f"Avg latency: {avg_lat:.0f}ms")
|
||
if self.entropies:
|
||
avg_ent = sum(self.entropies) / len(self.entropies)
|
||
lines.append(f"Avg mapping entropy: {avg_ent:.3f} bits/byte")
|
||
lines.append("")
|
||
lines.append("Top domains:")
|
||
for dom, cnt in sorted(self.domain_stats.items(), key=lambda x: -x[1])[:10]:
|
||
lines.append(f" {dom:25s} | {cnt:3d}")
|
||
return "\n".join(lines)
|
||
|
||
|
||
# ─── Main Pipeline ──────────────────────────────────────────────────────────
|
||
|
||
def main():
|
||
print("=" * 60)
|
||
print(f"Parallel Physics Remapper — {MAX_WORKERS} workers")
|
||
print("=" * 60)
|
||
|
||
print("\n[1/4] Loading moiré decoder...")
|
||
moire = MoireDecoder()
|
||
|
||
print("\n[2/4] Reading equations from /dev/shm...")
|
||
equations = load_equations()
|
||
print(f" → {len(equations)} equations")
|
||
|
||
print(f"\n[3/4] Mapping with {MAX_WORKERS} concurrent workers...")
|
||
print("-" * 60)
|
||
|
||
tracker = ProgressTracker(moire, len(equations))
|
||
|
||
# Open output file for incremental writes
|
||
with open(SHM_OUT, "w", encoding="utf-8") as out_f:
|
||
with ThreadPoolExecutor(max_workers=MAX_WORKERS) as executor:
|
||
futures = {}
|
||
for eq_num, title, domain, desc in equations:
|
||
future = executor.submit(call_ollama, title, domain, desc)
|
||
futures[future] = (eq_num, title, domain, desc, time.time())
|
||
|
||
for future in as_completed(futures):
|
||
eq_num, title, domain, desc, t0 = futures[future]
|
||
latency = (time.time() - t0) * 1000
|
||
mapping = future.result()
|
||
|
||
if mapping:
|
||
tracker.record(eq_num, title, domain, mapping, latency)
|
||
out_f.write(json.dumps({
|
||
'eq_number': eq_num,
|
||
'title': title,
|
||
'domain': domain,
|
||
'mapping': mapping,
|
||
}) + "\n")
|
||
out_f.flush()
|
||
print(f" ✓ #{eq_num:3d} [{domain:22s}] lat={latency:5.0f}ms | {title[:40]}...")
|
||
else:
|
||
tracker.record_fail()
|
||
print(f" ✗ #{eq_num:3d} FAILED")
|
||
|
||
print("-" * 60)
|
||
|
||
print("\n[4/4] Writing markdown to disk...")
|
||
lines = [
|
||
"# Physics Equations — Parallel Mapped\n",
|
||
f"**Workers:** {MAX_WORKERS} | **Equation:** Ω = Ψ [ B(θ) ⊗ C(n, α) ] ⊕ Δ(n, θ, α)\n\n",
|
||
"---\n\n",
|
||
]
|
||
with open(SHM_OUT, "r", encoding="utf-8") as f:
|
||
for row in f:
|
||
obj = json.loads(row)
|
||
m = obj['mapping']
|
||
lines.append(f"## Eq {obj['eq_number']}. {obj['title']}\n")
|
||
lines.append(f"**Domain:** {obj['domain']}\n")
|
||
lines.append("| Symbol | Mapping |\n")
|
||
lines.append("|--------|---------|\n")
|
||
for sym in ['Ω', 'Ψ', 'B', 'C', 'Δ']:
|
||
lines.append(f"| {sym} | {m.get(sym, 'N/A')} |\n")
|
||
lines.append("\n---\n\n")
|
||
|
||
with open(OUTPUT_MD, "w", encoding="utf-8") as f:
|
||
f.writelines(lines)
|
||
|
||
# Compression analysis
|
||
data = Path(OUTPUT_MD).read_bytes()
|
||
ent = moire.entropy(data)
|
||
ratio = moire.ratio(data)
|
||
|
||
with open(COMPRESSION_LOG, "w") as f:
|
||
f.write(f"file={OUTPUT_MD}\n")
|
||
f.write(f"size_bytes={len(data)}\n")
|
||
f.write(f"moire_entropy={ent:.6f}\n")
|
||
f.write(f"theoretical_ratio={ratio:.6f}\n")
|
||
f.write(f"workers={MAX_WORKERS}\n")
|
||
f.write(f"total_mapped={tracker.success}\n")
|
||
f.write(f"total_failed={tracker.fail}\n")
|
||
|
||
print(f"\nOutput: {OUTPUT_MD} ({len(data):,} bytes)")
|
||
print(f"Moiré entropy: {ent:.4f} bits/byte")
|
||
print(f"Compression ratio: {ratio:.2f}x")
|
||
print("")
|
||
print(tracker.summary())
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|