DQC Examples

Examples demonstrating simulation of quantum circuit execution across distributed quantum nodes with communication protocols.

DQC QFT

dqc_qft_example.py
  1from typing import Any
  2
  3import numpy as np
  4
  5from kosmos.dqc_scheduling.space_time_matrix import SpaceTimeMatrix
  6from kosmos.partitioning.mqt_bench.bench_circuit import MQTBenchCircuit
  7from kosmos.simulator.dqc_simulator import DQCSimulator
  8from kosmos.topology.link import LinkId, QuantumLink
  9from kosmos.topology.net import Network
 10from kosmos.topology.node import NodeId, NodeRole, QuantumNode
 11
 12
 13def dqc_qft_example() -> None:
 14    """Run a distributed quantum computing simulation with QFT circuit."""
 15    # Create desired network
 16    network = create_dqc_network()
 17
 18    # Define circuit to be executed
 19    bench_circuit = MQTBenchCircuit(circuit_type="qft", num_qubits=4)
 20    circuit = bench_circuit.circuit()
 21
 22    # Set up the Simulator for DQC
 23    # Optional: specify custom entanglement generation protocol and qubit assignment strategy
 24    simulator = DQCSimulator(network=network, seed=1)
 25    simulator.load_circuit(circuit)
 26    simulator.schedule_protocols()
 27
 28    # Run simulation and get results
 29    results = simulator.run()
 30
 31    # Retrieve qubit assignment from simulator
 32    qubit_assignment_matrix = simulator.execution_scheduler.space_time_matrix
 33
 34    # Retrieve measurement results
 35    measurement_counts = results["counts"]
 36    num_qubits = circuit.num_qubits
 37
 38    # Retrieve density matrix
 39    density_matrix = results["density_matrix"]
 40
 41    # Retrieve additional information
 42    additional_simulation_results = [
 43        results["computation_time"],
 44        results["communication_time"],
 45        results["total_time"],
 46        results["num_remote_gates"],
 47        results["num_teleportations"],
 48        results.get("expectation_values"),
 49    ]
 50
 51    # Print retrieved results
 52    print_space_time_matrix(qubit_assignment_matrix)
 53    print_measurement_counts(measurement_counts, num_qubits)
 54    print_density_matrix_diagonal(density_matrix, num_qubits)
 55    print_additional_simulation_results(additional_simulation_results)
 56
 57
 58def create_dqc_network() -> Network:
 59    """Create a quantum network with multiple QPUs.
 60
 61    Returns:
 62        Network: A network with 3 quantum nodes connected by quantum links.
 63
 64    """
 65    network = Network()
 66
 67    qpu_1 = QuantumNode(
 68        id=NodeId("QPU_1"),
 69        roles=[NodeRole.END_USER],
 70        num_qubits=4,
 71        num_data_qubits=3,  # If not provided, defaults to half of num_qubits (rounded down)
 72        coherence_time=100e-6,  # 100 microseconds
 73        gate_fid=0.99,
 74        meas_fid=0.98,
 75        has_transceiver=True,
 76    )
 77    qpu_2 = QuantumNode(
 78        id=NodeId("QPU_2"),
 79        roles=[NodeRole.END_USER],
 80        num_qubits=4,
 81        num_data_qubits=3,
 82        coherence_time=100e-6,
 83        gate_fid=0.99,
 84        meas_fid=0.98,
 85        has_transceiver=True,
 86    )
 87    qpu_3 = QuantumNode(
 88        id=NodeId("QPU_3"),
 89        roles=[NodeRole.END_USER],
 90        num_qubits=5,
 91        num_data_qubits=3,
 92        coherence_time=100e-6,
 93        gate_fid=0.99,
 94        meas_fid=0.98,
 95        has_transceiver=True,
 96    )
 97
 98    link_1_2 = QuantumLink(
 99        id=LinkId("QPU_1_2"),
100        src=qpu_1,
101        dst=qpu_2,
102        distance=50000.0,  # 50 km
103        attenuation=0.0002,
104        signal_speed=200.0,
105        repetition_rate=1e6,  # 1 MHz
106        polarization_fidelity=0.98,
107    )
108    link_1_3 = QuantumLink(
109        id=LinkId("QPU_1_3"),
110        src=qpu_1,
111        dst=qpu_3,
112        distance=50000.0,  # 50 km
113        attenuation=0.0002,
114        signal_speed=200.0,
115        repetition_rate=1e6,  # 1 MHz
116        polarization_fidelity=0.98,
117    )
118    link_2_3 = QuantumLink(
119        id=LinkId("QPU_2_3"),
120        src=qpu_2,
121        dst=qpu_3,
122        distance=100000.0,  # 100 km
123        attenuation=0.0002,
124        signal_speed=200.0,
125        repetition_rate=1e6,  # 1 MHz
126        polarization_fidelity=0.98,
127    )
128
129    network.add_node(qpu_1)
130    network.add_node(qpu_2)
131    network.add_node(qpu_3)
132    network.add_link(link_1_2)
133    network.add_link(link_2_3)
134    network.add_link(link_1_3)
135
136    return network
137
138
139def print_space_time_matrix(qubit_assignment_matrix: SpaceTimeMatrix) -> None:
140    """Visualize the space-time matrix showing qubit-node assignments over time.
141
142    Args:
143        qubit_assignment_matrix (SpaceTimeMatrix): The space-time matrix containing qubit to node
144            assignments.
145
146    """
147    output = ["\n" + "=" * 60, "SPACE-TIME MATRIX (Qubit → Node Assignment)\n"]
148
149    header = "Qubit |"
150    for t in range(min(qubit_assignment_matrix.num_timesteps, 20)):
151        header += f" T{t:2d} |"
152    output.append(header)
153    output.append("-" * len(header))
154
155    for qubit in range(qubit_assignment_matrix.num_qubits):
156        row = f"  {qubit:2d}  |"
157        for t in range(min(qubit_assignment_matrix.num_timesteps, 20)):
158            node = qubit_assignment_matrix.get_node_at(qubit, t)
159            if node:
160                # Use short node ID (e.g., QPU_1 -> "1")
161                node_short = node.id.value.split("_")[-1]
162                row += f" {node_short:3s} |"
163            else:
164                row += "  -  |"
165        output.append(row)
166
167    output.append("\nLegend:")
168    nodes_used = set()
169    for partition in qubit_assignment_matrix.partition_info.values():
170        if partition.node:
171            nodes_used.add(partition.node)
172        if partition.is_multi_node:
173            nodes_used.update(partition.qubit_to_node_mapping.values())
174
175    for node in sorted(nodes_used, key=lambda n: n.id.value):
176        node_short = node.id.value.split("_")[-1]
177        output.append(f"  {node_short} = {node.id.value}")
178
179    print("\n".join(output))  # noqa: T201
180
181
182def print_measurement_counts(
183    counts: dict[str, int], num_qubits: int, max_display: int = 16
184) -> None:
185    """Print measurement counts in a cleaner format.
186
187    Args:
188        counts (dict[str, int]): Dictionary of measurement counts.
189        num_qubits (int): Number of qubits in the circuit.
190        max_display (int): Maximum number of bitstrings to display.
191
192    """
193    output = ["\n" + "=" * 60, "MEASUREMENT COUNTS"]
194
195    total_shots = sum(counts.values())
196    output.append(f"\nTotal shots: {total_shots}")
197    output.append(f"Number of qubits: {num_qubits}")
198
199    if len(counts) <= max_display:
200        output.append("\nAll measurement outcomes:")
201    else:
202        output.append(f"\nTop {max_display} most probable outcomes:")
203
204    sorted_counts = sorted(counts.items(), key=lambda x: x[1], reverse=True)
205
206    for bitstring, count in sorted_counts[:max_display]:
207        probability = count / total_shots
208        output.append(f"  {bitstring}: {count:5d} ({probability:.4f})")
209
210    if len(counts) > max_display:
211        output.append(f"\n... and {len(counts) - max_display} more outcomes")
212
213    print("\n".join(output))  # noqa: T201
214
215
216def print_density_matrix_diagonal(
217    density_matrix: np.ndarray, num_qubits: int, max_display: int = 16
218) -> None:
219    """Print the diagonal of the resultant density matrix.
220
221    Args:
222        density_matrix: The density matrix to display.
223        num_qubits: Number of qubits.
224        max_display: Maximum number of basis states to display.
225
226    """
227    output = ["\n" + "=" * 60, "DENSITY MATRIX"]
228
229    dim = 2**num_qubits
230
231    output.append("\nDiagonal Elements (Populations):")
232    for i in range(min(dim, max_display)):
233        bitstring = format(i, f"0{num_qubits}b")
234        val = density_matrix[i, i]
235        output.append(f"  |{bitstring}⟩: {val.real:.6f}")
236
237    if dim > max_display:
238        output.append(f"  ... ({dim - max_display} more states)")
239
240    print("\n".join(output))  # noqa: T201
241
242
243def print_additional_simulation_results(additional_simulation_results: list[Any]) -> None:
244    """Print the simulation results.
245
246    Args:
247        additional_simulation_results (list[Any]): List of additional simulation results.
248
249    """
250    (
251        computation_time,
252        communication_time,
253        total_time,
254        num_remote_gates,
255        num_teleportations,
256        expectation_values,
257    ) = additional_simulation_results
258    output = [
259        "\n" + "=" * 60,
260        "SIMULATION RESULTS",
261        "\nTiming:",
262        f"  Computation time: {computation_time:.2f} ps",
263        f"  Communication time: {communication_time:.2f} ps",
264        f"  Total execution time: {total_time:.2f} ps",
265    ]
266
267    if communication_time > 0:
268        overhead_pct = (communication_time / total_time) * 100
269        output.append(f"  Communication overhead: {overhead_pct:.1f}%")
270
271    output.append("\nRemote Operations:")
272    output.append(f"  Remote CNOT gates: {num_remote_gates}")
273    output.append(f"  Teleportations: {num_teleportations}")
274
275    if expectation_values is not None:
276        output.append("\nExpectation Values (Z basis):")
277        for qubit_idx, exp_val in enumerate(expectation_values):
278            output.append(f"  Qubit {qubit_idx}: {exp_val:+.6f}")
279
280    print("\n".join(output))  # noqa: T201
281
282
283if __name__ == "__main__":
284    dqc_qft_example()