SamplerV2 options

qiskit_ibm_runtime/__init__.py

@@ -189,8 +189,8 @@ def result_callback(job_id, result):
 from .version import __version__
 
 from .estimator import EstimatorV2, EstimatorV1 as Estimator
-from .sampler import SamplerV1 as Sampler
-from .options import Options, EstimatorOptions
+from .sampler import SamplerV2, SamplerV1 as Sampler
+from .options import Options, EstimatorOptions, SamplerOptions
 
 # Setup the logger for the IBM Quantum Provider package.
 logger = logging.getLogger(__name__)

qiskit_ibm_runtime/estimator.py

@@ -69,7 +69,6 @@
 class Estimator:
     """Base class for Qiskit Runtime Estimator."""
 
-    _PROGRAM_ID = "estimator"
     version = 0
 
 

qiskit_ibm_runtime/options/__init__.py

@@ -51,6 +51,7 @@
    SimulatorOptions
    TwirlingOptions
    EstimatorOptions
+   SamplerOptions
 
 """
 
@@ -62,3 +63,4 @@
 from .resilience_options import ResilienceOptionsV1 as ResilienceOptions
 from .twirling_options import TwirlingOptions
 from .estimator_options import EstimatorOptions
+from .sampler_options import SamplerOptions

qiskit_ibm_runtime/options/estimator_options.py

@@ -13,18 +13,14 @@
 """Primitive options."""
 
 from typing import Union, Literal
-import copy
 
-from qiskit.transpiler import CouplingMap
 from pydantic.dataclasses import dataclass as pydantic_dataclass
 from pydantic import Field, ConfigDict, field_validator
 
 from .utils import (
     Dict,
     Unset,
     UnsetType,
-    _remove_dict_unset_values,
-    merge_options,
     skip_unset_validation,
 )
 from .execution_options import ExecutionOptionsV2
@@ -74,6 +70,8 @@ class EstimatorOptions(OptionsV2):
             Allowed values are ``"XX"``, ``"XpXm"``, ``"XY4"``.
             Default: None
 
+        seed_estimator: Seed used to control sampling.
+
         transpilation: Transpilation options. See :class:`TranspilationOptions` for all
             available options.
 
@@ -90,14 +88,14 @@ class EstimatorOptions(OptionsV2):
 
     """
 
-    _VERSION: int = Field(2, frozen=True)  # pylint: disable=invalid-name
     _MAX_OPTIMIZATION_LEVEL: int = Field(3, frozen=True)  # pylint: disable=invalid-name
     _MAX_RESILIENCE_LEVEL: int = Field(3, frozen=True)  # pylint: disable=invalid-name
 
     # Sadly we cannot use pydantic's built in validation because it won't work on Unset.
     optimization_level: Union[UnsetType, int] = Unset
     resilience_level: Union[UnsetType, int] = Unset
     dynamical_decoupling: Union[UnsetType, DDSequenceType] = Unset
+    seed_estimator: Union[UnsetType, int] = Unset
     transpilation: Union[TranspilationOptions, Dict] = Field(default_factory=TranspilationOptions)
     resilience: Union[ResilienceOptionsV2, Dict] = Field(default_factory=ResilienceOptionsV2)
     execution: Union[ExecutionOptionsV2, Dict] = Field(default_factory=ExecutionOptionsV2)
@@ -127,113 +125,3 @@ def _validate_resilience_level(cls, resilience_level: int) -> int:
                 f"0-{EstimatorOptions._MAX_RESILIENCE_LEVEL}"
             )
         return resilience_level
-
-    @staticmethod
-    def _get_program_inputs(options: dict) -> dict:
-        """Convert the input options to program compatible inputs.
-
-        Returns:
-            Inputs acceptable by primitives.
-        """
-
-        sim_options = options.get("simulator", {})
-        inputs = {}
-        inputs["transpilation"] = copy.copy(options.get("transpilation", {}))
-        inputs["skip_transpilation"] = inputs["transpilation"].pop("skip_transpilation")
-        coupling_map = sim_options.get("coupling_map", None)
-        # TODO: We can just move this to json encoder
-        if isinstance(coupling_map, CouplingMap):
-            coupling_map = list(map(list, coupling_map.get_edges()))
-        inputs["transpilation"].update(
-            {
-                "optimization_level": options.get("optimization_level"),
-                "coupling_map": coupling_map,
-                "basis_gates": sim_options.get("basis_gates", None),
-            }
-        )
-
-        inputs["resilience_level"] = options.get("resilience_level")
-        inputs["resilience"] = options.get("resilience", {})
-
-        inputs["twirling"] = options.get("twirling", {})
-
-        inputs["execution"] = options.get("execution", {})
-        inputs["execution"].update(
-            {
-                "noise_model": sim_options.get("noise_model", Unset),
-                "seed_simulator": sim_options.get("seed_simulator", Unset),
-            }
-        )
-
-        # Add arbitrary experimental options
-        if isinstance(options.get("experimental", None), dict):
-            inputs = merge_options(inputs, options.get("experimental"))
-
-        # Remove image
-        inputs.pop("image", None)
-
-        inputs["_experimental"] = True
-        inputs["version"] = EstimatorOptions._VERSION
-        _remove_dict_unset_values(inputs)
-
-        return inputs
-
-
-# @dataclass(frozen=True)
-# class _ResilienceLevel0Options:
-#     resilience_level: int = 0
-#     resilience: ResilienceOptions = field(
-#         default_factory=lambda: ResilienceOptions(
-#             measure_noise_mitigation=False, zne_mitigation=False, pec_mitigation=False
-#         )
-#     )
-#     twirling: TwirlingOptions = field(
-#         default_factory=lambda: TwirlingOptions(gates=False, measure=False)
-#     )
-
-
-# @dataclass(frozen=True)
-# class _ResilienceLevel1Options:
-#     resilience_level: int = 1
-#     resilience: ResilienceOptions = field(
-#         default_factory=lambda: ResilienceOptions(
-#             measure_noise_mitigation=True, zne_mitigation=False, pec_mitigation=False
-#         )
-#     )
-#     twirling: TwirlingOptions = field(
-#         default_factory=lambda: TwirlingOptions(gates=False, measure=True, strategy="active-accum")
-#     )
-
-
-# @dataclass(frozen=True)
-# class _ResilienceLevel2Options:
-#     resilience_level: int = 2
-#     resilience: ResilienceOptions = field(
-#         default_factory=lambda: ResilienceOptions(
-#             measure_noise_mitigation=True, pec_mitigation=False, **asdict(_ZneOptions())
-#         )
-#     )
-#     twirling: TwirlingOptions = field(
-#         default_factory=lambda: TwirlingOptions(gates=True, measure=True, strategy="active-accum")
-#     )
-
-
-# @dataclass(frozen=True)
-# class _ResilienceLevel3Options:
-#     resilience_level: int = 3
-#     resilience: ResilienceOptions = field(
-#         default_factory=lambda: ResilienceOptions(
-#             measure_noise_mitigation=True, zne_mitigation=False, **asdict(_PecOptions())
-#         )
-#     )
-#     twirling: TwirlingOptions = field(
-#         default_factory=lambda: TwirlingOptions(gates=True, measure=True, strategy="active")
-#     )
-
-
-# _DEFAULT_RESILIENCE_LEVEL_OPTIONS = {
-#     0: _ResilienceLevel0Options(),
-#     1: _ResilienceLevel1Options(),
-#     2: _ResilienceLevel2Options(),
-#     3: _ResilienceLevel3Options(),
-# }

qiskit_ibm_runtime/options/options.py

@@ -22,7 +22,7 @@
 from pydantic.dataclasses import dataclass as pydantic_dataclass
 from pydantic import Field, ConfigDict
 
-from .utils import Dict, _to_obj, UnsetType, Unset, _remove_dict_unset_values
+from .utils import Dict, _to_obj, UnsetType, Unset, _remove_dict_unset_values, merge_options
 from .environment_options import EnvironmentOptions
 from .execution_options import ExecutionOptionsV1 as ExecutionOptions
 from .simulator_options import SimulatorOptions
@@ -93,10 +93,77 @@ class OptionsV2(BaseOptions):
             :class:`SimulatorOptions` for all available options.
     """
 
+    _VERSION: int = Field(2, frozen=True)  # pylint: disable=invalid-name
+
+    # Options not really related to primitives.
     max_execution_time: Union[UnsetType, int] = Unset
     environment: Union[EnvironmentOptions, Dict] = Field(default_factory=EnvironmentOptions)
     simulator: Union[SimulatorOptions, Dict] = Field(default_factory=SimulatorOptions)
 
+    @staticmethod
+    def _get_program_inputs(options: dict) -> dict:
+        """Convert the input options to program compatible inputs.
+
+        Returns:
+            Inputs acceptable by primitives.
+        """
+
+        def _set_if_exists(name: str, _inputs: dict, _options: dict) -> None:
+            if name in _options:
+                _inputs[name] = _options[name]
+
+        options_copy = copy.deepcopy(options)
+        sim_options = options_copy.get("simulator", {})
+        inputs = {}
+        inputs["transpilation"] = options_copy.get("transpilation", {})
+        inputs["skip_transpilation"] = inputs["transpilation"].pop("skip_transpilation")
+        coupling_map = sim_options.get("coupling_map", Unset)
+        # TODO: We can just move this to json encoder
+        if isinstance(coupling_map, CouplingMap):
+            coupling_map = list(map(list, coupling_map.get_edges()))
+        inputs["transpilation"].update(
+            {
+                "optimization_level": options_copy.get("optimization_level", Unset),
+                "coupling_map": coupling_map,
+                "basis_gates": sim_options.get("basis_gates", Unset),
+            }
+        )
+
+        for fld in [
+            "resilience_level",
+            "resilience",
+            "twirling",
+            "dynamical_decoupling",
+            "seed_estimator",
+        ]:
+            _set_if_exists(fld, inputs, options_copy)
+
+        inputs["execution"] = options_copy.get("execution", {})
+        inputs["execution"].update(
+            {
+                "noise_model": sim_options.get("noise_model", Unset),
+                "seed_simulator": sim_options.get("seed_simulator", Unset),
+            }
+        )
+
+        # Add arbitrary experimental options
+        if isinstance(options_copy.get("experimental", None), dict):
+            inputs = merge_options(inputs, options_copy.get("experimental"))
+
+        # Remove image
+        inputs.pop("image", None)
+
+        inputs["_experimental"] = True
+        inputs["version"] = OptionsV2._VERSION
+        _remove_dict_unset_values(inputs)
+
+        # Remove empty dictionaries
+        for key, val in list(inputs.items()):
+            if isinstance(val, dict) and not val:
+                del inputs[key]
+
+        return inputs
+
 
 @dataclass
 class Options(BaseOptions):

qiskit_ibm_runtime/options/sampler_options.py

@@ -0,0 +1,91 @@
+# This code is part of Qiskit.
+#
+# (C) Copyright IBM 2022.
+#
+# 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.
+
+"""Sampler options."""
+
+from typing import Union, Literal
+
+from pydantic.dataclasses import dataclass as pydantic_dataclass
+from pydantic import Field, ConfigDict, field_validator
+
+from .utils import (
+    Dict,
+    Unset,
+    UnsetType,
+    skip_unset_validation,
+)
+from .execution_options import ExecutionOptionsV2
+from .transpilation_options import TranspilationOptions
+from .twirling_options import TwirlingOptions
+from .options import OptionsV2
+
+DDSequenceType = Literal["XX", "XpXm", "XY4"]
+
+
+@pydantic_dataclass(
+    config=ConfigDict(validate_assignment=True, arbitrary_types_allowed=True, extra="forbid")
+)
+class SamplerOptions(OptionsV2):
+    """Options for v2 Sampler.
+
+    Args:
+        optimization_level: How much optimization to perform on the circuits.
+            Higher levels generate more optimized circuits,
+            at the expense of longer transpilation times. This is based on the
+            ``optimization_level`` parameter in qiskit-terra but may include
+            backend-specific optimization. Default: 1.
+
+            * 0: no optimization
+            * 1: light optimization
+            * 2: heavy optimization
+            * 3: even heavier optimization
+
+        dynamical_decoupling: Optional, specify a dynamical decoupling sequence to use.
+            Allowed values are ``"XX"``, ``"XpXm"``, ``"XY4"``.
+            Default: None
+
+        transpilation: Transpilation options. See :class:`TranspilationOptions` for all
+            available options.
+
+        execution: Execution time options. See :class:`ExecutionOptionsV2` for all available options.
+
+        twirling: Pauli-twirling related options. See :class:`TwirlingOptions` for all available options.
+
+        environment: Options related to the execution environment. See
+            :class:`EnvironmentOptions` for all available options.
+
+        simulator: Simulator options. See
+            :class:`SimulatorOptions` for all available options.
+
+    """
+
+    _MAX_OPTIMIZATION_LEVEL: int = Field(3, frozen=True)  # pylint: disable=invalid-name
+
+    # Sadly we cannot use pydantic's built in validation because it won't work on Unset.
+    optimization_level: Union[UnsetType, int] = Unset
+    dynamical_decoupling: Union[UnsetType, DDSequenceType] = Unset
+    transpilation: Union[TranspilationOptions, Dict] = Field(default_factory=TranspilationOptions)
+    execution: Union[ExecutionOptionsV2, Dict] = Field(default_factory=ExecutionOptionsV2)
+    twirling: Union[TwirlingOptions, Dict] = Field(default_factory=TwirlingOptions)
+    experimental: Union[UnsetType, dict] = Unset
+
+    @field_validator("optimization_level")
+    @classmethod
+    @skip_unset_validation
+    def _validate_optimization_level(cls, optimization_level: int) -> int:
+        """Validate optimization_leve."""
+        if not 0 <= optimization_level <= SamplerOptions._MAX_OPTIMIZATION_LEVEL:
+            raise ValueError(
+                "Invalid optimization_level. Valid range is "
+                f"0-{SamplerOptions._MAX_OPTIMIZATION_LEVEL}"
+            )
+        return optimization_level

qiskit_ibm_runtime/qiskit/primitives/__init__.py

@@ -14,3 +14,4 @@
 
 from .base_estimator import BaseEstimatorV2  # type: ignore
 from .base_primitive import BasePrimitiveOptions  # type: ignore
+from .base_sampler import BaseSamplerV2  # type: ignore