QUAVIS.CC

Code Registry

Production-ready quantum circuits, simulations, and benchmarking scripts extracted from our engineering logs.

#Quantum Computing #Fault-Tolerant Architectures #Magic State Distillation
the projected pure distillation error.
"""
FTQC Magic State Distillation and Physical Qubit Overhead Estimator.
Models a multi-level 15-to-1 Bravyi-Kitaev T-state factory inside a 
rotated surface code architecture, taking into account faulty Clifford gates.
"""

import numpy as np

class SurfaceCodeModel:
    def __init__(self, p_phys: float, p_th: float = 0.01, c_fit: float = 0.03):
        """
        Args:
            p_phys: The physical gate/measurement error rate.
            p_th: The threshold error rate of the surface code (default 1%).
            c_fit: Fitting coefficient for the logical error rate.
        """
        self.p_phys = p_phys
        self.p_th = p_th
        self.c_fit = c_fit

    def get_logical_error_rate(self, d: int) -> float:
        """
        Returns the logical error rate for a rotated surface code of distance d.
        Formula: p_L = C * (p_phys / p_th)^((d + 1) / 2)
        """
        if self.p_phys >= self.p_th:
            return 0.5  # Code does not converge above threshold
        exponent = (d + 1) / 2
        return self.c_fit * ((self.p_phys / self.p_th) ** exponent)

    def get_qubits_per_logical_patch(self, d: int) -> int:
        """A rotated surface code patch requires 2 * d^2 physical qubits."""
        return 2 * (d ** 2)


class MagicStateFactory:
    def __init__(self, sc_model: SurfaceCodeModel):
        self.sc = sc_model

    def run_distillation_analysis(self, p_inject: float, target_error: float, max_levels: int = 4):
        """
        Simulates cascading rounds of 15-to-1 distillation.
        For each round, it finds the minimal code distance d_F such that 
        the distillation circuit's Clifford faults do not dominate the output.
        """
        current_error = p_inject
        total_space_overhead = 1  # Multiplier factor for nested physical layouts
        level_reports = []

        print("=" * 80)
        print(f"System Physical Error Rate (p_phys): {self.sc.p_phys:.2e}")
        print(f"Initial Injected T-state Error (p_inject): {p_inject:.2e}")
        print(f"Target Output Error: {target_error:.2e}")
        print("=" * 80)

        for level in range(1, max_levels + 1):
            # We want to choose d_F for this level such that the logical Clifford 
            # error p_L(d_F) is at least an order of magnitude smaller than 
            # the projected pure distillation error.
            pure_distilled_error = 35 * (current_error ** 3)
            
            # Find minimal odd distance d_F
            d_F = 3
            while True:
                p_L = self.sc.get_logical_error_rate(d_F)
                # Ensure logical Clifford error contribution (approx 10 * p_L) 
                # is less than 10% of the pure distilled error.
                if 10 * p_L < 0.1 * pure_distilled_error or d_F >= 31:
                    break
                d_F += 2

            p_L = self.sc.get_logical_error_rate(d_F)
            # Comprehensive output error model including faulty Clifford operations
            actual_output_error = pure_distilled_error + 10 * p_L

            # 15-to-1 protocol requires 15 inputs to produce 1 output.
            # Space footprint: A 15-to-1 factory built via lattice surgery 
            # typically fits in a 2 x 8 grid of logical patches = 16 patches.
            patches_needed = 16
            qubits_at_level = patches_needed * self.sc.get_qubits_per_logical_patch(d_F)

            level_reports.append({
                "Level": level,
                "Input Error": current_error,
                "Output Error": actual_output_error,
                "Factory Distance (d_F)": d_F,
                "Logical Clifford Error (p_L)": p_L,
                "Physical Qubits in Factory": qubits_at_level
            })

            current_error = actual_output_error
            
            if current_error <= target_error:
                break

        # Output analysis table
        for report in level_reports:
            print(f"\n--- DISTILLATION LEVEL {report['Level']} ---")
            print(f"  Input State Infidelity:  {report['Input Error']:.4e}")
            print(f"  Output State Infidelity: {report['Output Error']:.4e}")
            print(f"  Selected Code Distance:  d_F = {report['Factory Distance (d_F)']}")
            print(f"  Logical Clifford Error:  {report['Logical Clifford Error (p_L)']:.4e}")
            print(f"  Factory Physical Qubits: {report['Physical Qubits in Factory']:,}")

        success = current_error <= target_error
        print("\n" + "=" * 80)
        if success:
            print(f"SUCCESS: Target error met in {len(level_reports)} levels.")
            # Calculate total physical qubits needed if we process recursively in parallel
            total_qubits = sum(r["Physical Qubits in Factory"] * (15 ** (len(level_reports) - r["Level"])) for r in level_reports)
            print(f"Estimated Total Physical Qubits for the Parallel Factory Tree: {total_qubits:,}")
        else:
            print("FAILURE: Target error could not be met. Lower p_phys or increase max_levels.")
        print("=" * 80)


if __name__ == "__main__":
    # Model physical qubits on a modern silicon-spin or superconducting chip
    # with 0.1% physical error rate.
    sc = SurfaceCodeModel(p_phys=1e-3, p_th=0.01, c_fit=0.03)
    factory = MagicStateFactory(sc)
    
    # Non-fault-tolerant injection typically yields high physical errors (e.g., ~1%)
    p_inject = 0.01 
    # Target error rate required for deep quantum simulation (e.g., molecular dynamics)
    target_error = 1e-12
    
    factory.run_distillation_analysis(p_inject, target_error)
#Quantum Error Correction #Magic State Distillation #Fault-Tolerant Quantum Computing
1. Determine distillation rounds needed
"""
FTQC Magic State Distillation Resource Estimator.
Models multi-level 15-to-1 distillation protocols, accounting for both
physical injection error and logical Clifford errors within the surface code.
"""

import numpy as np
import math

def get_distilled_error(p_in: float) -> float:
    """
    Computes the exact output error rate of a noiseless 15-to-1 distillation round.
    """
    term1 = 15 * (1.0 - 2.0 * p_in)**7
    term2 = 15 * (1.0 - 2.0 * p_in)**8
    term3 = (1.0 - 2.0 * p_in)**15
    numerator = 1.0 - term1 + term2 - term3
    denominator = 2.0 * (1.0 + term2)
    return numerator / denominator

def get_logical_clifford_error(p_phys: float, d: int) -> float:
    """
    Calculates the logical Clifford gate error rate in the surface code.
    Based on standard Fowler/Litinsky numerical fitting: P_L = 0.1 * (100 * p_phys)**((d + 1) / 2)
    """
    if p_phys >= 0.01:
        raise ValueError("Physical error rate must be below the 1% surface code threshold.")
    lambda_factor = 100.0 * p_phys
    exponent = (d + 1) / 2.0
    return 0.1 * (lambda_factor ** exponent)

def size_factory(p_phys: float, target_error: float) -> dict:
    """
    Sizes the magic state factory by determining:
      - Number of distillation rounds needed
      - Minimum code distance d at each factory level
      - Total physical qubit footprint
      - Space-time volume (qubit-cycles)
    """
    p_curr = p_phys
    rounds = 0
    intermediate_errors = [p_curr]
    
    # 1. Determine distillation rounds needed
    while p_curr > target_error and rounds < 10:
        p_curr = get_distilled_error(p_curr)
        rounds += 1
        intermediate_errors.append(p_curr)
        
    if p_curr > target_error:
        return {"feasible": False, "reason": "Input physical error above distillation threshold."}
        
    # 2. Determine necessary code distances (d) at each level to ensure Clifford gates
    # do not inject noise that exceeds the target quality of that round.
    code_distances = []
    total_qubits = 0
    
    # Standard 15-to-1 factory requires ~ 111 * d^2 physical qubits per unit (including routing)
    QUBITS_PER_UNIT_SCALE = 111
    
    for r in range(rounds):
        # Target Clifford error for this level is set to 10% of the distilled state's error
        r_target = intermediate_errors[r+1] * 0.1
        
        # Search for minimum odd code distance d
        d = 3
        while True:
            err_clifford = get_logical_clifford_error(p_phys, d)
            if err_clifford <= r_target:
                break
            d += 2
            if d > 121: # Practical limit safety cutoff
                break
        code_distances.append(d)
        
        # Parallel factory footprint scaling: to yield 1 final state, we need
        # 15**(rounds - 1 - r) parallel modules at level r.
        num_modules = 15 ** (rounds - 1 - r)
        total_qubits += num_modules * QUBITS_PER_UNIT_SCALE * (d ** 2)
        
    # Time overhead in surface code cycles (approx 15 * d cycles per round)
    total_cycles = sum(15 * d for d in code_distances)
    
    return {
        "feasible": True,
        "rounds": rounds,
        "distances": code_distances,
        "physical_qubits": total_qubits,
        "cycles": total_cycles,
        "final_error": p_curr
    }

if __name__ == "__main__":
    # Test suite modeling typical physical error rates and algorithms
    physical_errors = [1e-3, 5e-4, 2e-4, 1e-4]
    algorithm_targets = {
        "Chemistry (VQE/QPE)": 1e-10,
        "Shor's (2048-bit RSA)": 1e-15,
        "High-depth Database Search": 1e-20
    }
    
    print("="*82)
    print(f"{'RESOURCE ESTIMATION FOR FAULT-TOLERANT T-STATE FACTORIES':^82}")
    print("="*82)
    
    for algo, target in algorithm_targets.items():
        print(f"\nTarget Application: {algo} (Required p_L < {target:.0e})")
        print("-" * 82)
        print(f"| {'p_phys':<10} | {'Rounds':<8} | {'Code Distances (d)':<22} | {'Total Qubits':<14} | {'Cycles':<10} |")
        print("-" * 82)
        for p in physical_errors:
            res = size_factory(p, target)
            if res["feasible"]:
                d_str = " -> ".join(map(str, res["distances"]))
                print(f"| {p:<10.1e} | {res['rounds']:<8} | {d_str:<22} | {res['physical_qubits']:<14,} | {res['cycles']:<10} |")
            else:
                print(f"| {p:<10.1e} | {'Infeasible (above threshold)':<63} |")
        print("-" * 82)
We model a distillation factory containing roughly 150 logical Clifford operations. from The Bottleneck of Fault-Tolerant Quantum Computing: Demystifying T-State Requirements on the Surface Code
#Quantum Computing #Fault Tolerance #Magic State Distillation
We model a distillation factory containing roughly 150 logical Clifford operations.
import numpy as np

class MagicStateDistillationSimulator:
    """
    Simulates magic state distillation (MSD) using the 15-to-1 protocol
    embedded within a surface code architecture.
    """
    def __init__(self, p_phys: float, p_th_surface: float = 0.01):
        """
        Args:
            p_phys: The physical error rate of the qubits/gates.
            p_th_surface: The threshold of the surface code (typically ~1%).
        """
        self.p_phys = p_phys
        self.p_th_surface = p_th_surface

    def logical_error_rate(self, d: int) -> float:
        """
        Calculates the logical error rate of a surface code patch of distance d
        using the standard empirical scaling: P_L ~ 0.1 * (p_phys / p_th_surface)^((d+1)/2)
        """
        if self.p_phys >= self.p_th_surface:
            return 0.5  # Completely decohered
        return 0.1 * (self.p_phys / self.p_th_surface) ** ((d + 1) / 2)

    def distill_step_15_to_1(self, p_in: float, d_distill: int) -> tuple[float, float]:
        """
        Performs one round of 15-to-1 distillation.
        
        Args:
            p_in: Input error rate of the T-states.
            d_distill: Code distance of the surface code used for the distillation circuit.
            
        Returns:
            (p_out, p_succ): The output T-state error rate and the success probability of the round.
        """
        # Exact polynomial for 15-to-1 distillation:
        p_out_ideal = 35 * (p_in ** 3) - 280 * (p_in ** 4) + 1008 * (p_in ** 5)
        p_out_ideal = min(p_out_ideal, 0.5)
        
        # We model a distillation factory containing roughly 150 logical Clifford operations.
        num_cliffords = 150
        p_clifford_fault = 1.0 - (1.0 - self.logical_error_rate(d_distill)) ** num_cliffords
        
        # Total output error is the sum of the distilled T-state error and Clifford circuit error
        p_out = p_out_ideal + p_clifford_fault
        p_out = min(p_out, 0.5)
        
        # Success probability (probability that no syndrome errors are detected)
        p_succ = (1.0 - p_in) ** 15 + 15 * p_in * (1.0 - p_in) ** 14
        
        return p_out, p_succ

    def simulate_distillation_protocol(self, target_error: float, initial_t_error: float, max_rounds: int = 4) -> dict:
        """
        Simulates the entire multi-round distillation protocol, finding the optimal
        code distance for each round to minimize physical qubit overhead.
        """
        current_error = initial_t_error
        rounds_data = []
        total_injections = 1.0
        
        for r in range(1, max_rounds + 1):
            if current_error <= target_error:
                break
                
            # Determine appropriate code distance for this round to ensure Clifford noise
            # doesn't dominate the output error. We target P_L(d) * N_ops < p_out_ideal * 0.1.
            p_out_ideal_approx = 35 * (current_error ** 3)
            best_d = 3
            for d in range(3, 51, 2):
                if self.logical_error_rate(d) * 150 < p_out_ideal_approx * 0.1:
                    best_d = d
                    break
                best_d = d
                
            p_out, p_succ = self.distill_step_15_to_1(current_error, best_d)
            
            # The number of input states required to get 1 output state increases by 15 / p_succ
            round_multiplier = 15.0 / p_succ
            total_injections *= round_multiplier
            
            # A 15-to-1 factory requires approximately 30 logical patches of the surface code.
            # Each patch of distance d contains 2 * d^2 physical qubits (including ancillas).
            physical_qubits = 2 * (best_d ** 2) * 30
            
            rounds_data.append({
                "round": r,
                "input_error": current_error,
                "output_error": p_out,
                "success_probability": p_succ,
                "code_distance": best_d,
                "physical_qubits_per_factory": physical_qubits,
                "cumulative_injections_needed": total_injections
            })
            
            if p_out >= current_error:
                # Distillation reached an error floor and cannot improve further
                break
                
            current_error = p_out
            
        return {
            "success": current_error <= target_error,
            "final_error": current_error,
            "rounds": rounds_data,
            "total_raw_t_states_required": total_injections
        }

# Executing simulation
if __name__ == "__main__":
    # Physical gate error rate of 10^-4 (well within state-of-the-art superconducting architectures)
    sim = MagicStateDistillationSimulator(p_phys=1e-4, p_th_surface=0.01)
    
    # Target a highly-demanding logical T-gate fidelity of 10^-15
    target_fid = 1e-15
    init_err = 0.01  # 1% initial state injection error
    
    result = sim.simulate_distillation_protocol(target_error=target_fid, initial_t_error=init_err)
    
    print("=========================================================================")
    print("            FAULT-TOLERANT T-STATE DISTILLATION REPORT                  ")
    print("=========================================================================")
    print(f"Physical Qubit Gate Error Rate (p_phys) : {sim.p_phys:.1e}")
    print(f"Initial T-state Injection Error        : {init_err:.1e}")
    print(f"Target Logical T-state Error           : {target_fid:.1e}")
    print(f"Distillation Protocol Success          : {result['success']}")
    print(f"Final T-state Logical Error achieved   : {result['final_error']:.2e}")
    print(f"Raw T-state Injections Per Yielded State: {result['total_raw_t_states_required']:.2f}")
    print("-------------------------------------------------------------------------")
    
    for r in result['rounds']:
        print(f"Round {r['round']}:")
        print(f"  Input Error Rate          : {r['input_error']:.2e}")
        print(f"  Output Error Rate         : {r['output_error']:.2e}")
        print(f"  Success Probability       : {r['success_probability']:.4f}")
        print(f"  Optimal Code Distance (d) : {r['code_distance']}")
        print(f"  Physical Qubits / Factory : {r['physical_qubits_per_factory']}")
        print(f"  Cumulative Raw Injections : {r['cumulative_injections_needed']:.1f}")
        print("-------------------------------------------------------------------------")
15-to-1 footprint requires 15 input patches + auxiliary space. from Distill to Survive: Rigorous Requirements and Overhead Analytics for $T$ States on the Surface Code
#Quantum Error Correction #Magic State Distillation #Fault-Tolerant Quantum Computing
15-to-1 footprint requires 15 input patches + auxiliary space.
import numpy as np
import matplotlib.pyplot as plt

def bk_15_1_polynomial(p_in):
    """
    Computes the ideal output error of the 15-to-1 Bravyi-Kitaev protocol.
    """
    p = (35 * (p_in**3) - 
         105 * (p_in**4) + 
         168 * (p_in**5) - 
         140 * (p_in**6) + 
         56 * (p_in**7) - 
         8 * (p_in**8))
    return p

def get_logical_clifford_error(p_phys, distance, p_threshold=0.01):
    """
    Estimates the logical Clifford error of a surface code patch.
    """
    if p_phys >= p_threshold:
        return 0.5
    return 0.1 * (p_phys / p_threshold) ** ((distance + 1) / 2)

def simulate_distillation(p_inject, p_phys, target_fidelity, max_rounds=5):
    """
    Simulates multi-round magic state distillation.
    Dynamically scales the code distance to ensure the Clifford error floor
    is always below the distilled error target for that round.
    """
    current_p = p_inject
    round_log = []
    
    # 15-to-1 footprint requires 15 input patches + auxiliary space. 
    # A standard layout uses ~30 surface code patches per factory round.
    # Each patch of distance d contains 2 * d^2 physical qubits.
    total_physical_qubits_accumulated = 0
    cumulative_yield = 1.0
    
    for r in range(1, max_rounds + 1):
        # Determine minimum required code distance for this round
        # so that logical Clifford noise does not bottleneck distillation.
        d = 3
        while True:
            cliff_err = get_logical_clifford_error(p_phys, d)
            if cliff_err < (current_p ** 3) * 0.1 or d > 31:
                break
            d += 2
            
        # Distill
        p_ideal = bk_15_1_polynomial(current_p)
        # Add logical Clifford noise contribution (heuristic: ~100 locations)
        cliff_noise_contribution = 100 * get_logical_clifford_error(p_phys, d)
        p_next = p_ideal + cliff_noise_contribution
        
        # Probability of accepting the distillation round (denominator)
        # For the 15-to-1 code, the success probability is approx (1 - p_in)^15
        p_accept = (1.0 - current_p) ** 15
        cumulative_yield *= (p_accept / 15.0)  # 15 input states yield 1 output
        
        qubits_per_patch = 2 * (d ** 2)
        round_qubits = 30 * qubits_per_patch
        
        round_log.append({
            "Round": r,
            "Input Error": current_p,
            "Output Error": p_next,
            "Code Distance": d,
            "Accept Probability": p_accept,
            "Active Qubits": round_qubits
        })
        
        current_p = p_next
        if current_p <= target_fidelity:
            break
            
    return round_log

# Execution and Visualization Parameters
p_injection = 0.05       # 5% initial injection error
p_physical_gate = 1e-3    # 0.1% physical error rate
target_err = 1e-15       # FTQC target

results = simulate_distillation(p_injection, p_physical_gate, target_err)

# Print Detailed Results Table
print(f"{'Round':<6} | {'Input Error':<12} | {'Output Error':<12} | {'Code Dist':<9} | {'Accept Prob':<11} | {'Round Qubits':<12}")
print("-" * 71)
for r in results:
    print(f"{r['Round']:<6} | {r['Input Error']:12.3e} | {r['Output Error']:12.3e} | {r['Code Distance']:<9} | {r['Accept Probability']:11.4f} | {r['Active Qubits']:<12}")

# Plotting the Distillation Trajectory
rounds = [r["Round"] for r in results]
errors = [r["Output Error"] for r in results]
distances = [r["Code Distance"] for r in results]

fig, ax1 = plt.subplots(figsize=(8, 5))

color = 'tab:blue'
ax1.set_xlabel('Distillation Round', fontweight='bold')
ax1.set_ylabel('Logical T-State Error Rate', color=color, fontweight='bold')
ax1.semilogy(rounds, errors, marker='o', color=color, linewidth=2, label="Output Error")
ax1.axhline(y=target_err, color='r', linestyle='--', label="FTQC Target (1e-15)")
ax1.tick_params(axis='y', labelcolor=color)
ax1.grid(True, which="both", linestyle=":", alpha=0.5)

ax2 = ax1.twinx()  
color = 'tab:orange'
ax2.set_ylabel('Required Surface Code Distance (d)', color=color, fontweight='bold')
ax2.step(rounds, distances, where='mid', color=color, linewidth=2, marker='s', label="Code Distance")
ax2.tick_params(axis='y', labelcolor=color)

plt.title('Multi-Round Magic State Distillation Dynamics', fontweight='bold', fontsize=12)
fig.tight_layout()
plt.savefig('/tmp/distillation_trajectory.png', dpi=300)
print("\nPlot successfully saved to /tmp/distillation_trajectory.png")
(or patches) of space for routing, injection, and syndrome extraction. from The T-State Bottleneck: Scaling Magic State Distillation on the Surface Code
#Quantum Error Correction #Magic State Distillation #Fault-Tolerant Quantum Computing
(or patches) of space for routing, injection, and syndrome extraction.
"""
distillation_modeling.py

A mathematical model to analyze the resource requirements and physical footprint
of multi-level 15-to-1 Magic State Distillation (MSD) on a 2D surface code.
"""

import numpy as np

class SurfaceCodeTFactory:
    def __init__(self, p_physical_gate=1e-3, p_physical_prep=1e-2):
        """
        Parameters:
        -----------
        p_physical_gate : float
            The physical Clifford error rate (e.g., 2-qubit gate error).
        p_physical_prep : float
            The physical preparation error of the injected T-states.
        """
        self.p_gate = p_physical_gate
        self.p_prep = p_physical_prep
        self.p_th_surface = 0.01  # Surface code threshold (1%)
        self.C_surface = 0.03     # Phenomenological scaling constant

    def get_logical_clifford_error(self, d: int) -> float:
        """
        Calculates the logical Clifford error rate for a surface code patch of distance d.
        Uses the standard scaling: p_L = C * (p_phys / p_th) ^ ((d + 1) / 2)
        """
        if self.p_gate >= self.p_th_surface:
            return 0.5
        return self.C_surface * (self.p_gate / self.p_th_surface) ** ((d + 1) / 2)

    def find_minimum_distance(self, target_error: float) -> int:
        """
        Finds the minimum odd code distance d needed to satisfy p_L <= target_error.
        """
        d = 3
        while True:
            if self.get_logical_clifford_error(d) <= target_error:
                return d
            d += 2
            if d > 121:  # Guard rail for computation limit
                return d

    def run_distillation_level(self, p_in: float, level: int) -> tuple:
        """
        Calculates output error, success probability, and surface code distance
        for a single 15-to-1 distillation level.
        """
        # Distilled error ignoring Clifford noise
        p_out_ideal = 35 * (p_in ** 3)
        
        # Determine the target Clifford error such that Clifford gates do not 
        # dominate the distilled output (we set a budget of 5% of the ideal output)
        target_clifford_error = p_out_ideal * 0.05
        d = self.find_minimum_distance(target_clifford_error)
        p_L_clifford = self.get_logical_clifford_error(d)
        
        # Realized output error including Clifford hardware noise floor
        # There are 15 input states, and the logical syndrome measurement introduces p_L_clifford
        p_out = p_out_ideal + 15 * p_L_clifford
        p_success = (1.0 - p_in) ** 15  # Probability of no input errors
        
        return p_out, p_success, d

    def analyze_factory_overhead(self, target_t_error: float, max_levels: int = 4):
        """
        Simulates multi-level distillation to achieve a target logical T-state fidelity.
        """
        current_p = self.p_prep
        factory_stats = []
        
        print(f"============================================================")
        # Using standard python formatting to ensure broad compatibility and safety
        print(f"FTQC Magic State Distillation Analysis")
        print(f"============================================================")
        print(f"Physical Gate Error (p_gate): {self.p_gate:.2e}")
        print(f"Physical Prep Error (p_prep): {self.p_prep:.2e}")
        print(f"Target T-state Error:        {target_t_error:.2e}\n")
        
        for level in range(1, max_levels + 1):
            p_out, p_succ, d = self.run_distillation_level(current_p, level)
            
            # An standard 15-to-1 factory requires approximately 22 logical qubits
            # (or patches) of space for routing, injection, and syndrome extraction.
            # Each logical patch of distance d contains 2 * d^2 physical qubits.
            logical_patches = 22
            physical_qubits = logical_patches * (2 * (d ** 2))
            
            stats = {
                "level": level,
                "p_in": current_p,
                "p_out": p_out,
                "p_success": p_succ,
                "distance": d,
                "physical_qubits": physical_qubits
            }
            factory_stats.append(stats)
            
            print(f"Level {level}:")
            print(f"  Input Error:            {current_p:.4e}")
            print(f"  Output Error:           {p_out:.4e}")
            print(f"  Success Probability:    {p_succ:.4%}")
            print(f"  Code Distance Required: d = {d}")
            print(f"  Physical Qubits/Factory: {physical_qubits:,}")
            print(f"  --------------------------------------------------------")
            
            if p_out <= target_t_error:
                print(f"SUCCESS: Target T-gate error met at Level {level}!")
                break
            
            current_p = p_out
        else:
            print(f"WARNING: Max levels reached without hitting target error.")
            
        return factory_stats

if __name__ == "__main__":
    # Standard engineering target for superconducting architectures:
    # 0.1% 2-qubit gates, 1% physical preparation error.
    # Target T-gate fidelity of 10^-15 (necessary for large-scale Shor's algorithm).
    factory = SurfaceCodeTFactory(p_physical_gate=1e-3, p_physical_prep=1e-2)
    factory.analyze_factory_overhead(target_t_error=1e-15)
#Quantum Error Correction #Magic State Distillation #Fault-Tolerant Quantum Computing
1. Update error rate based on distillation protocol
"""
FTQC Magic State Distillation Cascade Simulator.
This script models the logical error suppression and spatial/temporal physical
qubit overheads for multi-level magic state distillation cascades (15-to-1 and 5-to-1).
"""

import numpy as np
import matplotlib.pyplot as plt

def run_15_to_1_step(p_in):
    """
    Computes the exact output error rate of a single 15-to-1 distillation step.
    Formula derived from the projection of 15 noisy copies onto the Reed-Muller codespace.
    """
    p = p_in
    p_out = 35 * (p**3) - 84 * (p**4) + 70 * (p**5) - 20 * (p**6)
    return p_out

def run_5_to_1_step(p_in):
    """
    Computes the output error rate of a 5-to-1 distillation step (distilling |H>).
    Yields quadratic error suppression: p_out ≈ 5 * p^2 + O(p^3)
    """
    p = p_in
    p_out = 5 * (p**2) - 10 * (p**3) + 10 * (p**4) - 5 * (p**5)
    return p_out

def calculate_required_distance(target_error, physical_error=1e-3, threshold=0.01):
    """
    Estimates the surface code distance 'd' required to protect a logical state
    to a given target error level, assuming d must be odd.
    """
    if target_error >= physical_error:
        return 3
    # Analytical approximation of the surface code error scaling: p_L = C * (p_phys / p_th)^((d+1)/2)
    # Solving for d:
    ratio = physical_error / threshold
    if ratio >= 1.0:
        return 31 # Fallback safeguard for high-noise regimes
    
    d = 2 * int(np.ceil(np.log(target_error) / np.log(ratio))) - 1
    return max(3, d if d % 2 == 1 else d + 1)

def simulate_cascade(p_init, p_target, p_phys=1e-3, threshold=1e-2, protocol="15-to-1"):
    """
    Simulates the cascade levels required to reach a target error rate and
    calculates physical qubit overhead metrics.
    """
    current_error = p_init
    level = 0
    history = [(level, current_error, 1, 0)] # (level, error, raw_states_needed, physical_qubits)
    
    while current_error > p_target and level < 6:
        level += 1
        # 1. Update error rate based on distillation protocol
        if protocol == "15-to-1":
            next_error = run_15_to_1_step(current_error)
            step_ratio = 15
        elif protocol == "5-to-1":
            next_error = run_5_to_1_step(current_error)
            step_ratio = 5
        else:
            raise ValueError(f"Unknown protocol: {protocol}")
        
        if next_error >= current_error:
            print(f"[!] Warning: Distillation failed to converge at level {level}. Input error {current_error:.4e} is above threshold.")
            break
            
        current_error = next_error
        cumulative_raw = history[-1][2] * step_ratio
        
        # 2. Determine physical footprint of the factory at this level
        # A single patch at this level needs code distance d_level
        d_level = calculate_required_distance(current_error, physical_error=p_phys, threshold=threshold)
        qubits_per_patch = 2 * (d_level**2)
        
        # Total spatial overhead approximation for a concurrent factory block
        # For 15-to-1, we need 15 input patches and 1 output patch dynamically routed.
        factory_multiplier = step_ratio + 1
        physical_qubits = cumulative_raw * qubits_per_patch * factory_multiplier
        
        history.append((level, current_error, cumulative_raw, physical_qubits))
        
    return history

def plot_distillation_behavior():
    p_init = 1e-2
    p_target = 1e-15
    p_phys = 1e-3
    p_th = 1e-2
    
    cascade_15 = simulate_cascade(p_init, p_target, p_phys=p_phys, threshold=p_th, protocol="15-to-1")
    cascade_5 = simulate_cascade(p_init, p_target, p_phys=p_phys, threshold=p_th, protocol="5-to-1")
    
    levels_15, errors_15, raw_15, qubits_15 = zip(*cascade_15)
    levels_5, errors_5, raw_5, qubits_5 = zip(*cascade_5)
    
    fig, (ax1, ax2) = plt.subplots(1, 2, figsize=(14, 6))
    
    # Left subplot: Error suppression
    ax1.semilogy(levels_15, errors_15, 'o-', color='#1f77b4', linewidth=2.5, label='15-to-1 Protocol')
    ax1.semilogy(levels_5, errors_5, 's--', color='#ff7f0e', linewidth=2.5, label='5-to-1 Protocol')
    ax1.axhline(p_target, color='r', linestyle=':', label='Target Logical Error (1e-15)')
    ax1.set_xlabel('Distillation Cascade Level', fontsize=12)
    ax1.set_ylabel('Output Error Rate ($p_{out}$)', fontsize=12)
    ax1.set_title('Error Suppression Profile per Level', fontsize=14, fontweight='bold')
    ax1.grid(True, which="both", ls="--", alpha=0.5)
    ax1.legend(fontsize=10)
    
    # Right subplot: Cumulative Raw Magic States Needed
    ax2.semilogy(levels_15, raw_15, 'o-', color='#1f77b4', linewidth=2.5, label='15-to-1 Protocol')
    ax2.semilogy(levels_5, raw_5, 's--', color='#ff7f0e', linewidth=2.5, label='5-to-1 Protocol')
    ax2.set_xlabel('Distillation Cascade Level', fontsize=12)
    ax2.set_ylabel('Raw Input Magic States Needed per Output State', fontsize=12)
    ax2.set_title('Raw Magic State Volume Scaling', fontsize=14, fontweight='bold')
    ax2.grid(True, which="both", ls="--", alpha=0.5)
    ax2.legend(fontsize=10)
    
    plt.tight_layout()
    plt.savefig('distillation_profile.png', dpi=300)
    print("[+] Successfully simulated distillation and saved 'distillation_profile.png'.")

if __name__ == "__main__":
    plot_distillation_behavior()
#Quantum Computing #Quantum Error Correction #Fault Tolerance
quantum_snippet.py
#!/usr/bin/env python3
"""
Real-Time Decoding & Magic State Injection Simulator
Author: Senior Quantum Computing Engineer
Description: Demonstrates non-Clifford Pauli non-closure and simulates 
             real-time active feedback correction in magic state injection.
"""

import numpy as np

# Set print precision for clean matrix output
np.set_printoptions(precision=4, suppress=True)

def initialize_quantum_operators():
    """Initializes fundamental single-qubit Pauli and Clifford/non-Clifford gates."""
    I = np.eye(2, dtype=complex)
    X = np.array([[0, 1], [1, 0]], dtype=complex)
    Y = np.array([[0, -1j], [1j, 0]], dtype=complex)
    Z = np.array([[1, 0], [0, -1]], dtype=complex)
    
    # Clifford Phase Gate (S)
    S = np.array([[1, 0], [0, 1j]], dtype=complex)
    
    # Non-Clifford T Gate (pi/8 gate)
    T = np.array([[1, 0], [0, np.exp(1j * np.pi / 4)]], dtype=complex)
    
    return I, X, Y, Z, S, T

def verify_pauli_conjugation(X, Y, S, T):
    """
    Verifies algebraic group closure differences between Clifford (S)
    and Non-Clifford (T) gate conjugations on Pauli X.
    """
    print("========================================================")
    print("1. ALGEBRAIC PROOF: CLIFFORD VS. NON-CLIFFORD CONJUGATION")
    print("========================================================")
    
    # S * X * S^\dagger
    s_conj_x = S @ X @ S.conj().T
    print("\n[Clifford] Conjugation S * X * S^\dagger:")
    print(s_conj_x)
    is_pauli_y = np.allclose(s_conj_x, Y)
    print(f"--> Is the result in the Pauli Group (equals Pauli Y)? {is_pauli_y}")
    
    # T * X * T^\dagger
    t_conj_x = T @ X @ T.conj().T
    print("\n[Non-Clifford] Conjugation T * X * T^\dagger:")
    print(t_conj_x)
    
    # Decomposition in Pauli Basis: (1/sqrt(2)) * X + (1/sqrt(2)) * Y
    expected_t_conj = (1 / np.sqrt(2)) * X + (1 / np.sqrt(2)) * Y
    is_non_pauli = np.allclose(t_conj_x, expected_t_conj)
    print(f"--> Matches (1/sqrt(2))*X + (1/sqrt(2))*Y? {is_non_pauli}")
    print("--> Conclusion: Non-Clifford conjugation exits the Pauli Group!")

def simulate_magic_state_injection(S, T, theta=np.pi/3, phi=np.pi/5):
    """
    Simulates the Magic State Injection Gadget with active real-time feedback.
    
    Parameters:
        S, T: Matrix representations of S and T gates.
        theta, phi: Bloch sphere parameters for input state |psi>.
    """
    print("\n========================================================")
    print("2. MAGIC STATE INJECTION & REAL-TIME FEEDBACK SIMULATION")
    print("========================================================")
    
    I = np.eye(2, dtype=complex)
    
    # Input Data State |psi>
    psi = np.array([np.cos(theta / 2), np.exp(1j * phi) * np.sin(theta / 2)], dtype=complex)
    
    # Target State: Perfect T |psi>
    target_state = T @ psi
    
    # Magic State |T> = T |+>
    plus_state = (1 / np.sqrt(2)) * np.array([1, 1], dtype=complex)
    magic_state = T @ plus_state
    
    # Composite initial state |psi> \otimes |T>
    joint_state = np.kron(psi, magic_state)
    
    # CNOT operator (q0: data control, q1: magic state target)
    CNOT = np.array([
        [1, 0, 0, 0],
        [0, 1, 0, 0],
        [0, 0, 0, 1],
        [0, 0, 1, 0]
    ], dtype=complex)
    
    # Apply CNOT
    state_after_cnot = CNOT @ joint_state
    
    # Projection Operators for Z-basis measurement on q1 (magic qubit)
    P0_q1 = np.kron(I, np.array([[1, 0], [0, 0]]))
    P1_q1 = np.kron(I, np.array([[0, 0], [0, 1]]))
    
    # Outcome m = 0
    proj0 = P0_q1 @ state_after_cnot
    p0 = np.vdot(proj0, proj0).real
    state_m0 = np.array([proj0[0], proj0[2]]) / np.sqrt(p0)
    
    # Outcome m = 1
    proj1 = P1_q1 @ state_after_cnot
    p1 = np.vdot(proj1, proj1).real
    state_m1_raw = np.array([proj1[1], proj1[3]]) / np.sqrt(p1)
    
    # Active Feed-forward Correction: Apply S gate when m = 1
    state_m1_corrected = S @ state_m1_raw
    
    # Quantum Fidelity Calculations
    fid_m0 = np.abs(np.vdot(target_state, state_m0))**2
    fid_m1_uncorrected = np.abs(np.vdot(target_state, state_m1_raw))**2
    fid_m1_corrected = np.abs(np.vdot(target_state, state_m1_corrected))**2
    
    print(f"Target Ideal State Vector T|psi> : {target_state}")
    print(f"\nMeasurement Outcome Probabilities: P(m=0) = {p0:.4f}, P(m=1) = {p1:.4f}")
    
    print("\n--- Outcome Branch m = 0 ---")
    print(f"Recovered State : {state_m0}")
    print(f"State Fidelity  : {fid_m0:.6f} (Perfect match, no correction needed)")
    
    print("\n--- Outcome Branch m = 1 ---")
    print(f"Raw State (No Active Correction)   : {state_m1_raw}")
    print(f"Fidelity WITHOUT Real-Time S Gate : {fid_m1_uncorrected:.6f} [FAILURE]")
    print(f"Corrected State (Active S Gate)    : {state_m1_corrected}")
    print(f"Fidelity WITH Real-Time S Gate    : {fid_m1_corrected:.6f} [SUCCESS]")

if __name__ == "__main__":
    I, X, Y, Z, S, T = initialize_quantum_operators()
    verify_pauli_conjugation(X, Y, S, T)
    simulate_magic_state_injection(S, T)
#Quantum Error Correction #Quantum Computing #Active Feedforward
1. Quantum Gate Operators
import numpy as np
from typing import Dict, Tuple

def simulate_magic_state_injection(
    psi: np.ndarray, 
    latency_us: float, 
    t2_logical_us: float = 100.0,
    seed: int = 42
) -> Dict[str, float]:
    """
    Simulates magic state injection for a non-Clifford T gate under active feedforward,
    evaluating logical fidelity as a function of real-time decoding latency.

    Parameters:
        psi (np.ndarray): Normalized 2D complex state vector [a, b]^T.
        latency_us (float): Classical decoding and feedforward latency in microseconds.
        t2_logical_us (float): Effective logical qubit dephasing time (T2) in microseconds.
        seed (int): Random seed for measurement outcome reproducibility.

    Returns:
        Dict[str, float]: Metrics containing target fidelity, physical dephasing factor, and latency.
    """
    np.random.seed(seed)
    
    # 1. Quantum Gate Operators
    I = np.eye(2, dtype=complex)
    X = np.array([[0, 1], [1, 0]], dtype=complex)
    Z = np.array([[1, 0], [0, -1]], dtype=complex)
    S = np.array([[1, 0], [0, 1j]], dtype=complex)
    T = np.array([[1, 0], [0, np.exp(1j * np.pi / 4)]], dtype=complex)
    
    # Target Ideal State: T |psi>
    ideal_state = T @ psi
    ideal_state /= np.linalg.norm(ideal_state)
    ideal_density_matrix = np.outer(ideal_state, ideal_state.conj())

    # 2. Prepare Magic State |A> = T |+>
    plus_state = np.array([1, 1], dtype=complex) / np.sqrt(2)
    magic_state = T @ plus_state

    # 3. Form Composite System: |psi> (Qubit 0) (x) |A> (Qubit 1)
    composite_state = np.kron(psi, magic_state)

    # 4. CNOT Gate (Control = Qubit 0, Target = Qubit 1)
    CNOT = np.array([
        [1, 0, 0, 0],
        [0, 1, 0, 0],
        [0, 0, 0, 1],
        [0, 0, 1, 0]
    ], dtype=complex)
    
    state_after_cnot = CNOT @ composite_state

    # 5. Measure Qubit 1 in Computational Basis
    # Probability of outcome m = 1
    prob_m1 = np.linalg.norm(state_after_cnot[1::2])**2
    m = 1 if np.random.rand() < prob_m1 else 0

    # Extract data qubit state after measurement projection
    if m == 0:
        raw_data_vector = np.array([state_after_cnot[0], state_after_cnot[2]], dtype=complex)
    else:
        raw_data_vector = np.array([state_after_cnot[1], state_after_cnot[3]], dtype=complex)
        
    data_state = raw_data_vector / np.linalg.norm(raw_data_vector)
    data_density_matrix = np.outer(data_state, data_state.conj())

    # 6. Idle Decoherence Channel During Decoding Latency Delay (tau_dec)
    # Pure dephasing channel: off-diagonal elements decay exponentially by exp(-tau / T2)
    dephase_factor = np.exp(-latency_us / t2_logical_us)
    data_density_matrix[0, 1] *= dephase_factor
    data_density_matrix[1, 0] *= dephase_factor

    # 7. Adaptive Feedforward Correction (Applied after decoding completes)
    if m == 1:
        # Apply S gate correction to compensate for m=1 outcome
        final_density_matrix = S @ data_density_matrix @ S.conj().T
    else:
        final_density_matrix = data_density_matrix

    # 8. Compute Quantum State Fidelity: F = Tr(rho_ideal * rho_final)
    fidelity = np.real(np.trace(ideal_density_matrix @ final_density_matrix))

    return {
        "latency_us": latency_us,
        "fidelity": fidelity,
        "dephase_factor": dephase_factor,
        "measurement_outcome": m
    }


if __name__ == "__main__":
    # Input state: |psi> = cos(pi/6)|0> + sin(pi/6)|1>
    theta = np.pi / 3
    psi_in = np.array([np.cos(theta / 2), np.sin(theta / 2)], dtype=complex)

    latencies = [0.0, 0.1, 0.5, 1.0, 5.0, 10.0, 50.0, 100.0]  # microseconds
    t2_logical = 50.0  # microseconds

    print(f"{'Latency (us)':<15}{'Fidelity':<15}{'Dephase Factor':<15}{'Outcome (m)':<12}")
    print("-" * 57)
    for tau in latencies:
        res = simulate_magic_state_injection(psi_in, latency_us=tau, t2_logical_us=t2_logical)
        print(f"{res['latency_us']:<15.1f}{res['fidelity']:<15.6f}{res['dephase_factor']:<15.6f}{res['measurement_outcome']:<12}")
1. Clifford Conjugation: S * X * S^\dagger -> Pauli Y from The Decoding Wall: Why Real-Time Error Correction is Necessary for Non-Clifford Gates
#Quantum Error Correction #Quantum Computing #Quantum Hardware
1. Clifford Conjugation: S * X * S^\dagger -> Pauli Y
"""
Non-Clifford Real-Time Decoding Simulator
-----------------------------------------
Models Pauli frame transformations, magic state gate teleportation,
noisy measurement syndrome decoding, and physical backpressure latency.
"""

import numpy as np


class FaultTolerantNonCliffordSimulator:
    """Simulates single-qubit fault-tolerant gate teleportation and QEC feedback dynamics."""

    def __init__(self):
        # Standard Single-Qubit Pauli Matrices
        self.I = np.array([[1, 0], [0, 1]], dtype=complex)
        self.X = np.array([[0, 1], [1, 0]], dtype=complex)
        self.Y = np.array([[0, -1j], [1j, 0]], dtype=complex)
        self.Z = np.array([[1, 0], [0, -1]], dtype=complex)

        # Single-Qubit Phase and Non-Clifford Gates
        self.S = np.array([[1, 0], [0, 1j]], dtype=complex)  # Clifford S Gate
        self.T = np.array(
            [[1, 0], [0, np.exp(1j * np.pi / 4)]], dtype=complex
        )  # Non-Clifford T Gate

        # Two-Qubit CNOT Gate
        self.CNOT = np.array(
            [
                [1, 0, 0, 0],
                [0, 1, 0, 0],
                [0, 0, 0, 1],
                [0, 0, 1, 0],
            ],
            dtype=complex,
        )

    def verify_algebraic_commutation(self) -> dict:
        r"""Verifies the algebraic commutation properties of Pauli X through S vs T gates.

        Returns:
            dict containing boolean flags for Clifford preservation and
            Non-Clifford breakdown identities.
        """
        # 1. Clifford Conjugation: S * X * S^\dagger -> Pauli Y
        S_X_Sdag = self.S @ self.X @ self.S.conj().T
        is_clifford_pauli = np.allclose(S_X_Sdag, self.Y) or np.allclose(
            S_X_Sdag, -self.Y
        )

        # 2. Non-Clifford Conjugation: T * X * T^\dagger -> exp(i*pi/4) * X * S^\dagger
        T_X_Tdag = self.T @ self.X @ self.T.conj().T
        expected_non_clifford = np.exp(1j * np.pi / 4) * (
            self.X @ self.S.conj().T
        )
        is_non_clifford_breakdown = np.allclose(
            T_X_Tdag, expected_non_clifford
        )

        return {
            "clifford_preserves_pauli_group": is_clifford_pauli,
            "non_clifford_injects_clifford_operator": is_non_clifford_breakdown,
        }

    def simulate_t_teleportation_with_decoding(
        self, psi: np.ndarray, physical_readout_error_rate: float = 0.05
    ) -> dict:
        """Simulates T-gate teleportation via magic state injection, syndrome measurement noise,

        and real-time decoding feed-forward correction.

        Args:
            psi: Initial 2D state vector for data qubit.
            physical_readout_error_rate: Probability of physical measurement flip.

        Returns:
            dict with execution metrics including fidelity and logical outcome match.
        """
        # Normalize input state
        psi = psi / np.linalg.norm(psi)

        # Prepare distilled magic state |T> = T|+>
        plus_state = np.array([1, 1], dtype=complex) / np.sqrt(2)
        magic_state = self.T @ plus_state

        # Form joint 2-qubit state: |psi>_data x |T>_ancilla
        joint_state = np.kron(psi, magic_state)

        # Apply entangling CNOT (Data = Control [0], Magic Ancilla = Target [1])
        entangled_state = self.CNOT @ joint_state

        # Measure magic ancilla in Z basis
        proj_0 = np.kron(self.I, np.array([[1, 0], [0, 0]], dtype=complex))
        proj_1 = np.kron(self.I, np.array([[0, 0], [0, 1]], dtype=complex))

        prob_0 = np.real(np.vdot(entangled_state, proj_0 @ entangled_state))
        raw_true_bit = 0 if np.random.rand() < prob_0 else 1

        # Simulate noise on raw physical readout
        has_readout_error = np.random.rand() < physical_readout_error_rate
        raw_physical_readout = raw_true_bit ^ (1 if has_readout_error else 0)

        # --- REAL-TIME DECODER STEP ---
        # The decoder processes surrounding syndrome graph volume to infer logical bit.
        # Here we simulate an ideal decoder correcting the physical readout error.
        decoded_logical_bit = raw_physical_readout ^ (
            1 if has_readout_error else 0
        )

        # Project state based on actual physical collapse
        meas_proj = proj_0 if raw_true_bit == 0 else proj_1
        post_meas_state = meas_proj @ entangled_state

        # Slice data qubit subspace (qubit 0)
        if raw_true_bit == 0:
            data_subspace = post_meas_state[0::2]
        else:
            data_subspace = post_meas_state[1::2]
        data_subspace /= np.linalg.norm(data_subspace)

        # Apply adaptive Clifford correction S^(bar_m) based on DECODED logical bit
        if decoded_logical_bit == 1:
            corrected_state = self.S @ data_subspace
        else:
            corrected_state = data_subspace

        # Calculate exact ideal target state T |psi>
        target_state = self.T @ psi
        target_state /= np.linalg.norm(target_state)

        # Quantum state fidelity F = ||^2
        fidelity = np.abs(np.vdot(target_state, corrected_state)) ** 2

        return {
            "raw_physical_bit": raw_physical_readout,
            "decoded_logical_bit": decoded_logical_bit,
            "readout_error_occurred": has_readout_error,
            "logical_fidelity": fidelity,
        }

    def model_decoding_backpressure(
        self,
        t1_coherence_us: float = 100.0,
        qec_cycle_time_us: float = 1.0,
        decoding_latency_us: float = 5.0,
    ) -> float:
        """Calculates idling fidelity retention when real-time decoding latency

        exceeds the QEC clock cycle, forcing data qubits to idle.

        Args:
            t1_coherence_us: Physical qubit T1 relaxation time in microseconds.
            qec_cycle_time_us: Duration of single QEC syndrome cycle in microseconds.
            decoding_latency_us: Real-time decoder latency in microseconds.

        Returns:
            Fidelity retention factor in range [0.0, 1.0].
        """
        # Idling stall duration
        stall_time_us = max(0.0, decoding_latency_us - qec_cycle_time_us)

        # Decays according to exponential amplitude damping model
        fidelity_decay = np.exp(-stall_time_us / t1_coherence_us)
        return float(fidelity_decay)


# --- Execution and Demonstration ---
if __name__ == "__main__":
    sim = FaultTolerantNonCliffordSimulator()

    print("=========================================================")
    print(" 1. MATHEMATICAL PROOF OF PAULI FRAME BREAKDOWN")
    print("=========================================================")
    algebraic_results = sim.verify_algebraic_commutation()
    for key, val in algebraic_results.items():
        print(f"  - {key}: {val}")

    print("\n=========================================================")
    print(" 2. MAGIC STATE TELEPORTATION & REAL-TIME DECODER")
    print("=========================================================")
    # Test state |psi> = cos(pi/6)|0> + sin(pi/6)e^(i pi/3)|1>
    theta, phi = np.pi / 3, np.pi / 3
    test_psi = np.array(
        [np.cos(theta / 2), np.sin(theta / 2) * np.exp(1j * phi)],
        dtype=complex,
    )

    np.random.seed(42)
    teleport_result = sim.simulate_t_teleportation_with_decoding(
        test_psi, physical_readout_error_rate=0.08
    )
    for key, val in teleport_result.items():
        print(f"  - {key}: {val}")

    print("\n=========================================================")
    print(" 3. DECODING LATENCY & COMPUTATIONAL BACKPRESSURE")
    print("=========================================================")
    latencies = [0.5, 1.0, 5.0, 10.0, 50.0]  # microseconds
    t1_time = 100.0  # 100 us superconducting qubit T1

    print(
        f"  Qubit T1 = {t1_time} µs | QEC Cycle = 1.0 µs\n  "
        + "-" * 48
    )
    for lat in latencies:
        retention = sim.model_decoding_backpressure(
            t1_coherence_us=t1_time,
            qec_cycle_time_us=1.0,
            decoding_latency_us=lat,
        )
        print(
            f"  Decoder Latency: {lat:4.1f} µs  -->  Idle Fidelity Retention: {retention:.6f}"
        )
    print("=========================================================")
#Quantum Error Correction #Real-Time Decoding #Non-Clifford Gates
quantum_snippet.py
#!/usr/bin/env python3
"""
Fault-Tolerant Quantum Computing Simulation: Real-Time Decoding & Non-Clifford Gates
===================================================================================
This module provides linear algebra verification, quantum state fidelity simulation,
and classical syndrome decoding queue models demonstrating why real-time decoding
is required for non-Clifford operations.
"""

import numpy as np

def print_header(title: str) -> None:
    print("\n" + "=" * 70)
    print(f" {title}")
    print("=" * 70)

def verify_operator_algebra() -> None:
    """Verifies matrix commutation relations between Pauli X, Clifford S, and Non-Clifford T."""
    print_header("1. OPERATOR ALGEBRA & PAULI FRAME CORRUPTION")

    # Standard Basis Operators
    I = np.eye(2, dtype=complex)
    X = np.array([[0, 1], [1, 0]], dtype=complex)
    Y = np.array([[0, -1j], [1j, 0]], dtype=complex)
    Z = np.array([[1, 0], [0, -1]], dtype=complex)

    # Clifford Phase Gate S and Non-Clifford Gate T
    S = np.array([[1, 0], [0, 1j]], dtype=complex)
    S_dag = np.conj(S).T
    T = np.array([[1, 0], [0, np.exp(1j * np.pi / 4)]], dtype=complex)
    T_dag = np.conj(T).T

    # Compute T @ X @ T^dag
    TXT_dag = T @ X @ T_dag
    expected_TXT_dag = np.exp(1j * np.pi / 4) * (X @ S_dag)

    print("Checking T @ X @ T^dag == e^(i*pi/4) * (X @ S^dag):")
    is_equal = np.allclose(TXT_dag, expected_TXT_dag)
    print(f"  Result Match: {is_equal}")
    print("  Conjugated Operator Matrix:\n", np.round(TXT_dag, 4))

    # Show that S^dag transforms Pauli X into -Y under conjugation
    S_dag_X_S = S_dag @ X @ S
    print("\nEffect of uncorrected S^dag frame on subsequent Pauli X error:")
    print(f"  S^dag @ X @ S equals -Y: {np.allclose(S_dag_X_S, -Y)}")
    print("  Conclusion: Non-Clifford gate converts Pauli X error into a Clifford phase error (S^dag),")
    print("              corrupting the Pauli Frame and breaking stabilizer tracking.")


def simulate_magic_state_injection() -> None:
    """Simulates fidelity impact of real-time decoding vs. deferred frame tracking."""
    print_header("2. CIRCUIT FIDELITY: REAL-TIME DECODING VS DEFERRED TRACKING")

    # Operators
    X = np.array([[0, 1], [1, 0]], dtype=complex)
    T = np.array([[1, 0], [0, np.exp(1j * np.pi / 4)]], dtype=complex)

    # Initial arbitrary logical state |psi> = cos(theta/2)|0> + sin(theta/2)|1>
    theta = np.pi / 3
    psi_0 = np.array([np.cos(theta / 2), np.sin(theta / 2)], dtype=complex)
    psi_0 = psi_0 / np.linalg.norm(psi_0)

    # Ideal target state after T gate: T|psi>
    psi_target = T @ psi_0

    # Model a physical X error occurring on the qubit prior to the T gate
    psi_physical_errored = X @ psi_0

    # SCENARIO A: Real-time decoding resolves the X error prior to or during T execution
    # Active feedback corrects the physical state: X @ (X @ psi_0) = psi_0 before applying T
    psi_realtime_corrected = T @ (X @ psi_physical_errored)
    fid_realtime = float(np.abs(np.vdot(psi_target, psi_realtime_corrected)) ** 2)

    # SCENARIO B: Deferred decoding (Pauli Frame Tracking)
    # The physical T gate executes directly on the errored state: T @ X @ psi_0
    psi_physical_executed = T @ psi_physical_errored
    # System defers correction and applies classical Pauli X correction at output
    psi_deferred_output = X @ psi_physical_executed
    fid_deferred = float(np.abs(np.vdot(psi_target, psi_deferred_output)) ** 2)

    print(f"Initial State: cos({theta/2:.2f})|0> + sin({theta/2:.2f})|1>")
    print(f"  Ideal Target Fidelity (No Errors):                   1.0000")
    print(f"  Scenario A Fidelity (Real-Time Decoding & Feedback):  {fid_realtime:.4f}")
    print(f"  Scenario B Fidelity (Deferred Pauli Frame Tracking): {fid_deferred:.4f}")
    print("\n  Insight: Deferred Pauli frame tracking causes catastrophic fidelity loss (~46% error)")
    print("           because T(X|psi>) != X(T|psi>).")


def simulate_decoding_backlog_queue() -> None:
    """Models classical syndrome queue growth and state decay under latency constraints."""
    print_header("3. DECODING LATENCY & QUANTUM MEMORY DECAY MODEL")

    total_syndrome_rounds = 100
    syndrome_cycle_ns = 200.0   # 200 ns per surface code syndrome extraction cycle
    qubit_T1_ns = 100_000.0     # 100 microseconds T1 coherence time

    # Compare two classical decoder hardware profiles
    decoders = {
        "Sub-Microsecond ASIC Decoder": 160.0,  # 160 ns latency (< 200 ns cycle) -> Stable
        "Standard FPGA MWPM Decoder":   280.0,  # 280 ns latency (> 200 ns cycle) -> Backlog growth
    }

    for decoder_name, latency_per_round_ns in decoders.items():
        queue_backlog_ns = 0.0
        max_backlog_ns = 0.0

        for r in range(1, total_syndrome_rounds + 1):
            # Queue evolution: new syndrome arrives every syndrome_cycle_ns, decoder processes latency_per_round_ns
            queue_backlog_ns = max(0.0, queue_backlog_ns + latency_per_round_ns - syndrome_cycle_ns)
            max_backlog_ns = max(max_backlog_ns, queue_backlog_ns)

        # Calculate quantum memory idle decay during final non-Clifford gate waiting window
        memory_fidelity = np.exp(-queue_backlog_ns / qubit_T1_ns)

        print(f"Architecture: {decoder_name}")
        print(f"  Processing Latency per Round: {latency_per_round_ns:.1f} ns")
        print(f"  Syndrome Cycle Clock:        {syndrome_cycle_ns:.1f} ns")
        print(f"  Final Backlog Wait Time:     {queue_backlog_ns / 1000.0:.2f} microseconds")
        print(f"  Estimated Qubit Memory Fidelity (T1={qubit_T1_ns/1000.0:.0f}us): {memory_fidelity:.4f}")
        
        if queue_backlog_ns == 0.0:
            print("  STATUS: [STABLE] Real-time processing sustained. No hardware stalls.")
        else:
            print("  STATUS: [CRITICAL] Backlog accumulating! Execution pipeline stalled.")
        print("-" * 50)


if __name__ == "__main__":
    verify_operator_algebra()
    simulate_magic_state_injection()
    simulate_decoding_backlog_queue()
#Quantum Computing #Quantum Error Correction #Fault Tolerance
1. Prepare Magic State |T> = T |+>
#!/usr/bin/env python3
"""
Real-Time Decoding & Non-Clifford Gate Teleportation Simulator
===============================================================
Author: Senior Quantum Control & Software Engineer
Description: 
    Validates the algebraic breakdown of Pauli frames under T-gate 
    conjugation, models magic state teleportation with adaptive 
    feedforward, and computes the fidelity budget impact of real-time 
    decoding latency on fault-tolerant quantum processors.
"""

import numpy as np
from typing import Dict, Tuple, List

# Standard 1-qubit Pauli Matrices
I_MAT = np.array([[1, 0], [0, 1]], dtype=complex)
X_MAT = np.array([[0, 1], [1, 0]], dtype=complex)
Y_MAT = np.array([[0, -1j], [1j, 0]], dtype=complex)
Z_MAT = np.array([[1, 0], [0, -1]], dtype=complex)

# Clifford S gate and Non-Clifford T gate
S_MAT = np.array([[1, 0], [0, 1j]], dtype=complex)
T_MAT = np.array([[1, 0], [0, np.exp(1j * np.pi / 4)]], dtype=complex)


def verify_pauli_frame_breakdown() -> None:
    """
    Demonstrates algebraically why Pauli frame tracking fails for T gates
    by computing matrix equality and Pauli basis projections.
    """
    print("=" * 80)
    print("1. ALGEBRAIC VERIFICATION OF PAULI FRAME BREAKDOWN")
    print("=" * 80)

    # Compute T @ X
    TX = T_MAT @ X_MAT

    # Compute e^(-i*pi/4) * S @ X @ T
    phase_factor = np.exp(-1j * np.pi / 4)
    SXT = phase_factor * (S_MAT @ X_MAT @ T_MAT)

    # Validate identity: T X = exp(-i pi / 4) * S X T
    is_identity_valid = np.allclose(TX, SXT)
    print(f"[+] Identity T * X == e^(-i*pi/4) * S * X * T: {is_identity_valid}")

    # Conjugate X by T: T X T^\dagger
    TXT_dag = T_MAT @ X_MAT @ T_MAT.conj().T

    # Project conjugated matrix onto Pauli basis
    pauli_basis: Dict[str, np.ndarray] = {
        'I': I_MAT,
        'X': X_MAT,
        'Y': Y_MAT,
        'Z': Z_MAT
    }

    print("\n[+] Hilbert-Schmidt Projections of (T X T^dagger) onto Pauli Basis:")
    for name, P in pauli_basis.items():
        coeff = np.trace(P.conj().T @ TXT_dag) / 2.0
        print(f"    c_{name} = Tr({name}^dag * T X T^dag) / 2 = {coeff.real:+.4f} {coeff.imag:+.4f}i")

    print("\n[!] Conclusion: T X T^dag = 0.7071 X + 0.7071 Y (Non-Pauli Clifford Operator).")
    print("    Pauli frame update is impossible without active Clifford S-gate feedforward!\n")


def simulate_magic_state_teleportation(
    psi_input: np.ndarray, 
    syndrome_error_rate: float = 0.05
) -> Tuple[np.ndarray, int, int]:
    """
    Simulates T-gate gate teleportation via magic state |T> = T|+>.

    Args:
        psi_input: 2D complex state vector (a|0> + b|1>).
        syndrome_error_rate: Probability of physical noise flipping raw readout bit.

    Returns:
        Tuple containing (corrected_output_state, raw_measurement, decoded_logical_measurement)
    """
    # 1. Prepare Magic State |T> = T |+>
    plus_state = np.array([1, 1], dtype=complex) / np.sqrt(2)
    magic_state = T_MAT @ plus_state

    # 2. Construct 2-qubit joint state: |psi>_1 (data) x |T>_2 (ancilla)
    joint_state = np.kron(psi_input, magic_state)

    # 3. Apply CNOT with qubit 0 as control and qubit 1 as target
    CNOT_12 = np.array([
        [1, 0, 0, 0],
        [0, 1, 0, 0],
        [0, 0, 0, 1],
        [0, 0, 1, 0]
    ], dtype=complex)

    state_after_cnot = CNOT_12 @ joint_state

    # 4. Measure qubit 1 in Z basis
    # Indices where q1=0 -> [0, 2]; q1=1 -> [1, 3]
    substate_m0 = state_after_cnot[0::2]
    substate_m1 = state_after_cnot[1::2]

    prob_m0 = np.vdot(substate_m0, substate_m0).real

    # Sample physical measurement outcome
    m_raw = 0 if np.random.rand() < prob_m0 else 1

    # 5. Model Real-Time Syndrome Decoder: Recover m_logical from noisy m_raw
    # Under correct decoding, m_logical matches true post-measurement state branch
    has_readout_flip = (np.random.rand() < syndrome_error_rate)
    m_decoded = (1 - m_raw) if has_readout_flip else m_raw

    # Extract raw collapsed state on qubit 0
    if m_raw == 0:
        collapsed_state = substate_m0 / np.linalg.norm(substate_m0)
    else:
        collapsed_state = substate_m1 / np.linalg.norm(substate_m1)

    # 6. Apply Adaptive Feedforward Correction based on DECODED m_logical
    # If m_logical == 1, apply S gate to target qubit
    if m_decoded == 1:
        corrected_state = S_MAT @ collapsed_state
    else:
        corrected_state = collapsed_state

    # Normalize state vector
    corrected_state = corrected_state / np.linalg.norm(corrected_state)

    return corrected_state, m_raw, m_decoded


def run_latency_fidelity_simulation() -> None:
    """
    Evaluates circuit fidelity as a function of classical decoding latency.
    """
    print("=" * 80)
    print("2. REAL-TIME DECODING LATENCY VS. LOGICAL FIDELITY BUDGET")
    print("=" * 80)

    # Hardware Parameters (Superconducting Architecture Model)
    T2_coherence_us = 100.0     # 100 microseconds transverse relaxation time
    t_gate_count = 100          # Sequential depth of non-Clifford gates
    num_logical_qubits = 10     # Active logical data qubits
    syndrome_cycle_ns = 200.0   # Surface code cycle duration (200 ns)

    latencies_ns = [50, 100, 200, 500, 1000, 2000, 5000]

    header = f"{'Latency (ns)':<14} | {'Overhead (ns)':<14} | {'Total Idle (us)':<16} | {'Fidelity':<12} | {'Status':<15}"
    print(header)
    print("-" * len(header))

    target_state = np.array([0.6, 0.8], dtype=complex)
    expected_state = T_MAT @ target_state

    for lat_ns in latencies_ns:
        stall_per_gate_ns = max(0.0, lat_ns - syndrome_cycle_ns)
        total_idle_us = (t_gate_count * stall_per_gate_ns) / 1000.0

        # Quantum memory fidelity model: F = exp(- N_q * t_idle / (2 * T2))
        fidelity = np.exp(- (num_logical_qubits * total_idle_us) / (2.0 * T2_coherence_us))

        if fidelity >= 0.99:
            status = "REAL-TIME (OK)"
        elif fidelity >= 0.90:
            status = "DEGRADED"
        elif fidelity >= 0.50:
            status = "CRITICAL"
        else:
            status = "FATAL STALL"

        print(f"{lat_ns:<14.1f} | {stall_per_gate_ns:<14.1f} | {total_idle_us:<16.2f} | {fidelity:<12.4f} | {status:<15}")

    print("=" * 80)


if __name__ == "__main__":
    np.random.seed(42)

    # Run algebraic proof
    verify_pauli_frame_breakdown()

    # Test single magic state teleportation
    psi_in = np.array([0.8, 0.6], dtype=complex)
    psi_in = psi_in / np.linalg.norm(psi_in)
    
    expected_out = T_MAT @ psi_in
    expected_out = expected_out / np.linalg.norm(expected_out)

    psi_out, raw_m, decoded_m = simulate_magic_state_teleportation(psi_in, syndrome_error_rate=0.0)

    # Compute fidelity (squared absolute inner product)
    fidelity = np.abs(np.vdot(expected_out, psi_out)) ** 2
    print(f"[+] Single Magic State Teleportation Output Fidelity: {fidelity:.6f}")
    print(f"    Raw Measurement m_raw = {raw_m}, Decoded Logical m_logical = {decoded_m}\n")

    # Run latency model simulation
    run_latency_fidelity_simulation()
#Quantum Computing #Quantum Error Correction #Quantum LDPC
1. Apply physical depolarizing noise to data qubits
import numpy as np
import scipy.sparse as sp
import math
from typing import Tuple, List, Dict

class QuantumLDPC:
    """
    Constructs a CSS Quantum Low-Density Parity-Check (qLDPC) code 
    using the Hypergraph Product (HGP) construction from classical seed codes.
    """
    def __init__(self, H1: np.ndarray, H2: np.ndarray):
        self.H1 = np.array(H1, dtype=int)
        self.H2 = np.array(H2, dtype=int)
        
        r1, n1 = self.H1.shape
        r2, n2 = self.H2.shape
        
        self.num_data_qubits = n1 * n2 + r1 * r2
        self.num_X_checks = r1 * n2
        self.num_Z_checks = n1 * r2
        
        I_n1 = np.eye(n1, dtype=int)
        I_n2 = np.eye(n2, dtype=int)
        I_r1 = np.eye(r1, dtype=int)
        I_r2 = np.eye(r2, dtype=int)
        
        # CSS Parity Check Matrices: HX and HZ
        self.HX = np.hstack([np.kron(self.H1, I_n2), np.kron(I_r1, self.H2.T)]) % 2
        self.HZ = np.hstack([np.kron(I_n1, self.H2), np.kron(self.H1.T, I_r2)]) % 2
        
        # Verify CSS Commutation Relation: HX @ HZ^T = 0 (mod 2)
        commutation = (self.HX @ self.HZ.T) % 2
        if not np.all(commutation == 0):
            raise ValueError("CSS Commutation relation violated: HX @ HZ^T != 0 mod 2")

    def schedule_parallel_measurements(self) -> Tuple[List[List[int]], List[List[int]]]:
        """
        Schedules X and Z checks into non-conflicting parallel measurement layers
        using a greedy check-conflict graph coloring strategy.
        """
        x_layers = self._color_conflict_graph(self.HX)
        z_layers = self._color_conflict_graph(self.HZ)
        return x_layers, z_layers

    def _color_conflict_graph(self, H: np.ndarray) -> List[List[int]]:
        """
        Builds the check conflict graph for H and partitions checks into parallel layers.
        Two checks conflict if they share a physical data qubit.
        """
        num_checks, num_qubits = H.shape
        layers: List[List[int]] = []
        
        for check_idx in range(num_checks):
            qubits_in_check = set(np.where(H[check_idx] == 1)[0])
            placed = False
            
            for layer in layers:
                layer_qubits = set()
                for c in layer:
                    layer_qubits.update(np.where(H[c] == 1)[0])
                
                # If no overlap with current layer's qubits, add check to layer
                if len(qubits_in_check.intersection(layer_qubits)) == 0:
                    layer.append(check_idx)
                    placed = True
                    break
            
            if not placed:
                layers.append([check_idx])
                
        return layers


class ParallelPPMSimulator:
    """
    Monte Carlo Simulation Engine for Parallel Pauli Product Measurements
    under depolarizing and readout noise channels.
    """
    def __init__(self, qldpc: QuantumLDPC, p_depol: float, p_meas: float, seed: int = 42):
        self.code = qldpc
        self.p_depol = p_depol
        self.p_meas = p_meas
        self.rng = np.random.default_rng(seed)

    def extract_spatiotemporal_syndromes(self, T_rounds: int) -> Dict[str, np.ndarray]:
        """
        Simulates T rounds of parallel syndrome extraction.
        
        Returns a dictionary containing:
          - 'diff_X': Spatiotemporal syndrome difference for X-checks
          - 'diff_Z': Spatiotemporal syndrome difference for Z-checks
          - 'final_err_X': Final accumulated physical X errors on data qubits
          - 'final_err_Z': Final accumulated physical Z errors on data qubits
        """
        n = self.code.num_data_qubits
        mX = self.code.num_X_checks
        mZ = self.code.num_Z_checks
        
        # Cumulative physical Pauli errors on data qubits (0 or 1)
        data_err_X = np.zeros(n, dtype=int)
        data_err_Z = np.zeros(n, dtype=int)
        
        raw_syndromes_X = np.zeros((T_rounds, mX), dtype=int)
        raw_syndromes_Z = np.zeros((T_rounds, mZ), dtype=int)

        x_layers, z_layers = self.code.schedule_parallel_measurements()

        for t in range(T_rounds):
            # 1. Apply physical depolarizing noise to data qubits
            # P(X error) = p_depol, P(Z error) = p_depol
            rx = (self.rng.random(n) < self.p_depol).astype(int)
            rz = (self.rng.random(n) < self.p_depol).astype(int)
            
            data_err_X = (data_err_X + rx) % 2
            data_err_Z = (data_err_Z + rz) % 2

            # 2. Extract X-checks in parallel layers (detects Z data errors)
            ideal_synd_X = (self.code.HX @ data_err_Z) % 2
            meas_noise_X = (self.rng.random(mX) < self.p_meas).astype(int)
            raw_syndromes_X[t] = (ideal_synd_X + meas_noise_X) % 2

            # 3. Extract Z-checks in parallel layers (detects X data errors)
            ideal_synd_Z = (self.code.HZ @ data_err_X) % 2
            meas_noise_Z = (self.rng.random(mZ) < self.p_meas).astype(int)
            raw_syndromes_Z[t] = (ideal_synd_Z + meas_noise_Z) % 2

        # Compute Spatiotemporal Syndrome Differences: Delta s_t = s_t XOR s_{t-1}
        diff_X = np.zeros_like(raw_syndromes_X)
        diff_Z = np.zeros_like(raw_syndromes_Z)
        
        diff_X[0] = raw_syndromes_X[0]
        diff_Z[0] = raw_syndromes_Z[0]
        
        for t in range(1, T_rounds):
            diff_X[t] = (raw_syndromes_X[t] - raw_syndromes_X[t-1]) % 2
            diff_Z[t] = (raw_syndromes_Z[t] - raw_syndromes_Z[t-1]) % 2

        return {
            'diff_X': diff_X,
            'diff_Z': diff_Z,
            'final_err_X': data_err_X,
            'final_err_Z': data_err_Z
        }


def min_sum_bp_decoder(H: np.ndarray, syndrome: np.ndarray, p_error: float, max_iter: int = 30, alpha: float = 0.8) -> np.ndarray:
    """
    Attenuated Min-Sum Belief Propagation Decoder for binary Tanner graphs.
    
    Parameters:
      - H: Check matrix (m x n)
      - syndrome: Binary syndrome vector (m,)
      - p_error: Prior error probability on channel
      - max_iter: Maximum BP iterations
      - alpha: Min-Sum attenuation factor (scaling factor)
    """
    m, n = H.shape
    llr_prior = math.log((1.0 - p_error) / max(p_error, 1e-12))
    
    # Initialize variable-to-check messages
    var_to_check = np.zeros((m, n))
    for i in range(m):
        for j in range(n):
            if H[i, j] == 1:
                var_to_check[i, j] = llr_prior

    check_to_var = np.zeros((m, n))

    for iteration in range(max_iter):
        # 1. Update Check-to-Variable messages
        for i in range(m):
            var_indices = np.where(H[i] == 1)[0]
            for j in var_indices:
                other_vars = [v for v in var_indices if v != j]
                if not other_vars:
                    check_to_var[i, j] = 0.0
                    continue
                
                sign_prod = (-1.0) ** syndrome[i]
                min_magnitude = float('inf')
                
                for v in other_vars:
                    val = var_to_check[i, v]
                    sign_prod *= np.sign(val) if val != 0 else 1.0
                    min_magnitude = min(min_magnitude, abs(val))
                
                # Apply attenuation factor alpha
                check_to_var[i, j] = alpha * sign_prod * min_magnitude

        # 2. Marginalization and Hard Decision
        total_llr = np.full(n, llr_prior)
        for j in range(n):
            check_indices = np.where(H[:, j] == 1)[0]
            total_llr[j] += np.sum(check_to_var[check_indices, j])
        
        decoded_error = (total_llr < 0).astype(int)
        
        # Check syndrome convergence
        syn_check = (H @ decoded_error) % 2
        if np.array_equal(syn_check, syndrome):
            return decoded_error

        # 3. Update Variable-to-Check messages
        for j in range(n):
            check_indices = np.where(H[:, j] == 1)[0]
            for i in check_indices:
                other_checks = [c for c in check_indices if c != i]
                var_to_check[i, j] = llr_prior + np.sum(check_to_var[other_checks, j])

    return (total_llr < 0).astype(int)


# Execution Pipeline & Benchmark
if __name__ == "__main__":
    print("==================================================================")
    print(" Parallel Pauli Product Measurement (PPM) Simulator on qLDPC Code")
    print("==================================================================")
    
    # Define classical seed codes: H1 and H2 as Hamming-like / Repetition parity checks
    H_seed1 = np.array([
        [1, 1, 0, 1, 0, 0],
        [0, 1, 1, 0, 1, 0],
        [0, 0, 1, 1, 0, 1]
    ], dtype=int)
    
    H_seed2 = np.array([
        [1, 0, 1, 1],
        [0, 1, 1, 0]
    ], dtype=int)

    # Initialize Quantum LDPC Code via HGP Construction
    code = QuantumLDPC(H_seed1, H_seed2)
    
    print(f"\n[Code Structure Summary]")
    print(f" Physical Qubits (n) : {code.num_data_qubits}")
    print(f" X-Checks (mX)       : {code.num_X_checks} (Row weight: {np.sum(code.HX[0])})")
    print(f" Z-Checks (mZ)       : {code.num_Z_checks} (Row weight: {np.sum(code.HZ[0])})")
    
    # Benchmark Parallel Scheduling
    x_layers, z_layers = code.schedule_parallel_measurements()
    print(f"\n[Parallel PPM Scheduling]")
    print(f" Parallel X-Measurement Layers Required : {len(x_layers)}")
    print(f" Parallel Z-Measurement Layers Required : {len(z_layers)}")
    print(f" Checks per X-layer assignment          : {[len(l) for l in x_layers]}")
    
    # Run Monte Carlo Fault-Tolerant Simulation
    p_noise = 0.015
    p_readout = 0.01
    T_rounds = 5
    
    sim = ParallelPPMSimulator(code, p_depol=p_noise, p_meas=p_readout, seed=123)
    results = sim.extract_spatiotemporal_syndromes(T_rounds=T_rounds)
    
    print(f"\n[Spatiotemporal Simulation Results ({T_rounds} Rounds)]")
    print(f" X Syndrome Difference Matrix Shape : {results['diff_X'].shape}")
    print(f" Z Syndrome Difference Matrix Shape : {results['diff_Z'].shape}")
    print(f" Physical X Errors Accumulated      : {np.sum(results['final_err_X'])}")
    print(f" Physical Z Errors Accumulated      : {np.sum(results['final_err_Z'])}")
    
    # Test Min-Sum Decoder on Round 0 Syndrome
    test_synd_Z = (code.HZ @ results['final_err_X']) % 2
    decoded_X_err = min_sum_bp_decoder(code.HZ, test_synd_Z, p_error=p_noise)
    residual_synd = (code.HZ @ decoded_X_err) % 2
    
    print(f"\n[Decoder Verification]")
    print(f" Original Syndrome Weight : {np.sum(test_synd_Z)}")
    print(f" Residual Syndrome Weight : {np.sum((test_synd_Z - residual_synd) % 2)}")
    print(f" Decoder Converged        : {np.array_equal(test_synd_Z, residual_synd)}")
#Quantum Error Correction #qLDPC Codes #Quantum Simulation
quantum_snippet.py
"""
Numerical Simulation of Parallel Pauli Product Measurements on qLDPC Codes.
Constructs Hypergraph Product (HGP) codes and evaluates parallel syndrome
extraction under physical qubit and readout phenomenological noise models.
"""

import numpy as np
import scipy.sparse as sp
from typing import Dict, Tuple

class HypergraphProductCode:
    """
    Constructs a Quantum Low-Density Parity-Check (qLDPC) CSS code from two
    classical binary parity-check matrices H1 and H2.
    """
    def __init__(self, H1: np.ndarray, H2: np.ndarray):
        self.H1 = sp.csr_matrix(H1, dtype=np.int8)
        self.H2 = sp.csr_matrix(H2, dtype=np.int8)
        self._build_hgp_matrices()

    def _build_hgp_matrices(self) -> None:
        r1, n1 = self.H1.shape
        r2, n2 = self.H2.shape

        I_n1 = sp.eye(n1, dtype=np.int8, format='csr')
        I_n2 = sp.eye(n2, dtype=np.int8, format='csr')
        I_r1 = sp.eye(r1, dtype=np.int8, format='csr')
        I_r2 = sp.eye(r2, dtype=np.int8, format='csr')

        # Hx = [H1 ⊗ I_n2,  I_r1 ⊗ H2^T]
        hx_left = sp.kron(self.H1, I_n2, format='csr')
        hx_right = sp.kron(I_r1, self.H2.T, format='csr')
        self.Hx = sp.hstack([hx_left, hx_right], format='csr')

        # Hz = [I_n1 ⊗ H2,  H1^T ⊗ I_r2]
        hz_left = sp.kron(I_n1, self.H2, format='csr')
        hz_right = sp.kron(self.H1.T, I_r2, format='csr')
        self.Hz = sp.hstack([hz_left, hz_right], format='csr')

        self.num_qubits = self.Hx.shape[1]
        self.num_x_checks = self.Hx.shape[0]
        self.num_z_checks = self.Hz.shape[0]

    def verify_css_condition(self) -> bool:
        """
        Validates the orthogonality constraint Hx @ Hz^T == 0 (mod 2).
        """
        comm_matrix = (self.Hx @ self.Hz.T).tocoo()
        non_zero_mod2 = np.sum(comm_matrix.data % 2 != 0)
        return int(non_zero_mod2) == 0

    def get_sparsity_metrics(self) -> Dict[str, float]:
        """Calculates row/column weights for sparse check matrices."""
        hx_row_w = np.mean(self.Hx.sum(axis=1))
        hz_row_w = np.mean(self.Hz.sum(axis=1))
        return {
            "avg_x_check_degree": float(hx_row_w),
            "avg_z_check_degree": float(hz_row_w),
            "total_qubits": self.num_qubits,
            "total_checks": self.num_x_checks + self.num_z_checks
        }


class ParallelPauliMeasurementSimulator:
    """
    Simulates phenomenological noise injection and parallel Pauli stabilizer
    measurements across a qLDPC code grid.
    """
    def __init__(self, code: HypergraphProductCode):
        self.code = code

    def inject_depolarizing_noise(self, p_error: float) -> Tuple[np.ndarray, np.ndarray]:
        """
        Generates independent Pauli X and Z error vectors over physical qubits.
        """
        n = self.code.num_qubits
        e_x = (np.random.rand(n) < p_error).astype(np.int8)
        e_z = (np.random.rand(n) < p_error).astype(np.int8)
        return e_x, e_z

    def execute_measurement_cycle(
        self, e_x: np.ndarray, e_z: np.ndarray, p_readout: float
    ) -> Dict[str, np.ndarray]:
        """
        Performs parallel syndrome measurement with optional readout noise.
        
        s_z = Hx @ e_z (mod 2) -> detects X errors
        s_x = Hz @ e_x (mod 2) -> detects Z errors
        """
        # Exact linear algebra over GF(2) via CSR sparse matrices
        ideal_s_z = (self.code.Hx @ e_z) % 2
        ideal_s_x = (self.code.Hz @ e_x) % 2

        # Inject measurement readout errors
        readout_noise_z = (np.random.rand(len(ideal_s_z)) < p_readout).astype(np.int8)
        readout_noise_x = (np.random.rand(len(ideal_s_x)) < p_readout).astype(np.int8)

        observed_s_z = (ideal_s_z + readout_noise_z) % 2
        observed_s_x = (ideal_s_x + readout_noise_x) % 2

        return {
            "s_z": observed_s_z,
            "s_x": observed_s_x,
            "raw_s_z": ideal_s_z,
            "raw_s_x": ideal_s_x
        }


def run_monte_carlo_trial(
    p_error: float, p_readout: float, trials: int = 500
) -> Dict[str, float]:
    """Runs Monte Carlo simulations over a classical Hamming-derived HGP qLDPC code."""
    # Classical (7,4) Hamming parity-check matrix
    H_hamming = np.array([
        [1, 0, 1, 0, 1, 0, 1],
        [0, 1, 1, 0, 0, 1, 1],
        [0, 0, 0, 1, 1, 1, 1]
    ], dtype=np.int8)

    code = HypergraphProductCode(H_hamming, H_hamming)
    assert code.verify_css_condition(), "CSS condition failed!"

    sim = ParallelPauliMeasurementSimulator(code)

    z_syndrome_triggers = 0
    x_syndrome_triggers = 0

    for _ in range(trials):
        e_x, e_z = sim.inject_depolarizing_noise(p_error)
        results = sim.execute_measurement_cycle(e_x, e_z, p_readout)
        z_syndrome_triggers += np.sum(results["s_z"])
        x_syndrome_triggers += np.sum(results["s_x"])

    total_z_measurements = trials * code.num_x_checks
    total_x_measurements = trials * code.num_z_checks

    return {
        "z_defect_rate": z_syndrome_triggers / total_z_measurements,
        "x_defect_rate": x_syndrome_triggers / total_x_measurements,
        "code_qubits": code.num_qubits,
    }


if __name__ == "__main__":
    np.random.seed(42)
    p_physical = 0.015
    p_measurement = 0.005

    results = run_monte_carlo_trial(p_physical, p_measurement, trials=1000)
    print("--- SIMULATION RESULTS ---")
    print(f"Physical Qubits (N): {results['code_qubits']}")
    print(f"Physical Error Rate (p_err): {p_physical}")
    print(f"Readout Error Rate (p_read): {p_measurement}")
    print(f"Z-Syndrome Defect Density: {results['z_defect_rate']:.4f}")
    print(f"X-Syndrome Defect Density: {results['x_defect_rate']:.4f}")
#Quantum Computing #Quantum Simulation #Open Quantum Systems
1. Define topology: 6-node ring graph (Cycle Graph C_6)
"""
Continuous-Time Quantum Information Processing Simulator
Author: Senior Quantum Computing Engineer
Description: High-precision simulation of Continuous-Time Quantum Walks (CTQW)
             and Open-System Lindblad Dynamics using standard numerical methods.
"""

from typing import List, Tuple
import numpy as np


class ContinuousTimeQuantumSimulator:
    """Simulates unitary continuous-time quantum walks and non-unitary open quantum dynamics."""

    def __init__(self, adj_matrix: np.ndarray, hopping_rate: float = 1.0):
        """Initialize simulator with a graph adjacency matrix.

        Args:
            adj_matrix: Square, symmetric adjacency matrix representing the graph topology.
            hopping_rate: Tunneling amplitude gamma between connected nodes.
        """
        if adj_matrix.shape[0] != adj_matrix.shape[1]:
            raise ValueError("Adjacency matrix must be square.")
        if not np.allclose(adj_matrix, adj_matrix.T):
            raise ValueError("Adjacency matrix must be symmetric for undirected graphs.")

        self.num_nodes = adj_matrix.shape[0]
        self.gamma = hopping_rate
        self.adj_matrix = adj_matrix.astype(complex)
        
        # Construct graph Hamiltonian: H = -gamma * A
        self.H = -self.gamma * self.adj_matrix
        
        # Precompute eigensystem for exact unitary evolution
        self._evals, self._evecs = np.linalg.eigh(self.H)

    def simulate_unitary_ctqw(
        self, initial_node: int, total_time: float, num_steps: int
    ) -> Tuple[np.ndarray, np.ndarray]:
        """Simulates closed-system continuous-time quantum walk evolution.

        Args:
            initial_node: Node index where the quantum state is initially localized.
            total_time: Total integration time duration.
            num_steps: Number of discrete time sampling points.

        Returns:
            Tuple of (time_points, probability_distributions_over_time)
        """
        times = np.linspace(0.0, total_time, num_steps)
        psi0 = np.zeros(self.num_nodes, dtype=complex)
        psi0[initial_node] = 1.0 + 0.0j

        # Express initial state in Hamiltonian eigenbasis: |psi_0_diag> = V^\dagger |psi_0>
        psi0_diag = self._evecs.conj().T @ psi0
        probabilities = np.zeros((num_steps, self.num_nodes), dtype=float)

        for idx, t in enumerate(times):
            # Time-evolution diagonal propagator: exp(-i * E_k * t)
            phase_factors = np.exp(-1j * self._evals * t)
            psi_t_diag = phase_factors * psi0_diag
            # Transform back to spatial basis: |psi(t)> = V |psi_t_diag>
            psi_t = self._evecs @ psi_t_diag
            probabilities[idx, :] = np.abs(psi_t) ** 2

        return times, probabilities

    def simulate_lindblad_dynamics(
        self,
        initial_density_matrix: np.ndarray,
        jump_operators: List[np.ndarray],
        total_time: float,
        num_steps: int
    ) -> Tuple[np.ndarray, np.ndarray]:
        """Simulates open quantum system dynamics governed by the Lindblad master equation

        using explicit 4th-order Runge-Kutta (RK4) integration.

        Args:
            initial_density_matrix: Initial state density operator (N x N matrix).
            jump_operators: List of collapse operators L_k representing noise channels.
            total_time: Total simulation duration.
            num_steps: Number of time integration steps.

        Returns:
            Tuple of (time_points, density_matrices_over_time)
        """
        times = np.linspace(0.0, total_time, num_steps)
        dt = total_time / (num_steps - 1)
        
        # Precompute L^\dagger @ L for each dissipator
        L_dag_L = [L.conj().T @ L for L in jump_operators]

        def _lindblad_rhs(rho: np.ndarray) -> np.ndarray:
            """Evaluates d(rho)/dt = -i [H, rho] + sum_k D[L_k](rho)."""
            # Unitary term: -i (H rho - rho H)
            drho = -1j * (self.H @ rho - rho @ self.H)
            
            # Non-unitary dissipators
            for L, LdL in zip(jump_operators, L_dag_L):
                drho += L @ rho @ L.conj().T - 0.5 * (LdL @ rho + rho @ LdL)
            return drho

        density_matrices = np.zeros((num_steps, self.num_nodes, self.num_nodes), dtype=complex)
        rho_current = initial_density_matrix.copy().astype(complex)
        density_matrices[0] = rho_current.copy()

        for step in range(1, num_steps):
            # Classic RK4 Integration
            k1 = _lindblad_rhs(rho_current)
            k2 = _lindblad_rhs(rho_current + 0.5 * dt * k1)
            k3 = _lindblad_rhs(rho_current + 0.5 * dt * k2)
            k4 = _lindblad_rhs(rho_current + dt * k3)

            rho_current += (dt / 6.0) * (k1 + 2.0 * k2 + 2.0 * k3 + k4)
            
            # Numerical sanitization: enforce Hermiticity and unit trace
            rho_current = 0.5 * (rho_current + rho_current.conj().T)
            trace_val = np.trace(rho_current)
            if not np.isclose(trace_val, 0.0):
                rho_current /= trace_val

            density_matrices[step] = rho_current.copy()

        return times, density_matrices


# =====================================================================
# Verification and Demonstration Execution
# =====================================================================
if __name__ == "__main__":
    # 1. Define topology: 6-node ring graph (Cycle Graph C_6)
    num_nodes = 6
    adj_matrix = np.zeros((num_nodes, num_nodes))
    for i in range(num_nodes):
        adj_matrix[i, (i + 1) % num_nodes] = 1.0
        adj_matrix[i, (i - 1) % num_nodes] = 1.0

    sim = ContinuousTimeQuantumSimulator(adj_matrix=adj_matrix, hopping_rate=1.0)

    # 2. Simulate closed-system Unitary CTQW starting at Node 0
    t_unitary, p_unitary = sim.simulate_unitary_ctqw(initial_node=0, total_time=4.0, num_steps=5)
    
    print("=== Ideal CTQW Population Profiles (Unitary) ===")
    for idx, t in enumerate(t_unitary):
        probs_formatted = [f"{p:.4f}" for p in p_unitary[idx]]
        print(f"Time t={t:.2f}s | Probabilities: {probs_formatted} | Sum: {np.sum(p_unitary[idx]):.6f}")

    # 3. Simulate Open-System Lindblad Dynamics with Dephasing on Node 2
    gamma_dephasing = 0.3
    L_dephasing = np.zeros((num_nodes, num_nodes), dtype=complex)
    L_dephasing[2, 2] = np.sqrt(gamma_dephasing)  # Localized pure dephasing noise on Node 2

    # Initial density state: Pure state |0><0|
    psi0 = np.zeros(num_nodes, dtype=complex)
    psi0[0] = 1.0
    rho0 = np.outer(psi0, psi0.conj())

    t_open, rhos_open = sim.simulate_lindblad_dynamics(
        initial_density_matrix=rho0,
        jump_operators=[L_dephasing],
        total_time=4.0,
        num_steps=5
    )

    print("\n=== Open System CTQW Node Populations (with Local Dephasing) ===")
    for idx, t in enumerate(t_open):
        populations = np.real(np.diag(rhos_open[idx]))
        pops_formatted = [f"{p:.4f}" for p in populations]
        trace_val = np.real(np.trace(rhos_open[idx]))
        print(f"Time t={t:.2f}s | Populations: {pops_formatted} | Trace: {trace_val:.6f}")
#Quantum Computing #Qiskit #Quantum Information
Target IBM basis gate set (e.g., EagleHeron architecture)
"""
qiskit_ghz_pipeline.py
======================
Production-ready demonstration of state preparation, quantum state tomography,
transpilation optimization, and shot-based simulation using Qiskit 1.x standards.
"""

import numpy as np
from qiskit import QuantumCircuit, transpile
from qiskit.quantum_info import Statevector, DensityMatrix, state_fidelity
from qiskit.transpiler import CouplingMap
from qiskit_aer import AerSimulator
from qiskit_aer.noise import NoiseModel, depolarizing_error


def build_ghz_circuit(num_qubits: int) -> QuantumCircuit:
    """
    Constructs an n-qubit GHZ state preparation circuit.
    
    |000...0> -> (|000...0> + |111...1>) / sqrt(2)
    """
    if num_qubits < 2:
        raise ValueError("GHZ state requires at least 2 qubits.")
        
    qc = QuantumCircuit(num_qubits, num_qubits, name="GHZ_State")
    
    # Apply Hadamard to first qubit to enter superposition
    qc.h(0)
    
    # Entangle remaining qubits via CNOT cascade
    for q in range(num_qubits - 1):
        qc.cx(q, q + 1)
        
    qc.barrier()
    # Add measurement gates for execution phase
    qc.measure(range(num_qubits), range(num_qubits))
    
    return qc


def evaluate_ideal_state(qc_unmeasured: QuantumCircuit) -> DensityMatrix:
    """
    Computes theoretical statevector and density matrix prior to measurement.
    """
    # Extract statevector from unmeasured circuit
    sv = Statevector.from_instruction(qc_unmeasured)
    rho = DensityMatrix(sv)
    
    print(f"[+] Statevector Dimension: {sv.dim}")
    print(f"[+] Pure State Check: {rho.is_valid()}")
    print(f"[+] Density Matrix Purity: {np.real(rho.purity()):.4f}")
    
    return rho


def transpile_for_hardware(qc: QuantumCircuit, coupling_graph: list[list[int]]) -> QuantumCircuit:
    """
    Transpiles logical circuit to fit physical hardware topology constraints.
    """
    cmap = CouplingMap(couplinglist=coupling_graph)
    
    # Target IBM basis gate set (e.g., Eagle/Heron architecture)
    basis_gates = ['ecr', 'id', 'rz', 'x', 'sx']
    
    transpiled_qc = transpile(
        qc,
        coupling_map=cmap,
        basis_gates=basis_gates,
        optimization_level=3,
        seed_transpiler=42
    )
    
    print(f"[+] Original Circuit Depth: {qc.depth()}")
    print(f"[+] Transpiled Circuit Depth: {transpiled_qc.depth()}")
    print(f"[+] Transpiled Gate Count: {transpiled_qc.count_ops()}")
    
    return transpiled_qc


def execute_noisy_simulation(qc: QuantumCircuit, shots: int = 4096) -> dict[str, int]:
    """
    Executes circuit using AerSimulator with an artificial 1-qubit and 2-qubit noise model.
    """
    # Construct synthetic noise model
    noise_model = NoiseModel()
    p1_error = depolarizing_error(0.001, 1) # 0.1% single-qubit gate error
    p2_error = depolarizing_error(0.015, 2) # 1.5% two-qubit gate error
    
    noise_model.add_all_qubit_quantum_error(p1_error, ['x', 'sx', 'rz'])
    noise_model.add_all_qubit_quantum_error(p2_error, ['ecr', 'cx'])
    
    simulator = AerSimulator(noise_model=noise_model)
    
    # Run simulation
    job = simulator.run(qc, shots=shots)
    result = job.result()
    counts = result.get_counts(qc)
    
    return counts


def main():
    NUM_QUBITS = 3
    SHOTS = 8192
    
    print("==================================================")
    print(" QISKIT 1.x GHZ STATE PIPELINE ")
    print("==================================================")
    
    # 1. Build circuit without measurements for quantum_info analysis
    qc_base = QuantumCircuit(NUM_QUBITS)
    qc_base.h(0)
    for q in range(NUM_QUBITS - 1):
        qc_base.cx(q, q + 1)
        
    # 2. Mathematical validation
    rho_ideal = evaluate_ideal_state(qc_base)
    
    # 3. Build measurement circuit
    qc_full = build_ghz_circuit(NUM_QUBITS)
    
    # 4. Transpilation targeting linear physical topology: 0 -- 1 -- 2
    linear_topology = [[0, 1], [1, 0], [1, 2], [2, 1]]
    transpiled_qc = transpile_for_hardware(qc_full, linear_topology)
    
    # 5. Execution
    counts = execute_noisy_simulation(transpiled_qc, shots=SHOTS)
    
    print("\n[+] Execution Counts (Noisy Backend):")
    for state in sorted(counts.keys()):
        prob = counts[state] / SHOTS
        print(f"    State |{state}>: {counts[state]:5d} shots ({prob*100:5.2f}%)")
        
    # Calculate fidelity approximation against ideal |000> and |111> distribution
    ideal_shots = counts.get("000", 0) + counts.get("111", 0)
    raw_fidelity = ideal_shots / SHOTS
    print(f"\n[+] Raw GHZ Target State Population: {raw_fidelity*100:.2f}%")


if __name__ == "__main__":
    main()
#Quantum Error Correction #Quantum LDPC Codes #Fault-Tolerant Quantum Computing
quantum_snippet.py
import numpy as np
from typing import Tuple, Dict, List

class QLDPCCode:
    """
    Constructs a CSS Quantum LDPC code using the Hypergraph Product (HGP)
    from a classical parity-check matrix H_classical.
    """
    def __init__(self, H_classical: np.ndarray):
        self.H_class = (H_classical.copy() % 2).astype(np.uint8)
        self.r, self.m = self.H_class.shape
        self.n = self.m**2 + self.r**2
        
        # Build Hx and Hz matrices via Kronecker products
        I_m = np.eye(self.m, dtype=np.uint8)
        I_r = np.eye(self.r, dtype=np.uint8)
        
        # Hx = [H \otimes I_m,  I_r \otimes H^T]
        # Hz = [I_m \otimes H,  H^T \otimes I_r]
        self.Hx = np.hstack([
            np.kron(self.H_class, I_m),
            np.kron(I_r, self.H_class.T)
        ]) % 2
        
        self.Hz = np.hstack([
            np.kron(I_m, self.H_class),
            np.kron(self.H_class.T, I_r)
        ]) % 2
        
        self.m_x, _ = self.Hx.shape
        self.m_z, _ = self.Hz.shape
        self._verify_commutativity()

    def _verify_commutativity(self) -> None:
        """Verify Hx @ Hz.T == 0 over GF(2)."""
        comm = (self.Hx @ self.Hz.T) % 2
        if np.any(comm != 0):
            raise ValueError("Commutativity condition failed: Hx @ Hz^T != 0")

    @staticmethod
    def gf2_rank(matrix: np.ndarray) -> int:
        """Computes the rank of a binary matrix over GF(2)."""
        A = matrix.copy() % 2
        rows, cols = A.shape
        rank = 0
        for col in range(cols):
            pivot = None
            for row in range(rank, rows):
                if A[row, col] == 1:
                    pivot = row
                    break
            if pivot is None:
                continue
            A[[rank, pivot]] = A[[pivot, rank]]
            for row in range(rows):
                if row != rank and A[row, col] == 1:
                    A[row] = (A[row] + A[rank]) % 2
            rank += 1
        return rank

    def get_code_parameters(self) -> Tuple[int, int]:
        """Returns physical qubits (n) and logical qubits (k)."""
        rank_x = self.gf2_rank(self.Hx)
        rank_z = self.gf2_rank(self.Hz)
        k = self.n - rank_x - rank_z
        return self.n, k

    def generate_parallel_schedule(self) -> List[List[Tuple[str, int]]]:
        """
        Greedy edge-coloring to create parallel measurement rounds.
        Ensures non-overlapping qubit support per clock cycle.
        """
        checks = []
        for i in range(self.m_x):
            support = set(np.where(self.Hx[i] == 1)[0])
            checks.append(('X', i, support))
        for j in range(self.m_z):
            support = set(np.where(self.Hz[j] == 1)[0])
            checks.append(('Z', j, support))
            
        rounds: List[List[Tuple[str, int]]] = []
        round_occupied_qubits: List[set] = []
        
        for ctype, idx, supp in checks:
            placed = False
            for r_idx, occupied in enumerate(round_occupied_qubits):
                if occupied.isdisjoint(supp):
                    rounds[r_idx].append((ctype, idx))
                    occupied.update(supp)
                    placed = True
                    break
            if not placed:
                rounds.append([(ctype, idx)])
                round_occupied_qubits.append(set(supp))
                
        return rounds


class ParallelPPMSimulator:
    """
    Simulates Monte Carlo noise and parallel syndrome extraction
    on a given qLDPC code structure.
    """
    def __init__(self, code: QLDPCCode):
        self.code = code

    def run_syndrome_extraction(
        self, 
        p_data_error: float, 
        p_meas_error: float, 
        num_shots: int = 10000,
        seed: int = 42
    ) -> Dict[str, float]:
        """
        Executes parallel syndrome extraction across Monte Carlo shots.
        """
        rng = np.random.default_rng(seed)
        n = self.code.n
        m_x = self.code.m_x
        m_z = self.code.m_z
        
        trivial_syndromes = 0
        total_x_syndrome_flips = 0
        total_z_syndrome_flips = 0
        
        for _ in range(num_shots):
            # Physical Pauli errors on data qubits
            e_x = (rng.random(n) < p_data_error).astype(np.uint8)
            e_z = (rng.random(n) < p_data_error).astype(np.uint8)
            
            # Exact syndromes: Z errors detected by Hx, X errors detected by Hz
            raw_s_z = (self.code.Hx @ e_z) % 2
            raw_s_x = (self.code.Hz @ e_x) % 2
            
            # Parallel readout noise on syndrome bits
            y_z = (rng.random(m_x) < p_meas_error).astype(np.uint8)
            y_x = (rng.random(m_x if m_x == m_z else m_z) < p_meas_error).astype(np.uint8)
            
            s_z = (raw_s_z + y_z) % 2
            s_x = (raw_s_x + y_x) % 2
            
            num_flips = np.sum(s_z) + np.sum(s_x)
            total_z_syndrome_flips += np.sum(s_z)
            total_x_syndrome_flips += np.sum(s_x)
            
            if num_flips == 0:
                trivial_syndromes += 1

        return {
            "zero_syndrome_rate": trivial_syndromes / num_shots,
            "avg_z_syndrome_density": total_z_syndrome_flips / (num_shots * m_x),
            "avg_x_syndrome_density": total_x_syndrome_flips / (num_shots * m_z)
        }


# --- Example Execution ---
if __name__ == "__main__":
    # Classical seed matrix for a [7, 4, 3] Hamming code
    H_seed = np.array([
        [1, 1, 0, 1, 0, 0, 0],
        [0, 1, 1, 0, 1, 0, 0],
        [0, 0, 1, 1, 0, 1, 0]
    ], dtype=np.uint8)

    qldpc_code = QLDPCCode(H_seed)
    n_qubits, k_logical = qldpc_code.get_code_parameters()
    schedule = qldpc_code.generate_parallel_schedule()

    print(f"==================================================")
    print(f"qLDPC Code Parameterization (Hypergraph Product)")
    print(f"==================================================")
    print(f"Physical Qubits (n) : {n_qubits}")
    print(f"Logical Qubits (k)  : {k_logical}")
    print(f"Hx matrix shape     : {qldpc_code.Hx.shape}")
    print(f"Hz matrix shape     : {qldpc_code.Hz.shape}")
    print(f"Parallel Depth      : {len(schedule)} measurement rounds")
    
    sim = ParallelPPMSimulator(qldpc_code)
    results = sim.run_syndrome_extraction(
        p_data_error=0.005, 
        p_meas_error=0.005, 
        num_shots=5000
    )

    print(f"\n--- Phenomenological Simulation Results (p = 0.5%) ---")
    print(f"Zero Syndrome Rate    : {results['zero_syndrome_rate']:.4f}")
    print(f"Avg Z-Syndrome Density: {results['avg_z_syndrome_density']:.4f}")
    print(f"Avg X-Syndrome Density: {results['avg_x_syndrome_density']:.4f}")
#Quantum Computing #qLDPC Codes #Fault-Tolerant Quantum Computing
Commutation relation sign flip: (-1)^(err_x . obs_z + err_z . obs_x) mod 2
"""
qLDPC Parallel Pauli Product Measurement Simulator
===================================================
Models Hypergraph Product (HGP) code structures, constructs commuting
Pauli product measurement observables, and executes multi-round syndrome
extraction simulations under phenomenological noise.

Dependencies: numpy, scipy
"""

import numpy as np
import scipy.sparse as sp
from typing import Tuple, List, Dict, Any


def build_hgp_code(h1: sp.csr_matrix, h2: sp.csr_matrix) -> Tuple[sp.csr_matrix, sp.csr_matrix]:
    """
    Constructs Hypergraph Product (HGP) parity check matrices H_X and H_Z
    from two classical linear binary parity check matrices h1 and h2.
    
    Args:
        h1: Classical parity check matrix (shape: r1 x n1)
        h2: Classical parity check matrix (shape: r2 x n2)
        
    Returns:
        Tuple (H_X, H_Z) as sparse CSR matrices in uint8 format.
    """
    r1, n1 = h1.shape
    r2, n2 = h2.shape

    I_n1 = sp.eye(n1, dtype=np.uint8, format='csr')
    I_n2 = sp.eye(n2, dtype=np.uint8, format='csr')
    I_r1 = sp.eye(r1, dtype=np.uint8, format='csr')
    I_r2 = sp.eye(r2, dtype=np.uint8, format='csr')

    # H_X = [h1 \otimes I_n2 , I_r1 \otimes h2^T]
    hx_left = sp.kron(h1, I_n2, format='csr')
    hx_right = sp.kron(I_r1, h2.T, format='csr')
    hx = sp.hstack([hx_left, hx_right], format='csr')

    # H_Z = [I_n1 \otimes h2 , h1^T \otimes I_r2]
    hz_left = sp.kron(I_n1, h2, format='csr')
    hz_right = sp.kron(h1.T, I_r2, format='csr')
    hz = sp.hstack([hz_left, hz_right], format='csr')

    return hx, hz


class ParallelPPMSimulator:
    """
    Numerical simulator for parallel Pauli Product Measurements (PPM)
    and space-time syndrome extraction on qLDPC codes.
    """

    def __init__(self, hx: sp.csr_matrix, hz: sp.csr_matrix):
        """
        Initializes the simulator with X and Z parity check matrices.
        """
        self.hx = hx
        self.hz = hz
        self.num_x_checks, self.num_qubits = hx.shape
        self.num_z_checks = hz.shape[0]

        # Verify quantum commutativity: H_X @ H_Z^T = 0 (mod 2)
        comm = (self.hx.dot(self.hz.T)).astype(np.uint8)
        comm.data %= 2
        comm.eliminate_zeros()
        assert comm.nnz == 0, "Invalid qLDPC parity matrices: H_X and H_Z do not commute!"

    def generate_pauli_error(self, p_error: float) -> Tuple[np.ndarray, np.ndarray]:
        """
        Generates independent Pauli X and Z errors across all data qubits.
        
        Args:
            p_error: Physical probability of a bit/phase flip error per qubit.
            
        Returns:
            Tuple (err_x, err_z) of binary vectors of length `num_qubits`.
        """
        err_x = (np.random.rand(self.num_qubits) < p_error).astype(np.uint8)
        err_z = (np.random.rand(self.num_qubits) < p_error).astype(np.uint8)
        return err_x, err_z

    def extract_syndrome_history(
        self, err_x: np.ndarray, err_z: np.ndarray, p_meas: float, rounds: int = 3
    ) -> Tuple[np.ndarray, np.ndarray]:
        """
        Simulates multi-round syndrome extraction with readout errors.
        
        Args:
            err_x: Binary vector representing physical X errors.
            err_z: Binary vector representing physical Z errors.
            p_meas: Readout flip probability for check measurements.
            rounds: Number of syndrome extraction rounds.
            
        Returns:
            Tuple (syn_z_history, syn_x_history) with dimensions (rounds, num_checks).
        """
        syn_z_hist = []
        syn_x_hist = []

        # Z-checks detect X-errors: s_z = H_Z * e_x (mod 2)
        # X-checks detect Z-errors: s_x = H_X * e_z (mod 2)
        exact_syn_z = (self.hz.dot(err_x)) % 2
        exact_syn_x = (self.hx.dot(err_z)) % 2

        for _ in range(rounds):
            readout_noise_z = (np.random.rand(self.num_z_checks) < p_meas).astype(np.uint8)
            readout_noise_x = (np.random.rand(self.num_x_checks) < p_meas).astype(np.uint8)

            syn_z_hist.append((exact_syn_z + readout_noise_z) % 2)
            syn_x_hist.append((exact_syn_x + readout_noise_x) % 2)

        return np.array(syn_z_hist), np.array(syn_x_hist)

    def simulate_parallel_ppm(
        self,
        observables: List[Tuple[np.ndarray, np.ndarray]],
        p_err: float,
        p_meas: float,
        rounds: int = 3,
        trials: int = 1000
    ) -> Dict[str, Any]:
        """
        Simulates simultaneous measurement of multiple commuting Pauli observables
        alongside multi-round syndrome extraction.
        
        Args:
            observables: List of tuples (obs_x, obs_z) defining target observables.
            p_err: Physical data qubit error rate.
            p_meas: Readout measurement error rate.
            rounds: Number of syndrome extraction rounds per trial.
            trials: Number of Monte Carlo simulation iterations.
            
        Returns:
            Dict containing clean syndrome fraction, observable outcome stats, and metadata.
        """
        clean_syndrome_count = 0
        all_measured_outcomes = []

        for _ in range(trials):
            err_x, err_z = self.generate_pauli_error(p_err)
            syn_z_hist, syn_x_hist = self.extract_syndrome_history(err_x, err_z, p_meas, rounds=rounds)

            # Evaluate joint Pauli observable outcomes
            trial_outcomes = []
            for obs_x, obs_z in observables:
                # Commutation relation sign flip: (-1)^(err_x . obs_z + err_z . obs_x) mod 2
                phase_flip = (np.dot(err_x, obs_z) + np.dot(err_z, obs_x)) % 2
                readout_err = 1 if np.random.rand() < p_meas else 0
                measured_val = (phase_flip + readout_err) % 2
                trial_outcomes.append(measured_val)

            all_measured_outcomes.append(trial_outcomes)

            # Check if final round returned zero syndrome defect
            if np.sum(syn_z_hist[-1]) + np.sum(syn_x_hist[-1]) == 0:
                clean_syndrome_count += 1

        clean_rate = clean_syndrome_count / trials
        return {
            "num_qubits": self.num_qubits,
            "x_checks": self.num_x_checks,
            "z_checks": self.num_z_checks,
            "clean_syndrome_rate": clean_rate,
            "sample_outcomes": all_measured_outcomes[:5]
        }


# --- Example Execution & Verification ---
if __name__ == "__main__":
    # Define a 7-bit Hamming code parity check matrix [7, 4, 3]
    H_hamming = sp.csr_matrix([
        [1, 1, 1, 0, 1, 0, 0],
        [0, 1, 1, 1, 0, 1, 0],
        [1, 0, 1, 1, 0, 0, 1]
    ], dtype=np.uint8)

    # Generate Hypergraph Product qLDPC Code: [[49, 16, 3]]
    Hx, Hz = build_hgp_code(H_hamming, H_hamming)
    simulator = ParallelPPMSimulator(Hx, Hz)

    print("==================================================")
    print("qLDPC Parallel PPM Numerical Simulation Initialized")
    print("==================================================")
    print(f"Total Data Qubits (N) : {simulator.num_qubits}")
    print(f"X Check Operators    : {simulator.num_x_checks}")
    print(f"Z Check Operators    : {simulator.num_z_checks}")

    # Define two parallel commuting Pauli product observables
    # Observable 1: Z-type Pauli product on Qubit partition 1
    obs1_x = np.zeros(simulator.num_qubits, dtype=np.uint8)
    obs1_z = np.zeros(simulator.num_qubits, dtype=np.uint8)
    obs1_z[:7] = 1

    # Observable 2: X-type Pauli product on Qubit partition 2
    obs2_x = np.zeros(simulator.num_qubits, dtype=np.uint8)
    obs2_x[7:14] = 1
    obs2_z = np.zeros(simulator.num_qubits, dtype=np.uint8)

    parallel_observables = [(obs1_x, obs1_z), (obs2_x, obs2_z)]

    # Execute simulation under phenomenological noise (p_err = 0.2%, p_meas = 0.2%)
    results = simulator.simulate_parallel_ppm(
        observables=parallel_observables,
        p_err=0.002,
        p_meas=0.002,
        rounds=3,
        trials=1000
    )

    print("\n--- Simulation Results ---")
    print(f"Clean Syndrome Rate (3 rounds) : {results['clean_syndrome_rate'] * 100:.2f}%")
    print(f"Sample Parallel Measurement Outcomes (First 5 trials):")
    for idx, sample in enumerate(results["sample_outcomes"]):
        print(f"  Trial {idx+1}: Obs1 = {sample[0]}, Obs2 = {sample[1]}")
#Quantum Computing #Quantum Error Correction #qLDPC Codes
Verify CSS Commutativity: H_X @ H_Z.T == 0 mod 2
"""
Numerical Simulation of Parallel Pauli Product Measurements on qLDPC Codes
Author: Quantum Systems Engineering Group
Dependencies: numpy, scipy
"""

import numpy as np
import scipy.sparse as sp
from typing import List, Tuple, Dict


class SymplecticPauli:
    """
    Represents an n-qubit Pauli operator as a binary symplectic vector (x | z) in F_2^(2n).
    """
    def __init__(self, x_vec: np.ndarray, z_vec: np.ndarray, label: str = ""):
        self.x = np.array(x_vec, dtype=int) % 2
        self.z = np.array(z_vec, dtype=int) % 2
        self.n = len(self.x)
        self.label = label
        assert len(self.z) == self.n, "X and Z vector dimensions must match."

    def commutes_with(self, other: 'SymplecticPauli') -> bool:
        """
        Calculates the symplectic inner product: _symp = (x1 . z2 + z1 . x2) mod 2.
        Returns True if operators commute, False if they anti-commute.
        """
        symp_prod = (np.dot(self.x, other.z) + np.dot(self.z, other.x)) % 2
        return bool(symp_prod == 0)

    def weight(self) -> int:
        """Returns the support weight (number of non-identity terms)."""
        return int(np.sum(self.x | self.z))

    def __repr__(self) -> str:
        return f"Pauli({self.label}, weight={self.weight()})"


class QLDPCCode:
    """
    Constructs a CSS Quantum LDPC Code via Hypergraph Product (HGP) of two classical codes.
    """
    def __init__(self, h1: np.ndarray, h2: np.ndarray):
        m1, n1 = h1.shape
        m2, n2 = h2.shape
        
        # Hypergraph Product block matrix assembly
        # H_X = [H1 (x) I_n2,  I_m1 (x) H2^T]
        # H_Z = [I_n1 (x) H2,  H1^T (x) I_m2]
        I_n1 = np.eye(n1, dtype=int)
        I_n2 = np.eye(n2, dtype=int)
        I_m1 = np.eye(m1, dtype=int)
        I_m2 = np.eye(m2, dtype=int)
        
        self.H_X = np.block([[np.kron(h1, I_n2), np.kron(I_m1, h2.T)]]) % 2
        self.H_Z = np.block([[np.kron(I_n1, h2), np.kron(h1.T, I_m2)]]) % 2
        
        self.n = self.H_X.shape[1]
        self.num_X_checks = self.H_X.shape[0]
        self.num_Z_checks = self.H_Z.shape[0]
        
        # Verify CSS Commutativity: H_X @ H_Z.T == 0 mod 2
        css_check = (self.H_X @ self.H_Z.T) % 2
        assert np.all(css_check == 0), "CSS orthogonality condition failed!"

    def max_check_degrees(self) -> Tuple[int, int]:
        deg_X = int(np.max(np.sum(self.H_X, axis=1)))
        deg_Z = int(np.max(np.sum(self.H_Z, axis=1)))
        return deg_X, deg_Z


class ParallelPPMScheduler:
    """
    Schedules Pauli Product Measurements into parallel layers using greedy vertex coloring.
    """
    def __init__(self, code: QLDPCCode):
        self.code = code

    def schedule(self, pauli_ops: List[SymplecticPauli]) -> List[List[SymplecticPauli]]:
        """
        Partitions target Pauli measurements into mutually commuting parallel layers.
        """
        layers: List[List[SymplecticPauli]] = []
        
        for op in pauli_ops:
            placed = False
            for layer in layers:
                # Check if op commutes with every operator already in the layer
                if all(op.commutes_with(existing_op) for existing_op in layer):
                    layer.append(op)
                    placed = True
                    break
            if not placed:
                layers.append([op])
                
        return layers


class NormalizedMinSumDecoder:
    """
    Normalized Min-Sum Belief Propagation Decoder for qLDPC codes over binary symmetric channels.
    """
    def __init__(self, H: np.ndarray, alpha: float = 0.75, max_iter: int = 50):
        self.H = H
        self.m, self.n = H.shape
        self.alpha = alpha
        self.max_iter = max_iter
        
        self.check_adj = [np.where(H[i] == 1)[0] for i in range(self.m)]
        self.var_adj = [np.where(H[:, j] == 1)[0] for j in range(self.n)]

    def decode(self, syndrome: np.ndarray, p_error: float) -> Tuple[np.ndarray, bool]:
        llr_prior = np.log((1.0 - p_error) / p_error)
        M_vc = np.zeros((self.m, self.n))
        
        # Initialize variable-to-check messages
        for j in range(self.n):
            for i in self.var_adj[j]:
                M_vc[i, j] = llr_prior
                
        M_cv = np.zeros((self.m, self.n))
        
        for iteration in range(self.max_iter):
            # Check-to-Variable update
            for i in range(self.m):
                nbrs = self.check_adj[i]
                s_i = syndrome[i]
                for j in nbrs:
                    other_nbrs = [c for c in nbrs if c != j]
                    if len(other_nbrs) == 0:
                        sign = 1.0 - 2.0 * s_i
                        M_cv[i, j] = sign * 10.0
                    else:
                        signs = np.sign(M_vc[i, other_nbrs])
                        signs[signs == 0] = 1.0
                        sign_prod = np.prod(signs)
                        if s_i == 1:
                            sign_prod *= -1.0
                        min_mag = np.min(np.abs(M_vc[i, other_nbrs]))
                        M_cv[i, j] = self.alpha * sign_prod * min_mag

            # Variable node update & hard decision
            L_total = np.full(self.n, llr_prior)
            for j in range(self.n):
                L_total[j] += np.sum(M_cv[self.var_adj[j], j])
                
            e_hat = (L_total < 0).astype(int)
            
            # Check syndrome match
            if np.array_equal((self.H @ e_hat) % 2, syndrome):
                return e_hat, True
                
            # Update Variable-to-Check messages for next round
            for j in range(self.n):
                for i in self.var_adj[j]:
                    other_checks = [c for c in self.var_adj[j] if c != i]
                    M_vc[i, j] = llr_prior + np.sum(M_cv[other_checks, j])

        return (L_total < 0).astype(int), False


# =====================================================================
# Simulation Runner
# =====================================================================
if __name__ == "__main__":
    # Seed linear codes for Hypergraph Product construction
    h1 = np.array([[1, 1, 0, 1], [0, 1, 1, 1]], dtype=int)
    h2 = np.array([[1, 0, 1, 1], [1, 1, 1, 0]], dtype=int)

    code = QLDPCCode(h1, h2)
    deg_X, deg_Z = code.max_check_degrees()
    print(f"[qLDPC Code Built] Physical Qubits n = {code.n}")
    print(f"  Parity Checks: {code.num_X_checks} X-checks, {code.num_Z_checks} Z-checks")
    print(f"  Max Check Degrees: deg(H_X) = {deg_X}, deg(H_Z) = {deg_Z}")

    # Generate sample Pauli target measurements
    np.random.seed(42)
    target_ops = []
    for k in range(12):
        x_vec = np.random.choice([0, 1], size=code.n, p=[0.75, 0.25])
        z_vec = np.random.choice([0, 1], size=code.n, p=[0.75, 0.25])
        target_ops.append(SymplecticPauli(x_vec, z_vec, label=f"P_{k+1}"))

    # Parallel Scheduling
    scheduler = ParallelPPMScheduler(code)
    parallel_layers = scheduler.schedule(target_ops)

    print(f"\n[Parallel Scheduling Results]")
    print(f"  Total Target PPMs: {len(target_ops)}")
    print(f"  Parallel Measurement Depth: {len(parallel_layers)} execution layers")
    for idx, layer in enumerate(parallel_layers):
        print(f"    Layer {idx+1}: {[op.label for op in layer]}")

    # Monte Carlo Noise Simulation
    p_noise = 0.03
    trials = 500
    successes = 0

    decoder = NormalizedMinSumDecoder(code.H_Z, alpha=0.75, max_iter=40)

    for _ in range(trials):
        true_error = (np.random.rand(code.n) < p_noise).astype(int)
        syndrome = (code.H_Z @ true_error) % 2
        
        if np.sum(syndrome) == 0:
            successes += 1
            continue

        decoded_error, converged = decoder.decode(syndrome, p_error=p_noise)
        residual = (true_error + decoded_error) % 2
        residual_syndrome = (code.H_Z @ residual) % 2

        if np.sum(residual_syndrome) == 0:
            successes += 1

    print(f"\n[Monte Carlo BP Decoding Evaluation]")
    print(f"  Physical Noise Rate p = {p_noise}")
    print(f"  Monte Carlo Trials = {trials}")
    print(f"  Syndrome Recovery Success Rate = {successes / trials * 100:.2f}%")
#VQE #Algorithms #QuantumChemistry #DeepDive
quantum_snippet.py
import pennylane as qml
from pennylane import numpy as np

dev = qml.device('default.qubit', wires=2)
H = qml.Hamiltonian([1.0, 0.5], [qml.PauliZ(0), qml.PauliX(0) @ qml.PauliX(1)])

@qml.qnode(dev)
def circuit(params):
    qml.RY(params[0], wires=0)
    qml.CNOT(wires=[0, 1])
    return qml.expval(H)

params = np.array([0.1], requires_grad=True)
opt = qml.GradientDescentOptimizer(stepsize=0.4)
for step in range(20):
    params, energy = opt.step_and_cost(circuit, params)
print(f'Optimized Ground State Energy: {energy:.6f}')
#QNN #QML #Algorithms #Software
quantum_snippet.py
# Parameterized QNN Layer Definition
def qnn_layer(weights, wires):
    for i, wire in enumerate(wires):
        qml.Rot(*weights[i], wires=wire)
    for i in range(len(wires) - 1):
        qml.CNOT(wires=[wires[i], wires[i + 1]])
#QuantumKernel #QML #Algorithms #DeepDive
quantum_snippet.py
# Quantum Kernel Evaluation
@qml.qnode(dev)
def kernel_circuit(x1, x2):
    feature_map(x1)
    qml.adjoint(feature_map)(x2)
    return qml.probs(wires=range(wires))
quantum_snippet.py
"""
Parallel Pauli Product Measurement Simulator for qLDPC Codes
Author: Quantum Computing Engineering & Architecture Group
Description: Constructs Hypergraph Product (HGP) qLDPC codes, computes parallel
             syndrome extraction schedules using bipartite graph coloring, and
             runs numerical Monte Carlo depolarizing noise simulations.
"""

from dataclasses import dataclass
from typing import Dict, List, Set, Tuple
import matplotlib.pyplot as plt
import numpy as np


@dataclass
class SimulationResults:
    num_shots: int
    physical_error_rate: float
    syndrome_error_rate: float
    x_layers_count: int
    z_layers_count: int
    data_qubits_count: int
    total_checks_count: int


class HypergraphProductCode:
    """
    Constructs a CSS qLDPC code via the Hypergraph Product (HGP) of two classical codes.
    Given H1 (r1 x n1) and H2 (r2 x n2):
    n_data = n1*n2 + r1*r2
    n_checks_x = r1 * n2
    n_checks_z = n1 * r2
    """

    def __init__(self, h1: np.ndarray, h2: np.ndarray):
        self.h1 = np.array(h1, dtype=int)
        self.h2 = np.array(h2, dtype=int)

        r1, n1 = self.h1.shape
        r2, n2 = self.h2.shape

        # H_X = [H1 x I_n2 , I_r1 x H2^T]
        hx_part1 = np.kron(self.h1, np.eye(n2, dtype=int))
        hx_part2 = np.kron(np.eye(r1, dtype=int), self.h2.T)
        self.Hx = np.hstack([hx_part1, hx_part2]) % 2

        # H_Z = [I_n1 x H2 , H1^T x I_r2]
        hz_part1 = np.kron(np.eye(n1, dtype=int), self.h2)
        hz_part2 = np.kron(self.h1.T, np.eye(r2, dtype=int))
        self.Hz = np.hstack([hz_part1, hz_part2]) % 2

        self.num_checks_x, self.num_data_qubits = self.Hx.shape
        self.num_checks_z = self.Hz.shape[0]

        assert self.verify_css_orthogonality(), "CSS Commutativity Condition Failed!"

    def verify_css_orthogonality(self) -> bool:
        """Verifies H_X @ H_Z^T == 0 (mod 2)."""
        commutation = (self.Hx @ self.Hz.T) % 2
        return bool(np.all(commutation == 0))


class ParallelSyndromeScheduler:
    """
    Schedules Pauli Product Measurements into non-conflicting parallel execution layers
    using a greedy bipartite edge-coloring algorithm.
    """

    @staticmethod
    def generate_parallel_layers(
        check_matrix: np.ndarray,
    ) -> List[List[Tuple[int, int]]]:
        """
        Decomposes check-qubit CNOT interactions into parallel layers.
        Returns a list of layers, where each layer contains tuples (check_index, qubit_index).
        """
        num_checks, num_qubits = check_matrix.shape
        edges: List[Tuple[int, int]] = []

        for c in range(num_checks):
            for q in range(num_qubits):
                if check_matrix[c, q] == 1:
                    edges.append((c, q))

        # Graph coloring / Parallel layer assignment
        layers: List[List[Tuple[int, int]]] = []

        for check_idx, qubit_idx in edges:
            assigned = False
            for layer in layers:
                active_checks: Set[int] = {e[0] for e in layer}
                active_qubits: Set[int] = {e[1] for e in layer}

                # Ensure no resource contention
                if (
                    check_idx not in active_checks
                    and qubit_idx not in active_qubits
                ):
                    layer.append((check_idx, qubit_idx))
                    assigned = True
                    break

            if not assigned:
                layers.append([(check_idx, qubit_idx)])

        return layers


class ParallelSyndromeSimulator:
    """
    Monte Carlo simulator for parallel Pauli Product syndrome extraction
    subject to circuit-level depolarizing noise.
    """

    def __init__(self, code: HypergraphProductCode, noise_rate: float):
        self.code = code
        self.p = noise_rate
        self.x_schedule = ParallelSyndromeScheduler.generate_parallel_layers(
            code.Hx
        )
        self.z_schedule = ParallelSyndromeScheduler.generate_parallel_layers(
            code.Hz
        )

    def run_simulation(
        self, num_shots: int = 1000, seed: int = 42
    ) -> SimulationResults:
        rng = np.random.default_rng(seed)
        syndrome_failures = 0

        for _ in range(num_shots):
            # Qubit physical error registers (X and Z Pauli errors)
            data_x_errors = np.zeros(self.code.num_data_qubits, dtype=int)
            data_z_errors = np.zeros(self.code.num_data_qubits, dtype=int)

            ancilla_x_meas = np.zeros(self.code.num_checks_x, dtype=int)
            ancilla_z_meas = np.zeros(self.code.num_checks_z, dtype=int)

            # --- Execute X-Check Parallel Measurement Layers ---
            for layer in self.x_schedule:
                for check_idx, qubit_idx in layer:
                    # Depolarizing noise injection during parallel CNOT
                    if rng.random() < self.p:
                        data_z_errors[qubit_idx] ^= 1  # Phase error on data
                    if rng.random() < self.p:
                        ancilla_x_meas[
                            check_idx
                        ] ^= 1  # Measurement / Ancilla flip

                    # CNOT propagates X error on data to ancilla
                    ancilla_x_meas[check_idx] ^= data_x_errors[qubit_idx]

            # --- Execute Z-Check Parallel Measurement Layers ---
            for layer in self.z_schedule:
                for check_idx, qubit_idx in layer:
                    # Depolarizing noise injection during parallel CNOT
                    if rng.random() < self.p:
                        data_x_errors[qubit_idx] ^= 1  # Bit flip on data
                    if rng.random() < self.p:
                        ancilla_z_meas[
                            check_idx
                        ] ^= 1  # Measurement / Ancilla flip

                    # CNOT propagates Z error on data to ancilla
                    ancilla_z_meas[check_idx] ^= data_z_errors[qubit_idx]

            # Compute ideal expected syndromes from final data state
            ideal_x_syndrome = (self.code.Hx @ data_x_errors) % 2
            ideal_z_syndrome = (self.code.Hz @ data_z_errors) % 2

            # Check if measured syndrome matches ideal physical state syndrome
            x_mismatch = not np.array_equal(ancilla_x_meas, ideal_x_syndrome)
            z_mismatch = not np.array_equal(ancilla_z_meas, ideal_z_syndrome)

            if x_mismatch or z_mismatch:
                syndrome_failures += 1

        return SimulationResults(
            num_shots=num_shots,
            physical_error_rate=self.p,
            syndrome_error_rate=syndrome_failures / num_shots,
            x_layers_count=len(self.x_schedule),
            z_layers_count=len(self.z_schedule),
            data_qubits_count=self.code.num_data_qubits,
            total_checks_count=self.code.num_checks_x + self.code.num_checks_z,
        )


def main():
    print("==========================================================")
    print("Parallel Pauli Product Measurement Simulator (qLDPC Codes)")
    print("==========================================================")

    # Define classical components for HGP construction
    # H1: [7, 4, 3] Hamming Code parity check matrix
    h1_hamming = np.array(
        [
            [1, 1, 0, 1, 1, 0, 0],
            [1, 0, 1, 1, 0, 1, 0],
            [0, 1, 1, 1, 0, 0, 1],
        ],
        dtype=int,
    )

    # H2: [3, 1, 3] Repetition Code parity check matrix
    h2_rep = np.array([[1, 1, 0], [0, 1, 1]], dtype=int)

    # Construct quantum HGP code
    q_code = HypergraphProductCode(h1_hamming, h2_rep)
    print(f"Code Dimensions:")
    print(f" - Data Qubits (N) : {q_code.num_data_qubits}")
    print(f" - X Checks        : {q_code.num_checks_x}")
    print(f" - Z Checks        : {q_code.num_checks_z}")
    print(f" - CSS Validity    : {q_code.verify_css_orthogonality()}")

    # Sweep error rates
    physical_errors = np.linspace(0.0001, 0.005, 8)
    syndrome_error_rates = []

    print("\nRunning parallel schedule simulation sweep...")
    for p in physical_errors:
        sim = ParallelSyndromeSimulator(q_code, noise_rate=p)
        res = sim.run_simulation(num_shots=3000, seed=123)
        syndrome_error_rates.append(res.syndrome_error_rate)
        print(
            f"Physical Error Rate p={p:.4f} | Syndrome Error Rate={res.syndrome_error_rate:.4f} "
            f"| Schedule Depth (X:{res.x_layers_count}, Z:{res.z_layers_count})"
        )

    print("\nSimulation completed successfully.")


if __name__ == "__main__":
    main()
1. Quantum Circuit & Custom Gate Calibration Construction from Engineering Low-Level Control: Implementing Custom Gates via Qiskit Pulse in Qiskit 1.0
#Quantum Computing #Qiskit Pulse #Pulse Control
1. Quantum Circuit & Custom Gate Calibration Construction
import numpy as np
import scipy.linalg as la
import matplotlib.pyplot as plt

from qiskit import QuantumCircuit
from qiskit.circuit import Gate
from qiskit import pulse
from qiskit.pulse import DriveChannel, Drag, build


# =====================================================================
# 1. Quantum Circuit & Custom Gate Calibration Construction
# =====================================================================

class CustomSXGate(Gate):
    """Custom pi/2 rotation gate around X-axis (sqrt(X)) using DRAG."""
    def __init__(self, label=None):
        super().__init__('custom_sx', 1, [], label=label)

def build_custom_calibrated_circuit():
    """Builds a QuantumCircuit and attaches custom pulse calibration."""
    qc = QuantumCircuit(1, 1)
    custom_sx = CustomSXGate()
    
    # Append custom gate to circuit
    qc.append(custom_sx, [0])
    qc.measure(0, 0)

    # Hardware pulse parameters
    duration = 160        # pulse duration in samples (dt units)
    amp = 0.18           # peak pulse amplitude [0, 1]
    sigma = 40           # Gaussian standard deviation
    beta = 1.85          # DRAG parameter

    # Construct low-level pulse schedule using pulse builder
    with build(name="custom_sx_schedule") as custom_sched:
        d0 = DriveChannel(0)
        # Create DRAG pulse envelope
        drag_pulse = Drag(duration=duration, amp=amp, sigma=sigma, beta=beta)
        pulse.play(drag_pulse, d0)

    # Attach pulse calibration to circuit for qubit 0
    qc.add_calibration(custom_sx, (0,), custom_sched)
    
    return qc, custom_sched

# Generate circuit and schedule
circuit, schedule = build_custom_calibrated_circuit()
print("--- Quantum Circuit Representation ---")
print(circuit)
print("\n--- Attached Pulse Schedule ---")
print(schedule)


# =====================================================================
# 2. Physics Simulation: 3-Level Transmon Dynamics (DRAG vs Gaussian)
# =====================================================================

def simulate_transmon_dynamics(use_drag=True):
    """
    Simulates time-dependent Schrödinger equation for a 3-level transmon
    system subjected to a Gaussian or DRAG pulse envelope.
    """
    # System Constants
    dt = 0.222e-9                 # Qiskit sampling interval (222 ps)
    duration_samples = 160        # 160 dt = ~35.5 ns
    t_total = duration_samples * dt
    
    delta = -2 * np.pi * 330e6   # Transmon anharmonicity (-330 MHz in rad/s)
    rabi_scale = 2 * np.pi * 45e6 # Coupling scaling (45 MHz peak Rabi rate)
    
    # Time vector
    t = np.linspace(0, t_total, duration_samples)
    t0 = t_total / 2
    sigma = 40 * dt
    amp = 0.18
    beta = 1.85 / (2 * np.pi * 330e6) # Dimensional DRAG parameter
    
    # Envelope components
    gauss = amp * np.exp(-0.5 * ((t - t0) / sigma) ** 2)
    dgauss = -((t - t0) / (sigma ** 2)) * gauss
    
    Omega_x = gauss
    Omega_y = -beta * dgauss if use_drag else np.zeros_like(gauss)
    
    # 3-level Transmon Operators
    # |0> = [1,0,0]^T, |1> = [0,1,0]^T, |2> = [0,0,1]^T
    H_0 = np.diag([0.0, 0.0, delta]) # Static Hamiltonian in rotating frame
    
    a = np.array([[0, 1, 0],
                  [0, 0, np.sqrt(2)],
                  [0, 0, 0]], dtype=complex)
    a_dag = a.T.conj()
    
    # Time evolution starting from ground state |0>
    psi = np.array([1.0, 0.0, 0.0], dtype=complex)
    step_dt = dt
    
    pop_0, pop_1, pop_2 = [], [], []
    
    for i in range(duration_samples):
        # Time-dependent drive Hamiltonian: H_d = 0.5 * (Omega * a^\dagger + Omega* * a)
        Omega_t = rabi_scale * (Omega_x[i] + 1j * Omega_y[i])
        H_drive = 0.5 * (Omega_t * a_dag + np.conj(Omega_t) * a)
        H_total = H_0 + H_drive
        
        # Propagation via matrix exponential over timestep dt
        U_step = la.expm(-1j * H_total * step_dt)
        psi = U_step @ psi
        
        # Record populations
        pop_0.append(np.abs(psi[0])**2)
        pop_1.append(np.abs(psi[1])**2)
        pop_2.append(np.abs(psi[2])**2)
        
    return t * 1e9, np.array(pop_0), np.array(pop_1), np.array(pop_2)

# Run simulations
t_ns, p0_g, p1_g, p2_g = simulate_transmon_dynamics(use_drag=False)
_, p0_d, p1_d, p2_d = simulate_transmon_dynamics(use_drag=True)

print("\n--- Simulation Results ---")
print(f"Standard Gaussian Peak Leakage (|2> Population): {np.max(p2_g):.6e}")
print(f"DRAG Corrected Peak Leakage (|2> Population)    : {np.max(p2_d):.6e}")
print(f"Leakage Suppression Ratio                      : {np.max(p2_g) / np.max(p2_d):.2f}x")
#Quantum Computing #Qiskit #Quantum Hardware
1. Define Custom Gate Structure
import numpy as np
from scipy.linalg import expm
from qiskit import QuantumCircuit
from qiskit.circuit import Gate, Parameter
from qiskit.pulse import DriveChannel, Drag, builder

# =====================================================================
# 1. Define Custom Gate Structure
# =====================================================================
class CustomRxGate(Gate):
    """Custom single-qubit RX gate parameterized by rotation angle theta."""
    def __init__(self, theta):
        super().__init__('custom_rx', 1, [theta])

    def _define(self):
        """Standard gate decomposition fallback for ideal simulator."""
        qc = QuantumCircuit(1)
        qc.rx(self.params[0], 0)
        self.definition = qc


# =====================================================================
# 2. Build Pulse Schedule using Qiskit Pulse Builder
# =====================================================================
def build_drag_schedule(qubit: int, theta_val: float, duration: int = 160, sigma: float = 40):
    """
    Constructs a DRAG pulse schedule for a given qubit and target angle.
    
    Parameters:
        qubit: Drive channel index (qubit index).
        theta_val: Target rotation angle (radians).
        duration: Pulse duration in dtm units (must be integer multiple of 16).
        sigma: Standard deviation of the Gaussian profile.
    """
    # Linear calibration scaling: pi rotation corresponds to amp = 0.25
    amp_max = 0.25
    amp = (theta_val / np.pi) * amp_max
    beta = 0.4  # Anharmonicity correction factor
    
    with builder.build(name=f"drag_rx_{theta_val:.2f}_sched") as pulse_sched:
        chan = DriveChannel(qubit)
        drag_wave = Drag(
            duration=duration,
            amp=amp,
            sigma=sigma,
            beta=beta,
            name=f"drag_wave_{theta_val:.2f}"
        )
        builder.play(drag_wave, chan)
        
    return pulse_sched


# =====================================================================
# 3. Attach Calibration to QuantumCircuit
# =====================================================================
def main():
    theta_target = np.pi / 2  # Target RX(pi/2) operation
    custom_gate = CustomRxGate(theta_target)

    # Initialize Circuit
    qc = QuantumCircuit(1, 1)
    qc.append(custom_gate, [0])
    qc.measure(0, 0)

    # Generate calibrated pulse schedule
    pulse_sched = build_drag_schedule(qubit=0, theta_val=theta_target)

    # Attach pulse schedule as calibration in Qiskit 1.0.0
    qc.add_calibration(
        gate=custom_gate, 
        qubits=(0,), 
        schedule=pulse_sched, 
        params=[theta_target]
    )

    print("=== Compiled Quantum Circuit ===")
    print(qc)

    print("\n=== Registered Calibration Instructions ===")
    for time_step, instruction in pulse_sched.instructions:
        print(f"Time {time_step:04d} dtm: {instruction}")

    # =================================================================
    # 4. Semi-Classical Hamiltonian Unitary Verification
    # =================================================================
    # Pauli matrix basis
    sigma_x = np.array([[0, 1], [1, 0]], dtype=complex)
    sigma_y = np.array([[0, -1j], [1j, 0]], dtype=complex)

    # Time discretization
    duration = 160
    t = np.arange(duration)
    t0 = duration / 2.0
    sigma = 40.0
    amp = (theta_target / np.pi) * 0.25
    beta = 0.4

    # Envelopes
    env_i = amp * np.exp(-0.5 * ((t - t0) / sigma)**2)
    env_q = -beta * ((t - t0) / (sigma**2)) * env_i

    # Numerical integration of Schrödinger equation
    U = np.eye(2, dtype=complex)
    dt_scale = 0.0392  # Scaling factor to align pulse amplitude with integrated rotation angle
    
    for i in range(duration):
        H_d = 0.5 * (env_i[i] * sigma_x + env_q[i] * sigma_y)
        U = expm(-1j * H_d * dt_scale) @ U

    print("\n=== Numerically Integrated Unitary Matrix ===")
    print(np.round(U, 3))
    
    # Expected target unitary for RX(pi/2): cos(pi/4)I - i*sin(pi/4)X
    target_u = np.array([
        [np.cos(np.pi/4), -1j * np.sin(np.pi/4)],
        [-1j * np.sin(np.pi/4), np.cos(np.pi/4)]
    ])
    print("\n=== Theoretical Target RX(pi/2) Unitary ===")
    print(np.round(target_u, 3))

if __name__ == "__main__":
    main()
#Quantum Computing #Neutral Atoms #Fault Tolerance
1. Blockade Potential vs Distance
#!/usr/bin/env python3
"""
Rydberg Blockade & Neutral-Atom Fault-Tolerance Simulator
==========================================================
Models two-qubit dynamics under the Rydberg blockade Hamiltonian, computes 
CZ gate unitaries, and evaluates loss-to-erasure syndrome thresholds.
"""

import numpy as np
from scipy.linalg import expm
from scipy.integrate import solve_ivp


class NeutralAtomRydbergSimulator:
    def __init__(self, omega_mhz: float = 10.0, c6_ghz_um6: float = 5000.0):
        """
        Parameters:
        -----------
        omega_mhz : float
            Rabi frequency in MHz.
        c6_ghz_um6 : float
            Rydberg C6 dispersion coefficient in GHz * um^6.
        """
        self.omega = 2.0 * np.pi * omega_mhz * 1e6      # rad/s
        self.c6 = 2.0 * np.pi * c6_ghz_um6 * 1e9 * 1e-36 # rad/s * m^6
        
    def interaction_potential(self, distance_um: float) -> float:
        """Calculates V_rr potential in rad/s for inter-atomic distance in micrometers."""
        r_m = distance_um * 1e-6
        return self.c6 / (r_m ** 6)

    def build_hamiltonian(self, distance_um: float, delta_mhz: float = 0.0) -> np.ndarray:
        """
        Builds 9-level system Hamiltonian for 2 atoms with levels {|0>, |1>, |r>}.
        Basis mapping: 00, 01, 0r, 10, 11, 1r, r0, r1, rr
        """
        delta = 2.0 * np.pi * delta_mhz * 1e6
        v_rr = self.interaction_potential(distance_um)
        
        basis = ['00', '01', '0r', '10', '11', '1r', 'r0', 'r1', 'rr']
        dim = len(basis)
        H = np.zeros((dim, dim), dtype=np.complex128)
        
        for i, s1 in enumerate(basis):
            for j, s2 in enumerate(basis):
                # Detuning and Rydberg interaction (diagonal elements)
                if i == j:
                    r_count = s1.count('r')
                    H[i, i] -= delta * r_count
                    if s1 == 'rr':
                        H[i, i] += v_rr
                
                # Laser drive on atom 1 (|1> <-> |r>)
                if s1[1] == s2[1]:
                    if (s1[0] == '1' and s2[0] == 'r') or (s1[0] == 'r' and s2[0] == '1'):
                        H[i, j] += self.omega / 2.0
                        
                # Laser drive on atom 2 (|1> <-> |r>)
                if s1[0] == s2[0]:
                    if (s1[1] == '1' and s2[1] == 'r') or (s1[1] == 'r' and s2[1] == '1'):
                        H[i, j] += self.omega / 2.0

        return H

    def simulate_cz_unitary(self, distance_um: float, pulse_duration_ns: float) -> np.ndarray:
        """Computes effective 4x4 computational basis matrix after Rydberg pulse."""
        H = self.build_hamiltonian(distance_um)
        t_sec = pulse_duration_ns * 1e-9
        U_full = expm(-1j * H * t_sec)
        
        basis = ['00', '01', '0r', '10', '11', '1r', 'r0', 'r1', 'rr']
        comp_indices = [basis.index('00'), basis.index('01'), 
                        basis.index('10'), basis.index('11')]
        
        U_comp = U_full[np.ix_(comp_indices, comp_indices)]
        return U_comp


def evaluate_erasure_decoder_performance(p_erasure: float, p_pauli: float, n_trials: int = 100000):
    """
    Monte Carlo evaluation of logical error suppression in surface code 
    syndrome extraction under erasure-converted noise channels.
    """
    np.random.seed(42)
    # Simulate a distance-3 code distance check (4 data qubits per stabilizer)
    data_qubits = 4
    
    # Generate random errors
    erasure_events = np.random.rand(n_trials, data_qubits) < p_erasure
    pauli_events = np.random.rand(n_trials, data_qubits) < p_pauli
    
    # Logical failure criterion: >1 unknown Pauli OR >2 Erasures
    erasure_counts = np.sum(erasure_events, axis=1)
    pauli_counts = np.sum(pauli_events & (~erasure_events), axis=1)
    
    # Erasure errors are known: 1 known erasure requires 1 bit to correct (MWPM decoder)
    failures = (pauli_counts >= 2) | (erasure_counts >= 3) | ((erasure_counts >= 1) & (pauli_counts >= 1))
    logical_error_rate = np.mean(failures)
    
    return logical_error_rate


if __name__ == '__main__':
    print("=" * 70)
    print("NEUTRAL-ATOM RYDBERG BLOCKADE & FAULT-TOLERANCE MODEL")
    print("=" * 70)
    
    sim = NeutralAtomRydbergSimulator(omega_mhz=10.0, c6_ghz_um6=5000.0)
    
    # 1. Blockade Potential vs Distance
    print("\n[1] Rydberg Blockade Strength vs Distance:")
    for dist in [2.0, 3.0, 4.0, 6.0, 10.0]:
        v_mhz = sim.interaction_potential(dist) / (2 * np.pi * 1e6)
        blockade_ratio = v_mhz / 10.0  # Omega = 10 MHz
        print(f"  Distance: {dist:4.1f} μm | V_rr: {v_mhz:10.2f} MHz | V_rr/Ω: {blockade_ratio:8.2f}")

    # 2. CZ Gate Unitary Matrix Evaluation
    # Optimal pulse time for Levine-Lukin protocol ~ sqrt(2)*pi / Omega
    t_pulse_ns = (np.sqrt(2) * np.pi / sim.omega) * 1e9
    U_cz = sim.simulate_cz_unitary(distance_um=3.0, pulse_duration_ns=t_pulse_ns)
    
    print(f"\n[2] Calculated CZ Gate Computational Subspace Unitary (at r = 3.0 μm, t = {t_pulse_ns:.2f} ns):")
    print("  Magnitudes of U_comp:\n", np.round(np.abs(U_cz), 4))
    print("  Phases (radians):\n", np.round(np.angle(U_cz), 4))

    # 3. Erasure Conversion Decoding Performance
    print("\n[3] Logical Error Rate Simulation (Erasure Conversion vs Pure Pauli):")
    p_erasure_val = 0.04
    p_pauli_val = 0.005
    
    p_fail_erasure = evaluate_erasure_decoder_performance(p_erasure=p_erasure_val, p_pauli=p_pauli_val)
    p_fail_pauli_only = evaluate_erasure_decoder_performance(p_erasure=0.0, p_pauli=p_erasure_val + p_pauli_val)
    
    print(f"  Scenario A (Converted: {p_erasure_val*100}% Erasure, {p_pauli_val*100}% Pauli):")
    print(f"    --> Logical Failure Rate: {p_fail_erasure:.5f}")
    print(f"  Scenario B (Unconverted: {(p_erasure_val+p_pauli_val)*100}% Standard Pauli):")
    print(f"    --> Logical Failure Rate: {p_fail_pauli_only:.5f}")
    print(f"  Error Suppression Factor: {p_fail_pauli_only / p_fail_erasure:.2f}x improvement!")
    print("=" * 70)
#Quantum Computing #Neutral Atoms #Quantum Error Correction
quantum_snippet.py
import numpy as np
from scipy.integrate import solve_ivp
import matplotlib.pyplot as plt

def simulate_rydberg_blockade(
    omega_mhz: float = 5.0,          # Rabi frequency (\Omega / 2\pi) in MHz
    c6_ghz_um6: float = 860.0,       # C6 coefficient in GHz * \mu m^6 (e.g., 87Rb 70S1/2)
    r_microns: float = 3.5,          # Interatomic separation distance R in \mu m
    t_max_us: float = 0.5,           # Simulation duration in microseconds
    num_points: int = 500
):
    """
    Simulates time evolution of two neutral atoms under laser excitation to Rydberg state.
    
    Subspace basis:
      |00>: Both atoms in ground state |11>
      |W>:  Symmetric single-excitation state (|1r> + |r1>) / sqrt(2)
      |rr>: Double Rydberg excitation state |rr>
    """
    # Convert parameters to angular frequency units (rad / microsecond)
    omega = 2.0 * np.pi * omega_mhz
    
    # Calculate C6 in MHz * \mu m^6
    c6_mhz = c6_ghz_um6 * 1000.0
    v_blockade_mhz = c6_mhz / (r_microns ** 6)
    v_rr = 2.0 * np.pi * v_blockade_mhz

    # Effective Hamiltonian in {|00>, |W>, |rr>} basis
    H_eff = np.array([
        [0.0,                     np.sqrt(2) * omega / 2.0, 0.0],
        [np.sqrt(2) * omega / 2.0, 0.0,                      np.sqrt(2) * omega / 2.0],
        [0.0,                     np.sqrt(2) * omega / 2.0, v_rr]
    ], dtype=complex)

    def schrodinger_system(t, psi):
        return -1j * (H_eff @ psi)

    # Initial state: Both atoms in state |11> (represented as state index 0)
    psi_0 = np.array([1.0 + 0j, 0.0 + 0j, 0.0 + 0j])
    t_span = (0.0, t_max_us)
    t_eval = np.linspace(0.0, t_max_us, num_points)

    # Numerical integration of Schrödinger equation
    sol = solve_ivp(
        schrodinger_system, 
        t_span, 
        psi_0, 
        t_eval=t_eval, 
        rtol=1e-9, 
        atol=1e-11
    )

    probabilities = np.abs(sol.y) ** 2
    
    return sol.t, probabilities, v_blockade_mhz

if __name__ == "__main__":
    t_us, probs, v_blockade = simulate_rydberg_blockade(
        omega_mhz=5.0, 
        c6_ghz_um6=860.0, 
        r_microns=3.5, 
        t_max_us=0.5
    )

    print("--- Rydberg Blockade Numerical Simulation ---")
    print(f"Rabi Frequency (\Omega / 2\pi): 5.0 MHz")
    print(f"Calculated Blockade Shift (V_rr / 2\pi): {v_blockade:.2f} MHz")
    print(f"Peak Double-Excitation Population P(|rr>): {np.max(probs[2, :]):.6e}")
    print(f"Peak Symmetric Excitation Population P(|W>): {np.max(probs[1, :]):.4f}")

    # Output verification check
    assert np.max(probs[2, :]) < 1e-3, "Blockade failed to suppress double excitation!"
    print("Verification passed: Rydberg blockade strongly suppresses double excitation.")
1. Single-qubit subspace dynamics {|1>, |r>} from Architecting Fault-Tolerant Quantum Processors with Reconfigurable Neutral-Atom Arrays
#Quantum Computing #Neutral Atoms #Fault Tolerance
1. Single-qubit subspace dynamics {|1>, |r>}
#!/usr/bin/env python3
"""
Neutral-Atom Fault-Tolerant Architecture Simulation Toolkit
============================================================
This module simulates:
1. Hamiltonian dynamics of a 2-qubit Rydberg blockade CZ gate.
2. Logical error rate scaling for Surface Codes under Erasure Error Conversion.

Author: Senior Quantum Systems Engineer & Technical Writer
License: MIT
"""

import numpy as np
from scipy.linalg import expm
import matplotlib.pyplot as plt
from typing import Dict, Tuple, Any


class RydbergCZSimulator:
    """
    Simulates the continuous-time Hamiltonian dynamics of two neutral-atom qubits
    coupled via Rydberg interaction potentials V = C6 / R^6.
    """

    def __init__(self, omega_mhz: float = 15.0, blockade_v_mhz: float = 300.0):
        """
        Initialize simulator parameters.

        Parameters:
            omega_mhz: Single-atom Rabi frequency \Omega in MHz.
            blockade_v_mhz: Inter-atomic Rydberg blockade interaction energy V in MHz.
        """
        self.omega = 2.0 * np.pi * omega_mhz          # Convert to rad/us
        self.V = 2.0 * np.pi * blockade_v_mhz          # Convert to rad/us

    def build_4d_blockade_hamiltonian(self, detuning_mhz: float) -> np.ndarray:
        """
        Constructs the 4x4 Hamiltonian matrix in the blockaded subspace:
        Basis: {|11>, |1r>, |r1>, |rr>}

        Parameters:
            detuning_mhz: Laser detuning \Delta in MHz.

        Returns:
            4x4 complex numpy array representing H_4d in rad/us.
        """
        delta = 2.0 * np.pi * detuning_mhz
        H = np.zeros((4, 4), dtype=complex)

        # Off-diagonal Rabi couplings (\Omega / 2)
        H[0, 1] = H[1, 0] = self.omega / 2.0  # |11> <-> |1r>
        H[0, 2] = H[2, 0] = self.omega / 2.0  # |11> <-> |r1>
        H[1, 3] = H[3, 1] = self.omega / 2.0  # |1r> <-> |rr>
        H[2, 3] = H[3, 2] = self.omega / 2.0  # |r1> <-> |rr>

        # Diagonal detunings and interaction shift V
        H[1, 1] = -delta
        H[2, 2] = -delta
        H[3, 3] = -2.0 * delta + self.V

        return H

    def simulate_gate(self, detuning_mhz: float = 0.0, pulse_time_us: float = None) -> Dict[str, Any]:
        """
        Executes time evolution under the Rydberg Hamiltonian for time t.

        Parameters:
            detuning_mhz: Laser detuning \Delta in MHz.
            pulse_time_us: Pulse duration in microseconds. If None, computes
                           ideal 2\pi single-qubit duration.

        Returns:
            Dictionary containing pulse metrics, phases, leakage, and gate fidelity.
        """
        if pulse_time_us is None:
            # Ideal single-qubit 2\pi Rabi pulse condition
            pulse_time_us = (2.0 * np.pi) / np.sqrt(self.omega**2 + (2.0 * np.pi * detuning_mhz)**2)

        # 1. Single-qubit subspace dynamics {|1>, |r>}
        delta = 2.0 * np.pi * detuning_mhz
        H_1q = np.array([
            [0.0, self.omega / 2.0],
            [self.omega / 2.0, -delta]
        ], dtype=complex)

        U_1q = expm(-1j * H_1q * pulse_time_us)
        phase_1q = np.angle(U_1q[0, 0])

        # 2. Blockaded 4D subspace dynamics {|11>, |1r>, |r1>, |rr>}
        H_4d = self.build_4d_blockade_hamiltonian(detuning_mhz)
        U_4d = expm(-1j * H_4d * pulse_time_us)

        # Evolve initial state |11> (index 0)
        state_11_init = np.array([1.0, 0.0, 0.0, 0.0], dtype=complex)
        state_11_final = U_4d @ state_11_init

        phase_2q = np.angle(state_11_final[0])
        leakage = 1.0 - np.abs(state_11_final[0])**2

        # Net non-local entangling phase: \Delta\Phi = \Phi_{11} - 2*\Phi_1
        net_entangling_phase = (phase_2q - 2.0 * phase_1q) % (2.0 * np.pi)

        # Entangling phase deviation from ideal \pi
        phase_error = np.abs(net_entangling_phase - np.pi)
        cz_fidelity = (1.0 - leakage) * (np.cos(phase_error / 2.0)**2)

        return {
            "pulse_duration_ns": pulse_time_us * 1e3,
            "single_qubit_phase_rad": phase_1q,
            "two_qubit_phase_rad": phase_2q,
            "net_entangling_phase_rad": net_entangling_phase,
            "rydberg_population_leakage": leakage,
            "cz_gate_fidelity": cz_fidelity
        }


class ErasureSurfaceCodeModel:
    """
    Evaluates fault-tolerant Surface Code thresholds and logical error rates
    with erasure conversion in reconfigurable neutral atom arrays.
    """

    def __init__(self, code_distance: int = 7, erasure_ratio: float = 0.98):
        """
        Parameters:
            code_distance: Surface code distance 'd' (must be odd integer).
            erasure_ratio: Fraction \eta = p_erasure / p_total (0 <= \eta <= 1).
        """
        if code_distance % 2 == 0:
            raise ValueError("Code distance 'd' must be an odd integer.")
        self.d = code_distance
        self.eta = erasure_ratio

        # Threshold limits (analytical and numerical MWPM decoders)
        self.p_th_pauli = 0.010    # Standard physical Pauli threshold (~1.0%)
        self.p_th_erasure = 0.043  # Erasure-converted threshold (~4.3%)

    def calculate_logical_error(self, p_physical: float) -> float:
        """
        Calculates logical error probability P_L under dual-noise threshold scaling.

        Parameters:
            p_physical: Total physical error rate per gate step.

        Returns:
            Logical error probability P_L.
        """
        p_pauli = p_physical * (1.0 - self.eta)
        p_erasure = p_physical * self.eta

        # Effective distance scaling parameter
        effective_noise = (p_pauli / self.p_th_pauli) + (p_erasure / self.p_th_erasure)

        if effective_noise >= 1.0:
            return 0.5  # Chaotic error regime above threshold

        # Below-threshold exponential suppression: P_L \sim C * (p_eff)^((d+1)/2)
        exponent = (self.d + 1) / 2.0
        prefactor = 0.03
        p_logical = prefactor * (effective_noise ** exponent)

        return float(np.clip(p_logical, 0.0, 0.5))


def execute_full_simulation_pipeline():
    """
    Executes the end-to-end physics simulation and threshold generation pipeline.
    """
    print("=========================================================================")
    print("   NEUTRAL-ATOM QUANTUM ARCHITECTURE: PHYSICS & ERROR CODE SIMULATION")
    print("=========================================================================\n")

    # --- Part 1: Rydberg CZ Gate Dynamics ---
    rabi_freq_mhz = 15.0
    blockade_shift_mhz = 350.0

    sim = RydbergCZSimulator(omega_mhz=rabi_freq_mhz, blockade_v_mhz=blockade_shift_mhz)
    gate_results = sim.simulate_gate(detuning_mhz=0.0)

    print("[1] RYDBERG CZ GATE DYNAMICS SIMULATION")
    print(f"  Rabi Frequency (\u03a9)          : {rabi_freq_mhz:.2f} MHz")
    print(f"  Blockade Interaction (V)    : {blockade_shift_mhz:.2f} MHz")
    print(f"  Pulse Duration (\u03c4)          : {gate_results['pulse_duration_ns']:.3f} ns")
    print(f"  Single-Qubit Phase (\u03a6_1)     : {gate_results['single_qubit_phase_rad']:.6f} rad")
    print(f"  Two-Qubit Phase (\u03a6_11)     : {gate_results['two_qubit_phase_rad']:.6f} rad")
    print(f"  Net Entangling Phase (\u0394\u03a6) : {gate_results['net_entangling_phase_rad']:.6f} rad (Target: \u03c0 = {np.pi:.6f})")
    print(f"  Rydberg State Leakage     : {gate_results['rydberg_population_leakage']:.4e}")
    print(f"  Simulated CZ Gate Fidelity : {gate_results['cz_gate_fidelity'] * 100:.4f}%\n")

    # --- Part 2: Erasure Surface Code Thresholds ---
    print("[2] ERASURE-CONVERTED SURFACE CODE SCALING MODEL")
    physical_noise_levels = np.linspace(0.001, 0.025, 6)
    code_distances = [3, 5, 7, 9]
    erasure_ratio = 0.95  # 95% of errors converted to erasures

    print(f"  Erasure Error Ratio (\u03b7) : {erasure_ratio * 100:.1f}%\n")
    print(f"  {'Physical Error (p)':<20} | " + " | ".join([f"d = {d:<6}" for d in code_distances]))
    print("  " + "-" * 68)

    for p in physical_noise_levels:
        row_str = f"  {p * 100:6.3f}%               | "
        for d in code_distances:
            model = ErasureSurfaceCodeModel(code_distance=d, erasure_ratio=erasure_ratio)
            p_log = model.calculate_logical_error(p)
            row_str += f"{p_log:8.2e} | "
        print(row_str)

    print("\n=========================================================================")


if __name__ == "__main__":
    execute_full_simulation_pipeline()
quantum_snippet.py
#!/usr/bin/env python3
"""
Rydberg Blockade CZ Gate & Erasure Conversion Simulator

Simulates two neutral-atom qubits driven by laser pulses targeting 
the ground-to-Rydberg transition (|1> -> |r>). Integrates non-Hermitian
decay terms to track coherent gate dynamics, population leakage, 
and erasure conversion probability.
"""

import numpy as np
from scipy.integrate import solve_ivp
import matplotlib.pyplot as plt


class RydbergGateSimulator:
    """
    Simulates time-dependent Hamiltonian evolution for two neutral atoms
    under Rydberg blockade conditions with decay channel tracking.
    """
    def __init__(self, omega_max: float, v_blockade: float, gamma_rydberg: float):
        """
        Parameters:
            omega_max (float): Peak Rabi frequency (rad/s)
            v_blockade (float): Van der Waals interaction energy V_rr (rad/s)
            gamma_rydberg (float): Spontaneous decay rate from Rydberg state |r> (s^-1)
        """
        self.omega_max = omega_max
        self.v_blockade = v_blockade
        self.gamma_rydberg = gamma_rydberg
        
        # Subspace basis: [|11>, |1r>, |r1>, |rr>]
        self.dim = 4

    def pulse_profile(self, t: float, gate_duration: float) -> tuple[float, float, float]:
        """
        Generates smooth pulse envelopes for Rabi frequency Ω(t), phase φ(t), and detuning Δ(t).
        Uses a phase-discontinuous Levine-Pichler-style protocol.
        """
        # Smooth pulse envelope (sine squared)
        omega = self.omega_max * (np.sin(np.pi * t / gate_duration) ** 2)
        
        # Detuning tuned to optimal single-qubit excitation path
        delta = 0.375 * self.omega_max
        
        # Phase jump at mid-point of gate execution
        if t < (gate_duration / 2.0):
            phi = 0.0
        else:
            phi = 2.15  # Phase step in radians for Levine-Pichler protocol
            
        return omega, phi, delta

    def system_derivatives(self, t: float, state: np.ndarray, gate_duration: float) -> np.ndarray:
        """
        Calculates d|ψ>/dt using effective non-Hermitian Hamiltonian:
        H_eff = H_sys - (i * ħ * γ / 2) * (|1r><1r| + |r1>, 1 for |1r>, etc.
        """
        psi0 = np.zeros(self.dim, dtype=np.complex128)
        psi0[initial_state_idx] = 1.0 + 0.0j

        t_eval = np.linspace(0, gate_duration, 500)
        
        sol = solve_ivp(
            fun=lambda t, y: self.system_derivatives(t, y, gate_duration),
            t_span=(0, gate_duration),
            y0=psi0,
            t_eval=t_eval,
            method='RK45',
            rtol=1e-9,
            atol=1e-11
        )

        # Analysis of final state
        final_state = sol.y[:, -1]
        norm_survival = np.real(np.vdot(final_state, final_state))
        erasure_probability = 1.0 - norm_survival
        
        # Calculate conditional phase shift on state |11>
        accumulated_phase = np.angle(final_state[0])

        return {
            'times': sol.t,
            'state_trajectories': sol.y,
            'final_state': final_state,
            'norm_survival': norm_survival,
            'erasure_prob': erasure_probability,
            'phase_rad': accumulated_phase
        }


def main():
    # Physical parameter configuration (Scaled in MHz / microseconds)
    omega_max = 2.0 * np.pi * 12.0      # 12 MHz Rabi frequency
    v_blockade = 2.0 * np.pi * 180.0    # 180 MHz strong blockade limit
    gamma_rydberg = 1.0 / 150.0         # Lifetime tau = 150 microseconds

    # Target gate execution time for 2pi-equivalent pulse
    gate_duration = 0.22  # microseconds

    sim = RydbergGateSimulator(
        omega_max=omega_max, 
        v_blockade=v_blockade, 
        gamma_rydberg=gamma_rydberg
    )

    results = sim.run_simulation(gate_duration=gate_duration, initial_state_idx=0)

    print("==========================================================")
    print("NEUTRAL ATOM RYDBERG CZ GATE SIMULATION RESULTS")
    print("==========================================================")
    print(f"Rabi Frequency (Omega / 2pi)   : {omega_max / (2*np.pi):.2f} MHz")
    print(f"Blockade Energy (V_rr / 2pi)   : {v_blockade / (2*np.pi):.2f} MHz")
    print(f"Gate Duration                  : {gate_duration:.3f} us")
    print(f"State Population |11> (Final)  : {np.abs(results['final_state'][0])**2:.5f}")
    print(f"Accumulated Phase on |11>      : {results['phase_rad']:.4f} rad (Target: ~3.1416 rad)")
    print(f"Norm Survival Probability      : {results['norm_survival']:.6f}")
    print(f"Converted Erasure Error Prob   : {results['erasure_prob'] * 100:.4f} %")
    print("==========================================================")


if __name__ == "__main__":
    main()
quantum_snippet.py
#!/usr/bin/env python3
"""
Rydberg Blockade & Levine-Pillar CZ Gate Simulator
===================================================
Models two-atom state dynamics under laser excitation and Van der Waals interaction.
Calculates state populations, computational leakage, and process fidelity.
"""

import numpy as np
from scipy.linalg import expm
from typing import Dict, Tuple, Any


class RydbergBlockadeSimulator:
    """
    Simulates two-qubit Rydberg gate dynamics for neutral-atom quantum processors.
    
    Hilbert Space Representation (9 basis states):
        0: |00>, 1: |01>, 2: |0r>
        3: |10>, 4: |11>, 5: |1r>
        6: |r0>, 7: |r1>, 8: |rr>
    """
    
    def __init__(self, omega_mhz: float = 10.0, c6_ghz_um6: float = 50.0):
        """
        Initialize simulator parameters.
        
        Args:
            omega_mhz: Laser Rabi frequency in MHz (2*pi * MHz)
            c6_ghz_um6: Van der Waals dispersion coefficient C6 in GHz * um^6
        """
        self.omega = 2.0 * np.pi * omega_mhz  # rad / microsecond
        self.c6 = 2.0 * np.pi * c6_ghz_um6 * 1e3  # rad / microsecond * um^6
        
    def v_vdw(self, distance_um: float) -> float:
        """Computes Van der Waals interaction strength V_vdw at distance R."""
        return self.c6 / (distance_um ** 6)

    def build_hamiltonian(
        self, 
        omega1: float, 
        omega2: float, 
        delta1: float, 
        delta2: float, 
        phase1: float, 
        phase2: float, 
        v_interaction: float
    ) -> np.ndarray:
        """
        Constructs the 9x9 matrix Hamiltonian for two driven 3-level atoms (|0>, |1>, |r>).
        """
        w1 = omega1 * np.exp(1j * phase1)
        w2 = omega2 * np.exp(1j * phase2)
        
        H = np.zeros((9, 9), dtype=complex)
        
        # Single-atom detunings for state |r>
        H[2, 2] = -delta2
        H[5, 5] = -delta2
        H[6, 6] = -delta1
        H[7, 7] = -delta1
        H[8, 8] = -(delta1 + delta2) + v_interaction  # Shifted double-Rydberg state
        
        # Atom 1 Laser Couplings (|1> <-> |r>)
        H[3, 6], H[6, 3] = 0.5 * w1, 0.5 * np.conj(w1)  # |10> <-> |r0>
        H[4, 7], H[7, 4] = 0.5 * w1, 0.5 * np.conj(w1)  # |11> <-> |r1>
        H[5, 8], H[8, 5] = 0.5 * w1, 0.5 * np.conj(w1)  # |1r> <-> |rr>
        
        # Atom 2 Laser Couplings (|1> <-> |r>)
        H[1, 2], H[2, 1] = 0.5 * w2, 0.5 * np.conj(w2)  # |01> <-> |0r>
        H[4, 5], H[5, 4] = 0.5 * w2, 0.5 * np.conj(w2)  # |11> <-> |1r>
        H[7, 8], H[8, 7] = 0.5 * w2, 0.5 * np.conj(w2)  # |r1> <-> |rr>
        
        return H

    def simulate_cz_gate(self, distance_um: float) -> Dict[str, Any]:
        """
        Executes the Levine-Pillar two-pulse CZ gate sequence.
        
        Args:
            distance_um: Separation distance between atoms in micrometers.
            
        Returns:
            Dictionary containing process metrics (fidelity, leakage, phase).
        """
        v_int = self.v_vdw(distance_um)
        
        # Levine-Pillar optimal pulse parameters
        delta = 0.377371 * self.omega
        tau = 4.29268 / self.omega
        xi = 3.90242  # Laser phase shift for second pulse
        
        # Pulse 1: Phase = 0
        H1 = self.build_hamiltonian(self.omega, self.omega, delta, delta, 0.0, 0.0, v_int)
        U1 = expm(-1j * H1 * tau)
        
        # Pulse 2: Phase = xi
        H2 = self.build_hamiltonian(self.omega, self.omega, delta, delta, xi, xi, v_int)
        U2 = expm(-1j * H2 * tau)
        
        # Total unitary time evolution
        U_total = U2 @ U1
        
        # Extract 4x4 computational subspace (|00>, |01>, |10>, |11>)
        comp_idx = [0, 1, 3, 4]
        U_comp = U_total[np.ix_(comp_idx, comp_idx)]
        
        # Measure computational subspace loss (leakage to Rydberg states)
        leakage = float(1.0 - np.mean(np.sum(np.abs(U_comp)**2, axis=0)))
        
        # Extract phase accumulation
        phases = np.angle(np.diag(U_comp))
        cond_phase = float((phases[3] - phases[2] - phases[1] + phases[0]) % (2 * np.pi))
        
        # Compensate single-qubit phase shifts via virtual Rz rotations
        Rz_corr = np.diag([
            np.exp(-1j * phases[0]),
            np.exp(-1j * phases[1]),
            np.exp(-1j * phases[2]),
            np.exp(-1j * (phases[1] + phases[2] - phases[0]))
        ])
        U_corrected = Rz_corr @ U_comp
        
        # Ideal Target CZ Gate
        U_target = np.diag([1.0, 1.0, 1.0, -1.0])
        
        # Compute Process Fidelity F = |Tr(U_target^dag * U_corrected)|^2 / 16
        fidelity = float(np.abs(np.trace(U_target.conj().T @ U_corrected))**2 / 16.0)
        
        return {
            "distance_um": distance_um,
            "v_vdw_mhz": v_int / (2.0 * np.pi),
            "v_omega_ratio": v_int / self.omega,
            "cond_phase_rad": cond_phase,
            "gate_fidelity": fidelity,
            "subspace_leakage": leakage
        }


def main():
    """Run parameter sweep across atom separations and print technical report."""
    sim = RydbergBlockadeSimulator(omega_mhz=10.0, c6_ghz_um6=50.0)
    
    print("=" * 82)
    print(f"{'NEUTRAL-ATOM RYDBERG CZ GATE DYNAMICS SIMULATION':^82}")
    print("=" * 82)
    print(f"Rabi Frequency (Omega): 2pi x 10.0 MHz | C6 Coefficient: 50.0 GHz*um^6\n")
    
    header = f"{'Dist (um)':<10}{'V_vdw/Omega':<14}{'V_vdw (MHz)':<14}{'Cond Phase':<14}{'Fidelity':<14}{'Leakage':<12}"
    print(header)
    print("-" * 82)
    
    distances = [6.0, 4.5, 3.5, 2.8, 2.2]
    for r in distances:
        res = sim.simulate_cz_gate(r)
        print(
            f"{res['distance_um']:<10.2f}"
            f"{res['v_omega_ratio']:<14.1f}"
            f"{res['v_vdw_mhz']:<14.1f}"
            f"{res['cond_phase_rad']:<14.4f}"
            f"{res['gate_fidelity']:<14.5f}"
            f"{res['subspace_leakage']:<12.2e}"
        )
    print("=" * 82)

if __name__ == "__main__":
    main()
#Neutral-Atom Quantum Computing #Rydberg Blockade #Quantum Error Correction
quantum_snippet.py
"""
Rydberg Blockade Two-Qubit CZ Gate Simulator.
Models two 3-level neutral atoms (|0>, |1>, |r>) interacting via van der Waals forces.
Author: Senior Quantum Computing Engineer
"""

import numpy as np
from scipy.linalg import expm
from dataclasses import dataclass
from typing import Dict, Tuple, List


@dataclass(frozen=True)
class SimulationResult:
    distance_um: float
    interaction_v_mhz: float
    fidelity: float
    conditional_phase_rad: float
    subspace_leakage: float
    unitary_computational: np.ndarray


class RydbergGateSimulator:
    """
    Simulates the microscopic Hamiltonian dynamics of a 2-atom neutral atom system
    executing a controlled-phase (CZ) gate via Rydberg blockade.
    
    Hilbert Space (9 states):
    Index | State
    0     | |00>
    1     | |01>
    2     | |0r>
    3     | |10>
    4     | |11>
    5     | |1r>
    6     | |r0>
    7     | |r1>
    8     | |rr>
    """

    def __init__(self, c6_ghz_um6: float = 862.0, rabi_freq_mhz: float = 5.0):
        """
        Initialize the simulator with physical atomic properties.
        
        Args:
            c6_ghz_um6: van der Waals C6 coefficient in GHz * um^6 (e.g., 862 GHz*um^6 for Rb87 70S).
            rabi_freq_mhz: Laser coupling Rabi frequency Omega in MHz.
        """
        self.c6_mhz_um6: float = c6_ghz_um6 * 1e3  # Convert to MHz * um^6
        self.omega_rad_us: float = 2.0 * np.pi * rabi_freq_mhz  # Convert to rad/us (Angular MHz)
        
        # Define 9-dimensional state basis
        self.levels: List[str] = ['0', '1', 'r']
        self.basis: List[Tuple[str, str]] = [(s1, s2) for s1 in self.levels for s2 in self.levels]
        self.state_to_idx: Dict[Tuple[str, str], int] = {s: i for i, s in enumerate(self.basis)}
        
        # Computational subspace indices (|00>, |01>, |10>, |11>)
        self.comp_states: List[Tuple[str, str]] = [('0', '0'), ('0', '1'), ('1', '0'), ('1', '1')]
        self.comp_indices: List[int] = [self.state_to_idx[s] for s in self.comp_states]

    @property
    def blockade_radius_um(self) -> float:
        """Calculate the theoretical blockade radius Rb = (C6 / Omega)^(1/6) in micrometers."""
        return (self.c6_mhz_um6 / (self.omega_rad_us / (2.0 * np.pi))) ** (1.0 / 6.0)

    def interaction_energy_mhz(self, distance_um: float) -> float:
        """Calculates V(R) = C6 / R^6 in linear frequency (MHz)."""
        if distance_um <= 0:
            raise ValueError("Distance must be strictly positive.")
        return self.c6_mhz_um6 / (distance_um ** 6)

    def build_hamiltonian(
        self,
        omega1: float,
        phi1: float,
        delta1: float,
        omega2: float,
        phi2: float,
        delta2: float,
        distance_um: float
    ) -> np.ndarray:
        """
        Constructs the 9x9 Hamiltonian matrix in the rotating frame under the RWA.
        All parameters are in angular units (rad/us).
        """
        h_matrix = np.zeros((9, 9), dtype=complex)
        v_rad_us = 2.0 * np.pi * self.interaction_energy_mhz(distance_um)

        for i, (s1, s2) in enumerate(self.basis):
            # Diagonal: Laser detunings
            if s1 == 'r':
                h_matrix[i, i] -= delta1
            if s2 == 'r':
                h_matrix[i, i] -= delta2
                
            # Diagonal: Rydberg van der Waals interaction
            if s1 == 'r' and s2 == 'r':
                h_matrix[i, i] += v_rad_us

            # Off-diagonal: Atom 1 optical drive (|1> <-> |r>)
            if s1 == '1':
                target_idx = self.state_to_idx[('r', s2)]
                coupling = 0.5 * omega1 * np.exp(1j * phi1)
                h_matrix[target_idx, i] += coupling
                h_matrix[i, target_idx] += np.conj(coupling)

            # Off-diagonal: Atom 2 optical drive (|1> <-> |r>)
            if s2 == '1':
                target_idx = self.state_to_idx[(s1, 'r')]
                coupling = 0.5 * omega2 * np.exp(1j * phi2)
                h_matrix[target_idx, i] += coupling
                h_matrix[i, target_idx] += np.conj(coupling)

        return h_matrix

    def execute_jcz_cz_gate(self, distance_um: float) -> np.ndarray:
        """
        Simulates the 3-pulse Jaksch-Cirac-Zoller protocol:
          Pulse 1: pi-pulse on Atom 1 (Omega1 = Omega, Omega2 = 0)
          Pulse 2: 2pi-pulse on Atom 2 (Omega1 = 0, Omega2 = Omega)
          Pulse 3: pi-pulse on Atom 1 with pi-phase shift (Omega1 = Omega, phase = pi)
          
        Returns:
            U_comp: 4x4 unitary matrix in the computational basis {|00>, |01>, |10>, |11>}.
        """
        t_pi = np.pi / self.omega_rad_us
        t_2pi = 2.0 * np.pi / self.omega_rad_us

        # Pulse 1: pi pulse on Atom 1
        h1 = self.build_hamiltonian(
            omega1=self.omega_rad_us, phi1=0.0, delta1=0.0,
            omega2=0.0, phi2=0.0, delta2=0.0,
            distance_um=distance_um
        )
        u1 = expm(-1j * h1 * t_pi)

        # Pulse 2: 2pi pulse on Atom 2
        h2 = self.build_hamiltonian(
            omega1=0.0, phi1=0.0, delta1=0.0,
            omega2=self.omega_rad_us, phi2=0.0, delta2=0.0,
            distance_um=distance_um
        )
        u2 = expm(-1j * h2 * t_2pi)

        # Pulse 3: pi pulse on Atom 1 with pi phase shift
        h3 = self.build_hamiltonian(
            omega1=self.omega_rad_us, phi1=np.pi, delta1=0.0,
            omega2=0.0, phi2=0.0, delta2=0.0,
            distance_um=distance_um
        )
        u3 = expm(-1j * h3 * t_pi)

        # Total unitary propagator across all 9 levels
        u_total = u3 @ u2 @ u1

        # Project into 4x4 computational subspace
        u_comp = u_total[np.ix_(self.comp_indices, self.comp_indices)]
        return u_comp

    def evaluate_performance(self, distance_um: float) -> SimulationResult:
        """
        Evaluates gate fidelity, subspace leakage, and conditional phase accumulation.
        """
        u_comp = self.execute_jcz_cz_gate(distance_um)
        diag = np.diag(u_comp)
        phases = np.angle(diag)

        # Conditional phase: phi_cond = phi_00 - phi_01 - phi_10 + phi_11
        phi_cond = (phases[0] - phases[1] - phases[2] + phases[3]) % (2.0 * np.pi)

        # Target CZ matrix: diag(1, 1, 1, -1)
        target_cz = np.diag([1.0, 1.0, 1.0, -1.0])

        # Account for single-qubit Z-rotations / global phases
        phase_corr = np.diag([
            np.exp(-1j * phases[0]),
            np.exp(-1j * phases[1]),
            np.exp(-1j * phases[2]),
            np.exp(-1j * (phases[1] + phases[2] - phases[0]))
        ])
        u_corrected = phase_corr @ u_comp

        # Gate fidelity via Hilbert-Schmidt inner product
        fidelity = float(np.abs(np.trace(target_cz.conj().T @ u_corrected)) ** 2 / 16.0)

        # Subspace population retention
        pop_retained = np.sum(np.abs(u_comp) ** 2, axis=0)
        subspace_leakage = float(1.0 - np.mean(pop_retained))

        return SimulationResult(
            distance_um=distance_um,
            interaction_v_mhz=self.interaction_energy_mhz(distance_um),
            fidelity=fidelity,
            conditional_phase_rad=phi_cond,
            subspace_leakage=subspace_leakage,
            unitary_computational=u_comp
        )


def main():
    print("=" * 78)
    print(" NEUTRAL-ATOM RYDBERG CZ GATE SIMULATION (JAKSCH PROTOCOL) ")
    print("=" * 78)

    simulator = RydbergGateSimulator(c6_ghz_um6=862.0, rabi_freq_mhz=5.0)
    rb = simulator.blockade_radius_um
    print(f"[+] Computed Rydberg Blockade Radius (Rb): {rb:.3f} um\n")

    test_distances = [2.5, 3.0, 4.0, 6.0, 8.0, 15.0]
    print(f"{'Distance (um)':<14}{'V(R)/Omega':<14}{'Cond. Phase (pi)':<18}{'Leakage':<14}{'Fidelity (%)':<12}")
    print("-" * 78)

    for dist in test_distances:
        res = simulator.evaluate_performance(dist)
        ratio_v_omega = res.interaction_v_mhz / 5.0
        phase_in_pi = res.conditional_phase_rad / np.pi
        print(
            f"{res.distance_um:<14.2f}"
            f"{ratio_v_omega:<14.2f}"
            f"{phase_in_pi:<18.4f}"
            f"{res.subspace_leakage:<14.2e}"
            f"{res.fidelity * 100.0:<12.4f}"
        )

    print("\n[+] Verification Complete: Within the blockade regime (R < 3.0 um),")
    print("    the interaction shifts |rr> out of resonance, yielding >99.99% CZ gate fidelity.")


if __name__ == "__main__":
    main()
#Quantum Computing #Neutral Atoms #Quantum Error Correction
1. Detunings
"""
Neutral Atom 2-Qubit Rydberg Blockade CZ Gate Simulator
======================================================
Simulates the 9-dimensional Hilbert space of two three-level atoms 
(|0>, |1>, |r>) interacting via van der Waals Rydberg blockade under 
a symmetric two-pulse Levine-Pichler protocol.

Author: Senior Quantum Systems Engineer
"""

import numpy as np
from scipy.linalg import expm
from dataclasses import dataclass
from typing import Dict, Tuple, List


@dataclass
class SimulationParameters:
    omega_mhz: float = 4.0          # Rydberg Rabi frequency Omega / 2pi (MHz)
    c6_ghz_um6: float = 5000.0      # van der Waals coefficient C6 (GHz * um^6)
    interatomic_dist_um: float = 3.0 # Distance between the two atoms (um)


class RydbergCZSimulator:
    def __init__(self, params: SimulationParameters):
        self.params = params
        
        # Convert to angular frequencies in units of rad / microsecond
        self.omega = 2.0 * np.pi * params.omega_mhz
        
        # Interaction energy V = C6 / R^6 in rad / us
        c6_rad_us = 2.0 * np.pi * (params.c6_ghz_um6 * 1e3)
        self.v_rr = c6_rad_us / (params.interatomic_dist_um ** 6)
        
        # Blockade radius: R_b = (C6 / Omega)^(1/6)
        self.r_blockade = (c6_rad_us / self.omega) ** (1.0 / 6.0)
        
        # Analytical Levine-Pichler optimal parameters for symmetric 2-pulse CZ
        self.delta = 0.37737095 * self.omega
        self.tau = 4.29268182 / self.omega
        self.xi = 2.38076312  # Laser phase shift on pulse 2 (radians)
        
        # 9-dimensional Hilbert basis: |atom1, atom2>
        self.basis: List[str] = ['00', '01', '0r', '10', '11', '1r', 'r0', 'r1', 'rr']
        self.s2i: Dict[str, int] = {s: i for i, s in enumerate(self.basis)}
        self.comp_states: List[str] = ['00', '01', '10', '11']
        self.comp_indices: List[int] = [self.s2i[s] for s in self.comp_states]

    def build_hamiltonian(self, laser_phase: float) -> np.ndarray:
        """
        Constructs the 9x9 rotating-frame Hamiltonian:
        H = -Delta sum(|r><1| + h.c.)
        """
        h = np.zeros((9, 9), dtype=complex)
        
        # 1. Detunings
        h[self.s2i['0r'], self.s2i['0r']] = -self.delta
        h[self.s2i['1r'], self.s2i['1r']] = -self.delta
        h[self.s2i['r0'], self.s2i['r0']] = -self.delta
        h[self.s2i['r1'], self.s2i['r1']] = -self.delta
        h[self.s2i['rr'], self.s2i['rr']] = -2.0 * self.delta + self.v_rr
        
        # 2. Laser Drive Couplings: (Omega / 2) * exp(i * phi) |r><1| + h.c.
        coupling = 0.5 * self.omega * np.exp(1j * laser_phase)
        
        # Atom 1 transitions: (|10> <-> |r0>), (|11> <-> |r1>), (|1r> <-> |rr>)
        atom1_transitions = [('10', 'r0'), ('11', 'r1'), ('1r', 'rr')]
        for g_state, r_state in atom1_transitions:
            ig, ir = self.s2i[g_state], self.s2i[r_state]
            h[ir, ig] += coupling
            h[ig, ir] += np.conj(coupling)
            
        # Atom 2 transitions: (|01> <-> |0r>), (|11> <-> |1r>), (|r1> <-> |rr>)
        atom2_transitions = [('01', '0r'), ('11', '1r'), ('r1', 'rr')]
        for g_state, r_state in atom2_transitions:
            ig, ir = self.s2i[g_state], self.s2i[r_state]
            h[ir, ig] += coupling
            h[ig, ir] += np.conj(coupling)
            
        return h

    def run_simulation(self) -> Dict[str, any]:
        """
        Simulates the two-pulse sequence and extracts the computational unitary.
        """
        # Pulse 1 (duration tau, phase 0)
        h_pulse1 = self.build_hamiltonian(laser_phase=0.0)
        u1 = expm(-1j * h_pulse1 * self.tau)
        
        # Pulse 2 (duration tau, phase xi)
        h_pulse2 = self.build_hamiltonian(laser_phase=self.xi)
        u2 = expm(-1j * h_pulse2 * self.tau)
        
        # Total full-space unitary
        u_full = u2 @ u1
        
        # Project onto 4x4 computational subspace
        u_comp = u_full[np.ix_(self.comp_indices, self.comp_indices)]
        
        # Single-qubit phase compensation:
        # Single atoms in |1> accumulate phase phi_1 = angle(U_comp[01, 01])
        phi_1 = np.angle(u_comp[1, 1])
        phi_2 = np.angle(u_comp[3, 3])
        
        # Phase correction operator: Rz(-phi_1) on both atoms
        z_correction = np.diag([
            1.0,
            np.exp(-1j * phi_1),
            np.exp(-1j * phi_1),
            np.exp(-2j * phi_1)
        ])
        
        u_corrected = z_correction @ u_comp
        
        # Ideal Controlled-Z operator
        cz_ideal = np.diag([1.0, 1.0, 1.0, -1.0])
        
        # Calculate process/average gate fidelity
        overlap = np.trace(cz_ideal.conj().T @ u_corrected)
        gate_fidelity = (np.abs(overlap)**2 + 4.0) / 20.0
        
        # Calculate conditional phase
        conditional_phase = (phi_2 - 2.0 * phi_1) % (2.0 * np.pi)
        
        return {
            "r_blockade_um": self.r_blockade,
            "v_rr_mhz": self.v_rr / (2.0 * np.pi),
            "total_gate_time_ns": 2.0 * self.tau * 1e3,
            "conditional_phase_rad": conditional_phase,
            "gate_fidelity": gate_fidelity,
            "u_corrected": u_corrected
        }


def main():
    params = SimulationParameters(
        omega_mhz=4.0,           # 4 MHz Rabi drive
        c6_ghz_um6=5000.0,       # Rubidium/Ytterbium Rydberg coefficient
        interatomic_dist_um=3.0  # 3.0 um separation (deep in blockade)
    )
    
    sim = RydbergCZSimulator(params)
    results = sim.run_simulation()
    
    print("=" * 65)
    print("     NEUTRAL ATOM RYDBERG BLOCKADE CZ GATE SIMULATION")
    print("=" * 65)
    print(f"Interatomic Distance:           {params.interatomic_dist_um:.2f} um")
    print(f"Rydberg Blockade Radius (Rb):   {results['r_blockade_um']:.2f} um")
    print(f"Rydberg Interaction Energy:     {results['v_rr_mhz']:.2f} MHz")
    print(f"Total Gate Duration (2*tau):    {results['total_gate_time_ns']:.2f} ns")
    print("-" * 65)
    print(f"Conditional Phase (phi_cond):   {results['conditional_phase_rad']:.6f} rad (Target: {np.pi:.6f})")
    print(f"Phase Error vs Ideal (pi):      {abs(results['conditional_phase_rad'] - np.pi):.6e} rad")
    print(f"Entangling Gate Fidelity:       {results['gate_fidelity'] * 100:.5f}%")
    print("-" * 65)
    print("Corrected Unitary Matrix in Computational Basis (|00>, |01>, |10>, |11>):")
    
    u_matrix = results["u_corrected"]
    for row in range(4):
        row_str = "  ".join([f"{u_matrix[row, col].real:+.4f}{u_matrix[row, col].imag:+.4f}j" for col in range(4)])
        print(f"  [ {row_str} ]")
    print("=" * 65)


if __name__ == "__main__":
    main()
P_L(Pauli) ~ C_p * (p p_th_pauli)^((d+1)2) where p_th ~ 0.01 from Architecting Fault Tolerance: A Deep Dive into Reconfigurable Neutral-Atom Quantum Computers
#Quantum Computing #Neutral Atoms #Quantum Error Correction
P_L(Pauli) ~ C_p * (p p_th_pauli)^((d+1)2) where p_th ~ 0.01
"""
Rydberg Blockade Controlled-Z (CZ) Gate Simulator
Author: Senior Quantum Computing Engineer
Platform: Python 3.12+ (NumPy, SciPy)

This module simulates the two-atom Hamiltonian dynamics during a 
two-pulse Levine-Pichler / Bluvstein CZ gate sequence in neutral atom processors.
"""

from dataclasses import dataclass
import numpy as np
from scipy.linalg import expm


@dataclass(frozen=True)
class RydbergParameters:
    """Physical parameters for neutral-atom gate simulation."""
    rabi_frequency_hz: float = 4.0e6           # Omega / (2*pi) = 4 MHz
    c6_coefficient_hz_um6: float = 860.0e9     # C6 / (2*pi) for Rb-87 |70S> in Hz*um^6
    interatomic_distance_um: float = 3.2       # Distance in micrometers
    detuning_ratio: float = 0.352328           # Delta / Omega (Optimal LP parameter)
    laser_phase_jump_rad: float = 4.195246     # xi (Optimal phase jump between pulses)


class RydbergCZSimulator:
    """Simulates coherent dynamics of a 2-qubit system under Rydberg excitation."""

    def __init__(self, params: RydbergParameters = RydbergParameters()):
        self.params = params
        self.omega = 2.0 * np.pi * params.rabi_frequency_hz
        self.delta = params.detuning_ratio * self.omega
        self.c6 = 2.0 * np.pi * params.c6_coefficient_hz_um6
        self.distance = params.interatomic_distance_um
        self.xi = params.laser_phase_jump_rad

        # Calculate van der Waals interaction shift
        self.v_rydberg = self.c6 / (self.distance ** 6)
        self.blockade_radius = (self.c6 / self.omega) ** (1.0 / 6.0)

        # Pulse duration for one half of the symmetric sequence
        self.omega_eff = np.sqrt(self.omega**2 + self.delta**2)
        self.tau_pulse = (2.0 * np.pi) / self.omega_eff

    def _single_atom_subspace_evolution(self, rabi: float, phase: float, duration: float) -> np.ndarray:
        """
        Evolves a 2-level subspace {|1>, |r>} driven by laser field:
        H = [[0, (Omega/2)*e^(-i*phi)], [(Omega/2)*e^(i*phi), -Delta]]
        """
        h_matrix = np.array([
            [0.0, 0.5 * rabi * np.exp(-1j * phase)],
            [0.5 * rabi * np.exp(1j * phase), -self.delta]
        ], dtype=complex)
        return expm(-1j * h_matrix * duration)

    def run_two_pulse_sequence(self) -> dict:
        """
        Executes the two-pulse Levine-Pichler protocol across all computational basis states.
        
        Subspaces:
        1. |00> : Invariant (Energy = 0) -> Phase = 0
        2. |01>, |10> : Single-atom driving with Rabi frequency Omega
        3. |11> : Collective driving with Rabi frequency sqrt(2)*Omega (Blockade limit)
        """
        # Step 1: Single-atom subspace {|1>, |r>} evolution (Pulse 1: phi=0, Pulse 2: phi=xi)
        u1_single = self._single_atom_subspace_evolution(self.omega, 0.0, self.tau_pulse)
        u2_single = self._single_atom_subspace_evolution(self.omega, self.xi, self.tau_pulse)
        u_single_total = u2_single @ u1_single

        # Step 2: Two-atom subspace {|11>, |W>} evolution under blockade
        u1_pair = self._single_atom_subspace_evolution(np.sqrt(2.0) * self.omega, 0.0, self.tau_pulse)
        u2_pair = self._single_atom_subspace_evolution(np.sqrt(2.0) * self.omega, self.xi, self.tau_pulse)
        u_pair_total = u2_pair @ u1_pair

        # Population retention on computational state |1> and |11>
        amp_1 = u_single_total[0, 0]
        amp_11 = u_pair_total[0, 0]

        phase_1 = np.angle(amp_1)
        phase_11 = np.angle(amp_11)

        # Net entangling phase after single-qubit Z-rotations: phi_CZ = phi_11 - 2*phi_1
        phi_cz = (phase_11 - 2.0 * phase_1) % (2.0 * np.pi)

        # Construct full diagonal computational unitary
        # Applying single-qubit counter-rotations R_z(-phase_1) to each qubit:
        u_comp = np.diag([
            1.0 + 0.0j,
            amp_1 * np.exp(-1j * phase_1),
            amp_1 * np.exp(-1j * phase_1),
            amp_11 * np.exp(-2j * phase_1)
        ])

        # Target ideal Controlled-Z matrix
        u_target = np.diag([1.0, 1.0, 1.0, -1.0])

        # Process Fidelity: F = |Tr(U_target^dagger @ U_comp)|^2 / (d^2) with d=4
        trace_overlap = np.trace(u_target.conj().T @ u_comp)
        process_fidelity = float(np.abs(trace_overlap)**2 / 16.0)

        return {
            "blockade_radius_um": self.blockade_radius,
            "interaction_shift_mhz": self.v_rydberg / (2.0 * np.pi * 1e6),
            "pulse_duration_ns": self.tau_pulse * 1e9,
            "total_gate_time_ns": 2.0 * self.tau_pulse * 1e9,
            "single_atom_retention": float(np.abs(amp_1)**2),
            "pair_retention": float(np.abs(amp_11)**2),
            "entangling_phase_rad": phi_cz,
            "process_fidelity": process_fidelity,
            "effective_unitary": np.round(u_comp, 4)
        }


def erasure_threshold_comparison():
    """
    Demonstrates analytical code performance under standard Pauli vs. Erasure noise channels.
    """
    p_errors = np.linspace(0.001, 0.05, 10)
    
    print("\n=======================================================")
    print("      QUANTUM ERROR SUPPRESSION (DISTANCE d=5 CODE)   ")
    print("=======================================================")
    print(f"{'Physical Error (p)':<20} | {'Logical Pauli P_L':<18} | {'Logical Erasure P_L':<18}")
    print("-" * 62)

    # Scaling approximations for distance d=5 surface code:
    # P_L(Pauli) ~ C_p * (p / p_th_pauli)^((d+1)/2)  where p_th ~ 0.01
    # P_L(Erasure) ~ C_e * (p / p_th_erasure)^d      where p_th ~ 0.05
    for p in p_errors:
        p_l_pauli = min(1.0, 0.1 * (p / 0.010) ** 3)
        p_l_erasure = min(1.0, 0.1 * (p / 0.050) ** 5)
        print(f"{p:<20.4f} | {p_l_pauli:<18.6e} | {p_l_erasure:<18.6e}")
    print("=======================================================\n")


if __name__ == "__main__":
    sim = RydbergCZSimulator()
    results = sim.run_two_pulse_sequence()

    print("\n=======================================================")
    print("     RYDBERG LEVINE-PICHLER CONTROLLED-Z SIMULATION    ")
    print("=======================================================")
    print(f"Blockade Radius (R_b)       : {results['blockade_radius_um']:.3f} um")
    print(f"Interatomic Distance        : {sim.distance:.3f} um")
    print(f"Interaction Shift V/(2*pi)  : {results['interaction_shift_mhz']:.2f} MHz")
    print(f"Single Pulse Duration (tau) : {results['pulse_duration_ns']:.2f} ns")
    print(f"Total Two-Pulse Gate Time   : {results['total_gate_time_ns']:.2f} ns")
    print(f"Single-Qubit Retention |1>  : {results['single_atom_retention']:.6f}")
    print(f"Two-Qubit Retention |11>    : {results['pair_retention']:.6f}")
    print(f"Entangling Phase Delta_Phi  : {results['entangling_phase_rad']:.5f} rad (Target: {np.pi:.5f})")
    print(f"Calculated CZ Gate Fidelity : {results['process_fidelity'] * 100.0:.4f}%")
    print("\nReconstructed Unitary Matrix (Computational Subspace):")
    print(results['effective_unitary'])

    erasure_threshold_comparison()
#Neutral-Atom Quantum Computing #Rydberg Blockade #Quantum Error Correction
1. Single-atom driving on Atom 1 (|1> <-> |r>)
#!/usr/bin/env python3
"""
Rydberg Blockade Controlled-Z (CZ) Gate Simulator
Author: Senior Quantum Computing Architect
Description: Numerically integrates the 9-state time-dependent Hamiltonian
             for two neutral atoms undergoing a Levine-Pichler Rydberg CZ gate.
"""

from typing import Dict, List, Tuple
import numpy as np
from scipy.linalg import expm

# ==============================================================================
# HILBERT SPACE DEFINITION
# ==============================================================================
SINGLE_ATOM_STATES = ['0', '1', 'r']
TWO_ATOM_BASIS: List[Tuple[str, str]] = [
    (s1, s2) for s1 in SINGLE_ATOM_STATES for s2 in SINGLE_ATOM_STATES
]
DIM: int = len(TWO_ATOM_BASIS)
STATE_TO_IDX: Dict[Tuple[str, str], int] = {state: i for i, state in enumerate(TWO_ATOM_BASIS)}
COMP_STATES: List[Tuple[str, str]] = [('0', '0'), ('0', '1'), ('1', '0'), ('1', '1')]
COMP_INDICES: List[int] = [STATE_TO_IDX[s] for s in COMP_STATES]


def construct_hamiltonian(
    omega: float,
    delta: float,
    phi: float,
    v_blockade: float
) -> np.ndarray:
    """
    Constructs the 9x9 Hamiltonian for two neutral atoms driven by a Rydberg laser.

    Parameters:
        omega (float): Rabi frequency coupling |1> <-> |r> (rad/s).
        delta (float): Laser detuning Delta = omega_laser - omega_0r (rad/s).
        phi (float): Optical phase of the driving laser field (rad).
        v_blockade (float): van der Waals interaction energy V_vdW = C6 / R^6 (rad/s).

    Returns:
        np.ndarray: 9x9 complex Hermitian Hamiltonian matrix.
    """
    h_mat = np.zeros((DIM, DIM), dtype=np.complex128)

    for idx_u, (s1, s2) in enumerate(TWO_ATOM_BASIS):
        # 1. Single-atom driving on Atom 1 (|1> <-> |r>)
        if s1 == '1':
            target_state = ('r', s2)
            idx_v = STATE_TO_IDX[target_state]
            coupling = 0.5 * omega * np.exp(1j * phi)
            h_mat[idx_v, idx_u] += coupling
            h_mat[idx_u, idx_v] += np.conj(coupling)
        if s1 == 'r':
            h_mat[idx_u, idx_u] -= delta

        # 2. Single-atom driving on Atom 2 (|1> <-> |r>)
        if s2 == '1':
            target_state = (s1, 'r')
            idx_v = STATE_TO_IDX[target_state]
            coupling = 0.5 * omega * np.exp(1j * phi)
            h_mat[idx_v, idx_u] += coupling
            h_mat[idx_u, idx_v] += np.conj(coupling)
        if s2 == 'r':
            h_mat[idx_u, idx_u] -= delta

        # 3. Two-body Rydberg blockade interaction (|r, r>)
        if s1 == 'r' and s2 == 'r':
            h_mat[idx_u, idx_u] += v_blockade

    return h_mat


def simulate_levine_pichler_cz(
    omega_mhz: float = 4.0,
    r_um: float = 2.8,
    c6_ghz_um6: float = 5000.0
) -> Dict[str, object]:
    """
    Simulates the Levine-Pichler CZ gate protocol and evaluates gate metrics.

    Parameters:
        omega_mhz (float): Bare Rabi frequency in MHz.
        r_um (float): Interatomic distance in micrometers.
        c6_ghz_um6 (float): van der Waals C6 coefficient in GHz * um^6.

    Returns:
        Dict[str, object]: Unitary matrix, fidelity, phase shift, and leakage metrics.
    """
    # Unit conversions to angular frequency in rad/microsecond
    omega = 2.0 * np.pi * omega_mhz
    v_blockade = 2.0 * np.pi * (c6_ghz_um6 * 1e3) / (r_um**6)

    # Calibrated Levine-Pichler pulse parameters
    # Optimal detuning ratio Delta / Omega ~ 0.377371
    delta = 0.377371 * omega
    tau = 2.0 * np.pi / np.sqrt(omega**2 + delta**2)
    xi = 3.90242  # Phase jump between pulses in radians

    # Pulse 1: Duration tau, Phase 0.0
    h1 = construct_hamiltonian(omega=omega, delta=-delta, phi=0.0, v_blockade=v_blockade)
    u1 = expm(-1j * h1 * tau)

    # Pulse 2: Duration tau, Phase xi
    h2 = construct_hamiltonian(omega=omega, delta=-delta, phi=xi, v_blockade=v_blockade)
    u2 = expm(-1j * h2 * tau)

    # Full 9x9 Unitary evolution operator
    u_full = u2 @ u1

    # Project into computational subspace {|00>, |01>, |10>, |11>}
    u_comp = u_full[np.ix_(COMP_INDICES, COMP_INDICES)]

    # Compute phase shifts acquired by computational basis states
    diag_elements = np.diag(u_comp)
    raw_phases = np.angle(diag_elements)
    
    # Remove single-qubit Z-rotations: Phi_ij -> Phi_ij - phi_00 - (phi_01 - phi_00) - (phi_10 - phi_00)
    phi_00 = raw_phases[0]
    phi_01 = raw_phases[1]
    phi_10 = raw_phases[2]
    phi_11 = raw_phases[3]
    
    conditional_phase = (phi_11 - phi_01 - phi_10 + phi_00) % (2 * np.pi)

    # Ideal CZ operator
    u_ideal_cz = np.diag([1.0, 1.0, 1.0, -1.0]).astype(np.complex128)
    
    # Correct computational unitary by factoring out local phases
    phase_0 = phi_01 - phi_00
    local_phase_correction = np.diag([
        1.0,
        np.exp(-1j * phase_0),
        np.exp(-1j * phase_0),
        np.exp(-1j * (2 * phase_0))
    ])
    u_corrected = np.exp(-1j * phi_00) * (local_phase_correction @ u_comp)

    # Compute Entangling Gate Metrics
    # 1. State leakage out of computational subspace
    leakage = 1.0 - np.mean(np.sum(np.abs(u_comp)**2, axis=0))
    
    # 2. Average gate fidelity: F_avg = (|Tr(U_ideal^dagger U)|^2 + d) / (d(d + 1))
    d = 4.0
    trace_overlap = np.trace(u_ideal_cz.conj().T @ u_corrected)
    f_avg = (np.abs(trace_overlap)**2 + d) / (d * (d + 1.0))

    return {
        "U_comp_raw": u_comp,
        "U_comp_corrected": u_corrected,
        "Conditional_Phase_pi": conditional_phase / np.pi,
        "Subspace_Leakage": leakage,
        "Average_Gate_Fidelity": f_avg,
        "Blockade_Ratio_V_over_Omega": v_blockade / omega,
    }


if __name__ == "__main__":
    print("=" * 70)
    print("  SIMULATING LEVINE-PICHLER RYDBERG CZ GATE ON NEUTRAL ATOMS")
    print("=" * 70)

    sim_results = simulate_levine_pichler_cz(
        omega_mhz=4.5,
        r_um=2.5,
        c6_ghz_um6=5200.0
    )

    print(f"Blockade Ratio (V_vdW / Omega)   : {sim_results['Blockade_Ratio_V_over_Omega']:.2f}")
    print(f"Acquired Entangling Phase       : {sim_results['Conditional_Phase_pi']:.6f} * pi (Target: 1.000000 * pi)")
    print(f"Computational Subspace Leakage  : {sim_results['Subspace_Leakage']:.4e}")
    print(f"Average Gate Fidelity F_avg     : {sim_results['Average_Gate_Fidelity'] * 100:.4f}%")
    print("-" * 70)
    print("Corrected Computational Unitary Matrix:")
    print(np.round(sim_results["U_comp_corrected"], 4))
    print("=" * 70)
#Quantum Computing #Circuit Knitting #Quasiprobability Decomposition
1. Fundamental Quantum Algebra and Quantum States
"""
Distributed Quantum Computing via Classical Communication
Simulation of Wire Cutting and Quasiprobability Decomposition (QPD).
"""

from dataclasses import dataclass
from typing import List, Tuple
import numpy as np


# =====================================================================
# 1. Fundamental Quantum Algebra and Quantum States
# =====================================================================

I2 = np.eye(2, dtype=np.complex128)
PAULI_X = np.array([[0.0, 1.0], [1.0, 0.0]], dtype=np.complex128)
PAULI_Y = np.array([[0.0, -1.0j], [1.0j, 0.0]], dtype=np.complex128)
PAULI_Z = np.array([[1.0, 0.0], [0.0, -1.0]], dtype=np.complex128)

# Basis Eigenstates
Z_PLUS = np.array([1.0, 0.0], dtype=np.complex128)
Z_MINUS = np.array([0.0, 1.0], dtype=np.complex128)
X_PLUS = (Z_PLUS + Z_MINUS) / np.sqrt(2.0)
X_MINUS = (Z_PLUS - Z_MINUS) / np.sqrt(2.0)
Y_PLUS = (Z_PLUS + 1.0j * Z_MINUS) / np.sqrt(2.0)
Y_MINUS = (Z_PLUS - 1.0j * Z_MINUS) / np.sqrt(2.0)

# Projector Helper
def projector(state: np.ndarray) -> np.ndarray:
    return np.outer(state, np.conjugate(state))


# =====================================================================
# 2. Gate Definitions
# =====================================================================

def rx_gate(theta: float) -> np.ndarray:
    return np.cos(theta / 2.0) * I2 - 1.0j * np.sin(theta / 2.0) * PAULI_X

def ry_gate(theta: float) -> np.ndarray:
    return np.cos(theta / 2.0) * I2 - np.sin(theta / 2.0) * (1.0j * PAULI_Y)

def cnot_gate() -> np.ndarray:
    return np.array(
        [
            [1.0, 0.0, 0.0, 0.0],
            [0.0, 1.0, 0.0, 0.0],
            [0.0, 0.0, 0.0, 1.0],
            [0.0, 0.0, 1.0, 0.0],
        ],
        dtype=np.complex128,
    )


# =====================================================================
# 3. Wire Cutting Quasiprobability Decomposition Frame
# =====================================================================

@dataclass(frozen=True)
class QPDComponent:
    """Represents a single quasi-probability channel element."""
    weight: float
    meas_projector: np.ndarray  # Operator applied to QPU A cut boundary
    prep_state: np.ndarray      # State prepared on QPU B cut boundary


def get_wire_cut_qpd_frame() -> List[QPDComponent]:
    """
    Constructs the complete 10-term quasiprobability decomposition frame
    for an identity wire cut (gamma = 4.0).
    
    Expansion identity:
    I(rho) = Tr(Pz+ rho) Pz+ + Tr(Pz- rho) Pz-
           + 0.5 * sum_{s,s' in {+,-}} s*s' * Tr(Px,s rho) Px,s'
           + 0.5 * sum_{s,s' in {+,-}} s*s' * Tr(Py,s rho) Py,s'
    """
    components = [
        # Z-basis direct projections (weight = +1.0)
        QPDComponent(1.0, projector(Z_PLUS), Z_PLUS),
        QPDComponent(1.0, projector(Z_MINUS), Z_MINUS),
        # X-basis cross terms (weight = +/- 0.5)
        QPDComponent(0.5, projector(X_PLUS), X_PLUS),
        QPDComponent(-0.5, projector(X_PLUS), X_MINUS),
        QPDComponent(-0.5, projector(X_MINUS), X_PLUS),
        QPDComponent(0.5, projector(X_MINUS), X_MINUS),
        # Y-basis cross terms (weight = +/- 0.5)
        QPDComponent(0.5, projector(Y_PLUS), Y_PLUS),
        QPDComponent(-0.5, projector(Y_PLUS), Y_MINUS),
        QPDComponent(-0.5, projector(Y_MINUS), Y_PLUS),
        QPDComponent(0.5, projector(Y_MINUS), Y_MINUS),
    ]
    return components


# =====================================================================
# 4. Monolithic Reference Simulator
# =====================================================================

def execute_monolithic_circuit(
    theta1: float, theta2: float, observable: np.ndarray
) -> float:
    """
    Executes the undivided 2-qubit circuit on a single monolithic processor.
    Q0: |0> -> Ry(theta1) -> Rx(theta2) -> Control CNOT -> Measure
    Q1: |0> -----------------------------> Target  CNOT -> Measure
    """
    # Initial state |00>
    psi_00 = np.kron(Z_PLUS, Z_PLUS)
    
    # Step 1: Ry(theta1) on Q0
    u1 = np.kron(ry_gate(theta1), I2)
    psi_1 = u1 @ psi_00
    
    # Step 2: Rx(theta2) on Q0
    u2 = np.kron(rx_gate(theta2), I2)
    psi_2 = u2 @ psi_1
    
    # Step 3: CNOT(Q0 -> Q1)
    psi_out = cnot_gate() @ psi_2
    
    # Expectation value  = 
    expval = np.real(np.conjugate(psi_out).T @ observable @ psi_out)
    return float(expval)


# =====================================================================
# 5. Distributed QPU Partitions
# =====================================================================

class DistributedQPUCluster:
    """Simulates two isolated QPUs communicating solely via classical channels."""

    @staticmethod
    def run_qpu_a(theta1: float, meas_proj: np.ndarray, shots: int) -> float:
        """
        QPU A: Prepares state, applies Ry(theta1), measures with projector.
        Returns the empirical probability of registering the cut projector.
        """
        psi_init = Z_PLUS
        psi_a = ry_gate(theta1) @ psi_init
        rho_a = projector(psi_a)
        
        # Exact quantum probability of projector match: Tr(M rho)
        prob = np.real(np.trace(meas_proj @ rho_a))
        prob = np.clip(prob, 0.0, 1.0)
        
        # Simulate shot measurement statistics
        counts = np.random.binomial(shots, prob)
        return counts / shots

    @staticmethod
    def run_qpu_b(
        prep_state: np.ndarray,
        theta2: float,
        observable: np.ndarray,
        shots: int,
    ) -> float:
        """
        QPU B: Injects prep_state on Q0, initializes Q1 to |0>,
        applies Rx(theta2) on Q0, executes CNOT, and measures observable.
        """
        psi_b_init = np.kron(prep_state, Z_PLUS)
        u_rx = np.kron(rx_gate(theta2), I2)
        psi_b_mid = u_rx @ psi_b_init
        psi_b_final = cnot_gate() @ psi_b_mid
        
        # Quantum expectation on QPU B
        exact_exp = np.real(
            np.conjugate(psi_b_final).T @ observable @ psi_b_final
        )
        
        # Add finite-shot measurement variance: Var(O) = 1 - ^2
        variance = max(0.0, 1.0 - exact_exp**2)
        shot_noise = np.random.normal(0.0, np.sqrt(variance / shots))
        return float(np.clip(exact_exp + shot_noise, -1.0, 1.0))


# =====================================================================
# 6. Distributed Execution and Classical Post-Processing
# =====================================================================

def execute_distributed_circuit(
    theta1: float,
    theta2: float,
    observable: np.ndarray,
    shots_per_variant: int,
) -> Tuple[float, float]:
    """
    Coordinates distributed quantum execution via LOCC and QPD reconstruction.
    Returns: (exact_reconstructed_expval, empirical_shot_expval)
    """
    frame = get_wire_cut_qpd_frame()
    exact_reconstructed_val = 0.0
    empirical_sampled_val = 0.0

    for comp in frame:
        # 1. Exact Analytical Subcircuit Pipeline
        psi_a = ry_gate(theta1) @ Z_PLUS
        prob_a_exact = np.real(np.trace(comp.meas_projector @ projector(psi_a)))
        
        psi_b = cnot_gate() @ (
            np.kron(rx_gate(theta2), I2) @ np.kron(comp.prep_state, Z_PLUS)
        )
        exp_b_exact = np.real(np.conjugate(psi_b).T @ observable @ psi_b)
        exact_reconstructed_val += comp.weight * prob_a_exact * exp_b_exact

        # 2. Distributed Classical LOCC Execution with Finite Shot Noise
        meas_prob_a = DistributedQPUCluster.run_qpu_a(
            theta1, comp.meas_projector, shots_per_variant
        )
        meas_exp_b = DistributedQPUCluster.run_qpu_b(
            comp.prep_state, theta2, observable, shots_per_variant
        )
        
        # Accumulate weighted classical estimate
        empirical_sampled_val += comp.weight * meas_prob_a * meas_exp_b

    return exact_reconstructed_val, empirical_sampled_val


# =====================================================================
# 7. Verification and Experimentation
# =====================================================================

if __name__ == "__main__":
    np.random.seed(42)

    theta_1 = 0.785398  # pi / 4
    theta_2 = 1.047198  # pi / 3
    shots = 50_000

    observables = [
        ("Z (x) Z", np.kron(PAULI_Z, PAULI_Z)),
        ("X (x) X", np.kron(PAULI_X, PAULI_X)),
        ("Z (x) I", np.kron(PAULI_Z, I2)),
        ("I (x) Z", np.kron(I2, PAULI_Z)),
    ]

    print("=" * 80)
    print("DISTRIBUTED QUANTUM COMPUTING VIA CLASSICAL COMMUNICATION (WIRE CUTTING)")
    print(f"Sampling Overhead per cut (gamma): {sum(abs(c.weight) for c in get_wire_cut_qpd_frame()):.1f}")
    print(f"Shot Multiplier (gamma^2):        {sum(abs(c.weight) for c in get_wire_cut_qpd_frame())**2:.1f}")
    print("=" * 80)
    print(f"{'Observable':<12} | {'Monolithic':<12} | {'QPD Analytical':<15} | {'QPD Distributed (Shots)':<22} | {'Error'}")
    print("-" * 80)

    for name, obs in observables:
        mono_val = execute_monolithic_circuit(theta_1, theta_2, obs)
        exact_qpd, sampled_qpd = execute_distributed_circuit(
            theta_1, theta_2, obs, shots_per_variant=shots
        )
        error = abs(mono_val - sampled_qpd)

        print(
            f"{name:<12} | {mono_val:+11.6f} | {exact_qpd:+14.6f} | "
            f"{sampled_qpd:+21.6f} | {error:.2e}"
        )

    print("=" * 80)
    print("Verification complete: Zero quantum interconnects used.")