Add support for noise model and level 1 data to local sampler

qiskit_ibm_runtime/fake_provider/local_service.py

@@ -216,11 +216,43 @@ def _run_backend_primitive_v2(
         options_copy = copy.deepcopy(options)
 
         prim_options = {}
-        if seed_simulator := options_copy.pop("simulator", {}).pop("seed_simulator", None):
+        sim_options = options_copy.get("simulator", {})
+        if seed_simulator := sim_options.pop("seed_simulator", None):
             prim_options["seed_simulator"] = seed_simulator
         if primitive == "sampler":
+            # Create a dummy primitive to check which options it supports
+            dummy_prim = BackendSamplerV2(backend=backend)
+            use_run_options = hasattr(dummy_prim.options, "run_options")
+
+            run_options = {}
+            if use_run_options and "run_options" in options_copy:
+                run_options = options_copy.pop("run_options")
+            if use_run_options and "noise_model" in sim_options:
+                run_options["noise_model"] = sim_options.pop("noise_model")
+
             if default_shots := options_copy.pop("default_shots", None):
                 prim_options["default_shots"] = default_shots
+            if use_run_options and (
+                meas_type := options_copy.get("execution", {}).pop("meas_type", None)
+            ):
+                if meas_type == "classified":
+                    run_options["meas_level"] = 2
+                elif meas_type == "kerneled":
+                    run_options["meas_level"] = 1
+                    run_options["meas_return"] = "single"
+                elif meas_type == "avg_kerneled":
+                    run_options["meas_level"] = 1
+                    run_options["meas_return"] = "avg"
+                else:
+                    # Put unexepcted meas_type back so it is in the warning below
+                    options_copy["execution"]["meas_type"] = meas_type
+
+                if not options_copy["execution"]:
+                    del options_copy["execution"]
+
+            if run_options:
+                prim_options["run_options"] = run_options
+
             primitive_inst = BackendSamplerV2(backend=backend, options=prim_options)
         else:
             if default_shots := options_copy.pop("default_shots", None):
@@ -229,6 +261,9 @@ def _run_backend_primitive_v2(
                 prim_options["default_precision"] = default_precision
             primitive_inst = BackendEstimatorV2(backend=backend, options=prim_options)
 
+        if not sim_options:
+            # Pop to avoid warning below if all contents were popped above
+            options_copy.pop("simulator", None)
         if options_copy:
             warnings.warn(f"Options {options_copy} have no effect in local testing mode.")
 

release-notes/unreleased/1990.feat.rst

@@ -0,0 +1,13 @@
+Add support for noise model and level 1 data to local sampler
+
+The ``simulator.noise_model`` option of :class:`~.SamplerV2` is now passed
+through to the :class:`~qiskit.primitives.BackendSamplerV2` as a `noise_model`
+option under `run_options` if the primitive supports the `run_options` option
+(support was added in Qiskit 1.3).
+
+Similarly, the ``execution.meas_type`` option of :class:`~.SamplerV2` is now
+translated into ``meas_level`` and ``meas_return`` options under
+``run_options`` of the :class:`~qiskit.primitives.BackendSamplerV2` if it
+supports ``run_options``. This change allows support for level 1 data in local
+testing mode, where previously the only level 2 (classified) data was
+supported.

test/unit/test_sampler.py

@@ -15,12 +15,18 @@
 from unittest.mock import MagicMock
 
 from ddt import data, ddt, named_data
+from packaging.version import Version, parse as parse_version
 import numpy as np
 
+from qiskit.version import get_version_info as get_qiskit_version_info
 from qiskit import QuantumCircuit, QuantumRegister, ClassicalRegister
 from qiskit.primitives.containers.sampler_pub import SamplerPub
 from qiskit.circuit import Parameter
 from qiskit.circuit.library import RealAmplitudes
+from qiskit.providers import BackendV2, Options
+from qiskit.result import Result
+from qiskit.result.models import ExperimentResult, ExperimentResultData
+from qiskit.transpiler import Target
 from qiskit_ibm_runtime import Session, SamplerV2, SamplerOptions, IBMInputValueError
 from qiskit_ibm_runtime.fake_provider import FakeFractionalBackend, FakeSherbrooke, FakeCusco
 
@@ -315,3 +321,110 @@ def test_rzz_validates_only_for_fixed_angles(self):
             circ.rzz(2 * param, 0, 1)
             # Should run without an error
             SamplerV2(backend).run(pubs=[(circ, [0.5])])
+
+    @data(
+        "classified",
+        "kerneled",
+        "avg_kerneled",
+    )
+    def test_backend_run_options(self, meas_type):
+        """Test translation of sampler options into backend run options"""
+
+        # This test is checking that meas_level, meas_return, and noise_model
+        # get through the backend's run() call when SamplerV2 falls back to
+        # BackendSamplerV2 in local mode. To do this, it creates a dummy
+        # backend class that returns a result of the right format so that the
+        # sampler execution completes successfully.
+
+        if parse_version(get_qiskit_version_info()) < Version("1.3.0rc1"):
+            self.skipTest("Feature not supported on this version of Qiskit")
+
+        class DummyJob:
+            """Enough of a job class to return a result"""
+
+            def __init__(self, run_options):
+                self.run_options = run_options
+
+            def result(self):
+                """Return result object"""
+                shots = self.run_options["shots"]
+
+                if self.run_options["meas_level"] == 1:
+                    counts = None
+                    if self.run_options["meas_return"] == "single":
+                        memory = [[[0.0, 0.0]] * shots]
+                    else:
+                        memory = [[0.0, 0.0]]
+                else:
+                    counts = {"0": shots}
+                    memory = ["0"] * shots
+                result = Result(
+                    backend_name="test_backend",
+                    backend_version="0.0",
+                    qobj_id="xyz",
+                    job_id="123",
+                    success=True,
+                    results=[
+                        ExperimentResult(
+                            shots=100,
+                            success=True,
+                            data=ExperimentResultData(memory=memory, counts=counts),
+                        )
+                    ],
+                )
+                return result
+
+        class DummyBackend(BackendV2):
+            """Test backend that saves run options into the result"""
+
+            max_circuits = 1
+            # The backend gets cloned inside of the sampler execution code, so
+            # it is difficult to get a handle on the actual backend used to run
+            # the job. Here we save the run options into a class level variable
+            # that can be checked after run() is called.
+            used_run_options = {}
+
+            def __init__(self, **kwargs):
+                super().__init__(**kwargs)
+
+                self._target = Target()
+
+            @classmethod
+            def _default_options(cls):
+                return Options()
+
+            @property
+            def target(self):
+                return self._target
+
+            def run(self, run_input, **run_options):
+                nonlocal used_run_options
+                DummyBackend.used_run_options = run_options
+                return DummyJob(run_options)
+
+        backend = DummyBackend()
+
+        circ = QuantumCircuit(1, 1)
+        circ.measure(0, 0)
+
+        sampler = SamplerV2(mode=backend)
+        sampler.options.simulator.noise_model = {"name": "some_model"}
+        sampler.options.execution.meas_type = meas_type
+
+        job = sampler.run([circ], shots=100)
+        result = job.result()
+
+        used_run_options = DummyBackend.used_run_options
+        self.assertDictEqual(used_run_options["noise_model"], {"name": "some_model"})
+
+        if meas_type == "classified":
+            self.assertEqual(used_run_options["meas_level"], 2)
+            self.assertDictEqual(result[0].data.c.get_counts(), {"0": 100})
+        elif meas_type == "kerneled":
+            self.assertEqual(used_run_options["meas_level"], 1)
+            self.assertEqual(used_run_options["meas_return"], "single")
+            self.assertTrue(np.array_equal(result[0].data.c, np.zeros((1, 100))))
+        else:  # meas_type == "avg_kerneled"
+            self.assertEqual(used_run_options["meas_level"], 1)
+            self.assertEqual(used_run_options["meas_return"], "avg")
+            self.assertTrue(np.array_equal(result[0].data.c, np.zeros((1,))))