braket.aws.aws_device module

class braket.aws.aws_device.AwsDeviceType(value)[source]

Bases: str, Enum

Possible AWS device types

SIMULATOR = 'SIMULATOR'

QPU = 'QPU'

class braket.aws.aws_device.AwsDevice(arn: str, aws_session: AwsSession | None = None, noise_model: NoiseModel | None = None)[source]

Bases: Device

Amazon Braket implementation of a device. Use this class to retrieve the latest metadata about the device and to run a quantum task on the device.

Initializes an AwsDevice.

Parameters:

arn (str) – The ARN of the device
aws_session (Optional[AwsSession]) – An AWS session object. Default is None.
noise_model (Optional[NoiseModel]) – The Braket noise model to apply to the circuit before execution. Noise model can only be added to the devices that support noise simulation.

Note

Some devices (QPUs) are physically located in specific AWS Regions. In some cases, the current aws_session connects to a Region other than the Region in which the QPU is physically located. When this occurs, a cloned aws_session is created for the Region the QPU is located in.

See braket.aws.aws_device.AwsDevice.REGIONS for the AWS regions provider devices are located in across the AWS Braket service. This is not a device specific tuple.

REGIONS = ('us-east-1', 'us-west-1', 'us-west-2', 'eu-west-2', 'eu-north-1')

DEFAULT_SHOTS_QPU = 1000

DEFAULT_SHOTS_SIMULATOR = 0

DEFAULT_MAX_PARALLEL = 10

run(task_specification: Circuit | Problem | OpenQasmProgram | BlackbirdProgram | PulseSequence | AnalogHamiltonianSimulation, s3_destination_folder: AwsSession.S3DestinationFolder | None = None, shots: int | None = None, poll_timeout_seconds: float = 432000, poll_interval_seconds: float | None = None, inputs: dict[str, float] | None = None, gate_definitions: dict[tuple[Gate, QubitSet], PulseSequence] | None = None, reservation_arn: str | None = None, *aws_quantum_task_args: Any, **aws_quantum_task_kwargs: Any) → AwsQuantumTask[source]

Run a quantum task specification on this device. A quantum task can be a circuit or an annealing problem.

Parameters:

task_specification (Union[Circuit, Problem, OpenQasmProgram, BlackbirdProgram, PulseSequence, AnalogHamiltonianSimulation]) – Specification of quantum task (circuit, OpenQASM program or AHS program) to run on device.
s3_destination_folder (Optional[S3DestinationFolder]) – The S3 location to save the quantum task’s results to. Default is <default_bucket>/tasks if evoked outside a Braket Hybrid Job, <Job Bucket>/jobs/<job name>/tasks if evoked inside a Braket Hybrid Job.
shots (Optional[int]) – The number of times to run the circuit or annealing problem. Default is 1000 for QPUs and 0 for simulators.
poll_timeout_seconds (float) – The polling timeout for AwsQuantumTask.result(), in seconds. Default: 5 days.
poll_interval_seconds (Optional[float]) – The polling interval for AwsQuantumTask.result(), in seconds. Defaults to the getTaskPollIntervalMillis value specified in self.properties.service (divided by 1000) if provided, otherwise 1 second.
inputs (Optional[dict[str, float]]) – Inputs to be passed along with the IR. If the IR supports inputs, the inputs will be updated with this value. Default: {}.
gate_definitions (Optional[dict[tuple[Gate, QubitSet], PulseSequence]]) – A dict[tuple[Gate, QubitSet], PulseSequence]] for a user defined gate calibration. The calibration is defined for a particular Gate on a particular QubitSet and is represented by a PulseSequence. Default: None.
reservation_arn (str | None) – The reservation ARN provided by Braket Direct to reserve exclusive usage for the device to run the quantum task on. Note: If you are creating tasks in a job that itself was created reservation ARN, those tasks do not need to be created with the reservation ARN. Default: None.
*aws_quantum_task_args (Any) – Arbitrary arguments.
**aws_quantum_task_kwargs (Any) – Arbitrary keyword arguments.

Returns:

AwsQuantumTask – An AwsQuantumTask that tracks the execution on the device.

Examples

>>> circuit = Circuit().h(0).cnot(0, 1)
>>> device = AwsDevice("arn1")
>>> device.run(circuit, ("bucket-foo", "key-bar"))

>>> circuit = Circuit().h(0).cnot(0, 1)
>>> device = AwsDevice("arn2")
>>> device.run(task_specification=circuit,
>>>     s3_destination_folder=("bucket-foo", "key-bar"))

>>> circuit = Circuit().h(0).cnot(0, 1)
>>> device = AwsDevice("arn3")
>>> device.run(task_specification=circuit,
>>>     s3_destination_folder=("bucket-foo", "key-bar"), disable_qubit_rewiring=True)

>>> problem = Problem(
>>>     ProblemType.ISING,
>>>     linear={1: 3.14},
>>>     quadratic={(1, 2): 10.08},
>>> )
>>> device = AwsDevice("arn4")
>>> device.run(problem, ("bucket-foo", "key-bar"),
>>>     device_parameters={
>>>         "providerLevelParameters": {"postprocessingType": "SAMPLING"}}
>>> )

Executes a batch of quantum tasks in parallel

Parameters:

task_specifications (Union[Union[Circuit, Problem, OpenQasmProgram, BlackbirdProgram, PulseSequence, AnalogHamiltonianSimulation], list[Union[ Circuit, Problem, OpenQasmProgram, BlackbirdProgram, PulseSequence, AnalogHamiltonianSimulation]]]) – # noqa Single instance or list of circuits, annealing problems, pulse sequences, or photonics program to run on device.
s3_destination_folder (Optional[S3DestinationFolder]) – The S3 location to save the quantum tasks’ results to. Default is <default_bucket>/tasks if evoked outside a Braket Job, <Job Bucket>/jobs/<job name>/tasks if evoked inside a Braket Job.
shots (Optional[int]) – The number of times to run the circuit or annealing problem. Default is 1000 for QPUs and 0 for simulators.
max_parallel (Optional[int]) – The maximum number of quantum tasks to run on AWS in parallel. Batch creation will fail if this value is greater than the maximum allowed concurrent quantum tasks on the device. Default: 10
max_connections (int) – The maximum number of connections in the Boto3 connection pool. Also the maximum number of thread pool workers for the batch. Default: 100
poll_timeout_seconds (float) – The polling timeout for AwsQuantumTask.result(), in seconds. Default: 5 days.
poll_interval_seconds (float) – The polling interval for AwsQuantumTask.result(), in seconds. Defaults to the getTaskPollIntervalMillis value specified in self.properties.service (divided by 1000) if provided, otherwise 1 second.
inputs (Optional[Union[dict[str, float], list[dict[str, float]]]]) – Inputs to be passed along with the IR. If the IR supports inputs, the inputs will be updated with this value. Default: {}.
gate_definitions (Optional[dict[tuple[Gate, QubitSet], PulseSequence]]) – A dict[tuple[Gate, QubitSet], PulseSequence]] for a user defined gate calibration. The calibration is defined for a particular Gate on a particular QubitSet and is represented by a PulseSequence. Default: None.
reservation_arn (Optional[str]) – The reservation ARN provided by Braket Direct to reserve exclusive usage for the device to run the quantum task on. Note: If you are creating tasks in a job that itself was created reservation ARN, those tasks do not need to be created with the reservation ARN. Default: None.

Returns:

AwsQuantumTaskBatch – A batch containing all of the quantum tasks run

refresh_metadata() → None[source]: Refresh the AwsDevice object with the most recent Device metadata.

property type: str

Return the device type

Type:: str

property provider_name: str

Return the provider name

Type:: str

property aws_session: AwsSession

property arn: str

Return the ARN of the device

Type:: str

property gate_calibrations: GateCalibrations | None

Calibration data for a QPU. Calibration data is shown for gates on particular gubits. If a QPU does not expose these calibrations, None is returned.

Returns:: Optional[GateCalibrations] – The calibration object. Returns None if the data is not present.

property is_available: bool

Returns true if the device is currently available.

Returns:: bool – Return if the device is currently available.

property properties: DeviceCapabilities

Return the device properties

Please see braket.device_schema in amazon-braket-schemas-python

Type:: DeviceCapabilities

property topology_graph: DiGraph

topology of device as a networkx DiGraph object.

Examples

>>> import networkx as nx
>>> device = AwsDevice("arn1")
>>> nx.draw_kamada_kawai(device.topology_graph, with_labels=True, font_weight="bold")

>>> topology_subgraph = device.topology_graph.subgraph(range(8))
>>> nx.draw_kamada_kawai(topology_subgraph, with_labels=True, font_weight="bold")

>>> print(device.topology_graph.edges)

Returns:: DiGraph – topology of QPU as a networkx DiGraph object. None if the topology is not available for the device.
Type:: DiGraph

property frames: dict[str, Frame]: Returns a dict mapping frame ids to the frame objects for predefined frames for this device.

property ports: dict[str, Port]: Returns a dict mapping port ids to the port objects for predefined ports for this device.

static get_devices(arns: list[str] | None = None, names: list[str] | None = None, types: list[AwsDeviceType] | None = None, statuses: list[str] | None = None, provider_names: list[str] | None = None, order_by: str = 'name', aws_session: AwsSession | None = None) → list[AwsDevice][source]

Get devices based on filters and desired ordering. The result is the AND of all the filters arns, names, types, statuses, provider_names.

Examples

>>> AwsDevice.get_devices(provider_names=["Rigetti"], statuses=["ONLINE"])
>>> AwsDevice.get_devices(order_by="provider_name")
>>> AwsDevice.get_devices(types=["SIMULATOR"])

Parameters:

arns (Optional[list[str]]) – device ARN filter, default is None
names (Optional[list[str]]) – device name filter, default is None
types (Optional[list[AwsDeviceType]]) – device type filter, default is None QPUs will be searched for all regions and simulators will only be searched for the region of the current session.
statuses (Optional[list[str]]) – device status filter, default is None. When None is used, RETIRED devices will not be returned. To include RETIRED devices in the results, use a filter that includes “RETIRED” for this parameter.
provider_names (Optional[list[str]]) – provider name filter, default is None
order_by (str) – field to order result by, default is name. Accepted values are [‘arn’, ‘name’, ‘type’, ‘provider_name’, ‘status’]
aws_session (Optional[AwsSession]) – An AWS session object. Default is None.

Raises:

ValueError – order_by not in [‘arn’, ‘name’, ‘type’, ‘provider_name’, ‘status’]

Returns:

list[AwsDevice] – list of AWS devices

static get_device_region(device_arn: str) → str[source]

Gets the region from a device arn.

Parameters:: device_arn (str) – The device ARN.
Raises:: ValueError – Raised if the ARN is not properly formatted
Returns:: str – the region of the ARN.

queue_depth() → QueueDepthInfo[source]

Task queue depth refers to the total number of quantum tasks currently waiting to run on a particular device.

Returns:: QueueDepthInfo – Instance of the QueueDepth class representing queue depth information for quantum tasks and hybrid jobs. Queue depth refers to the number of quantum tasks and hybrid jobs queued on a particular device. The normal tasks refers to the quantum tasks not submitted via Hybrid Jobs. Whereas, the priority tasks refers to the total number of quantum tasks waiting to run submitted through Amazon Braket Hybrid Jobs. These tasks run before the normal tasks. If the queue depth for normal or priority quantum tasks is greater than 4000, we display their respective queue depth as ‘>4000’. Similarly, for hybrid jobs if there are more than 1000 jobs queued on a device, display the hybrid jobs queue depth as ‘>1000’. Additionally, for QPUs if hybrid jobs queue depth is 0, we display information about priority and count of the running hybrid job.

Example

Queue depth information for a running job. >>> device = AwsDevice(Device.Amazon.SV1) >>> print(device.queue_depth()) QueueDepthInfo(quantum_tasks={<QueueType.NORMAL: ‘Normal’>: ‘0’, <QueueType.PRIORITY: ‘Priority’>: ‘1’}, jobs=’0 (1 prioritized job(s) running)’)

If more than 4000 quantum tasks queued on a device. >>> device = AwsDevice(Device.Amazon.DM1) >>> print(device.queue_depth()) QueueDepthInfo(quantum_tasks={<QueueType.NORMAL: ‘Normal’>: ‘>4000’, <QueueType.PRIORITY: ‘Priority’>: ‘2000’}, jobs=’100’)

refresh_gate_calibrations() → GateCalibrations | None[source]

Refreshes the gate calibration data upon request.

If the device does not have calibration data, None is returned.

Raises:: URLError – If the URL provided returns a non 2xx response.
Returns:: Optional[GateCalibrations] – the calibration data for the device. None is returned if the device does not have a gate calibrations URL associated.