feat: Add qctrl validation (#1027)

* dry run adding qctrl validation

* revert changes

* more reverts

* adds validation

* lint

* moves settings around to match check function

* adding tests

* undo change

* lint

* adds one test case

* lint

* lint changes

* fixes test

* fixes more tests

* python 3.8 fix

* adds release note

* overrides more qctrl safe options

* fixes optmization level

* Apply suggestions from code review

Co-authored-by: Kevin Tian <[email protected]>

* fixes var name

---------

Co-authored-by: Kevin Tian <[email protected]>
Co-authored-by: Kevin Tian <[email protected]>

qiskit_ibm_runtime/estimator.py

@@ -27,6 +27,7 @@
 from .ibm_backend import IBMBackend
 from .options import Options
 from .base_primitive import BasePrimitive
+from .utils.qctrl import validate as qctrl_validate
 
 # pylint: disable=unused-import,cyclic-import
 from .session import Session
@@ -195,6 +196,10 @@ def _validate_options(self, options: dict) -> None:
         if os.getenv("QISKIT_RUNTIME_SKIP_OPTIONS_VALIDATION"):
             return
 
+        if self._service._channel_strategy == "q-ctrl":
+            qctrl_validate(options)
+            return
+
         if not options.get("resilience_level") in list(
             range(Options._MAX_RESILIENCE_LEVEL_ESTIMATOR + 1)
         ):

qiskit_ibm_runtime/sampler.py

@@ -27,6 +27,7 @@
 
 # pylint: disable=unused-import,cyclic-import
 from .session import Session
+from .utils.qctrl import validate as qctrl_validate
 
 logger = logging.getLogger(__name__)
 
@@ -161,6 +162,10 @@ def _validate_options(self, options: dict) -> None:
         if os.getenv("QISKIT_RUNTIME_SKIP_OPTIONS_VALIDATION"):
             return
 
+        if self._service._channel_strategy == "q-ctrl":
+            qctrl_validate(options)
+            return
+
         if options.get("resilience_level") and not options.get("resilience_level") in [
             0,
             1,

qiskit_ibm_runtime/utils/qctrl.py

@@ -0,0 +1,145 @@
+# 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.
+
+"""Qctrl validation functions and helpers."""
+
+import logging
+from typing import Any, Optional, Dict, List
+
+from ..options import Options
+from ..options import EnvironmentOptions, ExecutionOptions, TranspilationOptions, SimulatorOptions
+
+logger = logging.getLogger(__name__)
+
+
+def validate(options: Dict[str, Any]) -> None:
+    """Validates the options for qctrl"""
+
+    # Raise error on bad options.
+    _raise_if_error_in_options(options)
+    # Override options and warn.
+    _warn_and_clean_options(options)
+
+    # Default validation otherwise.
+    TranspilationOptions.validate_transpilation_options(options.get("transpilation"))
+    execution_time = options.get("max_execution_time")
+    if not execution_time is None:
+        if (
+            execution_time < Options._MIN_EXECUTION_TIME
+            or execution_time > Options._MAX_EXECUTION_TIME
+        ):
+            raise ValueError(
+                f"max_execution_time must be between "
+                f"{Options._MIN_EXECUTION_TIME} and {Options._MAX_EXECUTION_TIME} seconds."
+            )
+
+    EnvironmentOptions.validate_environment_options(options.get("environment"))
+    ExecutionOptions.validate_execution_options(options.get("execution"))
+    SimulatorOptions.validate_simulator_options(options.get("simulator"))
+
+
+def _raise_if_error_in_options(options: Dict[str, Any]) -> None:
+    """Checks for settings that produce errors and raise a ValueError"""
+
+    # Fail on resilience_level set to 0
+    resilience_level = options.get("resilience_level", 1)
+    _check_argument(
+        resilience_level > 0,
+        description=(
+            "Q-CTRL Primitives do not support resilience level 0. Please "
+            "set resilience_level to 1 and re-try"
+        ),
+        arguments={},
+    )
+
+    optimization_level = options.get("optimization_level", 1)
+    _check_argument(
+        optimization_level > 0,
+        description="Q-CTRL Primitives do not support optimization level 0. Please\
+        set optimization_level to 3 and re-try",
+        arguments={},
+    )
+
+
+def _warn_and_clean_options(options: Dict[str, Any]) -> None:
+    """
+    Validate and update transpilation settings
+    """
+    # Issue a warning and override if any of these setting is not None
+    # or a different value than the default below
+    expected_options = {
+        "optimization_level": 3,
+        "resilience_level": 1,
+        "transpilation": {"approximation_degree": 0, "skip_transpilation": False},
+        "resilience": {
+            "noise_amplifier": None,
+            "noise_factors": None,
+            "extrapolator": None,
+        },
+    }
+
+    # Collect keys with miss-matching values
+    different_keys = _validate_values(expected_options, options)
+    # Override options
+    _update_values(expected_options, options)
+    if different_keys:
+        logger.warning(
+            "The following settings cannot be customized and will be overwritten: %s",
+            ",".join(sorted(different_keys)),
+        )
+
+
+def _validate_values(
+    expected_options: Dict[str, Any], current_options: Optional[Dict[str, Any]]
+) -> List[str]:
+    """Validates expected_options and current_options have the same values if the
+    keys of expected_options are present in current_options"""
+
+    if current_options is None:
+        return []
+
+    different_keys = []
+    for expected_key, expected_value in expected_options.items():
+        if isinstance(expected_value, dict):
+            different_keys.extend(
+                _validate_values(expected_value, current_options.get(expected_key, None))
+            )
+        else:
+            current_value = current_options.get(expected_key, None)
+            if current_value is not None and expected_value != current_value:
+                different_keys.append(expected_key)
+    return different_keys
+
+
+def _update_values(
+    expected_options: Dict[str, Any], current_options: Optional[Dict[str, Any]]
+) -> None:
+
+    if current_options is None:
+        return
+
+    for expected_key, expected_value in expected_options.items():
+        if isinstance(expected_value, dict):
+            _update_values(expected_value, current_options.get(expected_key, None))
+        else:
+            if expected_key in current_options:
+                current_options[expected_key] = expected_value
+
+
+def _check_argument(
+    condition: bool,
+    description: str,
+    arguments: Dict[str, str],
+) -> None:
+    if not condition:
+        error_str = f"{description} arguments={arguments}"
+        raise ValueError(error_str)

releasenotes/notes/q-ctrl-validation-08d249f1e84a43a5.yaml

@@ -0,0 +1,7 @@
+---
+features:
+  - |
+    The Sampler and Estimator primitives have been enhanced to incorporate custom validation procedures when 
+    the channel_strategy property within the :class:qiskit_ibm_runtime.QiskitRuntimeService is configured as "q-ctrl." 
+    This customized validation logic effectively rectifies incorrect input options and safeguards users against 
+    inadvertently disabling Q-CTRL's performance enhancements.

test/unit/test_ibm_primitives.py

@@ -891,3 +891,68 @@ def _assert_dict_partially_equal(self, dict1, dict2):
             dict_paritally_equal(dict1, dict2),
             f"{dict1} and {dict2} not partially equal.",
         )
+
+    def test_qctrl_supported_values_for_options(self):
+        """Test exception when options levels not supported."""
+        no_resilience_options = {
+            "noise_factors": None,
+            "extrapolator": None,
+        }
+
+        options_good = [
+            # Minium working settings
+            {},
+            # No warnings, we need resilience options here because by default they are getting populated.
+            {"resilience": no_resilience_options},
+            # Arbitrary approximation degree (issues warning)
+            {"approximation_degree": 1},
+            # Arbitrary resilience options(issue warning)
+            {
+                "resilience_level": 1,
+                "resilience": {"noise_factors": (1, 1, 3)},
+                "approximation_degree": 1,
+            },
+            # Resilience level > 1 (issue warning)
+            {"resilience_level": 2},
+            # Optimization level = 1,2 (issue warning)
+            {"optimization_level": 1},
+            {"optimization_level": 2},
+            # Skip transpilation level(issue warning)
+            {"skip_transpilation": True},
+        ]
+        session = MagicMock(spec=MockSession)
+        session.service._channel_strategy = "q-ctrl"
+        session.service.backend().configuration().simulator = False
+        primitives = [Sampler, Estimator]
+        for cls in primitives:
+            for options in options_good:
+                with self.subTest(msg=f"{cls}, {options}"):
+                    inst = cls(session=session)
+                    if isinstance(inst, Estimator):
+                        _ = inst.run(self.qx, observables=self.obs, **options)
+                    else:
+                        _ = inst.run(self.qx, **options)
+
+    def test_qctrl_unsupported_values_for_options(self):
+        """Test exception when options levels are not supported."""
+        options_bad = [
+            # Bad resilience levels
+            ({"resilience_level": 0}, "resilience level"),
+            # Bad optimization level
+            ({"optimization_level": 0}, "optimization level"),
+        ]
+        session = MagicMock(spec=MockSession)
+        session.service._channel_strategy = "q-ctrl"
+        session.service.backend().configuration().simulator = False
+        primitives = [Sampler, Estimator]
+        for cls in primitives:
+            for bad_opt, expected_message in options_bad:
+                with self.subTest(msg=bad_opt):
+                    inst = cls(session=session)
+                    with self.assertRaises(ValueError) as exc:
+                        if isinstance(inst, Sampler):
+                            _ = inst.run(self.qx, **bad_opt)
+                        else:
+                            _ = inst.run(self.qx, observables=self.obs, **bad_opt)
+
+                        self.assertIn(expected_message, str(exc.exception))

test/unit/test_options.py

@@ -22,6 +22,7 @@
 from qiskit_aer.noise import NoiseModel
 
 from qiskit_ibm_runtime import Options, RuntimeOptions
+from qiskit_ibm_runtime.utils.qctrl import _warn_and_clean_options
 
 from ..ibm_test_case import IBMTestCase
 from ..utils import dict_keys_equal, dict_paritally_equal, flat_dict_partially_equal
@@ -257,3 +258,56 @@ def test_simulator_set_backend(self, fake_backend):
             "seed_simulator": 42,
         }
         self.assertDictEqual(asdict(options), asdict(expected_options))
+
+    def test_qctrl_overrides(self):
+        """Test override of options"""
+        all_test_options = [
+            (
+                {
+                    "optimization_level": 2,
+                    "transpilation": {"approximation_degree": 1},
+                    "resilience_level": 3,
+                    "resilience": {
+                        "noise_factors": (1, 3, 5),
+                        "extrapolator": "Linear",
+                    },
+                },
+                {
+                    "optimization_level": 3,
+                    "transpilation": {"approximation_degree": 0},
+                    "resilience_level": 1,
+                    "resilience": {
+                        "noise_factors": None,
+                        "extrapolator": None,
+                    },
+                },
+            ),
+            (
+                {
+                    "optimization_level": 0,
+                    "transpilation": {"approximation_degree": 1, "skip_transpilation": True},
+                    "resilience_level": 1,
+                },
+                {
+                    "optimization_level": 3,
+                    "transpilation": {"approximation_degree": 0, "skip_transpilation": False},
+                    "resilience_level": 1,
+                },
+            ),
+            (
+                {
+                    "optimization_level": 0,
+                    "transpilation": {"skip_transpilation": True},
+                    "resilience_level": 1,
+                },
+                {
+                    "optimization_level": 3,
+                    "transpilation": {"skip_transpilation": False},
+                    "resilience_level": 1,
+                },
+            ),
+        ]
+        for option, expected_ in all_test_options:
+            with self.subTest(msg=f"{option}"):
+                _warn_and_clean_options(option)
+                self.assertEqual(expected_, option)

test/unit/test_runtime_ws.py

@@ -91,6 +91,7 @@ def _patched_run(callback, *args, **kwargs):  # pylint: disable=unused-argument
 
         service = MagicMock(spec=QiskitRuntimeService)
         service.run = _patched_run
+        service._channel_strategy = None
 
         circ = ReferenceCircuits.bell()
         obs = SparsePauliOp.from_list([("IZ", 1)])