Classical-communications scheme
The classical-communications directives, together with simple examples, are explained in the Classical Communications section. This section will present a couple of examples are presented to highlight the use of classical communications.
In this first example we present a cyclic exchange of classic bits between 3 circuits:
import os, sys
# In order to import cunqa, we append to the search path the cunqa installation path
sys.path.append(os.getenv("HOME")) # HOME as install path is specific to CESGA
from cunqa.qpu import get_QPUs, qraise, qdrop, run
from cunqa.circuit import CunqaCircuit
from cunqa.qjob import gather
try:
# 1. QPU deployment
NUM_NODES = 3
# If GPU execution is desired, just add "gpu = True" as another qraise argument
family_name = qraise(NUM_NODES,"00:10:00", simulator="Maestro", classical_comm=True, co_located = True)
except Exception as error:
raise error
try:
qpus = get_QPUs(co_located = True, family = family_name)
# 2. Circuit design
# We want to achieve the following scheme:
# --------------------------------------------------------
# ══════════════════
# ‖ ‖
# circuit0.q0: ─────────[X]──────[M]─ ‖
# ‖
# circuit0.q1: ───[H]───[M]───────── ‖
# ‖ ‖
# ‖ ‖
# circuit1.q0: ─────────[X]──────[M]─ ‖
# ‖
# circuit1.q1: ───[H]───[M]────────── ‖
# ‖ ‖
# : ‖
# : ‖
# ‖ ‖
# circuitn.q0: ─────────[X]──────[M]─ ‖
# ‖
# circuitn.q1: ───[H]───[M]────────── ‖
# ‖ ‖
# ══════════════════
# ----------------------------------------------------
classcal_comms_circuits = []
for i in range(NUM_NODES):
circuit = CunqaCircuit(2,2, id = str(i))
# Here we prepare a superposition state at qubit 1, we measure and send its result to the next circuit
circuit.h(1)
circuit.measure(1,1)
circuit.send(1, recving_circuit = str(i+1) if (i+1) != NUM_NODES else str(0))
# Here we recieve the bit sent by the prior circuit and use it for conditioning an x gate at qubit 0
circuit.recv(0, sending_circuit = str(i-1) if (i-1) != -1 else str(NUM_NODES-1))
with circuit.cif(clbits = 0) as cgates:
cgates.x(0)
# Adding final measurement of que qubit after the x gate
circuit.measure(0,0)
classcal_comms_circuits.append(circuit)
# 3. Execution
# The output bitstrings are "switched" in orther, clbit 0 corresponds to the last bit of the bitstring.
# We expect then for the first bit of a circuit's result to be equal to the last of the next circuit.
# If we set the execution to have more shots, this have to be checked as:
#
# If, for circuit0 we have {'(0)0': 5, '10': 4, '11': 1}, we see that there are 5 cases in which the first
# qubit is `0`.
#
# We spect output at circuit1 to have a total of 5 cases in which the second qubit is `0`, therefore:
# {'0(0)': 1, '01': 3, '1(0)': 4, '11': 2} is correct since 1+4 = 5 .
distr_jobs = run(classcal_comms_circuits, qpus, shots=1)
results_list = gather(distr_jobs)
for i, result in enumerate(results_list):
print(f"For circuit {i}: ", result.counts)
# 4. Release classical resources
qdrop(family_name)
except Exception as error:
# 4. Release resources even if an error is raised
qdrop(family_name)
raise error
In this second example we present the distributed iterative QPE in several vQPUs. Compare with the QPE presented in No-communications scheme and Quantum-communications scheme.
"""
Code implementing the Iterative Quantum Phase Estimation (iQPE) algorithm with classical communications. To understand the algorithm without communications check:
- Original paper (here referred to as Iterative Phase Estimation Algorithm): https://arxiv.org/abs/quant-ph/0610214
- TalentQ explanation (in spanish): https://talentq-es.github.io/Fault-Tolerant-Algorithms/docs/Part_01_Fault-tolerant_Algorithms/Chapter_01_01_IPE_portada_myst.html
"""
import os, sys
import time
import numpy as np
# In order to import cunqa, we append to the search path the cunqa installation path
sys.path.append(os.getenv("HOME")) # HOME as install path is specific to CESGA
from cunqa.qpu import get_QPUs, qraise, qdrop, run
from cunqa.circuit import CunqaCircuit
from cunqa.qjob import gather
# Global variables
N_QPUS = 16 # Determines the number of bits of the phase that will be computed
PHASE_TO_COMPUTE = 1/2**5
SHOTS = 1024
SEED = 18 # Set seed for reproducibility
try:
# 1. QPU deployment
family_name = qraise(N_QPUS, "03:00:00", simulator="Aer", classical_comm=True, co_located = True)
except Exception as error:
raise error
try:
qpus = get_QPUs(co_located = True, family = family_name)
# 2. Circuit design: multiple circuits implementing the classically distributed Iterative Phase Estimation
circuits = []
for i in range(N_QPUS):
theta = 2**(N_QPUS - 1 - i) * PHASE_TO_COMPUTE * 2 * np.pi
circuits.append(CunqaCircuit(2, 2, id= f"cc_{i}"))
circuits[i].h(0)
circuits[i].x(1)
circuits[i].crz(theta, 0, 1)
for j in range(i):
param = -np.pi * 2**(-j - 1)
recv_id = i - j - 1
circuits[i].recv(0, sending_circuit = f"cc_{recv_id}")
# Gate conditioned by the received bit
with circuits[i].cif(0) as cgates:
cgates.rz(param, 0)
circuits[i].h(0)
circuits[i].measure(0, 0)
for j in range(N_QPUS - i - 1):
circuits[i].send(0, recving_circuit = f"cc_{i + j + 1}")
circuits[i].measure(1, 1)
# 3. Execution
algorithm_starts = time.time()
distr_jobs = run(circuits, qpus, shots=SHOTS, seed=SEED)
result_list = gather(distr_jobs)
algorithm_ends = time.time()
algorithm_time = algorithm_ends - algorithm_starts
# 4. Post processing results
counts_list = []
for result in result_list:
counts_list.append(result.counts)
binary_string = ""
for counts in counts_list:
# Extract the most frequent measurement (the best estimate of theta)
most_frequent_output = max(counts, key=counts.get)
binary_string += most_frequent_output[1]
estimated_theta = 0.0
for i, digit in enumerate(reversed(binary_string)):
if digit == '1':
estimated_theta += 1 / (2**i)
print(f"Estimated angle: {estimated_theta}")
print(f"Real angle: {PHASE_TO_COMPUTE}")
# 5. Release resources
qdrop(family_name)
except Exception as error:
qdrop(family_name)
raise error