mirror of
https://github.com/allaunthefox/Research-Stack.git
synced 2026-07-31 03:05:21 +00:00
allaun
7b498b95e4
- capability_probe.py: detection shim for 9 quantum/optimization/compute backends - Each backend tested with import + functional test - Assigns capability slots and formulation modes - Directed routing analysis: MIP (QAP) recommended over QAOA (asymmetric loss) - All highspy/perceval/quimb/wgpu/opt_einsum API quirks resolved Receipt: 9/9 functional, 9 formulation modes, JSON schema v1
405 lines
13 KiB
Python
405 lines
13 KiB
Python
#!/usr/bin/env python3
|
|
"""Capability probe: detect available quantum/optimization/compute backends.
|
|
|
|
Probes each package, runs a minimal functional test, and emits a structured
|
|
receipt mapping capability slots → available backends + version + status.
|
|
|
|
Usage:
|
|
python3 capability_probe.py # stdout
|
|
python3 capability_probe.py -o probe_receipt.json # file
|
|
python3 capability_probe.py --json # compact JSON
|
|
"""
|
|
|
|
import importlib
|
|
import json
|
|
import math
|
|
import sys
|
|
import time
|
|
import traceback
|
|
from dataclasses import dataclass, field, asdict
|
|
from typing import Any, Optional
|
|
|
|
|
|
# =========================================================================
|
|
# Probe registry
|
|
# =========================================================================
|
|
|
|
@dataclass
|
|
class ProbeResult:
|
|
package: str
|
|
version: str = ""
|
|
import_ok: bool = False
|
|
functional_ok: bool = False
|
|
error: str = ""
|
|
test_detail: str = ""
|
|
elapsed_s: float = 0.0
|
|
|
|
|
|
PROBES: list[dict] = []
|
|
|
|
|
|
def register(category: str, slot: str, package: str,
|
|
import_name: Optional[str] = None,
|
|
min_version: Optional[str] = None,
|
|
test: Optional[callable] = None):
|
|
PROBES.append(dict(
|
|
category=category,
|
|
slot=slot,
|
|
package=package,
|
|
import_name=import_name or package,
|
|
min_version=min_version,
|
|
test=test,
|
|
))
|
|
|
|
|
|
def _try_import(modname: str) -> tuple[bool, str]:
|
|
try:
|
|
mod = importlib.import_module(modname)
|
|
ver = getattr(mod, "__version__", "")
|
|
return True, ver
|
|
except Exception:
|
|
return False, ""
|
|
|
|
|
|
def _run_test(modname: str, test_fn: callable) -> tuple[bool, str]:
|
|
try:
|
|
mod = importlib.import_module(modname)
|
|
result = test_fn(mod)
|
|
return True, result
|
|
except Exception as e:
|
|
tb = traceback.format_exc()
|
|
return False, f"{type(e).__name__}: {e}\n{tb[:200]}"
|
|
|
|
|
|
# — Quantum backends —
|
|
|
|
|
|
def _test_perceval(mod):
|
|
from perceval import BasicState, Circuit, components, Processor, SLOSBackend
|
|
import math
|
|
c = Circuit(2)
|
|
c.add(0, components.BS(theta=math.pi / 4))
|
|
backend = SLOSBackend()
|
|
proc = Processor(backend, c)
|
|
input_state = BasicState("|0,1>")
|
|
proc.with_input(input_state)
|
|
output = proc.probs()
|
|
results = output["results"]
|
|
return f"BS probabilities: {dict(results)}"
|
|
|
|
|
|
def _test_quimb(mod):
|
|
import quimb.tensor as qtn
|
|
import numpy as np
|
|
t1 = qtn.Tensor(np.array([[1.0, 0.0], [0.0, 1.0]]), inds=["a", "b"])
|
|
t2 = qtn.Tensor(np.array([[1.0, 0.5], [0.0, 1.0]]), inds=["b", "c"])
|
|
tn = qtn.TensorNetwork([t1, t2])
|
|
result = tn.contract()
|
|
return f"contracted 2 tensors, result shape={result.shape}"
|
|
|
|
|
|
def _test_cirq(mod):
|
|
import cirq
|
|
q = cirq.LineQubit(0)
|
|
circuit = cirq.Circuit([cirq.H(q), cirq.measure(q)])
|
|
sim = cirq.Simulator()
|
|
result = sim.run(circuit, repetitions=10)
|
|
return f"simulated {len(result.measurements)} outcomes"
|
|
|
|
|
|
def _test_qiskit(mod):
|
|
from qiskit import QuantumCircuit
|
|
from qiskit_aer import AerSimulator
|
|
qc = QuantumCircuit(2, 2)
|
|
qc.h(0)
|
|
qc.cx(0, 1)
|
|
qc.measure([0, 1], [0, 1])
|
|
sim = AerSimulator()
|
|
result = sim.run(qc, shots=10).result()
|
|
counts = result.get_counts()
|
|
return f"simulated {len(counts)} distinct outcomes"
|
|
|
|
|
|
# — Classical optimization —
|
|
|
|
|
|
def _test_highspy(mod):
|
|
import highspy
|
|
import numpy as np
|
|
# Minimize -2x0 - x1 subject to x0 + x1 >= 1, 0 <= x0, x1 <= 2
|
|
# Optimum: x0=2, x1=0 → obj=-4
|
|
lp = highspy.HighsLp()
|
|
lp.num_col_ = 2
|
|
lp.num_row_ = 1
|
|
lp.sense_ = highspy.ObjSense.kMinimize
|
|
lp.col_cost_ = np.array([-2.0, -1.0])
|
|
lp.col_lower_ = np.array([0.0, 0.0])
|
|
lp.col_upper_ = np.array([2.0, 2.0])
|
|
lp.row_lower_ = np.array([1.0])
|
|
lp.row_upper_ = np.array([float("inf")])
|
|
lp.a_matrix_ = highspy.HighsSparseMatrix()
|
|
lp.a_matrix_.start_ = np.array([0, 1, 2], dtype=np.int32)
|
|
lp.a_matrix_.index_ = np.array([0, 0], dtype=np.int32)
|
|
lp.a_matrix_.value_ = np.array([1.0, 1.0])
|
|
lp.a_matrix_.format_ = highspy.MatrixFormat.kColwise
|
|
h = highspy.Highs()
|
|
h.passModel(lp)
|
|
h.run()
|
|
info = h.getInfo()
|
|
sol = h.getSolution()
|
|
x = list(sol.col_value) if hasattr(sol, "col_value") else []
|
|
return f"LP solved, obj={info.objective_function_value:.2f}, x={x}"
|
|
|
|
|
|
def _test_scipy_optimize(mod):
|
|
from scipy.optimize import minimize
|
|
result = minimize(lambda x: x[0]**2 + x[1]**2, [1.0, 1.0], method="Nelder-Mead")
|
|
return f"minimized to ({result.x[0]:.4f}, {result.x[1]:.4f})"
|
|
|
|
|
|
def _test_networkx(mod):
|
|
import networkx as nx
|
|
g = nx.complete_graph(5)
|
|
path = nx.shortest_path(g, source=0, target=4)
|
|
return f"K_5 diameter={nx.diameter(g)}, path={path}"
|
|
|
|
|
|
# — GPU compute —
|
|
|
|
|
|
def _test_wgpu(mod):
|
|
import wgpu
|
|
adapter = wgpu.gpu.request_adapter_sync(power_preference="high-performance")
|
|
device = adapter.request_device_sync()
|
|
info = device.adapter_info
|
|
dev_name = getattr(info, "name", str(info))
|
|
dev_type = getattr(info, "adapter_type", "?")
|
|
backend_type = getattr(info, "backend_type", "?")
|
|
return f"adapter={dev_name}, type={dev_type}, backend={backend_type}"
|
|
|
|
|
|
# — Utility —
|
|
|
|
|
|
def _test_opt_einsum(mod):
|
|
import opt_einsum as oe
|
|
import numpy as np
|
|
a = np.random.rand(2, 3)
|
|
b = np.random.rand(3, 4)
|
|
c = oe.contract("ij,jk->ik", a, b)
|
|
# Compute path
|
|
path_info = oe.contract_path("ij,jk->ik", a, b)
|
|
return f"contraction shape={c.shape}, path={path_info[1]}"
|
|
|
|
|
|
# =========================================================================
|
|
# Register all probes
|
|
# =========================================================================
|
|
|
|
# Photonic quantum
|
|
register("quantum", "photonic_slos", "perceval-quandela",
|
|
import_name="perceval", test=_test_perceval)
|
|
# Bosonic tensor network
|
|
register("quantum", "bosonic_tn", "quimb", test=_test_quimb)
|
|
# QAOA circuit sim (Cirq)
|
|
register("quantum", "qaoa_cirq", "cirq", test=_test_cirq)
|
|
# QAOA circuit sim (Qiskit)
|
|
register("quantum", "qaoa_qiskit", "qiskit", test=_test_qiskit)
|
|
# Classical MIP
|
|
register("optimization", "mip_highs", "highspy", test=_test_highspy)
|
|
# Classical continuous
|
|
register("optimization", "continuous_scipy", "scipy.optimize",
|
|
import_name="scipy", test=_test_scipy_optimize)
|
|
# Graph algorithms
|
|
register("optimization", "graph_networkx", "networkx", test=_test_networkx)
|
|
# GPU compute
|
|
register("compute", "gpu_wgpu", "wgpu", test=_test_wgpu)
|
|
# Tensor contraction
|
|
register("compute", "tensor_opt_einsum", "opt_einsum", test=_test_opt_einsum)
|
|
|
|
|
|
# =========================================================================
|
|
# Probe runner
|
|
# =========================================================================
|
|
|
|
def run_all_probes() -> dict[str, list[dict]]:
|
|
results: dict[str, list[dict]] = {}
|
|
for entry in PROBES:
|
|
cat = entry["category"]
|
|
pkg = entry["package"]
|
|
modname = entry["import_name"]
|
|
test_fn = entry["test"]
|
|
|
|
t0 = time.time()
|
|
import_ok, version = _try_import(modname)
|
|
functional_ok = False
|
|
test_detail = ""
|
|
error = ""
|
|
|
|
if import_ok and test_fn:
|
|
functional_ok, test_detail = _run_test(modname, test_fn)
|
|
if not functional_ok:
|
|
error = test_detail
|
|
test_detail = ""
|
|
|
|
if not import_ok:
|
|
error = f"import failed for {modname}"
|
|
|
|
elapsed = time.time() - t0
|
|
|
|
result = {
|
|
"slot": entry["slot"],
|
|
"package": pkg,
|
|
"version": version,
|
|
"import_ok": import_ok,
|
|
"functional_ok": functional_ok,
|
|
"test_detail": test_detail if functional_ok else "",
|
|
"error": error,
|
|
"elapsed_s": round(elapsed, 3),
|
|
}
|
|
results.setdefault(cat, []).append(result)
|
|
|
|
return results
|
|
|
|
|
|
# =========================================================================
|
|
# Capability assignment
|
|
# =========================================================================
|
|
|
|
def assign_capabilities(probe_results: dict[str, list[dict]]) -> dict:
|
|
avail = {}
|
|
for cat, entries in probe_results.items():
|
|
for e in entries:
|
|
if e["functional_ok"]:
|
|
avail[e["slot"]] = {
|
|
"package": e["package"],
|
|
"version": e["version"],
|
|
}
|
|
return avail
|
|
|
|
|
|
def assign_formulation_modes(avail: dict) -> list[str]:
|
|
"""Map available backends to RRC-relevant formulation modes."""
|
|
modes = []
|
|
if "bosonic_tn" in avail:
|
|
modes.append("bosonic_tensor_network")
|
|
if "qaoa_cirq" in avail or "qaoa_qiskit" in avail:
|
|
modes.append("qaoa_variational")
|
|
if "mip_highs" in avail:
|
|
modes.append("qubo_classical_mip")
|
|
modes.append("qap_classical_mip")
|
|
if "photonic_slos" in avail:
|
|
modes.append("photonic_slos")
|
|
if "continuous_scipy" in avail:
|
|
modes.append("continuous_optimization")
|
|
if "graph_networkx" in avail:
|
|
modes.append("spectral_graph_pipeline")
|
|
if "gpu_wgpu" in avail:
|
|
modes.append("gpu_vulkan_compute")
|
|
if "tensor_opt_einsum" in avail:
|
|
modes.append("contraction_optimization")
|
|
return modes
|
|
|
|
|
|
def assign_directed_formulation_support(avail: dict) -> dict:
|
|
"""What's available for directed (asymmetric) routing formulations."""
|
|
support = {}
|
|
if "mip_highs" in avail:
|
|
support["qap"] = {
|
|
"available": True,
|
|
"comment": "HiGHS handles asymmetric QAP natively (directed costs preserved in MIP)",
|
|
"recommended": True,
|
|
"solver": "highspy",
|
|
}
|
|
else:
|
|
support["qap"] = {
|
|
"available": False,
|
|
"comment": "no MIP solver — use SA/Levy on symmetrized QUBO instead",
|
|
}
|
|
if "qaoa_cirq" in avail or "qaoa_qiskit" in avail:
|
|
support["qaoa"] = {
|
|
"available": True,
|
|
"comment": "QAOA on QUBO — asymmetry lost in x_i x_j = x_j x_i symmetrization",
|
|
"recommended": False,
|
|
"solver": "cirq" if "qaoa_cirq" in avail else "qiskit",
|
|
}
|
|
return support
|
|
|
|
|
|
# =========================================================================
|
|
# Receipt assembly
|
|
# =========================================================================
|
|
|
|
def build_receipt(probe_results: dict[str, list[dict]],
|
|
avail: dict, modes: list[str],
|
|
directed: dict) -> dict:
|
|
timestamp = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())
|
|
total_probes = sum(len(v) for v in probe_results.values())
|
|
functional_count = sum(
|
|
1 for entries in probe_results.values()
|
|
for e in entries if e["functional_ok"]
|
|
)
|
|
|
|
return {
|
|
"schema": "capability_probe_receipt_v1",
|
|
"generated_at_utc": timestamp,
|
|
"probe_summary": {
|
|
"total_probes": total_probes,
|
|
"functional": functional_count,
|
|
"failed": total_probes - functional_count,
|
|
},
|
|
"results": probe_results,
|
|
"available": avail,
|
|
"formulation_modes": modes,
|
|
"directed_formulation_support": directed,
|
|
"claim_boundary": "capability-probe-only;no-decision-logic",
|
|
}
|
|
|
|
|
|
# =========================================================================
|
|
# CLI
|
|
# =========================================================================
|
|
|
|
def main():
|
|
import argparse
|
|
parser = argparse.ArgumentParser(description="Capability probe")
|
|
parser.add_argument("-o", "--output", type=str, help="JSON output path")
|
|
parser.add_argument("--json", action="store_true", help="Compact JSON (one line)")
|
|
parser.add_argument("--human", action="store_true", help="Human-readable summary")
|
|
args = parser.parse_args()
|
|
|
|
probe_results = run_all_probes()
|
|
avail = assign_capabilities(probe_results)
|
|
modes = assign_formulation_modes(avail)
|
|
directed = assign_directed_formulation_support(avail)
|
|
receipt = build_receipt(probe_results, avail, modes, directed)
|
|
|
|
if args.output:
|
|
with open(args.output, "w") as f:
|
|
json.dump(receipt, f, indent=2)
|
|
print(f"Wrote receipt to {args.output}")
|
|
|
|
if args.json:
|
|
print(json.dumps(receipt))
|
|
|
|
if args.human or not args.json:
|
|
print(f"\n=== Capability Probe ===")
|
|
print(f" {receipt['probe_summary']['functional']}/{receipt['probe_summary']['total_probes']} probes functional")
|
|
print(f"\n Available backends ({len(avail)}):")
|
|
for slot, info in sorted(avail.items()):
|
|
print(f" {slot:30s} {info['package']:20s} v{info['version']}")
|
|
print(f"\n Formulation modes ({len(modes)}):")
|
|
for m in modes:
|
|
print(f" {m}")
|
|
print(f"\n Directed (asymmetric) routing support:")
|
|
for label, info in directed.items():
|
|
status = "✅" if info["available"] else "❌"
|
|
rec = " (recommended)" if info.get("recommended") else " (not recommended)" if info.get("recommended") is False else ""
|
|
print(f" {status} {label}: {info['comment']}{rec}")
|
|
|
|
if not (args.output or args.json or args.human):
|
|
print(json.dumps(receipt, indent=2))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|