Source code for qlauncher.routines.cirq

  1"""
  2Routine file for Cirq library
  3"""
  4
  5import math
  6import types
  7import typing
  8from collections.abc import Iterable, Sequence
  9from typing import Any, Literal
 10
 11import numpy as np
 12import qiskit
 13from qiskit.primitives import BaseSamplerV2, BitArray, DataBin, Sampler
 14from qiskit.primitives.base.base_primitive_job import BasePrimitiveJob
 15from qiskit.primitives.base.sampler_result import SamplerResult
 16from qiskit.primitives.containers.primitive_result import PrimitiveResult
 17from qiskit.primitives.containers.sampler_pub import SamplerPubLike
 18from qiskit.primitives.containers.sampler_pub_result import SamplerPubResult
 19from qiskit.primitives.primitive_job import PrimitiveJob
 20from qiskit.quantum_info import SparsePauliOp
 21from qiskit.result import QuasiDistribution
 22
 23from qlauncher.base import Backend
 24
 25# from qlauncher.base.translator import Translation
 26from qlauncher.exceptions import DependencyError
 27from qlauncher.routines.qiskit.adapters import TranslatingSampler, TranslatingSamplerV1
 28from qlauncher.routines.qiskit.backends.gate_circuit_backend import GateCircuitBackend
 29from qlauncher.routines.qiskit.mitigation_suppression.base import CircuitExecutionMethod
 30from qlauncher.routines.qiskit.mitigation_suppression.mitigation import NoMitigation
 31
 32try:
 33    import cirq
 34    from cirq.contrib.qasm_import.qasm import circuit_from_qasm
 35    from cirq.sim.sparse_simulator import Simulator
 36except ImportError as e:
 37    raise DependencyError(e, install_hint='cirq') from e
 38
 39

[docs]
 40def extract_bitstrings_from_result(result: cirq.Result) -> list[str]:
 41    measurements = result.measurements
 42
 43    sorted_keys = list(measurements.keys())
 44
 45    bitstrings = []
 46    num_shots = len(measurements[sorted_keys[0]])
 47
 48    for shot_index in range(num_shots):
 49        bits = []
 50        for key in sorted_keys:
 51            bits.extend(str(b) for b in measurements[key][shot_index])
 52
 53        bitstring = ''.join(bits)
 54        bitstrings.append(bitstring)
 55    return bitstrings

 56
 57

[docs]
 58def cirq_result_to_counts(result: cirq.Result) -> dict:
 59    bitstrings = extract_bitstrings_from_result(result)
 60
 61    counts = {}
 62    for bs in bitstrings:
 63        counts[bs] = counts.get(bs, 0) + 1
 64    return counts

 65
 66

[docs]
 67def cirq_result_to_probabilities(result: cirq.Result, integer_keys: bool = False) -> dict:
 68    counts = cirq_result_to_counts(result)
 69
 70    total_shots = sum(counts.values())
 71    return {int(k, 2): v / total_shots for k, v in counts.items()} if integer_keys else {k: v / total_shots for k, v in counts.items()}

 72
 73
 74class _CirqRunner:
 75    simulator = Simulator()
 76    repetitions = 1024
 77
 78    @classmethod
 79    def calculate_circuit(
 80        cls, cirq_circ: cirq.Circuit, return_type: Literal['counts', 'dist', 'raw'] = 'counts', shots: int | None = None
 81    ) -> dict | list[str]:
 82        if not cirq_circ.all_measurement_key_names():
 83            cirq_circ.append(cirq.measure_each(*cirq_circ.all_qubits()))
 84
 85        result = cls.simulator.run(cirq_circ, repetitions=cls.repetitions if shots is None else shots)
 86
 87        if return_type == 'raw':
 88            return extract_bitstrings_from_result(result)
 89
 90        return cirq_result_to_counts(result) if return_type == 'counts' else cirq_result_to_probabilities(result)
 91
 92

[docs]
 93class CirqSampler(Sampler):
 94    """Sampler adapter for Cirq"""
 95
 96    def _call(self, circuits: Sequence[int], parameter_values: Sequence[Sequence[float]], **run_options) -> SamplerResult:
 97        distributions = [_CirqRunner.calculate_circuit(self._circuits[i], 'dist') for i in circuits]
 98        quasi_dists = list(map(QuasiDistribution, distributions))
 99        return SamplerResult(quasi_dists, [{} for _ in range(len(parameter_values))])
100

[docs]
101    def run(self, circuits: tuple[cirq.Circuit, ...], parameter_values: tuple[tuple[float, ...], ...], **run_options):
102        self._circuits = list(circuits)
103
104        job = PrimitiveJob(self._call, range(len(self._circuits)), parameter_values, **run_options)
105        job._submit()
106        return job


107
108

[docs]
109class CirqSamplerV2(BaseSamplerV2):
110    def __init__(self) -> None:
111        super().__init__()
112        self.cirq_sampler_v1 = CirqSampler()
113
114    def _run(self, bound_circuits, shots):
115        bitstring_collections = [_CirqRunner.calculate_circuit(circuit, 'raw', shots) for circuit in bound_circuits]
116
117        results = []
118        for bsc in bitstring_collections:
119            arr = np.array([np.frombuffer(int(bs, 2).to_bytes(math.ceil(len(bs) / 8)), dtype=np.uint8) for bs in bsc])
120            bit_array = BitArray(arr, num_bits=len(bsc[0]))
121            results.append(SamplerPubResult(data=DataBin(meas=bit_array), metadata={'shots': len(bitstring_collections[0])}))
122        return PrimitiveResult(results, metadata={'version': 2})
123

[docs]
124    def run(self, pubs: Iterable[SamplerPubLike], *, shots: int | None = None) -> BasePrimitiveJob[PrimitiveResult[SamplerPubResult], Any]:
125        bound_circuits = []
126        for pub in pubs:
127            if isinstance(pub, cirq.Circuit):
128                bound_circuits.append(pub)
129            elif len(pub) == 1 and isinstance(pub[0], cirq.Circuit):
130                bound_circuits.append(pub[0])
131            elif len(pub) == 2:
132                bound_circuits.append(pub[0].assign_parameters(pub[1]))
133            else:
134                raise ValueError(
135                    f'Incorrect pub, expected QuantumCircuit, tuple[QuantumCircuit] or tuple[QuantumCircuit, Iterable[float]], got {type(pub)}'
136                )
137
138        job = PrimitiveJob(self._run, bound_circuits, shots)
139        job._submit()
140        return job


141
142

[docs]
143class CirqBackend(GateCircuitBackend[cirq.Circuit]):
144    """
145
146    Args:
147            Backend (_type_): _description_
148    """
149
150    basis_gates = ['x', 'y', 'z', 'cx', 'h', 'rx', 'ry', 'rz']
151
152    def __init__(
153        self,
154        name: Literal['local_simulator'] = 'local_simulator',
155        error_mitigation_strategy: CircuitExecutionMethod | None = None,
156    ):
157        self.sampler = TranslatingSampler(CirqSamplerV2(), self.compatible_circuit)
158        self.samplerV1 = TranslatingSamplerV1(CirqSampler(), self.compatible_circuit)
159        self._mitigation_strategy = error_mitigation_strategy if error_mitigation_strategy is not None else NoMitigation()
160        self.backendv1v2 = None
161        super().__init__(name)
162

[docs]
163    @staticmethod
164    def to_qasm(circuit: cirq.Circuit) -> str:
165        return circuit.to_qasm()

166

[docs]
167    @staticmethod
168    def from_qasm(qasm: str) -> cirq.Circuit:
169        return circuit_from_qasm(qasm)

170

[docs]
171    def sample_circuit(self, circuit: cirq.Circuit, shots: int = 1024) -> dict[str, int]:
172        compatible_circuit = self._mitigation_strategy.compatible_circuit
173        if not isinstance(circuit, compatible_circuit):
174            if isinstance(compatible_circuit, types.UnionType):
175                compatible_circuit = typing.get_args(compatible_circuit)[0]
176            circuit = GateCircuitBackend.get_translation(circuit, compatible_circuit)
177        return self._mitigation_strategy.sample(circuit, self, shots)

178

[docs]
179    def estimate_energy(self, circuit: qiskit.QuantumCircuit, observable: SparsePauliOp) -> float:
180        raise NotImplementedError