Support IBMBackend.run()

qiskit_ibm_runtime/ibm_backend.py

@@ -13,13 +13,16 @@
 """Module for interfacing with an IBM Quantum Backend."""
 
 import logging
-
-from typing import Iterable, Union, Optional, Any, List
+from typing import Iterable, Union, Optional, Any, List, Dict
 from datetime import datetime as python_datetime
 from copy import deepcopy
+from dataclasses import asdict
+import warnings
 
 from qiskit import QuantumCircuit
 from qiskit.qobj.utils import MeasLevel, MeasReturnType
+from qiskit.tools.events.pubsub import Publisher
+
 from qiskit.providers.backend import BackendV2 as Backend
 from qiskit.providers.options import Options
 from qiskit.providers.models import (
@@ -42,11 +45,19 @@
     defaults_from_server_data,
     properties_from_server_data,
 )
-from qiskit_ibm_provider.utils import local_to_utc
+from qiskit_ibm_provider.utils import local_to_utc, are_circuits_dynamic
+from qiskit_ibm_provider.utils.options import QASM2Options, QASM3Options
+from qiskit_ibm_provider.exceptions import IBMBackendValueError, IBMBackendApiError
+from qiskit_ibm_provider.api.exceptions import RequestsApiError
 
-from qiskit_ibm_runtime import (  # pylint: disable=unused-import,cyclic-import
-    qiskit_runtime_service,
-)
+# temporary until we unite the 2 Session classes
+from qiskit_ibm_provider.session import (
+    Session as ProviderSession,
+)  # temporary until we unite the 2 Session classes
+
+from .utils.utils import validate_job_tags
+from . import qiskit_runtime_service  # pylint: disable=unused-import,cyclic-import
+from .runtime_job import RuntimeJob
 
 from .api.clients import RuntimeClient
 from .api.clients.backend import BaseBackendClient
@@ -57,6 +68,9 @@
 
 logger = logging.getLogger(__name__)
 
+QOBJRUNNERPROGRAMID = "circuit-runner"
+QASM3RUNNERPROGRAMID = "qasm3-runner"
+
 
 class IBMBackend(Backend):
     """Backend class interfacing with an IBM Quantum backend.
@@ -180,6 +194,7 @@ def __init__(
         self._defaults = None
         self._target = None
         self._max_circuits = configuration.max_experiments
+        self._session: ProviderSession = None
         if (
             not self._configuration.simulator
             and hasattr(self.options, "noise_model")
@@ -492,10 +507,25 @@ def __call__(self) -> "IBMBackend":
         # For backward compatibility only, can be removed later.
         return self
 
-    def run(self, *args: Any, **kwargs: Any) -> None:
-        """Not supported method"""
-        # pylint: disable=arguments-differ
-        raise RuntimeError("IBMBackend.run() is not supported in the Qiskit Runtime environment.")
+    def _check_circuits_attributes(self, circuits: List[Union[QuantumCircuit, str]]) -> None:
+        """Check that circuits can be executed on backend.
+        Raises:
+            IBMBackendValueError:
+                - If one of the circuits contains more qubits than on the backend."""
+
+        if len(circuits) > self._max_circuits:
+            raise IBMBackendValueError(
+                f"Number of circuits, {len(circuits)} exceeds the "
+                f"maximum for this backend, {self._max_circuits})"
+            )
+        for circ in circuits:
+            if isinstance(circ, QuantumCircuit):
+                if circ.num_qubits > self._configuration.num_qubits:
+                    raise IBMBackendValueError(
+                        f"Circuit contains {circ.num_qubits} qubits, "
+                        f"but backend has only {self.num_qubits}."
+                    )
+                self.check_faulty(circ)
 
     def check_faulty(self, circuit: QuantumCircuit) -> None:
         """Check if the input circuit uses faulty qubits or edges.
@@ -549,6 +579,263 @@ def __deepcopy__(self, _memo: dict = None) -> "IBMBackend":
         cpy._options = deepcopy(self._options, _memo)
         return cpy
 
+    def run(
+        self,
+        circuits: Union[QuantumCircuit, str, List[Union[QuantumCircuit, str]]],
+        dynamic: bool = None,
+        job_tags: Optional[List[str]] = None,
+        init_circuit: Optional[QuantumCircuit] = None,
+        init_num_resets: Optional[int] = None,
+        header: Optional[Dict] = None,
+        shots: Optional[Union[int, float]] = None,
+        memory: Optional[bool] = None,
+        meas_level: Optional[Union[int, MeasLevel]] = None,
+        meas_return: Optional[Union[str, MeasReturnType]] = None,
+        rep_delay: Optional[float] = None,
+        init_qubits: Optional[bool] = None,
+        use_measure_esp: Optional[bool] = None,
+        noise_model: Optional[Any] = None,
+        seed_simulator: Optional[int] = None,
+        **run_config: Dict,
+    ) -> RuntimeJob:
+        """Run on the backend.
+        If a keyword specified here is also present in the ``options`` attribute/object,
+        the value specified here will be used for this run.
+
+        Args:
+            circuits: An individual or a
+                list of :class:`~qiskit.circuits.QuantumCircuit`.
+            dynamic: Whether the circuit is dynamic (uses in-circuit conditionals)
+            job_tags: Tags to be assigned to the job. The tags can subsequently be used
+                as a filter in the :meth:`jobs()` function call.
+            init_circuit: A quantum circuit to execute for initializing qubits before each circuit.
+                If specified, ``init_num_resets`` is ignored. Applicable only if ``dynamic=True``
+                is specified.
+            init_num_resets: The number of qubit resets to insert before each circuit execution.
+
+            The following parameters are applicable only if ``dynamic=False`` is specified or
+            defaulted to.
+
+            header: User input that will be attached to the job and will be
+                copied to the corresponding result header. Headers do not affect the run.
+                This replaces the old ``Qobj`` header.
+            shots: Number of repetitions of each circuit, for sampling. Default: 4000
+                or ``max_shots`` from the backend configuration, whichever is smaller.
+            memory: If ``True``, per-shot measurement bitstrings are returned as well
+                (provided the backend supports it). For OpenPulse jobs, only
+                measurement level 2 supports this option.
+            meas_level: Level of the measurement output for pulse experiments. See
+                `OpenPulse specification <https://arxiv.org/pdf/1809.03452.pdf>`_ for details:
+
+                * ``0``, measurements of the raw signal (the measurement output pulse envelope)
+                * ``1``, measurement kernel is selected (a complex number obtained after applying the
+                  measurement kernel to the measurement output signal)
+                * ``2`` (default), a discriminator is selected and the qubit state is stored (0 or 1)
+
+            meas_return: Level of measurement data for the backend to return. For ``meas_level`` 0 and 1:
+
+                * ``single`` returns information from every shot.
+                * ``avg`` returns average measurement output (averaged over number of shots).
+
+            rep_delay: Delay between programs in seconds. Only supported on certain
+                backends (if ``backend.configuration().dynamic_reprate_enabled=True``).
+                If supported, ``rep_delay`` must be from the range supplied
+                by the backend (``backend.configuration().rep_delay_range``). Default is given by
+                ``backend.configuration().default_rep_delay``.
+            init_qubits: Whether to reset the qubits to the ground state for each shot.
+                Default: ``True``.
+            use_measure_esp: Whether to use excited state promoted (ESP) readout for measurements
+                which are the terminal instruction to a qubit. ESP readout can offer higher fidelity
+                than standard measurement sequences. See
+                `here <https://arxiv.org/pdf/2008.08571.pdf>`_.
+                Default: ``True`` if backend supports ESP readout, else ``False``. Backend support
+                for ESP readout is determined by the flag ``measure_esp_enabled`` in
+                ``backend.configuration()``.
+            noise_model: Noise model. (Simulators only)
+            seed_simulator: Random seed to control sampling. (Simulators only)
+            **run_config: Extra arguments used to configure the run.
+
+        Returns:
+            The job to be executed.
+
+        Raises:
+            IBMBackendApiError: If an unexpected error occurred while submitting
+                the job.
+            IBMBackendApiProtocolError: If an unexpected value received from
+                 the server.
+            IBMBackendValueError:
+                - If an input parameter value is not valid.
+                - If ESP readout is used and the backend does not support this.
+        """
+        # pylint: disable=arguments-differ
+        validate_job_tags(job_tags)
+        if not isinstance(circuits, List):
+            circuits = [circuits]
+        self._check_circuits_attributes(circuits)
+
+        if use_measure_esp and getattr(self.configuration(), "measure_esp_enabled", False) is False:
+            raise IBMBackendValueError(
+                "ESP readout not supported on this device. Please make sure the flag "
+                "'use_measure_esp' is unset or set to 'False'."
+            )
+        actually_dynamic = are_circuits_dynamic(circuits)
+        if dynamic is False and actually_dynamic:
+            warnings.warn(
+                "Parameter 'dynamic' is False, but the circuit contains dynamic constructs."
+            )
+        dynamic = dynamic or actually_dynamic
+
+        if dynamic and "qasm3" not in getattr(self.configuration(), "supported_features", []):
+            warnings.warn(f"The backend {self.name} does not support dynamic circuits.")
+
+        status = self.status()
+        if status.operational is True and status.status_msg != "active":
+            warnings.warn(f"The backend {self.name} is currently paused.")
+
+        program_id = str(run_config.get("program_id", ""))
+        if program_id:
+            run_config.pop("program_id", None)
+        else:
+            program_id = QASM3RUNNERPROGRAMID if dynamic else QOBJRUNNERPROGRAMID
+
+        image: Optional[str] = run_config.get("image", None)  # type: ignore
+        if image is not None:
+            image = str(image)
+
+        if isinstance(init_circuit, bool):
+            raise IBMBackendApiError(
+                "init_circuit does not accept boolean values. "
+                "A quantum circuit should be passed in instead."
+            )
+
+        if isinstance(shots, float):
+            shots = int(shots)
+
+        run_config_dict = self._get_run_config(
+            program_id=program_id,
+            init_circuit=init_circuit,
+            init_num_resets=init_num_resets,
+            header=header,
+            shots=shots,
+            memory=memory,
+            meas_level=meas_level,
+            meas_return=meas_return,
+            rep_delay=rep_delay,
+            init_qubits=init_qubits,
+            use_measure_esp=use_measure_esp,
+            noise_model=noise_model,
+            seed_simulator=seed_simulator,
+            **run_config,
+        )
+
+        run_config_dict["circuits"] = circuits
+
+        return self._runtime_run(
+            program_id=program_id,
+            inputs=run_config_dict,
+            backend_name=self.name,
+            job_tags=job_tags,
+            image=image,
+        )
+
+    def _runtime_run(
+        self,
+        program_id: str,
+        inputs: Dict,
+        backend_name: str,
+        job_tags: Optional[List[str]] = None,
+        image: Optional[str] = None,
+    ) -> RuntimeJob:
+        """Runs the runtime program and returns the corresponding job object"""
+        hgp_name = None
+        if self._service._channel == "ibm_quantum":
+            hgp_name = self._instance or self._service._get_hgp().name
+
+        session = self._session
+        if session and not session.active:
+            raise RuntimeError(f"The session {session.session_id} is closed.")
+        session_id = session.session_id if session else None
+        start_session = session is not None and session_id is None
+
+        log_level = getattr(self.options, "log_level", None)  # temporary
+        try:
+            response = self._api_client.program_run(
+                program_id=program_id,
+                backend_name=backend_name,
+                params=inputs,
+                hgp=hgp_name,
+                log_level=log_level,
+                job_tags=job_tags,
+                session_id=session_id,
+                start_session=start_session,
+                image=image,
+            )
+        except RequestsApiError as ex:
+            raise IBMBackendApiError("Error submitting job: {}".format(str(ex))) from ex
+        session_id = response.get("session_id", None)
+        if self._session:
+            self._session._session_id = session_id
+        try:
+            job = RuntimeJob(
+                backend=self,
+                api_client=self._api_client,
+                client_params=self._service._client_params,
+                job_id=response["id"],
+                program_id=program_id,
+                session_id=session_id,
+                service=self.service,
+            )
+            logger.debug("Job %s was successfully submitted.", job.job_id())
+        except TypeError as err:
+            logger.debug("Invalid job data received: %s", response)
+            raise IBMBackendApiProtocolError(
+                "Unexpected return value received from the server "
+                "when submitting job: {}".format(str(err))
+            ) from err
+        Publisher().publish("ibm.job.start", job)
+        return job
+
+    def _get_run_config(self, program_id: str, **kwargs: Any) -> Dict:
+        """Return the consolidated runtime configuration."""
+        # Check if is a QASM3 like program id.
+        if program_id.startswith(QASM3RUNNERPROGRAMID):
+            fields = asdict(QASM3Options()).keys()
+            run_config_dict = QASM3Options().to_transport_dict()
+        else:
+            fields = asdict(QASM2Options()).keys()
+            run_config_dict = QASM2Options().to_transport_dict()
+        backend_options = self._options.__dict__
+        for key, val in kwargs.items():
+            if val is not None:
+                run_config_dict[key] = val
+                if key not in fields and not self.configuration().simulator:
+                    warnings.warn(  # type: ignore[unreachable]
+                        f"{key} is not a recognized runtime option and may be ignored by the backend.",
+                        stacklevel=4,
+                    )
+            elif backend_options.get(key) is not None and key in fields:
+                run_config_dict[key] = backend_options[key]
+        return run_config_dict
+
+    def open_session(self) -> ProviderSession:
+        """Open session"""
+        self._session = ProviderSession()
+        return self._session
+
+    @property
+    def session(self) -> ProviderSession:
+        """Return session"""
+        return self._session
+
+    def cancel_session(self) -> None:
+        """Cancel session. All pending jobs will be cancelled."""
+        if self._session:
+            self._session.cancel()
+            if self._session.session_id:
+                self._api_client.close_session(self._session.session_id)
+
+        self._session = None
+
 
 class IBMRetiredBackend(IBMBackend):
     """Backend class interfacing with an IBM Quantum device no longer available."""

qiskit_ibm_runtime/qiskit_runtime_service.py

@@ -34,6 +34,7 @@
 from qiskit_ibm_provider.utils.backend_decoder import configuration_from_server_data
 from qiskit_ibm_runtime import ibm_backend
 
+from .utils.utils import validate_job_tags
 from .accounts import AccountManager, Account, ChannelType
 from .api.clients import AuthClient, VersionClient
 from .api.clients.runtime import RuntimeClient
@@ -1441,6 +1442,8 @@ def jobs(
                     "The 'instance' keyword is only supported for ``ibm_quantum`` runtime."
                 )
             hub, group, project = from_instance_format(instance)
+        if job_tags:
+            validate_job_tags(job_tags)
 
         job_responses = []  # type: List[Dict[str, Any]]
         current_page_limit = limit or 20