Support local execution on simulators

qiskit_ibm_runtime/base_primitive.py

@@ -20,10 +20,13 @@
 from dataclasses import asdict
 import warnings
 
+from qiskit_aer import AerSimulator
+
 from qiskit.providers.options import Options as TerraOptions
 
 from qiskit_ibm_provider.session import get_cm_session as get_cm_provider_session
 
+from .fake_provider.fake_provider import FakeProviderForBackendV2
 from .options import Options
 from .options.utils import set_default_error_levels
 from .runtime_job import RuntimeJob
@@ -98,6 +101,13 @@ def __init__(
         if isinstance(backend, IBMBackend):
             self._service = backend.service
             self._backend = backend
+        elif isinstance(backend, AerSimulator):
+            self._backend = backend
+            self._service = (
+                QiskitRuntimeService()
+                if QiskitRuntimeService.global_service is None
+                else QiskitRuntimeService.global_service
+            )
         elif isinstance(backend, str):
             self._service = (
                 QiskitRuntimeService()
@@ -137,6 +147,11 @@ def _run_primitive(self, primitive_inputs: Dict, user_kwargs: Dict) -> RuntimeJo
         Returns:
             Submitted job.
         """
+        run_simulator = isinstance(self._backend, AerSimulator) or (
+            FakeProviderForBackendV2().get_backend(self._backend.name) is not None
+            and not isinstance(self._backend, IBMBackend)
+        )
+
         combined = Options._merge_options(self._options, user_kwargs)
 
         if self._backend:
@@ -154,16 +169,30 @@ def _run_primitive(self, primitive_inputs: Dict, user_kwargs: Dict) -> RuntimeJo
 
         combined = Options._set_default_resilience_options(combined)
         combined = Options._remove_none_values(combined)
-
         primitive_inputs.update(Options._get_program_inputs(combined))
 
-        if self._backend and combined["transpilation"]["skip_transpilation"]:
+        if self._backend and combined["transpilation"]["skip_transpilation"] and not run_simulator:
             for circ in primitive_inputs["circuits"]:
                 self._backend.check_faulty(circ)
 
         logger.info("Submitting job using options %s", combined)
 
         runtime_options = Options._get_runtime_options(combined)
+
+        if run_simulator:
+            # do we need to keep the other 'flat' options as well,
+            # for passing to terra directly?
+            primitive_inputs["optimization_level"] = combined["optimization_level"]
+            runtime_options["backend"] = self._backend
+
+            return self._service.run(
+                program_id=self._program_id(),
+                options=runtime_options,
+                inputs=primitive_inputs,
+                # are the following relevant for simulators?
+                # callback=combined.get("environment", {}).get("callback", None),
+                # result_decoder=DEFAULT_DECODERS.get(self._program_id()),
+            )
         if self._session:
             return self._session.run(
                 program_id=self._program_id(),

qiskit_ibm_runtime/fake_provider/fake_provider.py

@@ -16,8 +16,11 @@
 Fake provider class that provides access to fake backends.
 """
 
+from typing import List
+
 from qiskit.providers.provider import ProviderV1
 from qiskit.providers.exceptions import QiskitBackendNotFoundError
+from qiskit.providers.fake_provider.fake_backend import FakeBackendV2
 
 from .backends import *
 
@@ -69,18 +72,19 @@ class FakeProviderForBackendV2(ProviderV1):
     available in the :mod:`qiskit_ibm_runtime.fake_provider`.
     """
 
-    def get_backend(self, name=None, **kwargs):  # type: ignore
-        backend = self._backends[0]
+    def get_backend(self, name=None, **kwargs) -> FakeBackendV2:  # type: ignore
+        # backend = self._backends[0]
+        # when name=None, this will simply return Almaden. Is that what we want?
         if name:
-            filtered_backends = [backend for backend in self._backends if backend.name() == name]
-            if not filtered_backends:
-                raise QiskitBackendNotFoundError()
-
-            backend = filtered_backends[0]
+            filtered_backends = [backend for backend in self.backends() if backend.name == name]
+            # if not filtered_backends:
+            #     raise QiskitBackendNotFoundError()
+            # I prefer returning None rather than throwing an error. Any objection?
+            backend = filtered_backends[0] if len(filtered_backends) > 0 else None
 
         return backend
 
-    def backends(self, name=None, **kwargs):  # type: ignore
+    def backends(self, name=None, **kwargs) -> List[FakeBackendV2]:  # type: ignore
         return self._backends
 
     def __init__(self) -> None:

qiskit_ibm_runtime/fake_runtime_job.py

@@ -0,0 +1,80 @@
+# This code is part of Qiskit.
+#
+# (C) Copyright IBM 2023.
+#
+# 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.
+
+"""Qiskit fake runtime job."""
+
+from typing import Optional, Dict, Any
+
+from qiskit.primitives.primitive_job import PrimitiveJob
+from qiskit.providers import JobStatus, JobV1
+from qiskit.providers.fake_provider import FakeBackendV2 as FakeBackend
+
+# pylint: disable=cyclic-import
+# from .qiskit_runtime_service import QiskitRuntimeService
+
+
+class FakeRuntimeJob(JobV1):
+    """Representation of a runtime program execution on a simulator."""
+
+    def __init__(
+        self,
+        primitive_job: PrimitiveJob,
+        backend: FakeBackend,
+        job_id: str,
+        program_id: str,
+        # service: "qiskit_runtime_service.QiskitRuntimeService",
+        params: Optional[Dict] = None,
+        # creation_date: Optional[str] = None,
+        # user_callback: Optional[Callable] = None,
+        # result_decoder: Optional[Union[Type[ResultDecoder], Sequence[Type[ResultDecoder]]]] = None,
+        # tags: Optional[List] = None,
+    ) -> None:
+        """FakeRuntimeJob constructor."""
+        super().__init__(backend=backend, job_id=job_id)
+        # self._service = service  # do we need this?
+        self._primitive_job = primitive_job
+        self._job_id = job_id
+        self._params = params or {}
+        self._program_id = program_id
+
+    def result(self) -> Any:
+        """Return the results of the job."""
+        return self._primitive_job.result()
+
+    def cancel(self) -> None:
+        self._primitive_job.cancel()
+
+    def status(self) -> JobStatus:
+        return self._primitive_job.status()
+
+    @property
+    def inputs(self) -> Dict:
+        """Job input parameters.
+
+        Returns:
+            Input parameters used in this job.
+        """
+        return self._params
+
+    def submit(self) -> None:
+        """Unsupported method.
+        Note:
+            This method is not supported, please use
+            :meth:`~qiskit_ibm_runtime.QiskitRuntimeService.run`
+            to submit a job.
+        Raises:
+            NotImplementedError: Upon invocation.
+        """
+        raise NotImplementedError(
+            "job.submit() is not supported. Please use "
+            "QiskitRuntimeService.run() to submit a job."
+        )

qiskit_ibm_runtime/options/utils.py

@@ -12,6 +12,7 @@
 
 """Utility functions for options."""
 
+from qiskit.providers.fake_provider import fake_backend
 from ..ibm_backend import IBMBackend
 
 
@@ -32,20 +33,17 @@ def set_default_error_levels(
     Returns:
         options with correct error level defaults.
     """
+    is_simulator = (
+        isinstance(backend, fake_backend.FakeBackendV2) or backend.configuration().simulator
+    )
     if options.get("optimization_level") is None:
-        if (
-            backend.configuration().simulator
-            and options.get("simulator", {}).get("noise_model") is None
-        ):
+        if is_simulator and options.get("simulator", {}).get("noise_model") is None:
             options["optimization_level"] = 1
         else:
             options["optimization_level"] = default_optimization_level
 
     if options.get("resilience_level") is None:
-        if (
-            backend.configuration().simulator
-            and options.get("simulator", {}).get("noise_model") is None
-        ):
+        if is_simulator and options.get("simulator", {}).get("noise_model") is None:
             options["resilience_level"] = 0
         else:
             options["resilience_level"] = default_resilience_level