Add BackendSamplerV2

qiskit/primitives/__init__.py

@@ -404,6 +404,7 @@
 
    BaseSamplerV2
    StatevectorSampler
+   BackendSamplerV2
 
 Results V2
 ----------
@@ -473,3 +474,4 @@
 from .sampler import Sampler
 from .statevector_estimator import StatevectorEstimator
 from .statevector_sampler import StatevectorSampler
+from .backend_sampler_v2 import BackendSamplerV2

qiskit/primitives/backend_sampler_v2.py

@@ -0,0 +1,233 @@
+# This code is part of Qiskit.
+#
+# (C) Copyright IBM 2024.
+#
+# This code is licensed under the Apache License, Version 2.0. You may
+# obtain a copy of this license in the LICENSE.txt file in the root directory
+# of this source tree or at http://www.apache.org/licenses/LICENSE-2.0.
+#
+# Any modifications or derivative works of this code must retain this
+# copyright notice, and modified files need to carry a notice indicating
+# that they have been altered from the originals.
+
+"""Sampler V2 implementation for an arbitrary Backend object."""
+
+from __future__ import annotations
+
+import warnings
+from dataclasses import dataclass
+from typing import Iterable
+
+import numpy as np
+from numpy.typing import NDArray
+
+from qiskit.circuit import QuantumCircuit
+from qiskit.primitives.backend_estimator import _run_circuits
+from qiskit.primitives.base import BaseSamplerV2
+from qiskit.primitives.containers import (
+    BitArray,
+    PrimitiveResult,
+    PubResult,
+    SamplerPubLike,
+    make_data_bin,
+)
+from qiskit.primitives.containers.bit_array import _min_num_bytes
+from qiskit.primitives.containers.sampler_pub import SamplerPub
+from qiskit.primitives.primitive_job import PrimitiveJob
+from qiskit.providers.backend import BackendV1, BackendV2
+from qiskit.result import Result
+
+
+@dataclass
+class Options:
+    """Options for :class:`~.BackendSamplerV2`"""
+
+    default_shots: int = 1024
+    """The default shots to use if none are specified in :meth:`~.run`.
+    Default: 1024.
+    """
+
+    seed_simulator: int | None = None
+    """The seed to use in the simulator. If None, a random seed will be used.
+    Default: None.
+    """
+
+
+@dataclass
+class _MeasureInfo:
+    creg_name: str
+    num_bits: int
+    num_bytes: int
+    start: int
+
+
+class BackendSamplerV2(BaseSamplerV2):
+    """Evaluates bitstrings for provided quantum circuits
+
+    The :class:`~.BackendSamplerV2` class is a generic implementation of the
+    :class:`~.BaseSamplerV2` interface that is used to wrap a :class:`~.BackendV2`
+    (or :class:`~.BackendV1`) object in the class :class:`~.BaseSamplerV2` API. It
+    facilitates using backends that do not provide a native
+    :class:`~.BaseSamplerV2` implementation in places that work with
+    :class:`~.BaseSamplerV2`. However,
+    if you're using a provider that has a native implementation of
+    :class:`~.BaseSamplerV2`, it is a better choice to leverage that native
+    implementation as it will likely include additional optimizations and be
+    a more efficient implementation. The generic nature of this class
+    precludes doing any provider- or backend-specific optimizations.
+
+    This class does not perform any measurement or gate mitigation.
+
+    Each tuple of ``(circuit, <optional> parameter values, <optional> shots)``, called a sampler
+    primitive unified bloc (PUB), produces its own array-valued result. The :meth:`~run` method can
+    be given many pubs at once.
+
+    The options for :class:`~.BackendSamplerV2` consist of the following items.
+
+    * ``default_shots``: The default shots to use if none are specified in :meth:`~run`.
+      Default: 1024.
+
+    * ``seed_simulator``: The seed to use in the simulator. If None, a random seed will be used.
+      Default: None.
+
+    .. note::
+
+        This class requires a backend that supports ``memory`` option.
+
+    """
+
+    def __init__(
+        self,
+        *,
+        backend: BackendV1 | BackendV2,
+        options: dict | None = None,
+    ):
+        """
+        Args:
+            backend: The backend to run the primitive on.
+            options: The options to control the default shots (``default_shots``) and
+                the random seed for the simulator (``seed_simulator``).
+        """
+        self._backend = backend
+        self._options = Options(**options) if options else Options()
+
+    @property
+    def backend(self) -> BackendV1 | BackendV2:
+        """Returns the backend which this sampler object based on."""
+        return self._backend
+
+    @property
+    def options(self) -> Options:
+        """Return the options"""
+        return self._options
+
+    def run(
+        self, pubs: Iterable[SamplerPubLike], *, shots: int | None = None
+    ) -> PrimitiveJob[PrimitiveResult[PubResult]]:
+        if shots is None:
+            shots = self._options.default_shots
+        coerced_pubs = [SamplerPub.coerce(pub, shots) for pub in pubs]
+        self._validate_pubs(coerced_pubs)
+        job = PrimitiveJob(self._run, coerced_pubs)
+        job._submit()
+        return job
+
+    def _validate_pubs(self, pubs: list[SamplerPub]):
+        for i, pub in enumerate(pubs):
+            if len(pub.circuit.cregs) == 0:
+                warnings.warn(
+                    f"The {i}-th pub's circuit has no output classical registers and so the result "
+                    "will be empty. Did you mean to add measurement instructions?",
+                    UserWarning,
+                )
+
+    def _run(self, pubs: Iterable[SamplerPub]) -> PrimitiveResult[PubResult]:
+        results = [self._run_pub(pub) for pub in pubs]
+        return PrimitiveResult(results)
+
+    def _run_pub(self, pub: SamplerPub) -> PubResult:
+        meas_info, max_num_bytes = _analyze_circuit(pub.circuit)
+        bound_circuits = pub.parameter_values.bind_all(pub.circuit)
+        arrays = {
+            item.creg_name: np.zeros(
+                bound_circuits.shape + (pub.shots, item.num_bytes), dtype=np.uint8
+            )
+            for item in meas_info
+        }
+        flatten_circuits = np.ravel(bound_circuits).tolist()
+        result_memory, _ = _run_circuits(
+            flatten_circuits,
+            self._backend,
+            memory=True,
+            shots=pub.shots,
+            seed_simulator=self._options.seed_simulator,
+        )
+        memory_list = _prepare_memory(result_memory, max_num_bytes)
+
+        for samples, index in zip(memory_list, np.ndindex(*bound_circuits.shape)):
+            for item in meas_info:
+                ary = _samples_to_packed_array(samples, item.num_bits, item.start)
+                arrays[item.creg_name][index] = ary
+
+        data_bin_cls = make_data_bin(
+            [(item.creg_name, BitArray) for item in meas_info],
+            shape=bound_circuits.shape,
+        )
+        meas = {
+            item.creg_name: BitArray(arrays[item.creg_name], item.num_bits) for item in meas_info
+        }
+        data_bin = data_bin_cls(**meas)
+        return PubResult(data_bin, metadata={})
+
+
+def _analyze_circuit(circuit: QuantumCircuit) -> tuple[list[_MeasureInfo], int]:
+    meas_info = []
+    max_num_bits = 0
+    for creg in circuit.cregs:
+        name = creg.name
+        num_bits = creg.size
+        start = circuit.find_bit(creg[0]).index
+        meas_info.append(
+            _MeasureInfo(
+                creg_name=name,
+                num_bits=num_bits,
+                num_bytes=_min_num_bytes(num_bits),
+                start=start,
+            )
+        )
+        max_num_bits = max(max_num_bits, start + num_bits)
+    return meas_info, _min_num_bytes(max_num_bits)
+
+
+def _prepare_memory(results: list[Result], num_bytes: int) -> NDArray[np.uint8]:
+    lst = []
+    for res in results:
+        for exp in res.results:
+            if hasattr(exp.data, "memory") and exp.data.memory:
+                data = b"".join(int(i, 16).to_bytes(num_bytes, "big") for i in exp.data.memory)
+                data = np.frombuffer(data, dtype=np.uint8).reshape(-1, num_bytes)
+            else:
+                # no measure in a circuit
+                data = np.zeros((exp.shots, num_bytes), dtype=np.uint8)
+            lst.append(data)
+    ary = np.array(lst, copy=False)
+    return np.unpackbits(ary, axis=-1, bitorder="big")
+
+
+def _samples_to_packed_array(
+    samples: NDArray[np.uint8], num_bits: int, start: int
+) -> NDArray[np.uint8]:
+    # samples of `Backend.run(memory=True)` will be the order of
+    # clbit_last, ..., clbit_1, clbit_0
+    # place samples in the order of clbit_start+num_bits-1, ..., clbit_start+1, clbit_start
+    if start == 0:
+        ary = samples[:, -start - num_bits :]
+    else:
+        ary = samples[:, -start - num_bits : -start]
+    # pad 0 in the left to align the number to be mod 8
+    # since np.packbits(bitorder='big') pads 0 to the right.
+    pad_size = -num_bits % 8
+    ary = np.pad(ary, ((0, 0), (pad_size, 0)), constant_values=0)
+    # pack bits in big endian order
+    ary = np.packbits(ary, axis=-1, bitorder="big")
+    return ary

releasenotes/notes/add-backend-sampler-v2-5e40135781eebc7f.yaml

@@ -0,0 +1,28 @@
+---
+features:
+  - |
+    The implementation :class:`~.BackendSamplerV2` of :class:`~.BaseSamplerV2` was added.
+    This sampler supports :class:`~.BackendV1` and :class:`~.BackendV2` that allow
+    ``memory`` option to compute bitstrings.
+
+    .. code-block:: python
+
+      import numpy as np
+      from qiskit import transpile
+      from qiskit.circuit.library import IQP
+      from qiskit.primitives import BackendSamplerV2
+      from qiskit.providers.fake_provider import Fake7QPulseV1
+      from qiskit.quantum_info import random_hermitian
+
+      backend = Fake7QPulseV1()
+      sampler = BackendSamplerV2(backend=backend)
+      n_qubits = 5
+      mat = np.real(random_hermitian(n_qubits, seed=1234))
+      circuit = IQP(mat)
+      circuit.measure_all()
+      isa_circuit = transpile(circuit, backend=backend, optimization_level=1)
+      job = sampler.run([isa_circuit], shots=100)
+      result = job.result()
+      print(f"> bitstrings: {result[0].data.meas.get_bitstrings()}")
+      print(f"> counts: {result[0].data.meas.get_counts()}")
+      print(f"> Metadata: {result[0].metadata}")