Q-CTRL support

qiskit_ibm_runtime/accounts/account.py

@@ -44,6 +44,7 @@ def __init__(
         instance: Optional[str] = None,
         proxies: Optional[ProxyConfiguration] = None,
         verify: Optional[bool] = True,
+        channel_strategy: Optional[str] = None,
     ):
         """Account constructor.
 
@@ -54,17 +55,18 @@ def __init__(
             instance: Service instance to use.
             proxies: Proxy configuration.
             verify: Whether to verify server's TLS certificate.
+            channel_strategy: Error mitigation strategy.
         """
         resolved_url = url or (
             IBM_QUANTUM_API_URL if channel == "ibm_quantum" else IBM_CLOUD_API_URL
         )
-
         self.channel = channel
         self.token = token
         self.url = resolved_url
         self.instance = instance
         self.proxies = proxies
         self.verify = verify
+        self.channel_strategy = channel_strategy
 
     def to_saved_format(self) -> dict:
         """Returns a dictionary that represents how the account is saved on disk."""
@@ -84,6 +86,7 @@ def from_saved_format(cls, data: dict) -> "Account":
             instance=data.get("instance"),
             proxies=ProxyConfiguration(**proxies) if proxies else None,
             verify=data.get("verify", True),
+            channel_strategy=data.get("channel_strategy"),
         )
 
     def resolve_crn(self) -> None:
@@ -155,8 +158,19 @@ def validate(self) -> "Account":
         self._assert_valid_url(self.url)
         self._assert_valid_instance(self.channel, self.instance)
         self._assert_valid_proxies(self.proxies)
+        self._assert_valid_channel_strategy(self.channel_strategy)
         return self
 
+    @staticmethod
+    def _assert_valid_channel_strategy(channel_strategy: str) -> None:
+        """Assert that the channel strategy is valid."""
+        # add more strategies as they are implemented
+        if channel_strategy and channel_strategy not in ["q-ctrl"]:
+            raise InvalidAccountError(
+                f"Invalid `channel_strategy` value. Expected one of "
+                f"{['q-ctrl']}, got '{channel_strategy}'."
+            )
+
     @staticmethod
     def _assert_valid_channel(channel: ChannelType) -> None:
         """Assert that the channel parameter is valid."""

qiskit_ibm_runtime/accounts/management.py

@@ -49,6 +49,7 @@ def save(
         proxies: Optional[ProxyConfiguration] = None,
         verify: Optional[bool] = None,
         overwrite: Optional[bool] = False,
+        channel_strategy: Optional[str] = None,
     ) -> None:
         """Save account on disk."""
         cls.migrate(filename=filename)
@@ -67,6 +68,7 @@ def save(
                 channel=channel,
                 proxies=proxies,
                 verify=verify,
+                channel_strategy=channel_strategy,
             )
             # avoid storing invalid accounts
             .validate().to_saved_format(),

qiskit_ibm_runtime/api/clients/runtime.py

@@ -128,6 +128,7 @@ def program_run(
         max_execution_time: Optional[int] = None,
         start_session: Optional[bool] = False,
         session_time: Optional[int] = None,
+        channel_strategy: Optional[str] = None,
     ) -> Dict:
         """Run the specified program.
 
@@ -143,6 +144,7 @@ def program_run(
             max_execution_time: Maximum execution time in seconds.
             start_session: Set to True to explicitly start a runtime session. Defaults to False.
             session_time: Length of session in seconds.
+            channel_strategy: Error mitigation strategy.
 
         Returns:
             JSON response.
@@ -162,6 +164,7 @@ def program_run(
             max_execution_time=max_execution_time,
             start_session=start_session,
             session_time=session_time,
+            channel_strategy=channel_strategy,
             **hgp_dict,
         )
 

qiskit_ibm_runtime/api/rest/runtime.py

@@ -138,6 +138,7 @@ def program_run(
         max_execution_time: Optional[int] = None,
         start_session: Optional[bool] = False,
         session_time: Optional[int] = None,
+        channel_strategy: Optional[str] = None,
     ) -> Dict:
         """Execute the program.
 
@@ -155,6 +156,7 @@ def program_run(
             max_execution_time: Maximum execution time in seconds.
             start_session: Set to True to explicitly start a runtime session. Defaults to False.
             session_time: Length of session in seconds.
+            channel_strategy: Error mitigation strategy.
 
         Returns:
             JSON response.
@@ -183,6 +185,8 @@ def program_run(
             payload["hub"] = hub
             payload["group"] = group
             payload["project"] = project
+        if channel_strategy:
+            payload["channel_strategy"] = channel_strategy
         data = json.dumps(payload, cls=RuntimeEncoder)
         return self.session.post(url, data=data, timeout=900).json()
 

qiskit_ibm_runtime/qiskit_runtime_service.py

@@ -130,6 +130,7 @@ def __init__(
         instance: Optional[str] = None,
         proxies: Optional[dict] = None,
         verify: Optional[bool] = None,
+        channel_strategy: Optional[str] = None,
     ) -> None:
         """QiskitRuntimeService constructor
 
@@ -181,6 +182,7 @@ def __init__(
             name=name,
             proxies=ProxyConfiguration(**proxies) if proxies else None,
             verify=verify,
+            channel_strategy=channel_strategy,
         )
 
         self._client_params = ClientParameters(
@@ -192,6 +194,7 @@ def __init__(
             verify=self._account.verify,
         )
 
+        self._channel_strategy = channel_strategy or self._account.channel_strategy
         self._channel = self._account.channel
         self._programs: Dict[str, RuntimeProgram] = {}
         self._backends: Dict[str, "ibm_backend.IBMBackend"] = {}
@@ -231,10 +234,19 @@ def _discover_account(
         name: Optional[str] = None,
         proxies: Optional[ProxyConfiguration] = None,
         verify: Optional[bool] = None,
+        channel_strategy: Optional[str] = None,
     ) -> Account:
         """Discover account."""
         account = None
         verify_ = verify or True
+        if channel_strategy:
+            if channel_strategy not in ["q-ctrl"]:
+                raise ValueError(f"{channel_strategy} is not a valid channel strategy.")
+            if channel and channel != "ibm_cloud":
+                raise ValueError(
+                    f"The channel strategy {channel_strategy} is "
+                    "only supported on the ibm_cloud channel."
+                )
         if name:
             if filename:
                 if any([auth, channel, token, url]):
@@ -266,6 +278,7 @@ def _discover_account(
                     instance=instance,
                     proxies=proxies,
                     verify=verify_,
+                    channel_strategy=channel_strategy,
                 )
             else:
                 if url:
@@ -685,6 +698,7 @@ def save_account(
         proxies: Optional[dict] = None,
         verify: Optional[bool] = None,
         overwrite: Optional[bool] = False,
+        channel_strategy: Optional[str] = None,
     ) -> None:
         """Save the account to disk for future use.
 
@@ -704,6 +718,7 @@ def save_account(
                 authentication)
             verify: Verify the server's TLS certificate.
             overwrite: ``True`` if the existing account is to be overwritten.
+            channel_strategy: Error mitigation strategy.
         """
 
         AccountManager.save(
@@ -716,6 +731,7 @@ def save_account(
             proxies=ProxyConfiguration(**proxies) if proxies else None,
             verify=verify,
             overwrite=overwrite,
+            channel_strategy=channel_strategy,
         )
 
     @staticmethod
@@ -1006,6 +1022,7 @@ def run(
                 max_execution_time=qrt_options.max_execution_time,
                 start_session=start_session,
                 session_time=qrt_options.session_time,
+                channel_strategy=self._channel_strategy,
             )
         except RequestsApiError as ex:
             if ex.status_code == 404:

releasenotes/notes/q-ctrl-support-157170386477dfbd.yaml

@@ -0,0 +1,9 @@
+---
+features:
+  - |
+    There is a new parameter, ``channel_strategy`` that can be set in the initialization of 
+    :class:`qiskit_ibm_runtime.QiskitRuntimeService` or saved in 
+    :meth:`qiskit_ibm_runtime.QiskitRuntimeService.save_account`. If ``channel_strategy`` 
+    is set to ``q-ctrl``, all jobs within the service will use
+    the Q-CTRL error mitigation strategy.
+

test/unit/mock/fake_runtime_client.py

@@ -104,6 +104,7 @@ def __init__(
         session_id=None,
         max_execution_time=None,
         start_session=None,
+        channel_strategy=None,
     ):
         """Initialize a fake job."""
         self._job_id = job_id
@@ -129,6 +130,7 @@ def __init__(
         elif final_status == "COMPLETED":
             self._result = json.dumps("foo")
         self._final_status = final_status
+        self._channel_strategy = channel_strategy
 
     def _auto_progress(self):
         """Automatically update job status."""
@@ -365,6 +367,7 @@ def program_run(
         max_execution_time: Optional[int] = None,
         start_session: Optional[bool] = None,
         session_time: Optional[int] = None,
+        channel_strategy: Optional[str] = None,
     ) -> Dict[str, Any]:
         """Run the specified program."""
         _ = self._get_program(program_id)
@@ -396,6 +399,7 @@ def program_run(
             job_tags=job_tags,
             max_execution_time=max_execution_time,
             start_session=start_session,
+            channel_strategy=channel_strategy,
             **self._job_kwargs,
         )
         self.session_time = session_time

test/unit/test_account.py

@@ -150,6 +150,22 @@ def test_invalid_instance(self):
                     ).validate()
                 self.assertIn("Invalid `instance` value.", str(err.exception))
 
+    def test_invalid_channel_strategy(self):
+        """Test invalid values for channel_strategy"""
+        subtests = [
+            {"channel": "ibm_cloud", "channel_strategy": "test"},
+        ]
+        for params in subtests:
+            with self.subTest(params=params):
+                with self.assertRaises(InvalidAccountError) as err:
+                    Account(
+                        **params,
+                        token=self.dummy_token,
+                        url=self.dummy_ibm_cloud_url,
+                        instance="crn:v1:bluemix:public:quantum-computing:us-east:a/...::",
+                    ).validate()
+                self.assertIn("Invalid `channel_strategy` value.", str(err.exception))
+
     def test_invalid_proxy_config(self):
         """Test invalid values for proxy configuration."""
 
@@ -275,6 +291,7 @@ def test_save_channel_ibm_cloud_over_auth_cloud_with_overwrite(self):
             proxies=_TEST_IBM_CLOUD_ACCOUNT.proxies,
             name=None,
             overwrite=True,
+            channel_strategy="q-ctrl",
         )
         self.assertEqual(_TEST_IBM_CLOUD_ACCOUNT, AccountManager.get(channel="ibm_cloud"))
 
@@ -793,6 +810,17 @@ def test_enable_account_bad_channel(self):
             _ = FakeRuntimeService(channel=channel)
         self.assertIn("channel", str(err.exception))
 
+    def test_enable_account_bad_channel_strategy(self):
+        """Test initializing account by bad channel strategy."""
+        subtests = [
+            {"channel_strategy": "q-ctrl", "channel": "ibm_quantum"},
+            {"channel_strategy": "test"},
+        ]
+        for test in subtests:
+            with temporary_account_config_file() as _, self.assertRaises(ValueError) as err:
+                _ = FakeRuntimeService(**test)
+            self.assertIn("channel", str(err.exception))
+
     def test_enable_account_by_name_pref(self):
         """Test initializing account by name and preferences."""
         name = "foo"