1from collections.abc import Sequence
2from typing import Any, Iterable
3
4import math
5
6import numpy as np
7
8from qiskit import transpile, QuantumCircuit
9from qiskit.primitives.containers.primitive_result import PrimitiveResult
10from qiskit.primitives.containers.sampler_pub_result import SamplerPubResult
11from qiskit.result import QuasiDistribution
12from qiskit.primitives import SamplerResult, BasePrimitiveJob, BitArray, DataBin
13from qiskit.primitives.base import BaseSamplerV1, BaseSamplerV2
14from qiskit.primitives.containers.sampler_pub import SamplerPubLike
15from qiskit.primitives.primitive_job import PrimitiveJob
16
17
[docs]
18class RuntimeJobV2Adapter(BasePrimitiveJob):
19 def __init__(self, job, **kwargs):
20 super().__init__(job.job_id(), **kwargs)
21 self.job = job
22
[docs]
23 def result(self):
24 raise NotImplementedError()
25
[docs]
26 def cancel(self):
27 return self.job.cancel()
28
[docs]
29 def status(self):
30 return self.job.status()
31
[docs]
32 def done(self):
33 return self.job.done()
34
[docs]
35 def cancelled(self):
36 return self.job.cancelled()
37
[docs]
38 def running(self):
39 return self.job.running()
40
[docs]
41 def in_final_state(self):
42 return self.job.in_final_state()
43
44
[docs]
45class SamplerV2JobAdapter(RuntimeJobV2Adapter):
46 """
47 Dummy data holder, returns a v1 SamplerResult from v2 sampler job.
48 """
49
50 def __init__(self, job, **kwargs):
51 super().__init__(job, **kwargs)
52
53 def _get_quasi_meta(self, res):
54 data = BitArray.concatenate_bits(list(res.data.values()))
55 counts = data.get_int_counts()
56 probs = {k: v/data.num_shots for k, v in counts.items()}
57 quasi_dists = QuasiDistribution(probs, shots=data.num_shots)
58
59 metadata = res.metadata
60 metadata["sampler_version"] = 2 # might be useful for debugging
61
62 return quasi_dists, metadata
63
[docs]
64 def result(self):
65 res = self.job.result()
66 qd, metas = [], []
67 for r in res:
68 quasi_dist, metadata = self._get_quasi_meta(r)
69 qd.append(quasi_dist)
70 metas.append(metadata)
71
72 return SamplerResult(quasi_dists=qd, metadata=metas)
73
74
75def _transpile_circuits(circuits, backend):
76 # Transpile qaoa circuit to backend instruction set, if backend is provided
77 # ? I pass a backend into SamplerV2 as *mode* but here sampler_v2.mode returns None, why?
78 if not backend is None:
79 if isinstance(circuits, Sequence):
80 circuits = [transpile(circuit) for circuit in circuits]
81 else:
82 circuits = transpile(circuits)
83
84 return circuits
85
86
[docs]
87class SamplerV2ToSamplerV1Adapter(BaseSamplerV1):
88 """
89 Adapts a v2 sampler to a v1 interface.
90 """
91
92 def __init__(self, sampler_v2: BaseSamplerV2, backend=None):
93 """
94 Args:
95 sampler_v2 (BaseSamplerV2): V2 sampler to be adapted.
96 backend (Backend | None): Backend to transpile circuits to.
97 """
98 self.sampler_v2 = sampler_v2
99 self.backend = backend
100 super().__init__()
101
102 def _run(self, circuits, parameter_values=None, **run_options) -> SamplerV2JobAdapter:
103 circuits = _transpile_circuits(circuits, self.backend)
104 v2_list = list(zip(circuits, parameter_values))
105 job = self.sampler_v2.run(pubs=v2_list, **run_options)
106
107 return SamplerV2JobAdapter(job)
108
109
[docs]
110class SamplerV1ToSamplerV2Adapter(BaseSamplerV2):
111 """
112 Adapts a v1 sampler to a v2 interface.
113
114 Args:
115 BaseSamplerV2 (_type_): _description_
116 """
117
118 def __init__(self, sampler_v1: BaseSamplerV1) -> None:
119 super().__init__()
120 self.samplerv1 = sampler_v1
121
122 def _run(self, pubs: Iterable[SamplerPubLike], shots: int = 1024):
123 circuits, params = [], []
124 for pub in pubs:
125 if isinstance(pub, QuantumCircuit):
126 circuits.append(pub)
127 params.append([])
128 elif isinstance(pub, tuple):
129 circuits.append(pub[0])
130 params.append(pub[1] if len(pub) == 2 else [])
131
132 out = self.samplerv1.run(circuits, params, shots=shots).result()
133 results = []
134 for dist in out.quasi_dists:
135 vals: list[int] = []
136 max_val = 0
137 for k, v in dist.items():
138 vals += [k] * int(round(v*shots, 0))
139 max_val = max(max_val, k)
140
141 required_bits = math.ceil(math.log2(max_val))
142 required_bytes = math.ceil(required_bits/8)
143
144 arr = np.array(
145 [
146 np.frombuffer(v.to_bytes(required_bytes), dtype=np.uint8)
147 for v in vals
148 ])
149 bit_array = BitArray(
150 arr,
151 num_bits=required_bits
152 )
153 results.append(SamplerPubResult(data=DataBin(meas=bit_array), metadata={'shots': shots}))
154
155 return PrimitiveResult(results, metadata={'version': 2})
156
[docs]
157 def run(
158 self,
159 pubs: Iterable[SamplerPubLike],
160 *,
161 shots: int | None = None
162 ) -> BasePrimitiveJob[PrimitiveResult[SamplerPubResult], Any]:
163 job = PrimitiveJob(self._run, pubs, shots if shots is not None else 1024)
164 job._submit()
165 return job
