Add experimental options (#1067)

* add options

* fix mypy

* rename meas err mit

* Add additional resilience options

* Add additional execution options for twirling

* A twirling strategy option and validation

* handle default resilience options

* add _experimental

* Update default resilience options (#1062)

* remove default resilience options

* add reno

* add logic to override default

* add test

* purge None values from options

(cherry picked from commit 76603f2)

* add finalize options

* add tests

* Update qiskit_ibm_runtime/options/resilience_options.py

* add validation

* lint

* lint again

* lint again

* Allow None values for specific options to be passed through

* Fix parameter validation, allow computational basis

* black

* Fix ZneExtrapolatorType validation

* lint

* lint again

* fix mypy

* Fix ZNE extrapolator default option

* fix level options

* black

* use _isreal

* Disable gate twirling for default lvl 1 opts

* Support for legacy options

---------

Co-authored-by: Christopher J. Wood <cjwood@us.ibm.com>
Co-authored-by: Kevin Tian <kevin.tian@ibm.com>
Co-authored-by: mberna <mariana.bernagozzi@gmail.com>
Co-authored-by: Mariana C Bernagozzi <Mariana.Bernagozzi@ibm.com>

qiskit_ibm_runtime/__init__.py

@@ -175,6 +175,7 @@ def result_callback(job_id, result):
 """
 
 import logging
+import warnings
 
 from .qiskit_runtime_service import QiskitRuntimeService
 from .ibm_backend import IBMBackend
@@ -203,3 +204,5 @@ def result_callback(job_id, result):
 """The environment variable name that is used to set the level for the IBM Quantum logger."""
 QISKIT_IBM_RUNTIME_LOG_FILE = "QISKIT_IBM_RUNTIME_LOG_FILE"
 """The environment variable name that is used to set the file for the IBM Quantum logger."""
+
+warnings.warn("You are using the experimental branch. Stability is not guaranteed.")

qiskit_ibm_runtime/base_primitive.py

@@ -22,7 +22,6 @@
 from qiskit.providers.options import Options as TerraOptions
 
 from .options import Options
-from .options.utils import set_default_error_levels
 from .runtime_job import RuntimeJob
 from .ibm_backend import IBMBackend
 from .session import get_cm_session
@@ -75,15 +74,6 @@ def __init__(
         self._service: QiskitRuntimeService = None
         self._backend: Optional[IBMBackend] = None
 
-        if options is None:
-            self._options = asdict(Options())
-        elif isinstance(options, Options):
-            self._options = asdict(copy.deepcopy(options))
-        else:
-            options_copy = copy.deepcopy(options)
-            default_options = asdict(Options())
-            self._options = Options._merge_options(default_options, options_copy)
-
         if isinstance(session, Session):
             self._session = session
             self._service = self._session.service
@@ -148,6 +138,21 @@ def __init__(
                 raise ValueError(
                     "A backend or session must be specified when not using ibm_cloud channel."
                 )
+        self._simulator_backend = (
+            self._backend.configuration().simulator if self._backend else False
+        )
+
+        if options is None:
+            self._options = asdict(Options())
+        elif isinstance(options, Options):
+            self._options = asdict(copy.deepcopy(options))
+        else:
+            options_copy = copy.deepcopy(options)
+            default_options = asdict(Options())
+            self._options = Options._merge_options_with_defaults(
+                default_options, options_copy, is_simulator=self._simulator_backend
+            )
+
         # self._first_run = True
         # self._circuits_map = {}
         # if self.circuits:
@@ -169,20 +174,11 @@ def _run_primitive(self, primitive_inputs: Dict, user_kwargs: Dict) -> RuntimeJo
         Returns:
             Submitted job.
         """
-        combined = Options._merge_options(self._options, user_kwargs)
-
-        if self._backend:
-            combined = set_default_error_levels(
-                combined,
-                self._backend,
-                Options._DEFAULT_OPTIMIZATION_LEVEL,
-                Options._DEFAULT_RESILIENCE_LEVEL,
-            )
-        else:
-            combined["optimization_level"] = Options._DEFAULT_OPTIMIZATION_LEVEL
-            combined["resilience_level"] = Options._DEFAULT_RESILIENCE_LEVEL
-
+        combined = Options._merge_options_with_defaults(
+            self._options, user_kwargs, self._simulator_backend
+        )
         self._validate_options(combined)
+
         primitive_inputs.update(Options._get_program_inputs(combined))
 
         if self._backend and combined["transpilation"]["skip_transpilation"]:
@@ -238,7 +234,9 @@ def set_options(self, **fields: Any) -> None:
         Args:
             **fields: The fields to update the options
         """
-        self._options = Options._merge_options(self._options, fields)
+        self._options = Options._merge_options_with_defaults(
+            self._options, fields, self._simulator_backend
+        )
 
     @abstractmethod
     def _validate_options(self, options: dict) -> None:

qiskit_ibm_runtime/estimator.py

@@ -14,27 +14,51 @@
 
 from __future__ import annotations
 import os
-from typing import Optional, Dict, Sequence, Any, Union
+from typing import Optional, Dict, Sequence, Any, Union, Mapping
 import logging
 
+import numpy as np
+from numpy.typing import ArrayLike
+
 from qiskit.circuit import QuantumCircuit
-from qiskit.opflow import PauliSumOp
 from qiskit.quantum_info.operators.base_operator import BaseOperator
 from qiskit.primitives import BaseEstimator
+from qiskit.quantum_info import SparsePauliOp, Pauli
+from qiskit.primitives.utils import init_observable
+from qiskit.circuit import Parameter
+from qiskit.primitives.base.base_primitive import _isreal
 
 # TODO import _circuit_key from terra once 0.23 is released
 from .runtime_job import RuntimeJob
 from .ibm_backend import IBMBackend
 from .options import Options
 from .base_primitive import BasePrimitive
 from .utils.qctrl import validate as qctrl_validate
+from .utils.deprecation import issue_deprecation_msg
 
 # pylint: disable=unused-import,cyclic-import
 from .session import Session
 
 logger = logging.getLogger(__name__)
 
 
+BasisObservableLike = Union[str, Pauli, SparsePauliOp, Mapping[Union[str, Pauli], complex]]
+"""Types that can be natively used to construct a :const:`BasisObservable`."""
+
+ObservablesArrayLike = Union[ArrayLike, Sequence[BasisObservableLike], BasisObservableLike]
+
+ParameterMappingLike = Mapping[
+    Parameter, Union[float, np.ndarray, Sequence[float], Sequence[Sequence[float]]]
+]
+BindingsArrayLike = Union[
+    float,
+    np.ndarray,
+    ParameterMappingLike,
+    Sequence[Union[float, Sequence[float], np.ndarray, ParameterMappingLike]],
+]
+"""Parameter types that can be bound to a single circuit."""
+
+
 class Estimator(BasePrimitive, BaseEstimator):
     """Class for interacting with Qiskit Runtime Estimator primitive service.
 
@@ -85,6 +109,7 @@ class Estimator(BasePrimitive, BaseEstimator):
     """
 
     _PROGRAM_ID = "estimator"
+    _ALLOWED_BASIS: str = "IXYZ01+-rl"
 
     def __init__(
         self,
@@ -119,8 +144,11 @@ def __init__(
     def run(  # pylint: disable=arguments-differ
         self,
         circuits: QuantumCircuit | Sequence[QuantumCircuit],
-        observables: BaseOperator | PauliSumOp | Sequence[BaseOperator | PauliSumOp],
-        parameter_values: Sequence[float] | Sequence[Sequence[float]] | None = None,
+        observables: Sequence[ObservablesArrayLike]
+        | ObservablesArrayLike
+        | Sequence[BaseOperator]
+        | BaseOperator,
+        parameter_values: BindingsArrayLike | Sequence[BindingsArrayLike] | None = None,
         **kwargs: Any,
     ) -> RuntimeJob:
         """Submit a request to the estimator primitive.
@@ -155,7 +183,7 @@ def run(  # pylint: disable=arguments-differ
     def _run(  # pylint: disable=arguments-differ
         self,
         circuits: Sequence[QuantumCircuit],
-        observables: Sequence[BaseOperator | PauliSumOp],
+        observables: Sequence[ObservablesArrayLike],
         parameter_values: Sequence[Sequence[float]],
         **kwargs: Any,
     ) -> RuntimeJob:
@@ -220,6 +248,84 @@ def _validate_options(self, options: dict) -> None:
                 )
         Options.validate_options(options)
 
+    @staticmethod
+    def _validate_observables(
+        observables: Sequence[ObservablesArrayLike] | ObservablesArrayLike,
+    ) -> Sequence[ObservablesArrayLike]:
+        def _check_and_init(obs: Any) -> Any:
+            if isinstance(obs, str):
+                pass
+                if not all(basis in Estimator._ALLOWED_BASIS for basis in obs):
+                    raise ValueError(
+                        f"Invalid character(s) found in observable string. "
+                        f"Allowed basis are {Estimator._ALLOWED_BASIS}."
+                    )
+            elif isinstance(obs, Sequence):
+                return tuple(_check_and_init(obs_) for obs_ in obs)
+            elif not isinstance(obs, (Pauli, SparsePauliOp)) and isinstance(obs, BaseOperator):
+                issue_deprecation_msg(
+                    msg="Only Pauli and SparsePauliOp operators can be used as observables.",
+                    version="0.13",
+                    remedy="",
+                )
+                return init_observable(obs)
+            elif isinstance(obs, Mapping):
+                for key in obs.keys():
+                    _check_and_init(key)
+
+            return obs
+
+        if isinstance(observables, str) or not isinstance(observables, Sequence):
+            observables = (observables,)
+
+        if len(observables) == 0:
+            raise ValueError("No observables were provided.")
+
+        return tuple(_check_and_init(obs_array) for obs_array in observables)
+
+    @staticmethod
+    def _validate_parameter_values(
+        parameter_values: BindingsArrayLike | Sequence[BindingsArrayLike] | None,
+        default: Sequence[Sequence[float]] | Sequence[float] | None = None,
+    ) -> Sequence:
+
+        # Allow optional (if default)
+        if parameter_values is None:
+            if default is None:
+                raise ValueError("No default `parameter_values`, optional input disallowed.")
+            parameter_values = default
+
+        # Convert single input types to length-1 lists
+        if _isreal(parameter_values):
+            parameter_values = [[parameter_values]]
+        elif isinstance(parameter_values, Mapping):
+            parameter_values = [parameter_values]
+        elif isinstance(parameter_values, Sequence) and all(
+            _isreal(item) for item in parameter_values
+        ):
+            parameter_values = [parameter_values]
+        return tuple(parameter_values)  # type: ignore[arg-type]
+
+    @staticmethod
+    def _cross_validate_circuits_parameter_values(
+        circuits: tuple[QuantumCircuit, ...], parameter_values: tuple[tuple[float, ...], ...]
+    ) -> None:
+        if len(circuits) != len(parameter_values):
+            raise ValueError(
+                f"The number of circuits ({len(circuits)}) does not match "
+                f"the number of parameter value sets ({len(parameter_values)})."
+            )
+
+    @staticmethod
+    def _cross_validate_circuits_observables(
+        circuits: tuple[QuantumCircuit, ...], observables: tuple[ObservablesArrayLike, ...]
+    ) -> None:
+        if len(circuits) != len(observables):
+            raise ValueError(
+                f"The number of circuits ({len(circuits)}) does not match "
+                f"the number of observables ({len(observables)})."
+            )
+
     @classmethod
     def _program_id(cls) -> str:
         """Return the program ID."""

qiskit_ibm_runtime/options/__init__.py

@@ -49,6 +49,7 @@
    ExecutionOptions
    EnvironmentOptions
    SimulatorOptions
+   TwirlingOptions
 
 """
 
@@ -58,3 +59,4 @@
 from .simulator_options import SimulatorOptions
 from .transpilation_options import TranspilationOptions
 from .resilience_options import ResilienceOptions
+from .twirling_options import TwirlingOptions

qiskit_ibm_runtime/options/execution_options.py

@@ -13,13 +13,17 @@
 """Execution options."""
 
 from dataclasses import dataclass
-from typing import Literal, get_args
+from typing import Literal, get_args, Optional
+from numbers import Integral
 
 from .utils import _flexible
 
 ExecutionSupportedOptions = Literal[
     "shots",
     "init_qubits",
+    "samples",
+    "shots_per_sample",
+    "interleave_samples",
 ]
 
 
@@ -29,14 +33,34 @@ class ExecutionOptions:
     """Execution options.
 
     Args:
-        shots: Number of repetitions of each circuit, for sampling. Default: 4000.
+        shots: Number of repetitions of each circuit, for sampling. Default: 4096.
 
         init_qubits: Whether to reset the qubits to the ground state for each shot.
             Default: ``True``.
+
+        samples: The number of samples of each measurement circuit to run. This
+            is used when twirling or resilience levels 1, 2, 3. If None it will
+            be calculated automatically based on the ``shots`` and
+            ``shots_per_sample`` (if specified).
+            Default: None
+
+        shots_per_sample: The number of shots per sample of each measurement
+            circuit to run. This is used when twirling or resilience levels 1, 2, 3.
+            If None it will be calculated automatically based on the ``shots`` and
+            ``samples`` (if specified).
+            Default: None
+
+        interleave_samples: If True interleave samples from different measurement
+            circuits when running. If False run all samples from each measurement
+            circuit in order.
+            Default: False
     """
 
-    shots: int = 4000
+    shots: int = 4096
     init_qubits: bool = True
+    samples: Optional[int] = None
+    shots_per_sample: Optional[int] = None
+    interleave_samples: bool = False
 
     @staticmethod
     def validate_execution_options(execution_options: dict) -> None:
@@ -47,3 +71,33 @@ def validate_execution_options(execution_options: dict) -> None:
         for opt in execution_options:
             if not opt in get_args(ExecutionSupportedOptions):
                 raise ValueError(f"Unsupported value '{opt}' for execution.")
+
+        shots = execution_options.get("shots")
+        samples = execution_options.get("samples")
+        shots_per_sample = execution_options.get("shots_per_sample")
+        if (
+            shots is not None
+            and samples is not None
+            and shots_per_sample is not None
+            and shots != samples * shots_per_sample
+        ):
+            raise ValueError(
+                f"If shots ({shots}) != samples ({samples}) * shots_per_sample ({shots_per_sample})"
+            )
+        if shots is not None:
+            if not isinstance(shots, Integral):
+                raise ValueError(f"shots must be None or an integer, not {type(shots)}")
+            if shots < 1:
+                raise ValueError("shots must be None or >= 1")
+        if samples is not None:
+            if not isinstance(samples, Integral):
+                raise ValueError(f"samples must be None or an integer, not {type(samples)}")
+            if samples < 1:
+                raise ValueError("samples must be None or >= 1")
+        if shots_per_sample is not None:
+            if not isinstance(shots_per_sample, Integral):
+                raise ValueError(
+                    f"shots_per_sample must be None or an integer, not {type(shots_per_sample)}"
+                )
+            if shots_per_sample < 1:
+                raise ValueError("shots_per_sample must be None or >= 1")