Code Examples

Production-ready code for CV cluster state applications

Complete Cluster State Pipeline

Full Example

End-to-end workflow: design graph, optimize phases, analyze tolerance, export for fabrication.

#!/usr/bin/env python3
"""Complete CV cluster state design pipeline."""

import numpy as np
from cv_cluster import ClusterCompiler
from cv_cluster.graph import GraphState
from cv_cluster.analysis import MonteCarloAnalyzer
from cv_cluster.loss import LossSimulator
from cv_cluster.export import GDSExporter

# ============== STEP 1: DESIGN GRAPH ==============
print("Step 1: Defining graph topology...")

# Choose topology based on application
graph = GraphState.hexagonal(7)  # 7-mode hexagonal for MBQC

print(f"  Modes: {graph.num_modes}")
print(f"  Edges: {graph.num_edges}")
print(f"  Connectivity: {graph.avg_degree:.2f}")

# ============== STEP 2: CREATE COMPILER ==============
print("\nStep 2: Initializing compiler...")

compiler = ClusterCompiler(
    graph=graph,
    squeezing_dB=6.0,           # Input squeezed states
    mesh_architecture="clements"  # Interferometer type
)

print(f"  Architecture: Clements mesh")
print(f"  Phase parameters: {compiler.num_phases}")

# ============== STEP 3: OPTIMIZE PHASES ==============
print("\nStep 3: Optimizing phases...")

result = compiler.optimize_phases(
    method="L-BFGS-B",
    tol=1e-8,
    maxiter=1000
)

print(f"  Converged in {result.iterations} iterations")
print(f"  Avg nullifier variance: {result.avg_variance:.4f} SNU")
print(f"  Max nullifier variance: {result.max_variance:.4f} SNU")

# ============== STEP 4: MONTE CARLO ANALYSIS ==============
print("\nStep 4: Running Monte Carlo tolerance analysis...")

mc = MonteCarloAnalyzer(compiler)
mc_result = mc.run(
    num_trials=2000,
    phase_error_std=0.5,     # degrees
    bs_imbalance_std=0.02,   # 2% splitting error
    threshold=1.0             # SNU criterion
)

print(f"  Success rate: {mc_result.success_rate:.1%}")
print(f"  95th percentile variance: {mc_result.percentile_95:.4f} SNU")

# ============== STEP 5: LOSS ANALYSIS ==============
print("\nStep 5: Analyzing loss tolerance...")

loss_sim = LossSimulator(
    compiler,
    loss_per_layer_dB=0.3,
    num_layers=compiler.mesh_depth
)
loss_result = loss_sim.analyze()

print(f"  Total loss: {loss_result.total_loss_dB:.2f} dB")
print(f"  Entangled modes: {loss_result.entangled_count}/{graph.num_modes}")

# ============== STEP 6: EXPORT FOR FABRICATION ==============
print("\nStep 6: Exporting design...")

exporter = GDSExporter(compiler, result.optimal_phases)
exporter.export("hexagonal_cluster.gds")
exporter.export_phases("phase_settings.json")

print("  Generated: hexagonal_cluster.gds")
print("  Generated: phase_settings.json")
print("\n✓ Pipeline complete!")

Nullifier Variance Verification

Verification

Verify cluster state entanglement through nullifier operator measurements.

"""Nullifier-based entanglement verification."""

import numpy as np
from cv_cluster import ClusterCompiler
from cv_cluster.graph import GraphState

def verify_cluster_state(compiler, threshold=1.0):
    """
    Verify entanglement via nullifier variances.

    For an ideal cluster state: δ_j = p_j - Σ_k A_jk x_k → 0
    Entanglement witness: ⟨δ_j²⟩ < 1 SNU for all j
    """
    n = compiler.graph.num_modes
    adj = compiler.graph.adjacency_matrix
    V = compiler.get_covariance_matrix()

    variances = []
    for j in range(n):
        # Build nullifier coefficient vector
        # c = [0...0, 0...0] with c[n+j] = 1 (p_j)
        # and c[k] = -A_jk for all k (x terms)
        c = np.zeros(2 * n)
        c[n + j] = 1  # p_j coefficient
        for k in range(n):
            c[k] = -adj[j, k]  # -A_jk * x_k

        # Variance: ⟨δ_j²⟩ = c^T V c
        var = c @ V @ c
        variances.append(var)

    variances = np.array(variances)

    # Check entanglement criterion
    all_entangled = np.all(variances < threshold)
    avg_var = np.mean(variances)
    max_var = np.max(variances)

    return {
        'variances': variances,
        'average': avg_var,
        'maximum': max_var,
        'all_entangled': all_entangled,
        'entangled_count': np.sum(variances < threshold)
    }

# Example usage
graph = GraphState.square(4)
compiler = ClusterCompiler(graph, squeezing_dB=6.0)
compiler.optimize_phases()

result = verify_cluster_state(compiler)
print(f"Mode variances (SNU):")
for i, v in enumerate(result['variances']):
    status = "✓ entangled" if v < 1 else "✗ separable"
    print(f"  δ_{i}²: {v:.4f} {status}")
print(f"\nCluster valid: {result['all_entangled']}")

Symplectic Covariance Propagation

Theory

Propagate covariance matrix through interferometer using symplectic formalism.

"""Symplectic transformation of Gaussian states."""

import numpy as np

def beamsplitter_symplectic(theta, modes, n_total):
    """2N×2N symplectic matrix for a beamsplitter."""
    S = np.eye(2 * n_total)
    c, s = np.cos(theta), np.sin(theta)
    i, j = modes

    # x quadratures
    S[i, i] = c
    S[i, j] = s
    S[j, i] = -s
    S[j, j] = c

    # p quadratures
    S[n_total+i, n_total+i] = c
    S[n_total+i, n_total+j] = s
    S[n_total+j, n_total+i] = -s
    S[n_total+j, n_total+j] = c

    return S

def phase_shifter_symplectic(phi, mode, n_total):
    """2N×2N symplectic matrix for a phase shifter."""
    S = np.eye(2 * n_total)
    c, s = np.cos(phi), np.sin(phi)

    S[mode, mode] = c
    S[mode, n_total+mode] = s
    S[n_total+mode, mode] = -s
    S[n_total+mode, n_total+mode] = c

    return S

def clements_mesh(phases, n_modes):
    """Build full Clements mesh symplectic matrix."""
    S = np.eye(2 * n_modes)
    idx = 0

    # Triangular arrangement of beamsplitters
    for layer in range(n_modes - 1):
        start = layer % 2
        for i in range(start, n_modes - 1, 2):
            # Phase shifter
            S = phase_shifter_symplectic(phases[idx], i, n_modes) @ S
            idx += 1
            # Beamsplitter
            S = beamsplitter_symplectic(phases[idx], (i, i+1), n_modes) @ S
            idx += 1

    return S

def propagate_covariance(V_in, S):
    """Propagate covariance: V_out = S V_in S^T"""
    return S @ V_in @ S.T

# Example: 4-mode squeezed states through mesh
n = 4
r = 6.0 / (10 * np.log10(np.e) * 2)  # 6 dB squeezing

# Initial squeezed vacuum covariance
V0 = np.diag([np.exp(-2*r)/2]*n + [np.exp(2*r)/2]*n)

# Random phases
num_phases = 2 * (n * (n - 1))
phases = np.random.uniform(0, 2*np.pi, num_phases)

S = clements_mesh(phases, n)
V_out = propagate_covariance(V0, S)

print(f"Output covariance matrix shape: {V_out.shape}")
print(f"Symplecticity check: {np.allclose(np.linalg.det(S), 1.0)}")

CV GHZ State Generation

Application

Generate CV analog of GHZ states using linear cluster rail topology.

"""CV GHZ-type state generation via cluster rail."""

from cv_cluster import ClusterCompiler
from cv_cluster.graph import GraphState
import numpy as np

def create_ghz_cluster(num_modes, squeezing_dB):
    """
    Create GHZ-rail cluster state.

    Linear graph: 0 - 1 - 2 - ... - (N-1)
    Nullifiers: p_0 - x_1, p_1 - x_0 - x_2, ..., p_{N-1} - x_{N-2}
    """
    graph = GraphState.ghz_rail(num_modes)

    compiler = ClusterCompiler(
        graph=graph,
        squeezing_dB=squeezing_dB,
        mesh_architecture="clements"
    )

    # Optimize for minimal nullifier variance
    result = compiler.optimize_phases(tol=1e-10)

    return compiler, result

def measure_ghz_correlations(compiler):
    """Measure X-X and P-P correlations characteristic of GHZ states."""
    V = compiler.get_covariance_matrix()
    n = compiler.graph.num_modes

    # Extract X-X correlations (upper-left block)
    V_xx = V[:n, :n]

    # Extract P-P correlations (lower-right block)
    V_pp = V[n:, n:]

    # Sum variance: Var(X_total) for GHZ
    x_sum_var = np.sum(V_xx)
    p_diff_var = V_pp[0,0] + V_pp[-1,-1] - 2*V_pp[0,-1]

    return {
        'x_correlation_matrix': V_xx,
        'p_correlation_matrix': V_pp,
        'x_sum_variance': x_sum_var,
        'p_diff_variance': p_diff_var
    }

# Generate 8-mode GHZ cluster
compiler, opt_result = create_ghz_cluster(8, squeezing_dB=8.0)

print("8-Mode CV GHZ Cluster State")
print("=" * 40)
print(f"Average nullifier variance: {opt_result.avg_variance:.4f} SNU")
print(f"Optimization iterations: {opt_result.iterations}")

correlations = measure_ghz_correlations(compiler)
print(f"\nGHZ correlations:")
print(f"  Var(ΣX): {correlations['x_sum_variance']:.4f}")
print(f"  Var(P_0-P_7): {correlations['p_diff_variance']:.4f}")

Batch Parameter Sweep

Utility

Sweep design parameters and generate comprehensive performance reports.

"""Batch parameter sweep for design space exploration."""

import numpy as np
import pandas as pd
from itertools import product
from cv_cluster import ClusterCompiler
from cv_cluster.graph import GraphState
from cv_cluster.analysis import MonteCarloAnalyzer

def batch_sweep(topologies, squeezing_range, phase_errors, output_file):
    """
    Sweep over multiple parameters and save results.

    Args:
        topologies: List of graph topology names
        squeezing_range: Array of squeezing values (dB)
        phase_errors: Array of phase error std (degrees)
        output_file: CSV output path
    """
    results = []

    topology_funcs = {
        'square4': lambda: GraphState.square(4),
        'hexagonal7': lambda: GraphState.hexagonal(7),
        'ghz8': lambda: GraphState.ghz_rail(8),
        'square9': lambda: GraphState.square(9)
    }

    total = len(topologies) * len(squeezing_range) * len(phase_errors)
    count = 0

    for topo, sq, pe in product(topologies, squeezing_range, phase_errors):
        count += 1
        print(f"Processing {count}/{total}: {topo}, {sq}dB, σφ={pe}°")

        try:
            graph = topology_funcs[topo]()
            compiler = ClusterCompiler(graph, squeezing_dB=sq)
            opt = compiler.optimize_phases()

            mc = MonteCarloAnalyzer(compiler)
            mc_result = mc.run(
                num_trials=500,
                phase_error_std=pe,
                bs_imbalance_std=0.02
            )

            results.append({
                'topology': topo,
                'num_modes': graph.num_modes,
                'squeezing_dB': sq,
                'phase_error_deg': pe,
                'avg_variance': opt.avg_variance,
                'max_variance': opt.max_variance,
                'success_rate': mc_result.success_rate,
                'p95_variance': mc_result.percentile_95
            })
        except Exception as e:
            print(f"  Error: {e}")

    df = pd.DataFrame(results)
    df.to_csv(output_file, index=False)
    print(f"\nResults saved to {output_file}")
    return df

# Run sweep
df = batch_sweep(
    topologies=['square4', 'hexagonal7', 'ghz8'],
    squeezing_range=np.arange(4, 12, 2),
    phase_errors=np.array([0.1, 0.5, 1.0, 2.0]),
    output_file='cluster_sweep_results.csv'
)

# Summary statistics
print("\nSummary by topology:")
print(df.groupby('topology')['success_rate'].describe())