Source code for openqaoa_braket.backends.devices

import numpy as np
from typing import List
from boto3.session import Session
from botocore.exceptions import NoRegionError
from braket.aws import AwsDevice
from braket.aws.aws_session import AwsSession

from openqaoa.backends.devices_core import DeviceBase


[docs]class DeviceAWS(DeviceBase): """ Contains the required information and methods needed to access QPUs hosted on AWS Braket. Attributes ---------- available_qpus: `list` When connection to AWS is established, this attribute contains a list of device names which can be used to access the selected device by reinitialising the Access Object with the name of the available device as input to the device_name parameter. n_qubits: `int` The maximum number of qubits available for the selected backend. Only available if check_connection method is executed and a connection to the qpu and provider is established. """ def __init__( self, device_name: str, s3_bucket_name: str = None, aws_region: str = None, folder_name: str = "openqaoa", ): """Input the device arn and the name of the folder in which all the results for the QPU runs would be saved on the pre-defined s3 bucket. Note that the user is required to authenticate through the AWS CLI before being able to use this Device object. See: https://docs.aws.amazon.com/cli/latest/userguide/cli-configure-files.html for further details. Parameters ---------- device_name: `str` The ARN string of the braket QPU/simulator to be used s3_bucket_name: `str` The name of S3 Bucket where the Braket run results will be saved. aws_region: `str` The aws region in which the QPU/simulator is located. Defaults to the region set in aws cli config. folder_name: `str` The name of the folder in the s3 bucket that will contain the results from the tasks performed in this run. """ self.device_name = device_name self.device_location = "aws" self.s3_bucket_name = s3_bucket_name self.aws_region = aws_region self.folder_name = folder_name self.provider_connected = None self.qpu_connected = None self._connectivity = None
[docs] def check_connection(self) -> bool: self.provider_connected = self._check_provider_connection() if self.provider_connected == False: return self.provider_connected # Only QPUs that are available for the specified aws region on Braket # will be shown. We filter out QPUs that do not work with the circuit model sess_devices = self.aws_session.search_devices() device_filter = np.multiply( [each_dict["deviceStatus"] == "ONLINE" for each_dict in sess_devices], [ each_dict["providerName"] != "D-Wave Systems" for each_dict in sess_devices ], ) active_devices = np.array(sess_devices)[device_filter].tolist() self.available_qpus = [ backend_dict["deviceArn"] for backend_dict in active_devices ] if self.device_name == "": return self.provider_connected self.qpu_connected = self._check_backend_connection() if self.provider_connected and self.qpu_connected: return True else: return False
def _check_backend_connection(self) -> bool: if self.device_name in self.available_qpus: self.backend_device = AwsDevice(self.device_name, self.aws_session) else: print( """ These are the only available devices for this aws region: {}. Try a different aws region if the device you are looking for is not in the list.' """.format( self.available_qpus ) ) return False # Get the maximum number of qubits for that particular AWS Backend try: self.n_qubits = self.backend_device.properties.paradigm.qubitCount except AttributeError: print( "OpenQAOA is unable to retrieve the number of qubits available in the selected QPU." ) return False else: return True def _check_provider_connection(self) -> bool: try: sess = Session(region_name=self.aws_region) self.aws_session = AwsSession(sess, default_bucket=self.s3_bucket_name) self.aws_region = self.aws_session.region self.s3_bucket_name = self.aws_session.default_bucket() return True except NoRegionError: self.aws_session = None return True except Exception as e: print( "An Exception has occured when trying to connect with the provider." "You are required to authenticate through the AWS CLI in order to connect to the Braket QPUs." "Please check if you have properly set it up." "See: https://docs.aws.amazon.com/cli/latest/userguide/cli-configure-files.html for further details. : {}".format( e ) ) return False @property def connectivity(self) -> List[List[int]]: aws_device_properties = self.backend_device.properties if hasattr(aws_device_properties, "topology_graph"): aws_device_topology = aws_device_properties.topology_gprah if aws_device_topology is not None: self._connectivity = [list(edge) for edge in aws_device_topology.edges] # else: # return None return self._connectivity @connectivity.setter def connectivity(self, value: List[List[int]]): if isinstance(value, list): self._connectivity = value else: raise TypeError("Connectivity must be a list of lists")