Add run_options option and support for processing level 1 data to BackendSamplerV2

qiskit/primitives/backend_estimator.py

@@ -44,13 +44,15 @@
 def _run_circuits(
     circuits: QuantumCircuit | list[QuantumCircuit],
     backend: BackendV1 | BackendV2,
+    clear_metadata: bool = True,
     **run_options,
 ) -> tuple[list[Result], list[dict]]:
     """Remove metadata of circuits and run the circuits on a backend.
     Args:
         circuits: The circuits
         backend: The backend
-        monitor: Enable job minotor if True
+        clear_metadata: Clear circuit metadata before passing to backend.run if
+            True.
         **run_options: run_options
     Returns:
         The result and the metadata of the circuits
@@ -60,7 +62,8 @@ def _run_circuits(
     metadata = []
     for circ in circuits:
         metadata.append(circ.metadata)
-        circ.metadata = {}
+        if clear_metadata:
+            circ.metadata = {}
     if isinstance(backend, BackendV1):
         max_circuits = getattr(backend.configuration(), "max_experiments", None)
     elif isinstance(backend, BackendV2):

qiskit/primitives/backend_sampler_v2.py

@@ -17,12 +17,13 @@
 import warnings
 from collections import defaultdict
 from dataclasses import dataclass
-from typing import Iterable
+from typing import Any, Iterable, Union
 
 import numpy as np
 from numpy.typing import NDArray
 
 from qiskit.circuit import QuantumCircuit
+from qiskit.exceptions import QiskitError
 from qiskit.primitives.backend_estimator import _run_circuits
 from qiskit.primitives.base import BaseSamplerV2
 from qiskit.primitives.containers import (
@@ -53,6 +54,11 @@ class Options:
     Default: None.
     """
 
+    run_options: dict[str, Any] | None = None
+    """A dictionary of options to pass to the backend's ``run()`` method.
+    Default: None (no option passed to backend's ``run`` method)
+    """
+
 
 @dataclass
 class _MeasureInfo:
@@ -62,6 +68,16 @@ class _MeasureInfo:
     start: int
 
 
+ResultMemory = Union[list[str], list[list[float]], list[list[list[float]]]]
+"""Type alias for possible level 2 and level 1 result memory formats. For level
+2, the format is a list of bit strings. For level 1, format can be either a
+list of I/Q pairs (list with two floats) for each memory slot if using
+``meas_return=avg`` or a list of of lists of I/Q pairs if using
+``meas_return=single`` with the outer list indexing shot number and the inner
+list indexing memory slot.
+"""
+
+
 class BackendSamplerV2(BaseSamplerV2):
     """Evaluates bitstrings for provided quantum circuits
 
@@ -91,6 +107,9 @@ class BackendSamplerV2(BaseSamplerV2):
     * ``seed_simulator``: The seed to use in the simulator. If None, a random seed will be used.
       Default: None.
 
+    * ``run_options``: A dictionary of options to pass through to the ``run()``
+      method of the wrapped :class:`~.BackendV2` instance.
+
     .. note::
 
         This class requires a backend that supports ``memory`` option.
@@ -165,19 +184,27 @@ def _run_pubs(self, pubs: list[SamplerPub], shots: int) -> list[SamplerPubResult
         for circuits in bound_circuits:
             flatten_circuits.extend(np.ravel(circuits).tolist())
 
+        run_opts = self._options.run_options or {}
         # run circuits
         results, _ = _run_circuits(
             flatten_circuits,
             self._backend,
+            clear_metadata=False,
             memory=True,
             shots=shots,
             seed_simulator=self._options.seed_simulator,
+            **run_opts,
         )
         result_memory = _prepare_memory(results)
 
         # pack memory to an ndarray of uint8
         results = []
         start = 0
+        meas_level = (
+            None
+            if self._options.run_options is None
+            else self._options.run_options.get("meas_level")
+        )
         for pub, bound in zip(pubs, bound_circuits):
             meas_info, max_num_bytes = _analyze_circuit(pub.circuit)
             end = start + bound.size
@@ -189,6 +216,7 @@ def _run_pubs(self, pubs: list[SamplerPub], shots: int) -> list[SamplerPubResult
                     meas_info,
                     max_num_bytes,
                     pub.circuit.metadata,
+                    meas_level,
                 )
             )
             start = end
@@ -197,28 +225,43 @@ def _run_pubs(self, pubs: list[SamplerPub], shots: int) -> list[SamplerPubResult
 
     def _postprocess_pub(
         self,
-        result_memory: list[list[str]],
+        result_memory: list[ResultMemory],
         shots: int,
         shape: tuple[int, ...],
         meas_info: list[_MeasureInfo],
         max_num_bytes: int,
         circuit_metadata: dict,
+        meas_level: int | None,
     ) -> SamplerPubResult:
-        """Converts the memory data into an array of bit arrays with the shape of the pub."""
-        arrays = {
-            item.creg_name: np.zeros(shape + (shots, item.num_bytes), dtype=np.uint8)
-            for item in meas_info
-        }
-        memory_array = _memory_array(result_memory, max_num_bytes)
-
-        for samples, index in zip(memory_array, np.ndindex(*shape)):
-            for item in meas_info:
-                ary = _samples_to_packed_array(samples, item.num_bits, item.start)
-                arrays[item.creg_name][index] = ary
-
-        meas = {
-            item.creg_name: BitArray(arrays[item.creg_name], item.num_bits) for item in meas_info
-        }
+        """Converts the memory data into a sampler pub result
+
+        For level 2 data, the memory data are stored in an array of bit arrays
+        with the shape of the pub. For level 1 data, the data are stored in a
+        complex numpy array.
+        """
+        if meas_level == 2 or meas_level is None:
+            arrays = {
+                item.creg_name: np.zeros(shape + (shots, item.num_bytes), dtype=np.uint8)
+                for item in meas_info
+            }
+            memory_array = _memory_array(result_memory, max_num_bytes)
+
+            for samples, index in zip(memory_array, np.ndindex(*shape)):
+                for item in meas_info:
+                    ary = _samples_to_packed_array(samples, item.num_bits, item.start)
+                    arrays[item.creg_name][index] = ary
+
+            meas = {
+                item.creg_name: BitArray(arrays[item.creg_name], item.num_bits)
+                for item in meas_info
+            }
+        elif meas_level == 1:
+            raw = np.array(result_memory)
+            cplx = raw[..., 0] + 1j * raw[..., 1]
+            cplx = np.reshape(cplx, (*shape, *cplx.shape[1:]))
+            meas = {item.creg_name: cplx for item in meas_info}
+        else:
+            raise QiskitError(f"Unsupported meas_level: {meas_level}")
         return SamplerPubResult(
             DataBin(**meas, shape=shape),
             metadata={"shots": shots, "circuit_metadata": circuit_metadata},
@@ -248,7 +291,7 @@ def _analyze_circuit(circuit: QuantumCircuit) -> tuple[list[_MeasureInfo], int]:
     return meas_info, _min_num_bytes(max_num_bits)
 
 
-def _prepare_memory(results: list[Result]) -> list[list[str]]:
+def _prepare_memory(results: list[Result]) -> list[ResultMemory]:
     """Joins splitted results if exceeding max_experiments"""
     lst = []
     for res in results:

releasenotes/notes/backend-sampler-v2-level1-dc13af460cd38454.yaml

@@ -0,0 +1,17 @@
+---
+features:
+  - |
+    Support for level 1 data was added to :class:`~.BackendSamplerV2` as was
+    support for passing options through to the ``run()`` method of the wrapped
+    :class:`~.BackendV2`. The run options can be specified using a
+    ``"run_options"`` entry inside of the ``options`` dicitonary passed to
+    :class:`~.BackendSamplerV2`. The ``"run_options"`` entry should be a
+    dictionary mapping argument names to values for passing to the backend's
+    ``run()`` method. When a ``"meas_level"`` option with a value of 1 is set
+    in the run options, the results from the backend will be treated as level 1
+    results rather as bit arrays (the level 2 format).
+upgrade:
+  - |
+    When using :class:`~.BackendSamplerV2`, circuit metadata is no longer
+    cleared before passing circuits to the ``run()`` method of the wrapped
+    :class:`~.BackendV2` instance.

test/python/primitives/test_backend_sampler_v2.py

@@ -31,8 +31,10 @@
 from qiskit.primitives.containers.sampler_pub import SamplerPub
 from qiskit.providers import JobStatus
 from qiskit.providers.backend_compat import BackendV2Converter
-from qiskit.providers.basic_provider import BasicSimulator
+from qiskit.providers.basic_provider import BasicProviderJob, BasicSimulator
 from qiskit.providers.fake_provider import Fake7QPulseV1, GenericBackendV2
+from qiskit.result import Result
+from qiskit.qobj.utils import MeasReturnType, MeasLevel
 from qiskit.transpiler.preset_passmanagers import generate_preset_pass_manager
 from ..legacy_cmaps import LAGOS_CMAP
 
@@ -50,6 +52,64 @@
 BACKENDS = BACKENDS_V1 + BACKENDS_V2
 
 
+class Level1BackendV2(GenericBackendV2):
+    """Wrapper around GenericBackendV2 adding level 1 data support for testing
+
+    GenericBackendV2 is used to run the simulation. Then level 1 data (a
+    complex number per classical register per shot) is generated by mapping 0
+    to -1 and 1 to 1 with a random number added to each shot drawn from a
+    normal distribution with a standard deviation of ``level1_sigma``. Each
+    data point has ``1j * idx`` added to it where ``idx`` is the index of the
+    classical register. For ``meas_return="avg"``, the individual shot results
+    are still calculated and then averaged.
+    """
+
+    level1_sigma = 0.1
+
+    def run(self, run_input, **options):
+        # Validate level 1 options
+        if "meas_level" not in options or "meas_return" not in options:
+            raise ValueError(f"{type(self)} requires 'meas_level' and 'meas_return' run options!")
+        meas_level = options.pop("meas_level")
+        if meas_level != 1:
+            raise ValueError(f"'meas_level' must be 1, not {meas_level}")
+        meas_return = options.pop("meas_return")
+        if meas_return not in ("single", "avg"):
+            raise ValueError(f"Unexpected value for 'meas_return': {meas_return}")
+
+        options["memory"] = True
+
+        rng = np.random.default_rng(seed=options.get("seed_simulator"))
+
+        inner_job = super().run(run_input, **options)
+        result_dict = inner_job.result().to_dict()
+        for circ, exp_result in zip(run_input, result_dict["results"]):
+            num_clbits = sum(cr.size for cr in circ.cregs)
+            bitstrings = [
+                format(int(x, 16), f"0{num_clbits}b") for x in exp_result["data"]["memory"]
+            ]
+            new_data = [
+                [
+                    [2 * int(d) - 1 + rng.normal(scale=self.level1_sigma), i]
+                    for i, d in enumerate(reversed(bs))
+                ]
+                for bs in bitstrings
+            ]
+            if meas_return == "avg":
+                new_data = [
+                    [sum(shot[idx][0] for shot in new_data) / len(new_data), idx]
+                    for idx in range(num_clbits)
+                ]
+                exp_result["meas_return"] = MeasReturnType.AVERAGE
+            else:
+                exp_result["meas_return"] = MeasReturnType.SINGLE
+            exp_result["data"] = {"memory": new_data}
+            exp_result["meas_level"] = MeasLevel.KERNELED
+
+        result = Result.from_dict(result_dict)
+        return BasicProviderJob(self, inner_job.job_id(), result)
+
+
 @ddt
 class TestBackendSamplerV2(QiskitTestCase):
     """Test for BackendSamplerV2"""
@@ -942,6 +1002,55 @@ def test_run_shots_result_size_v1(self, backend):
         self.assertLessEqual(result[0].data.meas.num_shots, self._shots)
         self.assertEqual(sum(result[0].data.meas.get_counts().values()), self._shots)
 
+    def test_run_level1(self):
+        """Test running with meas_level=1"""
+        nq = 2
+        shots = 100
+
+        backend = Level1BackendV2(nq)
+        qc = QuantumCircuit(nq)
+        qc.x(1)
+        qc.measure_all()
+        pm = generate_preset_pass_manager(optimization_level=0, backend=backend)
+        qc = pm.run(qc)
+        options = {
+            "default_shots": shots,
+            "seed_simulator": self._seed,
+            "run_options": {
+                "meas_level": 1,
+                "meas_return": "single",
+            },
+        }
+        sampler = BackendSamplerV2(backend=backend, options=options)
+        result_single = sampler.run([qc]).result()
+
+        options = {
+            "default_shots": shots,
+            "seed_simulator": self._seed,
+            "run_options": {
+                "meas_level": 1,
+                "meas_return": "avg",
+            },
+        }
+        sampler = BackendSamplerV2(backend=backend, options=options)
+        result_avg = sampler.run([qc]).result()
+
+        # Check that averaging the meas_return="single" data matches the
+        # meas_return="avg" data.
+        averaged_singles = np.average(result_single[0].join_data(), axis=0)
+        average_data = result_avg[0].join_data()
+        self.assertLessEqual(
+            max(abs(averaged_singles - average_data)),
+            backend.level1_sigma / np.sqrt(shots) * 6,
+        )
+
+        # Check that average data matches expected form for the circuit
+        expected_average = np.array([-1, 1 + 1j])
+        self.assertLessEqual(
+            max(abs(expected_average - average_data)),
+            backend.level1_sigma / np.sqrt(shots) * 6,
+        )
+
     @combine(backend=BACKENDS_V2)
     def test_primitive_job_status_done(self, backend):
         """test primitive job's status"""