Allow user to select runtime image

qiskit/providers/ibmq/api/clients/runtime.py

@@ -125,7 +125,8 @@ def program_run(
             program_id: str,
             credentials: Credentials,
             backend_name: str,
-            params: str
+            params: str,
+            image: str
     ) -> Dict:
         """Run the specified program.
 
@@ -134,13 +135,15 @@ def program_run(
             credentials: Credentials used to run the program.
             backend_name: Name of the backend to run the program.
             params: Parameters to use.
+            image: The runtime image to use.
 
         Returns:
             JSON response.
         """
         return self.api.program_run(program_id=program_id, hub=credentials.hub,
                                     group=credentials.group, project=credentials.project,
-                                    backend_name=backend_name, params=params)
+                                    backend_name=backend_name, params=params,
+                                    image=image)
 
     def program_delete(self, program_id: str) -> None:
         """Delete the specified program.

qiskit/providers/ibmq/api/rest/runtime.py

@@ -124,6 +124,7 @@ def program_run(
             project: str,
             backend_name: str,
             params: str,
+            image: str
     ) -> Dict:
         """Execute the program.
 
@@ -134,6 +135,7 @@ def program_run(
             project: Project to be used.
             backend_name: Name of the backend.
             params: Program parameters.
+            image: Runtime image.
 
         Returns:
             JSON response.
@@ -145,7 +147,8 @@ def program_run(
             'group': group,
             'project': project,
             'backend': backend_name,
-            'params': [params]
+            'params': [params],
+            'runtime': image
         }
         data = json.dumps(payload)
         return self.session.post(url, data=data).json()

qiskit/providers/ibmq/runtime/ibm_runtime_service.py

@@ -16,6 +16,7 @@
 from typing import Dict, Callable, Optional, Union, List, Any, Type
 import json
 import copy
+import re
 
 from qiskit.providers.exceptions import QiskitBackendNotFoundError
 from qiskit.providers.ibmq import accountprovider  # pylint: disable=unused-import
@@ -199,7 +200,8 @@ def run(
             options: Dict,
             inputs: Union[Dict, ParameterNamespace],
             callback: Optional[Callable] = None,
-            result_decoder: Optional[Type[ResultDecoder]] = None
+            result_decoder: Optional[Type[ResultDecoder]] = None,
+            image: Optional[str] = ""
     ) -> RuntimeJob:
         """Execute the runtime program.
 
@@ -217,6 +219,8 @@ def run(
 
             result_decoder: A :class:`ResultDecoder` subclass used to decode job results.
                 ``ResultDecoder`` is used if not specified.
+            image: The runtime image used to execute the program, specified in the form
+                of image_name:tag. Not all accounts are authorized to select a different image.
 
         Returns:
             A ``RuntimeJob`` instance representing the execution.
@@ -231,21 +235,28 @@ def run(
             inputs.validate()
             inputs = vars(inputs)
 
+        if image and not \
+            re.match("[a-zA-Z0-9]+([/.\\-_][a-zA-Z0-9]+)*:[a-zA-Z0-9]+([.\\-_][a-zA-Z0-9]+)*$",
+                     image):
+            raise IBMQInputValueError('"image" needs to be in form of image_name:tag')
+
         backend_name = options['backend_name']
         params_str = json.dumps(inputs, cls=RuntimeEncoder)
         result_decoder = result_decoder or ResultDecoder
         response = self._api_client.program_run(program_id=program_id,
                                                 credentials=self._provider.credentials,
                                                 backend_name=backend_name,
-                                                params=params_str)
+                                                params=params_str,
+                                                image=image)
 
         backend = self._provider.get_backend(backend_name)
         job = RuntimeJob(backend=backend,
                          api_client=self._api_client,
                          credentials=self._provider.credentials,
                          job_id=response['id'], program_id=program_id, params=inputs,
                          user_callback=callback,
-                         result_decoder=result_decoder)
+                         result_decoder=result_decoder,
+                         image=image)
         return job
 
     def upload_program(

qiskit/providers/ibmq/runtime/runtime_job.py

@@ -84,7 +84,8 @@ def __init__(
             params: Optional[Dict] = None,
             creation_date: Optional[str] = None,
             user_callback: Optional[Callable] = None,
-            result_decoder: Type[ResultDecoder] = ResultDecoder
+            result_decoder: Type[ResultDecoder] = ResultDecoder,
+            image: Optional[str] = ""
     ) -> None:
         """RuntimeJob constructor.
 
@@ -98,6 +99,7 @@ def __init__(
             creation_date: Job creation date, in UTC.
             user_callback: User callback function.
             result_decoder: A :class:`ResultDecoder` subclass used to decode job results.
+            image: Runtime image used for this job: image_name:tag.
         """
         self._job_id = job_id
         self._backend = backend
@@ -109,6 +111,7 @@ def __init__(
         self._status = JobStatus.INITIALIZING
         self._error_message = None  # type: Optional[str]
         self._result_decoder = result_decoder
+        self._image = image
 
         # Used for streaming
         self._ws_client_future = None  # type: Optional[futures.Future]
@@ -385,6 +388,16 @@ def backend(self) -> Backend:
         """
         return self._backend
 
+    @property
+    def image(self) -> str:
+        """Return the runtime image used for the job.
+
+        Returns:
+            Runtime image: image_name:tag or "" if the default
+            image is used.
+        """
+        return self._image
+
     @property
     def inputs(self) -> Dict:
         """Job input parameters.

releasenotes/notes/support-runtime-image-selection-e2c26cac76e79b55.yaml

@@ -0,0 +1,6 @@
+---
+features:
+  - |
+    Runtime image can now be specified using the `image` parameter in
+    :meth:`qiskit.providers.ibmq.runtime.IBMRuntimeService.run`.
+    Note that not all accounts are authorized to select a different image.

test/ibmq/runtime/fake_runtime_client.py

@@ -15,6 +15,7 @@
 import time
 import uuid
 import json
+from typing import Optional
 from concurrent.futures import ThreadPoolExecutor
 
 from qiskit.providers.ibmq.credentials import Credentials
@@ -75,7 +76,7 @@ class BaseFakeRuntimeJob:
     _executor = ThreadPoolExecutor()  # pylint: disable=bad-option-value,consider-using-with
 
     def __init__(self, job_id, program_id, hub, group, project, backend_name, final_status,
-                 params):
+                 params, image):
         """Initialize a fake job."""
         self._job_id = job_id
         self._status = final_status or "QUEUED"
@@ -85,6 +86,7 @@ def __init__(self, job_id, program_id, hub, group, project, backend_name, final_
         self._project = project
         self._backend_name = backend_name
         self._params = params
+        self._image = image
         if final_status is None:
             self._future = self._executor.submit(self._auto_progress)
             self._result = None
@@ -110,7 +112,8 @@ def to_dict(self):
                 'backend': self._backend_name,
                 'status': self._status,
                 'params': [self._params],
-                'program': {'id': self._program_id}}
+                'program': {'id': self._program_id},
+                'image': self._image}
 
     def result(self):
         """Return job result."""
@@ -266,15 +269,17 @@ def program_run(
             program_id: str,
             credentials: Credentials,
             backend_name: str,
-            params: str
+            params: str,
+            image: Optional[str] = ""
     ):
         """Run the specified program."""
         job_id = uuid.uuid4().hex
         job_cls = self._job_classes.pop(0) if len(self._job_classes) > 0 else BaseFakeRuntimeJob
         job = job_cls(job_id=job_id, program_id=program_id,
                       hub=credentials.hub, group=credentials.group,
                       project=credentials.project, backend_name=backend_name,
-                      params=params, final_status=self._final_status, **self._job_kwargs)
+                      params=params, final_status=self._final_status, image=image,
+                      **self._job_kwargs)
         self._jobs[job_id] = job
         return {'id': job_id}
 

test/ibmq/runtime/test_runtime.py

@@ -307,6 +307,20 @@ def test_run_program(self):
         self.assertEqual(job.status(), JobStatus.DONE)
         self.assertTrue(job.result())
 
+    def test_run_program_with_custom_runtime_image(self):
+        """Test running program."""
+        params = {'param1': 'foo'}
+        image = "name:tag"
+        job = self._run_program(inputs=params, image=image)
+        self.assertTrue(job.job_id())
+        self.assertIsInstance(job, RuntimeJob)
+        self.assertIsInstance(job.status(), JobStatus)
+        self.assertEqual(job.inputs, params)
+        job.wait_for_final_state()
+        self.assertEqual(job.status(), JobStatus.DONE)
+        self.assertTrue(job.result())
+        self.assertEqual(job.image, image)
+
     def test_program_params_validation(self):
         """Test program parameters validation process"""
         program_id = self.runtime.upload_program(
@@ -596,7 +610,7 @@ def _upload_program(self, name=None, max_execution_time=300):
         return program_id
 
     def _run_program(self, program_id=None, inputs=None, job_classes=None, final_status=None,
-                     decoder=None):
+                     decoder=None, image=""):
         """Run a program."""
         options = {'backend_name': "some_backend"}
         if final_status is not None:
@@ -606,7 +620,8 @@ def _run_program(self, program_id=None, inputs=None, job_classes=None, final_sta
         if program_id is None:
             program_id = self._upload_program()
         job = self.runtime.run(program_id=program_id, inputs=inputs,
-                               options=options, result_decoder=decoder)
+                               options=options, result_decoder=decoder,
+                               image=image)
         return job
 
     def _populate_jobs_with_all_statuses(self, jobs, program_id):