#!/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()