Spaces:
Sleeping
Sleeping
| import gradio as gr | |
| import os | |
| import io | |
| import time | |
| import json | |
| import math | |
| import threading | |
| import traceback | |
| from datetime import datetime | |
| from dataclasses import dataclass, field | |
| from typing import Optional, Callable | |
| from collections import Counter | |
| import numpy as np | |
| import matplotlib | |
| matplotlib.use('Agg') | |
| import matplotlib.pyplot as plt | |
| from PIL import Image | |
| from qiskit import QuantumCircuit, QuantumRegister, ClassicalRegister, transpile | |
| from qiskit.quantum_info import Statevector, partial_trace, random_statevector | |
| from qiskit_aer import AerSimulator | |
| from qiskit.visualization import circuit_drawer, plot_histogram, plot_bloch_multivector | |
| from qiskit.transpiler.preset_passmanagers import generate_preset_pass_manager | |
| from qiskit.circuit.library import QFT, GroverOperator, MCMT | |
| IBM_AVAILABLE = False | |
| try: | |
| from qiskit_ibm_runtime import QiskitRuntimeService, SamplerV2 as Sampler | |
| IBM_AVAILABLE = True | |
| except ImportError: | |
| pass | |
| COMPANY = "Quantum Advanced LLC" | |
| VERSION = "3.0.0" | |
| QASM_BENCHMARK = """OPENQASM 2.0; | |
| include "qelib1.inc"; | |
| qreg q[4]; | |
| creg c[4]; | |
| rzz(pi/2) q[1],q[0]; | |
| rz(pi/2) q[0]; | |
| rx(pi/4) q[0]; | |
| rz(-pi/2) q[1]; | |
| rzz(pi/2) q[3],q[2]; | |
| rz(pi/2) q[2]; | |
| rzz(pi/2) q[2],q[1]; | |
| cz q[1],q[0]; | |
| rx(pi/4) q[0]; | |
| sx q[1]; | |
| cz q[1],q[0]; | |
| sx q[0]; | |
| sx q[1]; | |
| cz q[1],q[0]; | |
| sx q[0]; | |
| sx q[1]; | |
| cz q[1],q[0]; | |
| sx q[0]; | |
| sx q[1]; | |
| sx q[2]; | |
| cz q[2],q[1]; | |
| sx q[1]; | |
| sx q[2]; | |
| cz q[2],q[1]; | |
| sx q[1]; | |
| sx q[2]; | |
| cz q[2],q[1]; | |
| rz(pi/2) q[2]; | |
| sx q[2]; | |
| rz(pi/2) q[2]; | |
| rz(-pi/2) q[3]; | |
| cz q[3],q[2]; | |
| rx(pi/4) q[2]; | |
| cz q[1],q[2]; | |
| rx(-pi/4) q[2]; | |
| cz q[3],q[2]; | |
| rx(pi/4) q[2]; | |
| cz q[1],q[2]; | |
| sx q[1]; | |
| rz(0.9553166181245096) q[2]; | |
| rx(pi/3) q[2]; | |
| rz(-0.6154797086703869) q[2]; | |
| cz q[1],q[2]; | |
| sx q[1]; | |
| sx q[2]; | |
| cz q[1],q[2]; | |
| sx q[1]; | |
| sx q[2]; | |
| cz q[1],q[2]; | |
| rz(pi/4) q[1]; | |
| rx(pi/2) q[1]; | |
| rz(pi/2) q[1]; | |
| cz q[2],q[1]; | |
| rx(pi/2) q[1]; | |
| sx q[2]; | |
| cz q[2],q[1]; | |
| sx q[1]; | |
| sx q[2]; | |
| cz q[2],q[1]; | |
| sx q[1]; | |
| sx q[2]; | |
| cz q[2],q[1]; | |
| rz(pi/2) q[2]; | |
| sx q[2]; | |
| rz(pi/2) q[2]; | |
| rz(pi/2) q[3]; | |
| cz q[3],q[2]; | |
| rx(pi/4) q[2]; | |
| cz q[1],q[2]; | |
| rx(-pi/4) q[2]; | |
| cz q[3],q[2]; | |
| rx(pi/4) q[2]; | |
| cz q[1],q[2]; | |
| sx q[1]; | |
| cz q[1],q[0]; | |
| sx q[0]; | |
| sx q[1]; | |
| cz q[1],q[0]; | |
| sx q[0]; | |
| sx q[1]; | |
| cz q[1],q[0]; | |
| rz(2.1862760354652835) q[2]; | |
| rx(2*pi/3) q[2]; | |
| rz(0.9553166181245079) q[2]; | |
| cz q[1],q[2]; | |
| rx(-pi/4) q[2]; | |
| """ | |
| class SteaneCode: | |
| """Steane [[7,1,3]] Quantum Error Correction Code.""" | |
| def __init__(self): | |
| self.n = 7 | |
| self.k = 1 | |
| self.d = 3 | |
| self.stabilizers_x = [ | |
| [0, 0, 0, 1, 1, 1, 1], | |
| [0, 1, 1, 0, 0, 1, 1], | |
| [1, 0, 1, 0, 1, 0, 1], | |
| ] | |
| self.stabilizers_z = [ | |
| [0, 0, 0, 1, 1, 1, 1], | |
| [0, 1, 1, 0, 0, 1, 1], | |
| [1, 0, 1, 0, 1, 0, 1], | |
| ] | |
| self.logical_x = [1, 1, 1, 1, 1, 1, 1] | |
| self.logical_z = [1, 1, 1, 1, 1, 1, 1] | |
| def inject_error(self, qc, qubit, error_type): | |
| if error_type == 'X': qc.x(qubit) | |
| elif error_type == 'Z': qc.z(qubit) | |
| elif error_type == 'Y': qc.y(qubit) | |
| return qc | |
| def compute_syndrome(self, error_qubit, error_type): | |
| x_syn = '' | |
| z_syn = '' | |
| if error_type in ('X', 'Y'): | |
| for stab in self.stabilizers_z: | |
| x_syn += str(stab[error_qubit]) | |
| else: | |
| x_syn = '000' | |
| if error_type in ('Z', 'Y'): | |
| for stab in self.stabilizers_x: | |
| z_syn += str(stab[error_qubit]) | |
| else: | |
| z_syn = '000' | |
| return x_syn + z_syn | |
| def correct_error(self, syndrome): | |
| x_locations = {'000': -1, '111': 0, '011': 1, '101': 2, '110': 3, '100': 4, '010': 5, '001': 6} | |
| z_locations = x_locations.copy() | |
| x_syn, z_syn = syndrome[:3], syndrome[3:] | |
| x_loc = x_locations.get(x_syn, -1) | |
| z_loc = z_locations.get(z_syn, -1) | |
| correction = [] | |
| if z_loc >= 0: | |
| correction.append(('X', z_loc)) | |
| if x_loc >= 0: | |
| correction.append(('Z', x_loc)) | |
| return correction | |
| class IonTrapSimulator: | |
| """Ion trap simulator with AI optimization.""" | |
| def __init__(self, num_ions=4): | |
| self.num_ions = num_ions | |
| self.trap_frequency = 1.0 | |
| self.lamb_dicke = 0.1 | |
| self._mode_frequencies = self._compute_mode_spectrum() | |
| def _compute_mode_spectrum(self): | |
| return [self.trap_frequency * math.sqrt(p) for p in range(1, self.num_ions + 1)] | |
| def ms_gate_circuit(self, ion_a, ion_b): | |
| qr = QuantumRegister(self.num_ions, 'ion') | |
| qc = QuantumCircuit(qr) | |
| qc.h(ion_a); qc.h(ion_b) | |
| qc.cx(ion_a, ion_b) | |
| qc.h(ion_a); qc.h(ion_b) | |
| qc.rz(np.pi/4, ion_a); qc.rz(np.pi/4, ion_b) | |
| return qc | |
| def optimize_ion_placement(self, target_connectivity): | |
| best_order = list(range(self.num_ions)) | |
| best_cost = float('inf') | |
| for _ in range(100): | |
| order = list(range(self.num_ions)) | |
| np.random.shuffle(order) | |
| cost = sum(abs(order.index(a) - order.index(b)) for (a, b) in target_connectivity) | |
| if cost < best_cost: | |
| best_cost = cost | |
| best_order = order | |
| return best_order | |
| def compute_gate_fidelity(self, circuit, noise_level=0.01): | |
| two_q = sum(1 for inst in circuit.data if len(inst.qubits) == 2) | |
| single_q = circuit.size() - two_q | |
| fid_single = (1 - noise_level * 0.1) ** single_q | |
| fid_two = (1 - noise_level) ** two_q | |
| return float(fid_single * fid_two) | |
| class GuardianLog: | |
| timestamp: str | |
| level: str | |
| module: str | |
| message: str | |
| metric: Optional[float] = None | |
| class GuardianAI: | |
| """Guardian AI — Intelligent quantum lab management.""" | |
| def __init__(self): | |
| self.logs: list[GuardianLog] = [] | |
| self.metrics = {'circuits_executed': 0, 'errors_detected': 0, 'errors_corrected': 0} | |
| self._lock = threading.Lock() | |
| def log(self, level, module, message, metric=None): | |
| entry = GuardianLog( | |
| timestamp=datetime.now().strftime('%H:%M:%S.%f')[:-3], | |
| level=level, module=module, message=message, metric=metric | |
| ) | |
| with self._lock: | |
| self.logs.append(entry) | |
| return f"[{entry.timestamp}] [{level}] {module}: {message}" | |
| def get_status(self): | |
| with self._lock: | |
| total = self.metrics['circuits_executed'] | |
| errors = self.metrics['errors_detected'] | |
| corrected = self.metrics['errors_corrected'] | |
| return f""" | |
| ### 🛡️ Guardian AI Status | |
| | Métrica | Valor | | |
| |---------|-------| | |
| | Circuitos ejecutados | {total} | | |
| | Errores detectados | {errors} | | |
| | Errores corregidos | {corrected} | | |
| | Tasa de corrección | {corrected/max(errors,1)*100:.1f}% | | |
| | Logs activos | {len(self.logs)} | | |
| """ | |
| class IBMQuantumProxy: | |
| """Secure IBM Quantum proxy.""" | |
| def __init__(self): | |
| self._service = None | |
| self._connected = False | |
| self._backend_list = [] | |
| self._lock = threading.Lock() | |
| def connect(self, token, instance=None): | |
| if not IBM_AVAILABLE: | |
| return {"ok": False, "error": "qiskit-ibm-runtime not installed."} | |
| try: | |
| with self._lock: | |
| kwargs = {"channel": "ibm_quantum_platform", "token": token.strip()} | |
| if instance and instance.strip(): | |
| kwargs["instance"] = instance.strip() | |
| self._service = QiskitRuntimeService(**kwargs) | |
| backends = self._service.backends() | |
| self._backend_list = [ | |
| {"name": b.name, "qubits": b.num_qubits, "simulator": b.simulator, | |
| "operational": b.status().operational if hasattr(b, 'status') else True, | |
| "pending_jobs": b.status().pending_jobs if hasattr(b, 'status') else 0} | |
| for b in backends | |
| ] | |
| self._connected = True | |
| return {"ok": True, "backends": self._backend_list, "count": len(backends)} | |
| except Exception as e: | |
| self._connected = False | |
| return {"ok": False, "error": str(e)} | |
| def disconnect(self): | |
| with self._lock: | |
| self._service = None | |
| self._connected = False | |
| self._backend_list = [] | |
| def is_connected(self): | |
| return self._connected and self._service is not None | |
| def run_circuit(self, circuit, backend_name=None, shots=1024, optimization_level=1, progress_callback=None): | |
| if not self.is_connected: | |
| return {"ok": False, "error": "Not connected to IBM Quantum."} | |
| try: | |
| with self._lock: | |
| if backend_name and backend_name.strip(): | |
| backend = self._service.backend(backend_name.strip()) | |
| else: | |
| if progress_callback: | |
| progress_callback("Finding least busy backend...") | |
| backend = self._service.least_busy(operational=True, simulator=False, min_num_qubits=circuit.num_qubits) | |
| if progress_callback: | |
| progress_callback(f"Backend: {backend.name} ({backend.num_qubits} qubits)") | |
| progress_callback("Transpiling to native ECR/SX/RZ...") | |
| pm = generate_preset_pass_manager(backend=backend, optimization_level=optimization_level) | |
| isa_circuit = pm.run(circuit) | |
| if progress_callback: | |
| progress_callback(f"ISA: {isa_circuit.size()} gates, depth {isa_circuit.depth()}") | |
| progress_callback(f"Submitting job ({shots} shots)...") | |
| sampler = Sampler(mode=backend) | |
| sampler.options.default_shots = shots | |
| job = sampler.run([isa_circuit]) | |
| job_id = job.job_id() | |
| if progress_callback: | |
| progress_callback(f"Job {job_id} queued — waiting for execution...") | |
| result = job.result() | |
| pub_result = result[0] | |
| try: | |
| counts = pub_result.data.meas.get_counts() | |
| except Exception: | |
| counts = {} | |
| for creg_name in pub_result.data: | |
| try: | |
| counts = getattr(pub_result.data, creg_name).get_counts() | |
| break | |
| except Exception: | |
| continue | |
| return {"ok": True, "job_id": job_id, "backend": backend.name, "counts": counts, "shots": shots} | |
| except Exception as e: | |
| return {"ok": False, "error": str(e)} | |
| ibm_proxy = IBMQuantumProxy() | |
| guardian = GuardianAI() | |
| steane = SteaneCode() | |
| def build_benchmark_circuit(measure_all=True): | |
| qc = QuantumCircuit.from_qasm_str(QASM_BENCHMARK) | |
| if measure_all: | |
| qc.measure_all() | |
| return qc | |
| def build_benchmark_no_measure(): | |
| return QuantumCircuit.from_qasm_str(QASM_BENCHMARK) | |
| def run_syndrome_extraction_demo(error_qubit, error_type, shots=1024): | |
| qc = QuantumCircuit(7, 6) | |
| for i in [0, 1, 2]: | |
| qc.h(i) | |
| qc.cx(0, 4); qc.cx(0, 5); qc.cx(0, 6) | |
| qc.cx(1, 3); qc.cx(1, 5); qc.cx(1, 6) | |
| qc.cx(2, 3); qc.cx(2, 4); qc.cx(2, 6) | |
| for i in range(7): | |
| qc.h(i) | |
| qc.barrier() | |
| steane.inject_error(qc, error_qubit, error_type) | |
| qc.barrier() | |
| stab_x = steane.stabilizers_x | |
| stab_z = steane.stabilizers_z | |
| for i, stab in enumerate(stab_x): | |
| qc.h(i) | |
| for j, val in enumerate(stab): | |
| if val == 1: | |
| qc.cz(i, j) | |
| qc.h(i) | |
| qc.measure(i, i) | |
| for i, stab in enumerate(stab_z): | |
| anc = i + 3 | |
| for j, val in enumerate(stab): | |
| if val == 1: | |
| qc.cx(anc, j) | |
| qc.measure(anc, anc) | |
| expected = steane.compute_syndrome(error_qubit, error_type) | |
| sim = AerSimulator() | |
| result = sim.run(qc, shots=shots).result() | |
| counts = result.get_counts() | |
| most_common = max(counts, key=counts.get) if counts else "000000" | |
| correction = steane.correct_error(most_common) | |
| match = most_common == expected | |
| return qc, counts, most_common, expected, correction, match | |
| def optimize_circuit_cost(counts_per_job, cost_per_shot=0.00001): | |
| total_shots = sum(counts_per_job.values()) | |
| cost = total_shots * cost_per_shot | |
| max_prob = max(counts_per_job.values()) / total_shots if total_shots > 0 else 1.0 | |
| min_shots_needed = int(100 / max_prob) if max_prob > 0 else 10000 | |
| savings = max(0, total_shots - min_shots_needed) * cost_per_shot | |
| return { | |
| 'total_shots': total_shots, 'cost': cost, | |
| 'confidence': max_prob * 100, 'min_shots_needed': min_shots_needed, | |
| 'potential_savings': savings, | |
| 'recommendation': f"Reduce shots {total_shots} -> {min_shots_needed} to save ${savings:.4f} while maintaining statistical significance." | |
| } | |
| def draw_circuit_image(qc, title="Circuit"): | |
| fig = qc.draw(output="mpl", style="iqp", fold=-1, scale=0.55, plot_barriers=True) | |
| buf = io.BytesIO() | |
| fig.savefig(buf, format="png", dpi=180, bbox_inches="tight", pad_inches=0.1) | |
| plt.close(fig) | |
| buf.seek(0) | |
| return Image.open(buf) | |
| def draw_histogram(counts, title="Results", color="#1a1a6c"): | |
| fig, ax = plt.subplots(figsize=(10, 5)) | |
| states = sorted(counts.keys()) | |
| values = [counts[s] for s in states] | |
| colors = plt.cm.viridis(np.linspace(0.2, 0.9, len(states))) | |
| ax.bar(states, values, color=colors, edgecolor='white', linewidth=1.2) | |
| ax.set_title(title, fontsize=14, fontweight='bold') | |
| ax.set_ylabel('Counts', fontsize=12) | |
| ax.set_xlabel('State', fontsize=12) | |
| for bar, val in zip(ax.patches, values): | |
| ax.text(bar.get_x() + bar.get_width()/2, bar.get_height() + max(values)*0.01, | |
| str(val), ha='center', fontsize=9) | |
| plt.tight_layout() | |
| buf = io.BytesIO() | |
| fig.savefig(buf, format="png", dpi=150, bbox_inches="tight") | |
| plt.close(fig) | |
| buf.seek(0) | |
| return Image.open(buf) | |
| def draw_bloch_image(sv): | |
| fig = plot_bloch_multivector(sv) | |
| buf = io.BytesIO() | |
| fig.savefig(buf, format="png", dpi=130, bbox_inches="tight") | |
| plt.close(fig) | |
| buf.seek(0) | |
| return Image.open(buf) | |
| def draw_comparison(local_counts, ibm_counts, local_label="Simulator", ibm_label="IBM Q"): | |
| fig, (ax1, ax2) = plt.subplots(1, 2, figsize=(14, 5)) | |
| all_states = sorted(set(list(local_counts.keys()) + list(ibm_counts.keys()))) | |
| local_vals = [local_counts.get(s, 0) for s in all_states] | |
| ibm_vals = [ibm_counts.get(s, 0) for s in all_states] | |
| ax1.bar(all_states, local_vals, color='#4ECDC4', edgecolor='#2c3e50') | |
| ax1.set_title(f'{local_label} — Total: {sum(local_vals)}', fontweight='bold') | |
| ax2.bar(all_states, ibm_vals, color='#FF6B6B', edgecolor='#2c3e50') | |
| ax2.set_title(f'{ibm_label} — Total: {sum(ibm_vals)}', fontweight='bold') | |
| fig.suptitle(f'{COMPANY} — Local vs Real Hardware Comparison', fontsize=15, fontweight='bold') | |
| plt.tight_layout() | |
| buf = io.BytesIO() | |
| fig.savefig(buf, format="png", dpi=150, bbox_inches="tight") | |
| plt.close(fig) | |
| buf.seek(0) | |
| return Image.open(buf) | |
| def draw_cost_gauge(savings_pct, current_cost): | |
| fig, ax = plt.subplots(figsize=(6, 4), subplot_kw={'projection': 'polar'}) | |
| theta = np.linspace(0, 2*np.pi, 100) | |
| ax.fill(theta, np.ones_like(theta), color='#1a1a6c', alpha=0.3) | |
| fill_theta = np.linspace(0, 2*np.pi * (1 - min(savings_pct, 0.95)), 50) | |
| ax.fill(fill_theta, np.ones_like(fill_theta) * 0.8, color='#4ECDC4', alpha=0.7) | |
| ax.set_ylim(0, 1.2) | |
| ax.set_xticks([]); ax.set_yticks([]) | |
| ax.spines['polar'].set_visible(False) | |
| ax.text(0, 0, f'${current_cost:.4f}', ha='center', va='center', fontsize=20, fontweight='bold') | |
| ax.text(0, -0.3, f'Savings: {savings_pct*100:.0f}%', ha='center', va='center', fontsize=14, color='#4ECDC4') | |
| plt.tight_layout() | |
| buf = io.BytesIO() | |
| fig.savefig(buf, format="png", dpi=130, bbox_inches="tight", transparent=True) | |
| plt.close(fig) | |
| buf.seek(0) | |
| return Image.open(buf) | |
| def sim_local(qc, shots): | |
| sim = AerSimulator() | |
| result = sim.run(qc, shots=shots).result() | |
| return qc, result.get_counts() | |
| # ── Handlers ── | |
| def handle_benchmark_local(shots): | |
| guardian.log('INFO', 'Benchmark', f'Starting ({shots} shots)') | |
| qc = build_benchmark_circuit(True) | |
| _, counts = sim_local(qc, shots) | |
| sv = Statevector.from_instruction(build_benchmark_no_measure()) | |
| q3_probs = sv.probabilities([3]) | |
| circuit_img = draw_circuit_image(qc, "Benchmark (87 gates, depth 62)") | |
| bloch_img = draw_bloch_image(sv) | |
| hist_img = draw_histogram(counts, f"Benchmark — {shots} shots") | |
| cost = shots * 0.00001 | |
| cost_analysis = optimize_circuit_cost(counts) | |
| info = f""" | |
| ### 📊 Benchmark — Local Simulation ({COMPANY}) | |
| | Metric | Value | | |
| |--------|-------| | |
| | Shots | {shots} | | |
| | Gates | {qc.size()} | | |
| | Depth | {qc.depth()} | | |
| | P(|0000⟩) | {list(counts.values())[0]/shots*100:.1f}% | | |
| | Est. HW cost | ${cost:.4f} | | |
| | Min shots needed | {cost_analysis['min_shots_needed']} | | |
| **Counts:** `{json.dumps(dict(sorted(counts.items())))}` | |
| """ | |
| console = guardian.log('SUCCESS', 'Benchmark', f'Done: {counts}') | |
| return circuit_img, bloch_img, hist_img, info, console | |
| def handle_steane_demo(error_qubit, error_type, shots): | |
| guardian.log('INFO', 'Steane', f'Demo [[7,1,3]]: {error_type} on q{error_qubit}') | |
| qc, counts, syndrome, expected, correction, match = run_syndrome_extraction_demo(error_qubit, error_type, shots) | |
| circuit_img = draw_circuit_image(qc, f"Steane [[7,1,3]] — {error_type} Error on q{error_qubit}") | |
| hist_img = draw_histogram(counts, f"Syndromes — {shots} shots", "#6C3483") | |
| corr_str = ", ".join([f"{g} on q{q}" for g, q in correction]) if correction else "None (no error)" | |
| info = f""" | |
| ### 🛡️ Steane [[7,1,3]] — Error Correction | |
| | Parameter | Value | | |
| |-----------|-------| | |
| | Injected error | **{error_type}** on qubit **{error_qubit}** | | |
| | Expected syndrome | `{expected}` | | |
| | Measured syndrome | `{syndrome}` | | |
| | Match | {'✅ EXACT' if match else '⚠️ Mismatch'} | | |
| | Correction applied | {corr_str} | | |
| | Code | [[7,1,3]] — Distance d=3 | | |
| | Stabilizers | 6 (3 X-type, 3 Z-type) | | |
| """ | |
| guardian.metrics['errors_detected'] += 1 | |
| if match: | |
| guardian.metrics['errors_corrected'] += 1 | |
| console = f"[Steane] Syndrome: {syndrome} | Correction: {corr_str}" | |
| return circuit_img, hist_img, info, console | |
| def handle_ion_trap_sim(num_ions, noise): | |
| guardian.log('INFO', 'IonTrap', f'Simulating {num_ions} ions, noise={noise}') | |
| trap = IonTrapSimulator(num_ions) | |
| qc = QuantumCircuit(num_ions) | |
| for i in range(num_ions - 1): | |
| ms_gate = trap.ms_gate_circuit(i, i + 1) | |
| qc = qc.compose(ms_gate, qubits=list(range(num_ions))) | |
| fidelity = trap.compute_gate_fidelity(qc, noise) | |
| modes = trap._compute_mode_spectrum() | |
| target_conn = [(i, i+1) for i in range(num_ions-1)] | |
| optimal_order = trap.optimize_ion_placement(target_conn) | |
| circuit_img = draw_circuit_image(qc, f"Ion Trap — {num_ions} Ions") | |
| fig, ax = plt.subplots(figsize=(8, 3)) | |
| ax.stem(range(len(modes)), modes, basefmt=' ') | |
| ax.set_xlabel('Mode p'); ax.set_ylabel('Frequency (MHz)') | |
| ax.set_title('Axial Mode Spectrum') | |
| buf = io.BytesIO() | |
| fig.savefig(buf, format="png", dpi=120, bbox_inches="tight") | |
| plt.close(fig) | |
| buf.seek(0) | |
| spectrum_img = Image.open(buf) | |
| info = f""" | |
| ### ⚛️ Ion Trap Simulator ({COMPANY}) | |
| | Parameter | Value | | |
| |-----------|-------| | |
| | Number of ions | {num_ions} | | |
| | Trap frequency | {trap.trap_frequency:.1f} MHz | | |
| | Lamb-Dicke η | {trap.lamb_dicke} | | |
| | Noise (motional heating) | {noise:.3f} | | |
| | **Estimated fidelity** | **{fidelity*100:.2f}%** | | |
| | Optimal ion order | {optimal_order} | | |
| """ | |
| console = f"[IonTrap] {num_ions} ions | η={trap.lamb_dicke} | Fidelity={fidelity*100:.2f}%" | |
| return circuit_img, spectrum_img, info, console | |
| def handle_cost_reduction(counts_json, cost_per_shot): | |
| try: | |
| counts = json.loads(counts_json) | |
| except Exception: | |
| counts = {"0000": 8192} | |
| result = optimize_circuit_cost(counts, cost_per_shot) | |
| savings_pct = result['potential_savings'] / max(result['cost'], 0.0001) | |
| chart = draw_cost_gauge(savings_pct, result['cost']) | |
| hist = draw_histogram(counts, "Current Distribution") | |
| info = f""" | |
| ### 💰 Cost Reduction Analysis | |
| | Metric | Value | | |
| |--------|-------| | |
| | Shots executed | {result['total_shots']} | | |
| | Estimated cost | ${result['cost']:.6f} | | |
| | Statistical confidence | {result['confidence']:.1f}% | | |
| | Minimum shots needed | {result['min_shots_needed']} | | |
| | **Potential savings** | **${result['potential_savings']:.6f}** | | |
| > {result['recommendation']} | |
| """ | |
| return hist, chart, info | |
| def handle_connect_ibm(token, instance): | |
| if not token or not token.strip(): | |
| return "## ⚠️ Enter API Key", "⚫ IBM: Disconnected", gr.update(choices=[], visible=False) | |
| result = ibm_proxy.connect(token, instance if instance else None) | |
| if result["ok"]: | |
| choices = [b["name"] for b in result["backends"]] | |
| info = "\n".join([ | |
| f"• **{b['name']}** — {b['qubits']}q — {'✅' if b['operational'] else '⚠️'} — {b['pending_jobs']} jobs" | |
| for b in result["backends"][:15] | |
| ]) | |
| return ( | |
| f"## ✅ Connected — {result['count']} backends\n\n{info}", | |
| f"🔗 IBM: {result['count']} backends", | |
| gr.update(choices=choices, value=choices[0] if choices else None, visible=True), | |
| ) | |
| else: | |
| return (f"## ❌ Error\n```\n{result['error']}\n```", "❌ Error", gr.update(choices=[], visible=False)) | |
| def handle_run_ibm(shots, backend, optimization): | |
| if not ibm_proxy.is_connected: | |
| return None, None, "## ❌ Not connected to IBM", "❌ Disconnected" | |
| logs = [] | |
| def log(msg): | |
| logs.append(f"[{datetime.now().strftime('%H:%M:%S')}] {msg}") | |
| qc = build_benchmark_circuit(True) | |
| result = ibm_proxy.run_circuit(qc, backend_name=backend if backend else None, | |
| shots=shots, optimization_level=optimization, | |
| progress_callback=log) | |
| if result["ok"]: | |
| counts = result["counts"] | |
| hist_img = draw_histogram(counts, f"IBM {result['backend']} — {shots} shots", "#FF6B6B") | |
| _, local_counts = sim_local(qc, shots) | |
| cmp_img = draw_comparison(local_counts, counts) | |
| info = f""" | |
| ### 🖥️ IBM Quantum — {result['backend']} | |
| | Metric | Value | | |
| |--------|-------| | |
| | Backend | **{result['backend']}** | | |
| | Job ID | `{result['job_id']}` | | |
| | Shots | {shots} | | |
| **IBM Counts:** `{json.dumps(dict(sorted(counts.items())))}` | |
| **Local Counts:** `{json.dumps(dict(sorted(local_counts.items())))}` | |
| 💡 *Difference = real quantum hardware noise.* | |
| """ | |
| guardian.log('SUCCESS', 'IBM', f'Job {result["job_id"]} completed on {result["backend"]}') | |
| else: | |
| hist_img = cmp_img = None | |
| info = f"## ❌ Error\n```\n{result.get('error')}\n```" | |
| return hist_img, cmp_img, info, "\n".join(logs) | |
| def handle_guardian_status(): | |
| status = guardian.get_status() | |
| recent = "\n".join([f"[{l.timestamp}] [{l.level}] {l.module}: {l.message}" for l in guardian.logs[-20:]]) | |
| return status, recent | |
| # ═══════════════════════════════ | |
| # GRADIO UI | |
| # ═══════════════════════════════ | |
| THEME = gr.themes.Soft(primary_hue="indigo", secondary_hue="slate", neutral_hue="slate") | |
| CSS = """ | |
| .header-bar { background: linear-gradient(135deg, #0a0a2e 0%, #1a1a6c 50%, #16213e 100%); | |
| border-radius: 16px; padding: 28px 32px; margin-bottom: 16px; | |
| border: 1px solid rgba(255,255,255,0.1); } | |
| .company-badge { display: inline-block; background: linear-gradient(135deg, #f0c040, #ff8c00); | |
| color: #0a0a2e; padding: 4px 14px; border-radius: 20px; | |
| font-weight: 800; font-size: 13px; letter-spacing: 0.5px; } | |
| .guardian-pulse { animation: pulse 2s infinite; display: inline-block; width: 10px; height: 10px; | |
| background: #4ECDC4; border-radius: 50%; margin-right: 6px; } | |
| @keyframes pulse { 0%, 100% { opacity: 1; } 50% { opacity: 0.3; } } | |
| footer { visibility: hidden; } | |
| """ | |
| with gr.Blocks(title=f"{COMPANY} — Quantum Laboratory v{VERSION}", theme=THEME, css=CSS) as demo: | |
| gr.HTML(f""" | |
| <div class="header-bar"> | |
| <div style="display:flex; justify-content:space-between; align-items:center;"> | |
| <div> | |
| <h1 style="margin:0; color:white; font-size:2.2em;">⚛️ {COMPANY}</h1> | |
| <p style="margin:4px 0 0 0; color:#aab; font-size:1.1em;"> | |
| Quantum Error Correction • Ion Trap Simulation • Benchmarking | |
| </p> | |
| </div> | |
| <div style="text-align:right;"> | |
| <span class="company-badge">v{VERSION}</span> | |
| <div style="margin-top:8px; color:#4ECDC4;"> | |
| <span class="guardian-pulse"></span> Guardian AI Active | |
| </div> | |
| </div> | |
| </div> | |
| </div> | |
| """) | |
| with gr.Row(): | |
| ibm_status = gr.Textbox(value="⚫ IBM: Disconnected", label="IBM Quantum", interactive=False, scale=2) | |
| guardian_status_box = gr.Textbox(value="🛡️ Guardian: Online", label="Guardian AI", interactive=False, scale=2) | |
| job_id_box = gr.Textbox(value="—", label="Last Job", interactive=False, scale=1) | |
| with gr.Tabs(): | |
| with gr.Tab("🎯 Benchmark"): | |
| with gr.Row(): | |
| with gr.Column(scale=1): | |
| gr.Markdown("### ⚙️ Parameters") | |
| bench_shots = gr.Slider(256, 65536, 8192, step=256, label="Shots") | |
| bench_btn = gr.Button("💻 Run Local Benchmark", variant="primary", size="lg") | |
| gr.Markdown("---") | |
| gr.Markdown("### 🔑 IBM Quantum") | |
| ibm_token = gr.Textbox(label="API Key", type="password", placeholder="Paste IBM API key...") | |
| ibm_instance = gr.Textbox(label="CRN (optional)", placeholder="crn:v1:bluemix:...") | |
| ibm_backend = gr.Dropdown(label="Backend", choices=[], visible=False) | |
| connect_btn = gr.Button("🔗 Connect", variant="secondary") | |
| disconnect_btn = gr.Button("🔌 Disconnect", variant="stop") | |
| ibm_shots = gr.Slider(256, 32000, 1024, step=256, label="IBM Shots") | |
| ibm_optim = gr.Radio([0,1,2,3], value=1, label="Optimization") | |
| ibm_run_btn = gr.Button("☁️ Run on IBM", variant="primary") | |
| with gr.Column(scale=2): | |
| bench_circuit_img = gr.Image(label="Benchmark Circuit", type="pil") | |
| with gr.Row(): | |
| bench_bloch = gr.Image(label="Bloch Spheres", type="pil") | |
| bench_hist = gr.Image(label="Histogram", type="pil") | |
| with gr.Row(): | |
| ibm_hist = gr.Image(label="IBM Quantum", type="pil") | |
| cmp_hist = gr.Image(label="Local vs IBM", type="pil") | |
| bench_info = gr.Markdown("### Awaiting execution...") | |
| console = gr.Textbox(label="Console", lines=8, interactive=False) | |
| with gr.Tab("🛡️ Steane [[7,1,3]]"): | |
| gr.Markdown(""" | |
| ### Steane [[7,1,3]] Code — Quantum Error Correction | |
| Encodes **1 logical qubit in 7 physical qubits** with distance d=3. | |
| Corrects **any single-qubit error** (X, Z, or Y = XZ). | |
| **6 Stabilizers:** `IIIXXXX IXXIIXX XIXIXIX` (Z-error detection) / `IIIZZZZ IZZIIZZ ZIZIZIZ` (X-error detection) | |
| """) | |
| with gr.Row(): | |
| with gr.Column(scale=1): | |
| steane_qubit = gr.Slider(0, 6, 3, step=1, label="Error Qubit") | |
| steane_error = gr.Radio(['X', 'Z', 'Y'], value='X', label="Error Type") | |
| steane_shots = gr.Slider(256, 8192, 1024, step=256, label="Shots") | |
| steane_btn = gr.Button("🔬 Extract Syndrome", variant="primary", size="lg") | |
| with gr.Column(scale=2): | |
| steane_circuit = gr.Image(label="[[7,1,3]] + Syndromes", type="pil") | |
| steane_hist = gr.Image(label="Measured Syndromes", type="pil") | |
| steane_info = gr.Markdown("### Select parameters and run...") | |
| steane_console = gr.Textbox(label="Result", lines=4, interactive=False) | |
| with gr.Tab("⚛️ Ion Trap"): | |
| gr.Markdown(""" | |
| ### Ion Trap Simulator with AI Optimization | |
| Models trapped ion dynamics: phonon modes, spin-motion coupling, and Mølmer-Sørensen (MS) gates. | |
| Includes ion placement optimization to minimize SWAP operations. | |
| """) | |
| with gr.Row(): | |
| with gr.Column(scale=1): | |
| trap_ions = gr.Slider(2, 12, 5, step=1, label="Number of Ions") | |
| trap_noise = gr.Slider(0.001, 0.1, 0.01, step=0.001, label="Motional Heating Noise") | |
| trap_btn = gr.Button("⚛️ Simulate Ion Trap", variant="primary", size="lg") | |
| with gr.Column(scale=2): | |
| trap_circuit = gr.Image(label="MS Gate Circuit", type="pil") | |
| trap_spectrum = gr.Image(label="Mode Spectrum", type="pil") | |
| trap_info = gr.Markdown("### Adjust parameters and simulate...") | |
| trap_console = gr.Textbox(label="Console", lines=3, interactive=False) | |
| with gr.Tab("💰 Cost Reduction"): | |
| gr.Markdown(""" | |
| ### Quantum Cost Optimization | |
| Automatic cost analysis based on execution counts. | |
| Determines minimum shots needed while maintaining statistical significance. | |
| """) | |
| with gr.Row(): | |
| with gr.Column(scale=1): | |
| cost_json = gr.Textbox(label="Counts (JSON)", value='{"0000": 8180, "0001": 12}', lines=4) | |
| cost_per_shot = gr.Number(value=0.00001, label="Cost per shot ($)") | |
| cost_btn = gr.Button("💰 Analyze Costs", variant="primary") | |
| with gr.Column(scale=2): | |
| cost_dist = gr.Image(label="Distribution", type="pil") | |
| cost_gauge = gr.Image(label="Potential Savings", type="pil") | |
| cost_info = gr.Markdown("### Enter counts and analyze...") | |
| with gr.Tab("🛡️ Guardian AI"): | |
| gr.Markdown(f""" | |
| ### Guardian AI — Intelligent Management System | |
| Monitors, diagnoses, and automatically optimizes all laboratory operations. | |
| **Capabilities:** Auto circuit diagnosis, error detection & correction, parameter optimization, fidelity & cost estimation. | |
| """) | |
| guardian_btn = gr.Button("🔄 Refresh Status", variant="primary") | |
| guardian_status = gr.Markdown(guardian.get_status()) | |
| guardian_logs = gr.Textbox(label="Recent Logs", lines=15, interactive=False) | |
| with gr.Tab("📜 QASM"): | |
| gr.Code(value=QASM_BENCHMARK, language="python", label="Benchmark Circuit (OpenQASM 2.0)", lines=50) | |
| gr.HTML(f""" | |
| <div style="text-align:center; padding:20px; color:#666; font-size:0.85em;"> | |
| <strong>{COMPANY}</strong> — Quantum Laboratory v{VERSION}<br> | |
| <span style="color:#4ECDC4;">Real Quantum Results. Real Business Impact.</span><br> | |
| <span style="color:#888;">API keys processed via secure internal proxy. No data stored.</span> | |
| </div> | |
| """) | |
| bench_btn.click(fn=handle_benchmark_local, inputs=[bench_shots], outputs=[bench_circuit_img, bench_bloch, bench_hist, bench_info, console]) | |
| connect_btn.click(fn=handle_connect_ibm, inputs=[ibm_token, ibm_instance], outputs=[bench_info, ibm_status, ibm_backend]) | |
| disconnect_btn.click(fn=lambda: (ibm_proxy.disconnect(), "⚫ IBM: Disconnected", gr.update(choices=[], visible=False)), outputs=[ibm_status, ibm_backend]) | |
| ibm_run_btn.click(fn=handle_run_ibm, inputs=[ibm_shots, ibm_backend, ibm_optim], outputs=[ibm_hist, cmp_hist, bench_info, console]) | |
| steane_btn.click(fn=handle_steane_demo, inputs=[steane_qubit, steane_error, steane_shots], outputs=[steane_circuit, steane_hist, steane_info, steane_console]) | |
| trap_btn.click(fn=handle_ion_trap_sim, inputs=[trap_ions, trap_noise], outputs=[trap_circuit, trap_spectrum, trap_info, trap_console]) | |
| cost_btn.click(fn=handle_cost_reduction, inputs=[cost_json, cost_per_shot], outputs=[cost_dist, cost_gauge, cost_info]) | |
| guardian_btn.click(fn=handle_guardian_status, outputs=[guardian_status, guardian_logs]) | |
| if __name__ == "__main__": | |
| demo.queue(max_size=10, default_concurrency_limit=3).launch( | |
| server_name="0.0.0.0", | |
| theme=THEME, css=CSS, | |
| ) |