Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 2 additions & 2 deletions qiskit_ibm_runtime/api/clients/runtime.py
Original file line number Diff line number Diff line change
Expand Up @@ -281,13 +281,13 @@ def logout(self) -> None:

# IBM Cloud only functions

def list_backends(self) -> List[Dict[str, Any]]:
def list_backends(self) -> List[str]:
"""Return IBM Cloud backends available for this service instance.

Returns:
IBM Cloud backends available for this service instance.
"""
return self.api.backends()
return self.api.backends()["devices"]

def backend_configuration(self, backend_name: str) -> Dict[str, Any]:
"""Return the configuration of the IBM Cloud backend.
Expand Down
2 changes: 1 addition & 1 deletion qiskit_ibm_runtime/api/rest/runtime.py
Original file line number Diff line number Diff line change
Expand Up @@ -204,7 +204,7 @@ def backend(self, backend_name: str) -> CloudBackend:
"""
return CloudBackend(self.session, backend_name)

def backends(self, timeout: Optional[float] = None) -> List[Dict[str, Any]]:
def backends(self, timeout: Optional[float] = None) -> Dict[str, List[str]]:
"""Return a list of IBM Cloud backends.

Args:
Expand Down
2 changes: 1 addition & 1 deletion qiskit_ibm_runtime/hub_group_project.py
Original file line number Diff line number Diff line change
Expand Up @@ -120,7 +120,7 @@ def _discover_remote_backends(
configuration=config,
service=self._service,
credentials=self.credentials,
api_client=self._api_client,
account_client=self._api_client,
)
except Exception: # pylint: disable=broad-except
logger.warning(
Expand Down
29 changes: 20 additions & 9 deletions qiskit_ibm_runtime/ibm_backend.py
Original file line number Diff line number Diff line change
Expand Up @@ -31,7 +31,7 @@
# pylint: disable=unused-import, cyclic-import
from qiskit_ibm_runtime import ibm_runtime_service

from .api.clients import AccountClient
from .api.clients import AccountClient, RuntimeClient
from .backendreservation import BackendReservation
from .credentials import Credentials
from .exceptions import IBMBackendApiProtocolError
Expand Down Expand Up @@ -69,7 +69,8 @@ def __init__(
configuration: Union[QasmBackendConfiguration, PulseBackendConfiguration],
service: "ibm_runtime_service.IBMRuntimeService",
credentials: Credentials,
api_client: AccountClient,
account_client: Optional[AccountClient] = None,
runtime_client: Optional[RuntimeClient] = None,
) -> None:
"""IBMBackend constructor.

Expand All @@ -81,7 +82,8 @@ def __init__(
"""
super().__init__(provider=service, configuration=configuration)

self._api_client = api_client
self._account_client = account_client
self._runtime_client = runtime_client
self._credentials = credentials
self.hub = credentials.hub
self.group = credentials.group
Expand Down Expand Up @@ -150,9 +152,12 @@ def properties(
datetime = local_to_utc(datetime)

if datetime or refresh or self._properties is None:
api_properties = self._api_client.backend_properties(
self.name(), datetime=datetime
)
if self._account_client:
api_properties = self._account_client.backend_properties(
self.name(), datetime=datetime
)
elif self._runtime_client:
api_properties = self._runtime_client.backend_properties(self.name())
if not api_properties:
return None
decode_backend_properties(api_properties)
Expand All @@ -177,7 +182,10 @@ def status(self) -> BackendStatus:
Raises:
IBMBackendApiProtocolError: If the status for the backend cannot be formatted properly.
"""
api_status = self._api_client.backend_status(self.name())
if self._account_client:
api_status = self._account_client.backend_status(self.name())
elif self._runtime_client:
api_status = self._runtime_client.backend_status(self.name())

try:
return BackendStatus.from_dict(api_status)
Expand All @@ -202,7 +210,10 @@ def defaults(self, refresh: bool = False) -> Optional[PulseDefaults]:
The backend pulse defaults or ``None`` if the backend does not support pulse.
"""
if refresh or self._defaults is None:
api_defaults = self._api_client.backend_pulse_defaults(self.name())
if self._account_client:
api_defaults = self._api_client.backend_pulse_defaults(self.name())
elif self._runtime_client:
api_defaults = self._runtime_client.backend_pulse_defaults(self.name())
if api_defaults:
decode_pulse_defaults(api_defaults)
self._defaults = PulseDefaults.from_dict(api_defaults)
Expand Down Expand Up @@ -233,7 +244,7 @@ def reservations(
"""
start_datetime = local_to_utc(start_datetime) if start_datetime else None
end_datetime = local_to_utc(end_datetime) if end_datetime else None
raw_response = self._api_client.backend_reservations(
raw_response = self._account_client.backend_reservations(
self.name(), start_datetime, end_datetime
)
return convert_reservation_data(raw_response, self.name())
Expand Down
97 changes: 71 additions & 26 deletions qiskit_ibm_runtime/ibm_runtime_service.py
Original file line number Diff line number Diff line change
Expand Up @@ -26,6 +26,7 @@
from qiskit.providers.backend import BackendV1 as Backend
from qiskit.providers.exceptions import QiskitBackendNotFoundError
from qiskit.providers.providerutils import filter_backends
from qiskit.providers.models import PulseBackendConfiguration, QasmBackendConfiguration
from qiskit.transpiler import Layout
from typing_extensions import Literal

Expand Down Expand Up @@ -61,7 +62,7 @@
from .runtime_job import RuntimeJob
from .runtime_program import RuntimeProgram, ParameterNamespace
from .utils import RuntimeDecoder, to_base64_string, to_python_identifier
from .utils.backend import convert_reservation_data
from .utils.backend import convert_reservation_data, decode_backend_configuration

logger = logging.getLogger(__name__)

Expand Down Expand Up @@ -158,34 +159,32 @@ def __init__(
IBMProviderCredentialsInvalidToken: If the `token` is not a valid IBM Quantum token.
"""
super().__init__()

account_credentials, account_preferences = self._resolve_credentials(
self.account_credentials, account_preferences = self._resolve_credentials(
token=token, locator=locator, **kwargs
)

if auth == "cloud":
self._api_client = RuntimeClient(credentials=account_credentials)
self._programs: Dict[str, RuntimeProgram] = {}
return

self._initialize_hgps(
credentials=account_credentials, preferences=account_preferences
)
self._programs: Dict[str, RuntimeProgram] = {}
self._backends: Dict[str, "ibm_backend.IBMBackend"] = {}
self._api_client = None
hgps = self._get_hgps()
for hgp in hgps:
for name, backend in hgp.backends.items():
if name not in self._backends:
self._backends[name] = backend
if not self._api_client and hgp.has_service("runtime"):
self._default_hgp = hgp
self._api_client = RuntimeClient(self._default_hgp.credentials)
self._access_token = self._default_hgp.credentials.access_token
self._ws_url = self._default_hgp.credentials.runtime_url.replace(
"https", "wss"
)
self._programs = {}
if auth == "cloud":
self._api_client = RuntimeClient(credentials=self.account_credentials)
self._backends = self._discover_remote_backends()
else:
self._initialize_hgps(
credentials=self.account_credentials, preferences=account_preferences
)
self._api_client = None
hgps = self._get_hgps()
for hgp in hgps:
for name, backend in hgp.backends.items():
if name not in self._backends:
self._backends[name] = backend
if not self._api_client and hgp.has_service("runtime"):
self._default_hgp = hgp
self._api_client = RuntimeClient(self._default_hgp.credentials)
self._access_token = self._default_hgp.credentials.access_token
self._ws_url = self._default_hgp.credentials.runtime_url.replace(
"https", "wss"
)
self._programs = {}
self._discover_backends()

def _resolve_credentials(
Expand Down Expand Up @@ -249,6 +248,52 @@ def _resolve_credentials(
account_credentials = credentials_list[0]
return account_credentials, preferences

def _discover_remote_backends(self) -> Dict[str, "ibm_backend.IBMBackend"]:
"""Return the remote backends available for this service instance.

Returns:
A dict of the remote backend instances, keyed by backend name.
"""
ret = OrderedDict() # type: ignore[var-annotated]
backends_list = self._api_client.list_backends()
for backend_name in backends_list:
raw_config = self._api_client.backend_configuration(
backend_name=backend_name
)
# Make sure the raw_config is of proper type
if not isinstance(raw_config, dict):
logger.warning(
"An error occurred when retrieving backend "
"information. Some backends might not be available."
)
continue
try:
decode_backend_configuration(raw_config)
try:
config = PulseBackendConfiguration.from_dict(raw_config)
except (KeyError, TypeError):
config = QasmBackendConfiguration.from_dict(raw_config)
backend_cls = (
ibm_backend.IBMSimulator
if config.simulator
Comment thread
rathishcholarajan marked this conversation as resolved.
else ibm_backend.IBMBackend
)
ret[config.backend_name] = backend_cls(
configuration=config,
service=self,
credentials=self.account_credentials,
runtime_client=self._api_client,
)
except Exception: # pylint: disable=broad-except
logger.warning(
'Remote backend "%s" for service instance %s could not be instantiated due to an '
"invalid config: %s",
raw_config.get("backend_name", raw_config.get("name", "unknown")),
repr(self),
traceback.format_exc(),
)
return ret

def _initialize_hgps(
self, credentials: Credentials, preferences: Optional[Dict] = None
) -> None:
Expand Down