Code Examples
Production-ready code for CV cluster state applications
Complete Cluster State Pipeline
Full ExampleEnd-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
VerificationVerify 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
TheoryPropagate 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
ApplicationGenerate 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
UtilitySweep 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())