Source code for quantum_launcher.routines.qiskit_routines.v2_wrapper

 1from typing import Sequence
 2
 3from qiskit import transpile
 4from qiskit.primitives.base import BaseSamplerV1, BaseSamplerV2
 5from qiskit.primitives import SamplerResult, BasePrimitiveJob
 6from qiskit.result import QuasiDistribution
 7
 8

[docs]
 9class RuntimeJobV2Adapter(BasePrimitiveJob):
10    def __init__(self, job, **kwargs):
11        super().__init__(job.job_id(), **kwargs)
12        self.job = job
13

[docs]
14    def result(self):
15        raise NotImplementedError()

16

[docs]
17    def cancel(self):
18        return self.job.cancel()

19

[docs]
20    def status(self):
21        return self.job.status()

22

[docs]
23    def done(self):
24        return self.job.done()

25

[docs]
26    def cancelled(self):
27        return self.job.cancelled()

28

[docs]
29    def running(self):
30        return self.job.running()

31

[docs]
32    def in_final_state(self):
33        return self.job.in_final_state()


34
35

[docs]
36class SamplerV2JobAdapter(RuntimeJobV2Adapter):
37    """
38    Dummy data holder, returns a v1 SamplerResult from v2 sampler job.
39    """
40
41    def __init__(self, job, **kwargs):
42        super().__init__(job, **kwargs)
43
44    def _get_quasi_meta(self, res):
45        data = next(iter(res.data.values()))
46        counts = data.get_int_counts()
47        probs = {k: v/data.num_shots for k, v in counts.items()}
48        quasi_dists = QuasiDistribution(probs, shots=data.num_shots)
49
50        metadata = res.metadata
51        metadata["sampler_version"] = 2  # might be useful for debugging
52
53        return quasi_dists, metadata
54

[docs]
55    def result(self):
56        res = self.job.result()
57        qd, metas = [], []
58        for r in res:
59            quasi_dist, metadata = self._get_quasi_meta(r)
60            qd.append(quasi_dist)
61            metas.append(metadata)
62
63        return SamplerResult(quasi_dists=qd, metadata=metas)


64
65
66def _transpile_circuits(circuits, backend):
67    # Transpile qaoa circuit to backend instruction set, if backend is provided
68    # ? I pass a backend into SamplerV2 as *mode* but here sampler_v2.mode returns None, why?
69    if not backend is None:
70        if isinstance(circuits, Sequence):
71            circuits = [transpile(circuit) for circuit in circuits]
72        else:
73            circuits = transpile(circuits)
74
75    return circuits
76
77

[docs]
78class SamplerV2Adapter(BaseSamplerV1):
79    """
80    V1 adapter for V2 samplers.
81    """
82
83    def __init__(self, sampler_v2: BaseSamplerV2, backend=None):
84        """
85        Args:
86            sampler_v2 (BaseSamplerV2): V2 sampler to be adapted.
87            backend (Optional[Backend]): Backend to transpile circuits to.
88        """
89        self.sampler_v2 = sampler_v2
90        self.backend = backend
91        super().__init__()
92
93    def _run(self, circuits, parameter_values=None, **run_options) -> SamplerV2JobAdapter:
94        circuits = _transpile_circuits(circuits, self.backend)
95        v2_list = list(zip(circuits, parameter_values))
96        job = self.sampler_v2.run(pubs=v2_list, **run_options)
97
98        return SamplerV2JobAdapter(job)