| """ |
| METASTATE dual-input compiler. |
| |
| Parses the METASTATE DSL and emits TWO real targets from one source: |
| - CLASSICAL: executable Python/NumPy |
| - QUANTUM: OpenQASM 3.0 that runs on the live IBM worker (/v1/quantum/route) |
| |
| HONEST SCOPE: this is a real, working compiler for a defined DSL subset. It is |
| NOT a raw-CPU-register / AVX-512 assembler — that remains roadmap. The classical |
| target is NumPy (which itself uses vectorised CPU instructions under the hood); |
| the quantum target is genuine OpenQASM 3.0. |
| |
| Supported DSL: |
| hybrid_state psi[N] { target_cpu: ...; target_qpu: ...; } |
| minimize_energy(threshold: T) { psi.evaluate_causal_matrix(); } |
| let NAME = eml(A, B) # eml(x,y) = exp(x) - ln(y) |
| rotate psi[i] by ANGLE |
| entangle psi[i], psi[j] |
| measure psi |
| |
| Example: |
| hybrid_state psi[3] { target_cpu: avx512; target_qpu: ibm; } |
| rotate psi[0] by 0.78 |
| entangle psi[0], psi[1] |
| let r = eml(0.5, 2.0) |
| measure psi |
| """ |
| import re, math |
|
|
| class CompileError(Exception): |
| pass |
|
|
| def _tokenize(src): |
| lines = [] |
| for raw in src.splitlines(): |
| line = raw.split("#", 1)[0].strip() |
| if line: |
| lines.append(line) |
| return lines |
|
|
| def parse(src): |
| """Parse DSL into a simple instruction list + state declaration.""" |
| n_qubits = None |
| targets = {} |
| ops = [] |
| for line in _tokenize(src): |
| m = re.match(r"hybrid_state\s+(\w+)\s*\[\s*(\d+)\s*\]\s*\{(.*)\}", line) |
| if m: |
| n_qubits = int(m.group(2)) |
| body = m.group(3) |
| for part in body.split(";"): |
| if ":" in part: |
| k, v = part.split(":", 1) |
| targets[k.strip()] = v.strip() |
| continue |
| m = re.match(r"rotate\s+\w+\s*\[\s*(\d+)\s*\]\s*by\s+([-\d.]+)", line) |
| if m: |
| ops.append(("rotate", int(m.group(1)), float(m.group(2)))); continue |
| m = re.match(r"entangle\s+\w+\s*\[\s*(\d+)\s*\]\s*,\s*\w+\s*\[\s*(\d+)\s*\]", line) |
| if m: |
| ops.append(("entangle", int(m.group(1)), int(m.group(2)))); continue |
| m = re.match(r"let\s+(\w+)\s*=\s*eml\s*\(\s*([-\d.]+)\s*,\s*([-\d.]+)\s*\)", line) |
| if m: |
| ops.append(("eml", m.group(1), float(m.group(2)), float(m.group(3)))); continue |
| m = re.match(r"minimize_energy\s*\(\s*threshold\s*:\s*([-\d.]+)\s*\)", line) |
| if m: |
| ops.append(("minimize", float(m.group(1)))); continue |
| if re.match(r"measure\s+\w+", line): |
| ops.append(("measure",)); continue |
| |
| if line in ("{", "}") or "evaluate_causal_matrix" in line: |
| continue |
| raise CompileError(f"unparseable line: {line}") |
| if n_qubits is None: |
| raise CompileError("missing hybrid_state declaration") |
| return {"n_qubits": n_qubits, "targets": targets, "ops": ops} |
|
|
| def emit_qasm(ast): |
| """Emit real OpenQASM 3.0.""" |
| n = ast["n_qubits"] |
| out = ["OPENQASM 3.0;", 'include "stdgates.inc";', |
| f"qubit[{n}] q;", f"bit[{n}] c;"] |
| measured = False |
| for op in ast["ops"]: |
| if op[0] == "rotate": |
| out.append(f"ry({op[2]}) q[{op[1]}];") |
| elif op[0] == "entangle": |
| out.append(f"cx q[{op[1]}], q[{op[2]}];") |
| elif op[0] == "measure": |
| out.append("c = measure q;") |
| measured = True |
| if not measured: |
| out.append("c = measure q;") |
| return "\n".join(out) |
|
|
| def emit_numpy(ast): |
| """Emit executable Python/NumPy for the classical path.""" |
| n = ast["n_qubits"] |
| lines = ["# np is provided by the runtime namespace", |
| "def run():", |
| f" n = {n}", |
| " # statevector start |0...0>", |
| " state = np.zeros(2**n, dtype=complex); state[0] = 1.0", |
| " def ry(state, q, theta):", |
| " c, s = np.cos(theta/2), np.sin(theta/2)", |
| " new = state.copy()", |
| " for i in range(2**n):", |
| " if not (i >> q) & 1:", |
| " j = i | (1 << q)", |
| " a, b = state[i], state[j]", |
| " new[i] = c*a - s*b; new[j] = s*a + c*b", |
| " return new", |
| " def cx(state, ctrl, tgt):", |
| " new = state.copy()", |
| " for i in range(2**n):", |
| " if (i >> ctrl) & 1:", |
| " j = i ^ (1 << tgt); new[j] = state[i]", |
| " return new", |
| " results = {}"] |
| for op in ast["ops"]: |
| if op[0] == "rotate": |
| lines.append(f" state = ry(state, {op[1]}, {op[2]})") |
| elif op[0] == "entangle": |
| lines.append(f" state = cx(state, {op[1]}, {op[2]})") |
| elif op[0] == "eml": |
| lines.append(f" results['{op[1]}'] = float(np.exp({op[2]}) - np.log({op[3]}))") |
| elif op[0] == "minimize": |
| lines.append(f" results['energy_threshold'] = {op[1]}") |
| lines.append(" probs = np.abs(state)**2") |
| lines.append(" results['probabilities'] = {format(i,'0%db'%n): float(probs[i]) for i in range(2**n) if probs[i]>1e-9}") |
| lines.append(" return results") |
| return "\n".join(lines) |
|
|
| def compile_source(src): |
| ast = parse(src) |
| return {"ast": ast, |
| "classical_numpy": emit_numpy(ast), |
| "quantum_qasm": emit_qasm(ast), |
| "targets": ast["targets"], |
| "note": "Real dual-target compile. Classical=NumPy (CPU-vectorised), " |
| "quantum=OpenQASM 3.0 (runs on the IBM worker). Not a raw-register " |
| "assembler — that is roadmap."} |
|
|
| def run_classical(src): |
| """Actually execute the emitted NumPy in a sandboxed namespace.""" |
| out = compile_source(src) |
| import numpy as _np |
| safe_builtins = {"range": range, "format": format, "float": float, |
| "int": int, "abs": abs, "len": len, "dict": dict, |
| "complex": complex, "__import__": __import__} |
| ns = {} |
| exec(out["classical_numpy"], {"__builtins__": safe_builtins, "np": _np}, ns) |
| return ns["run"]() |
|
|
