Add support executing circuits with a SamplerV2 instance

qiskit_experiments/framework/base_experiment.py

@@ -23,6 +23,8 @@
 from qiskit.exceptions import QiskitError
 from qiskit.qobj.utils import MeasLevel
 from qiskit.providers.options import Options
+from qiskit.primitives.base import BaseSamplerV2
+from qiskit_ibm_runtime import SamplerV2 as Sampler
 from qiskit_experiments.framework import BackendData
 from qiskit_experiments.framework.store_init_args import StoreInitArgs
 from qiskit_experiments.framework.base_analysis import BaseAnalysis
@@ -40,6 +42,7 @@ def __init__(
         analysis: Optional[BaseAnalysis] = None,
         backend: Optional[Backend] = None,
         experiment_type: Optional[str] = None,
+        backend_run: Options[bool] = False,
     ):
         """Initialize the experiment object.
 
@@ -48,7 +51,7 @@ def __init__(
             analysis: Optional, the analysis to use for the experiment.
             backend: Optional, the backend to run the experiment on.
             experiment_type: Optional, the experiment type string.
-
+            backend_run: Optional, use backend run vs the sampler (temporary)
         Raises:
             QiskitError: If qubits contains duplicates.
         """
@@ -82,6 +85,7 @@ def __init__(
         # attributes created during initialization
         self._backend = None
         self._backend_data = None
+        self._backend_run = backend_run
         if isinstance(backend, Backend):
             self._set_backend(backend)
 
@@ -196,23 +200,26 @@ def from_config(cls, config: Union[ExperimentConfig, Dict]) -> "BaseExperiment":
 
     def run(
         self,
-        backend: Optional[Backend] = None,
+        run_obj: Optional[Union[Backend, BaseSamplerV2]] = None,
         analysis: Optional[Union[BaseAnalysis, None]] = "default",
         timeout: Optional[float] = None,
+        backend_run: Optional[bool] = None,
         **run_options,
     ) -> ExperimentData:
         """Run an experiment and perform analysis.
 
         Args:
-            backend: Optional, the backend to run the experiment on. This
-                     will override any currently set backends for the single
-                     execution.
+            run_obj: Optional, the element to run the experiment on. Either a
+                      backend or a session. Will override existing backend settings.
+                      If None then a session will be invoked from previously
+                      set backend
             analysis: Optional, a custom analysis instance to use for performing
                       analysis. If None analysis will not be run. If ``"default"``
                       the experiments :meth:`analysis` instance will be used if
                       it contains one.
             timeout: Time to wait for experiment jobs to finish running before
                      cancelling.
+            backend_run: Use backend run (temp option for testing)
             run_options: backend runtime options used for circuit execution.
 
         Returns:
@@ -223,11 +230,14 @@ def run(
                          ExperimentData container.
         """
 
-        if backend is not None or analysis != "default" or run_options:
+        if backend_run is not None:
+            self._backend_run = backend_run
+
+        if isinstance(run_obj, Backend) or analysis != "default" or run_options:
             # Make a copy to update analysis or backend if one is provided at runtime
             experiment = self.copy()
-            if backend:
-                experiment._set_backend(backend)
+            if isinstance(run_obj, Backend):
+                experiment._set_backend(run_obj)
             if isinstance(analysis, BaseAnalysis):
                 experiment.analysis = analysis
             if run_options:
@@ -250,8 +260,14 @@ def run(
         # Run options
         run_opts = experiment.run_options.__dict__
 
+        # see if sampler was sent in
+        if isinstance(run_obj, BaseSamplerV2):
+            sampler = run_obj
+        else:
+            sampler = None
+
         # Run jobs
-        jobs = experiment._run_jobs(transpiled_circuits, **run_opts)
+        jobs = experiment._run_jobs(transpiled_circuits, sampler=sampler, **run_opts)
         experiment_data.add_jobs(jobs, timeout=timeout)
 
         # Optionally run analysis
@@ -333,7 +349,9 @@ def job_info(self, backend: Backend = None):
             "Total number of jobs": num_jobs,
         }
 
-    def _run_jobs(self, circuits: List[QuantumCircuit], **run_options) -> List[Job]:
+    def _run_jobs(
+        self, circuits: List[QuantumCircuit], sampler: BaseSamplerV2 = None, **run_options
+    ) -> List[Job]:
         """Run circuits on backend as 1 or more jobs."""
         max_circuits = self._max_circuits(self.backend)
 
@@ -348,7 +366,39 @@ def _run_jobs(self, circuits: List[QuantumCircuit], **run_options) -> List[Job]:
             job_circuits = [circuits]
 
         # Run jobs
-        jobs = [self.backend.run(circs, **run_options) for circs in job_circuits]
+        if not self._backend_run:
+            if sampler is None:
+                # instantiate a sampler from the backend
+                sampler = Sampler(self.backend)
+
+                # have to hand set some of these options
+                # see https://docs.quantum.ibm.com/api/qiskit-ibm-runtime
+                # /qiskit_ibm_runtime.options.SamplerExecutionOptionsV2
+                if "init_qubits" in run_options:
+                    sampler.options.execution.init_qubits = run_options["init_qubits"]
+                if "rep_delay" in run_options:
+                    sampler.options.execution.rep_delay = run_options["rep_delay"]
+                if "meas_level" in run_options:
+                    if run_options["meas_level"] == 2:
+                        sampler.options.execution.meas_type = "classified"
+                    elif run_options["meas_level"] == 1:
+                        if "meas_return" in run_options:
+                            if run_options["meas_return"] == "avg":
+                                sampler.options.execution.meas_type = "avg_kerneled"
+                            else:
+                                sampler.options.execution.meas_type = "kerneled"
+                        else:
+                            # assume this is what is wanted if no  meas return specified
+                            sampler.options.execution.meas_type = "kerneled"
+                    else:
+                        raise QiskitError("Only meas level 1 + 2 supported by sampler")
+
+                if run_options.get("shots") is not None:
+                    sampler.options.default_shots = run_options.get("shots")
+
+            jobs = [sampler.run(circs) for circs in job_circuits]
+        else:
+            jobs = [self.backend.run(circs, **run_options) for circs in job_circuits]
 
         return jobs
 

qiskit_experiments/framework/composite/batch_experiment.py

@@ -18,6 +18,7 @@
 
 from qiskit import QuantumCircuit
 from qiskit.providers import Job, Backend, Options
+from qiskit_ibm_runtime import SamplerV2 as Sampler
 
 from .composite_experiment import CompositeExperiment, BaseExperiment
 from .composite_analysis import CompositeAnalysis
@@ -134,7 +135,11 @@ def _remap_qubits(self, circuit, qubit_mapping):
         return new_circuit
 
     def _run_jobs_recursive(
-        self, circuits: List[QuantumCircuit], truncated_metadata: List[Dict], **run_options
+        self,
+        circuits: List[QuantumCircuit],
+        truncated_metadata: List[Dict],
+        sampler: Sampler = None,
+        **run_options,
     ) -> List[Job]:
         # The truncated metadata is a truncation of the original composite metadata.
         # During the recursion, the current experiment (self) will be at the head of the truncated
@@ -161,19 +166,21 @@ def _run_jobs_recursive(
                 # even if they run in different jobs
                 if isinstance(exp, BatchExperiment):
                     new_jobs = exp._run_jobs_recursive(
-                        circs_by_subexps[index], truncated_metadata, **run_options
+                        circs_by_subexps[index], truncated_metadata, sampler, **run_options
                     )
                 else:
-                    new_jobs = exp._run_jobs(circs_by_subexps[index], **run_options)
+                    new_jobs = exp._run_jobs(circs_by_subexps[index], sampler, **run_options)
                 jobs.extend(new_jobs)
         else:
-            jobs = super()._run_jobs(circuits, **run_options)
+            jobs = super()._run_jobs(circuits, sampler, **run_options)
 
         return jobs
 
-    def _run_jobs(self, circuits: List[QuantumCircuit], **run_options) -> List[Job]:
+    def _run_jobs(
+        self, circuits: List[QuantumCircuit], sampler: Sampler = None, **run_options
+    ) -> List[Job]:
         truncated_metadata = [circ.metadata for circ in circuits]
-        jobs = self._run_jobs_recursive(circuits, truncated_metadata, **run_options)
+        jobs = self._run_jobs_recursive(circuits, truncated_metadata, sampler, **run_options)
         return jobs
 
     @classmethod

qiskit_experiments/framework/experiment_data.py

@@ -38,13 +38,15 @@
 from qiskit.exceptions import QiskitError
 from qiskit.providers import Job, Backend, Provider
 from qiskit.utils.deprecation import deprecate_arg
+from qiskit.primitives import BitArray, SamplerPubResult, BasePrimitiveJob
 
 from qiskit_ibm_experiment import (
     IBMExperimentService,
     ExperimentData as ExperimentDataclass,
     AnalysisResultData as AnalysisResultDataclass,
     ResultQuality,
 )
+
 from qiskit_experiments.framework.json import ExperimentEncoder, ExperimentDecoder
 from qiskit_experiments.database_service.utils import (
     plot_to_svg_bytes,
@@ -738,7 +740,7 @@ def add_data(
 
     def add_jobs(
         self,
-        jobs: Union[Job, List[Job]],
+        jobs: Union[Job, List[Job], BasePrimitiveJob, List[BasePrimitiveJob]],
         timeout: Optional[float] = None,
     ) -> None:
         """Add experiment data.
@@ -769,19 +771,22 @@ def add_jobs(
         # Add futures for extracting finished job data
         timeout_ids = []
         for job in jobs:
-            if self.backend is not None:
-                backend_name = BackendData(self.backend).name
-                job_backend_name = BackendData(job.backend()).name
-                if self.backend and backend_name != job_backend_name:
-                    LOG.warning(
-                        "Adding a job from a backend (%s) that is different "
-                        "than the current backend (%s). "
-                        "The new backend will be used, but "
-                        "service is not changed if one already exists.",
-                        job.backend(),
-                        self.backend,
-                    )
-            self.backend = job.backend()
+            if hasattr(job, "backend"):
+                if self.backend is not None:
+                    backend_name = BackendData(self.backend).name
+                    job_backend_name = BackendData(job.backend()).name
+                    if self.backend and backend_name != job_backend_name:
+                        LOG.warning(
+                            "Adding a job from a backend (%s) that is different "
+                            "than the current backend (%s). "
+                            "The new backend will be used, but "
+                            "service is not changed if one already exists.",
+                            job.backend(),
+                            self.backend,
+                        )
+                self.backend = job.backend()
+            else:
+                self.backend = None
 
             jid = job.job_id()
             if jid in self._jobs:
@@ -985,27 +990,58 @@ def _add_result_data(self, result: Result, job_id: Optional[str] = None) -> None
             job_id: The id of the job the result came from. If `None`, the
             job id in `result` is used.
         """
-        if job_id is None:
-            job_id = result.job_id
-        if job_id not in self._jobs:
-            self._jobs[job_id] = None
-            self.job_ids.append(job_id)
-        with self._result_data.lock:
-            # Lock data while adding all result data
-            for i, _ in enumerate(result.results):
-                data = result.data(i)
-                data["job_id"] = job_id
-                if "counts" in data:
-                    # Format to Counts object rather than hex dict
-                    data["counts"] = result.get_counts(i)
-                expr_result = result.results[i]
-                if hasattr(expr_result, "header") and hasattr(expr_result.header, "metadata"):
-                    data["metadata"] = expr_result.header.metadata
-                data["shots"] = expr_result.shots
-                data["meas_level"] = expr_result.meas_level
-                if hasattr(expr_result, "meas_return"):
-                    data["meas_return"] = expr_result.meas_return
-                self._result_data.append(data)
+        if hasattr(result, "results"):
+            # backend run results
+            if job_id is None:
+                job_id = result.job_id
+            if job_id not in self._jobs:
+                self._jobs[job_id] = None
+                self.job_ids.append(job_id)
+            with self._result_data.lock:
+                # Lock data while adding all result data
+                for i, _ in enumerate(result.results):
+                    data = result.data(i)
+                    data["job_id"] = job_id
+                    if "counts" in data:
+                        # Format to Counts object rather than hex dict
+                        data["counts"] = result.get_counts(i)
+                    expr_result = result.results[i]
+                    if hasattr(expr_result, "header") and hasattr(expr_result.header, "metadata"):
+                        data["metadata"] = expr_result.header.metadata
+                    data["shots"] = expr_result.shots
+                    data["meas_level"] = expr_result.meas_level
+                    if hasattr(expr_result, "meas_return"):
+                        data["meas_return"] = expr_result.meas_return
+                    self._result_data.append(data)
+        else:
+            # sampler results
+            if job_id is None:
+                raise QiskitError("job_id must be provided, not available in the sampler result")
+            if job_id not in self._jobs:
+                self._jobs[job_id] = None
+                self.job_ids.append(job_id)
+            with self._result_data.lock:
+                # Lock data while adding all result data
+                # Sampler results are a list
+                for i, _ in enumerate(result):
+                    data = {}
+                    # convert to a Sampler Pub Result (can remove this later when the bug is fixed)
+                    testres = SamplerPubResult(result[i].data, result[i].metadata)
+                    data["job_id"] = job_id
+                    if isinstance(testres.data[next(iter(testres.data))], BitArray):
+                        # bit results so has counts
+                        data["meas_level"] = 2
+                        data["meas_return"] = "avg"
+                        # join the data
+                        data["counts"] = testres.join_data(testres.data.keys()).get_counts()
+                        # number of shots
+                        data["shots"] = testres.data[next(iter(testres.data))].num_shots
+                    else:
+                        raise QiskitError("Sampler with meas level 1 support TBD")
+
+                    data["metadata"] = testres.metadata["circuit_metadata"]
+
+                    self._result_data.append(data)
 
     def _retrieve_data(self):
         """Retrieve job data if missing experiment data."""

requirements.txt

@@ -2,6 +2,7 @@ numpy>=1.17
 scipy>=1.4
 qiskit>=0.45
 qiskit-ibm-experiment>=0.4.6
+qiskit_ibm_runtime>=0.29.0
 matplotlib>=3.4
 uncertainties
 lmfit