diff --git a/src/braket/aws/aws_device.py b/src/braket/aws/aws_device.py index 041098f5a..309e23e7d 100644 --- a/src/braket/aws/aws_device.py +++ b/src/braket/aws/aws_device.py @@ -18,6 +18,7 @@ import os import urllib.request import warnings +from copy import deepcopy from datetime import datetime from enum import Enum from typing import Any, ClassVar, Optional, Union @@ -27,6 +28,13 @@ from braket.ahs.analog_hamiltonian_simulation import AnalogHamiltonianSimulation from braket.annealing.problem import Problem +from braket.aws.aws_emulation import ( + connectivity_validator, + gate_connectivity_validator, + gate_validator, + qubit_count_validator, +) +from braket.aws.aws_noise_models import device_noise_model from braket.aws.aws_quantum_task import AwsQuantumTask from braket.aws.aws_quantum_task_batch import AwsQuantumTaskBatch from braket.aws.aws_session import AwsSession @@ -39,14 +47,18 @@ # TODO: Remove device_action module once this is added to init in the schemas repo from braket.device_schema.pulse.pulse_device_action_properties_v1 import PulseDeviceActionProperties +from braket.devices import Devices from braket.devices.device import Device +from braket.emulation import Emulator from braket.ir.blackbird import Program as BlackbirdProgram from braket.ir.openqasm import Program as OpenQasmProgram from braket.parametric.free_parameter import FreeParameter from braket.parametric.free_parameter_expression import _is_float +from braket.passes import ProgramType from braket.pulse import ArbitraryWaveform, Frame, Port, PulseSequence from braket.pulse.waveforms import _parse_waveform_from_calibration_schema from braket.schema_common import BraketSchemaBase +from braket.tasks import QuantumTask class AwsDeviceType(str, Enum): @@ -855,3 +867,114 @@ def _parse_calibration_json( parsed_calibration_data[gate_qubit_key] = gate_qubit_pulse return parsed_calibration_data + + @property + def emulator(self) -> Emulator: + """ + A device emulator mimics the restrictions and noise of the AWS QPU by validating and + compiling programs before running them on a simulated backend. An emulator can be used + as a soft check that a program can run the target AwsDevice. + + Examples: + >>> device = AwsDevice(Devices.IQM.Garnet) + >>> circuit = Circuit().cnot(0, 1).h(2).cz(2, 3) + >>> device.validate(circuit) + >>> # validates, compiles and runs on the local simulator. + >>> result = device.emulator(circuit, shots=100) + >>> print(result.result().measurement_counts) + + Returns: + Emulator: An emulator for this device, if this is not a simulator device. Raises an + exception if an emulator is requested for al simulator device. + """ + if self._arn in [simulator_enum.value for simulator_enum in Devices.Amazon]: + raise ValueError( + "Creating an emulator from a Braket managed simulator is not supported." + ) + if not hasattr(self, "_emulator"): + self._emulator = self._setup_emulator() + return self._emulator + + def _setup_emulator(self) -> Emulator: + """ + Sets up an Emulator object whose properties mimic that of this AwsDevice, if the device is a + real QPU (not a simulator). + + Returns: + Emulator: An emulator with a noise model, compilation passes, and validation passes + based on this device's properites. + """ + emulator_noise_model = device_noise_model(self.properties, self._arn) + self._emulator = Emulator( + noise_model=emulator_noise_model, backend="braket_dm", name=self._name + ) + + self._emulator.add_pass(qubit_count_validator(self.properties)) + self._emulator.add_pass(gate_validator(self.properties)) + self._emulator.add_pass(connectivity_validator(self.properties, self.topology_graph)) + self._emulator.add_pass(gate_connectivity_validator(self.properties, self.topology_graph)) + return self._emulator + + def validate( + self, + task_specification: ProgramType, + ) -> None: + """ + Runs all non-modifying emulator passes on the input program and raises an + error if any device-specific criteria are not met by the program. If the + program meets all criteria, returns. + + Args: + task_specification (ProgramType): The quantum program to emulate against + this AwsDevice device properties. + + """ + self.emulator.validate(task_specification) + + def run_passes( + self, task_specification: ProgramType, apply_noise_model: bool = True + ) -> ProgramType: + """ + Runs all emulator passes and returns the modified program, which should be the same + type as the input program. + + Args: + task_specification (ProgramType): The quantum program to emulate against + this AwsDevice device properties. + + apply_noise_model (bool): If true, apply a device specific noise model to the program + before returning. + + Returns: + ProgramType: A validated and compiled program that may be augmented with noise + operations to mimic noise on this device. + """ + task_specification = deepcopy(task_specification) + return self.emulator.run_passes(task_specification, apply_noise_model) + + def emulate( + self, + task_specification: ProgramType, + shots: Optional[int] = None, + inputs: Optional[dict[str, float]] = None, + ) -> QuantumTask: + """Emulate a quantum task specification on this quantum device emulator. + A quantum task can be a circuit. Emulation + involves running all emulator passes on the input program before running + the program on the emulator's backend. + + Args: + task_specification (ProgramType): Specification of a quantum task + to run on device. + + shots (Optional[int]): The number of times to run the quantum task on the device. + Default is `None`. + + inputs (Optional[dict[str, float]]): Inputs to be passed along with the + IR. If IR is an OpenQASM Program, the inputs will be updated with this value. + Not all devices and IR formats support inputs. Default: {}. + Returns: + QuantumTask: The QuantumTask tracking task execution on this device emulator. + """ + task_specification = deepcopy(task_specification) + return self.emulator.run(task_specification, shots, inputs) diff --git a/src/braket/aws/aws_emulation.py b/src/braket/aws/aws_emulation.py new file mode 100644 index 000000000..75ff9ad58 --- /dev/null +++ b/src/braket/aws/aws_emulation.py @@ -0,0 +1,214 @@ +from collections.abc import Iterable +from functools import singledispatch +from typing import Union + +from networkx import DiGraph + +from braket.device_schema import DeviceActionType, DeviceCapabilities +from braket.device_schema.ionq import IonqDeviceCapabilities +from braket.device_schema.iqm import IqmDeviceCapabilities +from braket.device_schema.rigetti import RigettiDeviceCapabilities +from braket.emulation.emulation_passes.gate_device_passes import ( + ConnectivityValidator, + GateConnectivityValidator, + GateValidator, + QubitCountValidator, +) + + +def qubit_count_validator(properties: DeviceCapabilities) -> QubitCountValidator: + """ + Create a QubitCountValidator pass which checks that the number of qubits used in a program does + not exceed the number of qubits allowed by a QPU, as defined in the device properties. + + Args: + properties (DeviceCapabilities): QPU Device Capabilities object with a + QHP-specific schema. + + Returns: + QubitCountValidator: An emulator pass that checks that the number of qubits used in a + program does not exceed that of the max qubit count on the device. + """ + qubit_count = properties.paradigm.qubitCount + return QubitCountValidator(qubit_count) + + +def gate_validator(properties: DeviceCapabilities) -> GateValidator: + """ + Create a GateValidator pass which defines what supported and native gates are allowed in a + program based on the provided device properties. + + Args: + properties (DeviceCapabilities): QPU Device Capabilities object with a + QHP-specific schema. + + Returns: + GateValidator: An emulator pass that checks that a circuit only uses supported gates and + verbatim circuits only use native gates. + """ + + supported_gates = properties.action[DeviceActionType.OPENQASM].supportedOperations + native_gates = properties.paradigm.nativeGateSet + + return GateValidator(supported_gates=supported_gates, native_gates=native_gates) + + +def connectivity_validator( + properties: DeviceCapabilities, connectivity_graph: DiGraph +) -> ConnectivityValidator: + """ + Creates a ConnectivityValidator pass which validates that two-qubit gates are applied to + connected qubits based on this device's connectivity graph. + + Args: + properties (DeviceCapabilities): QPU Device Capabilities object with a + QHP-specific schema. + + connectivity_graph (DiGraph): Connectivity graph for this device. + + Returns: + ConnectivityValidator: An emulator pass that checks that a circuit only applies two-qubit + gates to connected qubits on the device. + """ + + return _connectivity_validator(properties, connectivity_graph) + + +@singledispatch +def _connectivity_validator( + properties: DeviceCapabilities, connectivity_graph: DiGraph +) -> ConnectivityValidator: + + connectivity_validator = ConnectivityValidator(connectivity_graph) + return connectivity_validator + + +@_connectivity_validator.register(IqmDeviceCapabilities) +def _(properties: IqmDeviceCapabilities, connectivity_graph: DiGraph) -> ConnectivityValidator: + """ + IQM qubit connectivity is undirected but the directed graph that represents qubit connectivity + does not include back-edges. Thus, we must explicitly introduce back edges before creating + the ConnectivityValidator for an IQM device. + """ + connectivity_graph = connectivity_graph.copy() + for edge in connectivity_graph.edges: + connectivity_graph.add_edge(edge[1], edge[0]) + return ConnectivityValidator(connectivity_graph) + + +def gate_connectivity_validator( + properties: DeviceCapabilities, connectivity_graph: DiGraph +) -> GateConnectivityValidator: + return _gate_connectivity_validator(properties, connectivity_graph) + + +@singledispatch +def _gate_connectivity_validator( + properties: DeviceCapabilities, connectivity_graph: DiGraph +) -> GateConnectivityValidator: + raise NotImplementedError + + +@_gate_connectivity_validator.register(IqmDeviceCapabilities) +@_gate_connectivity_validator.register(RigettiDeviceCapabilities) +def _( + properties: RigettiDeviceCapabilities, connectivity_graph: DiGraph +) -> GateConnectivityValidator: + """ + Both IQM and Rigetti have undirected connectivity graphs; Rigetti device capabilities + provide back edges, but the calibration data only provides edges in one direction. + Additionally, IQM does not provide back edges in its connectivity_graph (nor is this + resolved manually by AwsDevice at the moment). + """ + gate_connectivity_graph = connectivity_graph.copy() + edge_properties = properties.standardized.twoQubitProperties + for u, v in gate_connectivity_graph.edges: + edge_key = "-".join([str(qubit) for qubit in (u, v)]) + edge_property = edge_properties.get(edge_key) + + # Check that the QHP provided calibration data for this edge. + if not edge_property: + gate_connectivity_graph[u][v]["supported_gates"] = set() + continue + edge_supported_gates = _get_qpu_gate_translations( + properties, [property.gateName for property in edge_property.twoQubitGateFidelity] + ) + gate_connectivity_graph[u][v]["supported_gates"] = set(edge_supported_gates) + + # Add the reversed edge to ensure gates can be applied + # in both directions for a given qubit pair. + for u, v in gate_connectivity_graph.edges: + if (v, u) not in gate_connectivity_graph.edges or gate_connectivity_graph[v][u].get( + "supported_gates" + ) in [None, set()]: + gate_connectivity_graph.add_edge( + v, u, supported_gates=set(gate_connectivity_graph[u][v]["supported_gates"]) + ) + + return GateConnectivityValidator(gate_connectivity_graph) + + +@_gate_connectivity_validator.register(IonqDeviceCapabilities) +def _(properties: IonqDeviceCapabilities, connectivity_graph: DiGraph) -> GateConnectivityValidator: + """ + Qubits in IonQ's trapped ion devices are all fully connected with identical + gate-pair capabilities. IonQ does not expliclty provide a set of edges for + gate connectivity between qubit pairs in their trapped ion QPUs. + We extrapolate gate connectivity across all possible qubit edge pairs. + """ + gate_connectivity_graph = connectivity_graph.copy() + native_gates = _get_qpu_gate_translations(properties, properties.paradigm.nativeGateSet) + + for edge in gate_connectivity_graph.edges: + gate_connectivity_graph[edge[0]][edge[1]]["supported_gates"] = set(native_gates) + + return GateConnectivityValidator(gate_connectivity_graph) + + +def _get_qpu_gate_translations( + properties: DeviceCapabilities, gate_name: Union[str, Iterable[str]] +) -> Union[str, list[str]]: + """Returns the translated gate name(s) for a given QPU device capabilities schema type + and gate name(s). + + Args: + properties (DeviceCapabilities): Device capabilities object based on a + device-specific schema. + gate_name (Union[str, Iterable[str]]): The name(s) of the gate(s). If gate_name is a list + of string gate names, this function attempts to retrieve translations of all the gate + names. + + Returns: + Union[str, list[str]]: The translated gate name(s) + """ + if isinstance(gate_name, str): + return _get_qpu_gate_translation(properties, gate_name) + else: + return [_get_qpu_gate_translation(properties, name) for name in gate_name] + + +@singledispatch +def _get_qpu_gate_translation(properties: DeviceCapabilities, gate_name: str) -> str: + """Returns the translated gate name for a given QPU ARN and gate name. + + Args: + properties (DeviceCapabilities): QPU Device Capabilities object with a + QHP-specific schema. + gate_name (str): The name of the gate + + Returns: + str: The translated gate name + """ + return gate_name + + +@_get_qpu_gate_translation.register(RigettiDeviceCapabilities) +def _(properties: RigettiDeviceCapabilities, gate_name: str) -> str: + translations = {"CPHASE": "CPhaseShift"} + return translations.get(gate_name, gate_name) + + +@_get_qpu_gate_translation.register(IonqDeviceCapabilities) +def _(properties: IonqDeviceCapabilities, gate_name: str) -> str: + translations = {"GPI": "GPi", "GPI2": "GPi2"} + return translations.get(gate_name, gate_name) diff --git a/src/braket/aws/aws_noise_models.py b/src/braket/aws/aws_noise_models.py new file mode 100644 index 000000000..5a4f9223f --- /dev/null +++ b/src/braket/aws/aws_noise_models.py @@ -0,0 +1,282 @@ +from dataclasses import dataclass +from functools import singledispatch +from typing import Dict, List, Set, Tuple, Union + +import numpy as np + +from braket.aws.aws_emulation import _get_qpu_gate_translations +from braket.circuits import Gate +from braket.circuits.noise_model import GateCriteria, NoiseModel, ObservableCriteria +from braket.circuits.noises import ( + AmplitudeDamping, + BitFlip, + Depolarizing, + PhaseDamping, + TwoQubitDepolarizing, +) +from braket.device_schema import DeviceCapabilities +from braket.device_schema.ionq import IonqDeviceCapabilities +from braket.device_schema.iqm import IqmDeviceCapabilities +from braket.device_schema.rigetti import RigettiDeviceCapabilities +from braket.device_schema.standardized_gate_model_qpu_device_properties_v1 import ( + GateFidelity2Q, + OneQubitProperties, + StandardizedGateModelQpuDeviceProperties, +) +from braket.devices import Devices + +""" + The following gate duration values are not available through Braket device + calibration data and must be hardcoded. +""" +_QPU_GATE_DURATIONS = { + Devices.Rigetti.AspenM3: { + "single_qubit_gate_duration": 40e-9, + "two_qubit_gate_duration": 240e-9, + }, + Devices.IQM.Garnet: {"single_qubit_gate_duration": 32e-9, "two_qubit_gate_duration": 60e-9}, +} + + +@dataclass +class GateFidelity: + gate: Gate + fidelity: float + + +@dataclass +class GateDeviceCalibrationData: + single_qubit_gate_duration: float + two_qubit_gate_duration: float + qubit_labels: Set[int] + single_qubit_specs: Dict[int, Dict[str, float]] + two_qubit_edge_specs: Dict[Tuple[int, int], List[GateFidelity]] + + def _validate_single_qubit_specs(self) -> None: + """ + Checks single qubit specs and the input qubit labels are compatible with + one another. + + Raises: + ValueError: If a qubit in the single-qubit calibration data is not mentioned in the + provided qubit labels. + """ + for qubit in self.single_qubit_specs.keys(): + if qubit not in self.qubit_labels: + raise ValueError(f"Invalid qubit label {qubit}") + + def _validate_two_qubit_specs(self) -> None: + """ + Checks that the qubit edge specs and the input qubit labels are compatible + with one another. + + Raises: + ValueError: If a qubit in the two-qubit calibration data is not mentioned in the + provided qubit labels. + """ + for edge in self.two_qubit_edge_specs.keys(): + if edge[0] not in self.qubit_labels or edge[1] not in self.qubit_labels: + raise ValueError(f"Invalid qubit pair {edge}") + + def __post_init__(self): + self._validate_single_qubit_specs() + self._validate_two_qubit_specs() + + +def device_noise_model(properties: DeviceCapabilities, arn: str) -> NoiseModel: + """ + Create a device-specific noise model using the calibration data provided + in the device properties object for a QPU. + + Args: + properties (DeviceCapabilities): Data structure containing + device properties and calibration data to be used when creating the + device noise model for an emulator. + + arn (str): Amazon Braket ARN for the QPU. + + Returns: + NoiseModel: Returns a NoiseModel object created using the device's + calibration data created from the device properties. + """ + device_calibration_data = _setup_calibration_specs(properties, arn) + noise_model = _setup_basic_noise_model_strategy(device_calibration_data) + return noise_model + + +@singledispatch +def _setup_calibration_specs(properties: DeviceCapabilities, arn: str) -> NoiseModel: + raise NotImplementedError( + f"A noise model cannot be created from device capabilities with type {type(properties)}." + ) + + +@_setup_calibration_specs.register(RigettiDeviceCapabilities) +@_setup_calibration_specs.register(IqmDeviceCapabilities) +def _(properties: Union[RigettiDeviceCapabilities, IqmDeviceCapabilities], arn: str) -> NoiseModel: + gate_durations = _QPU_GATE_DURATIONS.get(arn, None) + if not gate_durations: + raise ValueError(f"Gate durations are not available for device {arn}") + single_qubit_gate_duration = gate_durations["single_qubit_gate_duration"] + two_qubit_gate_duration = gate_durations["two_qubit_gate_duration"] + + standardized_properties: StandardizedGateModelQpuDeviceProperties = properties.standardized + one_qubit_properties = standardized_properties.oneQubitProperties + qubit_labels = set(int(qubit) for qubit in one_qubit_properties.keys()) + single_qubit_specs = { + int(qubit): _create_qubit_specs(one_qubit_properties[qubit]) + for qubit in one_qubit_properties.keys() + } + + two_qubit_properties = standardized_properties.twoQubitProperties + two_qubit_edge_specs = { + tuple(int(qubit) for qubit in edge.split("-")[0:2]): _create_edge_specs( + properties, gate_fidelities.twoQubitGateFidelity + ) + for edge, gate_fidelities in two_qubit_properties.items() + } + + return GateDeviceCalibrationData( + single_qubit_gate_duration, + two_qubit_gate_duration, + qubit_labels, + single_qubit_specs, + two_qubit_edge_specs, + ) + + +@_setup_calibration_specs.register(IonqDeviceCapabilities) +def _(properties: IonqDeviceCapabilities, arn: str) -> NoiseModel: + """ + IonQ's Trapped Ion Devices do not have per-qubit calibration data and instead + provide averaged qubit and two-qubit gate fidelities across the device. All qubits + are connected in a trapped ion device. + + We instead copy the averaged fidelities to each qubit and qubit edge pair. + """ + calibration_data = properties.provider + fidelity_data = calibration_data.fidelity + timing_data = calibration_data.timing + qubit_count = properties.paradigm.qubitCount + native_gates = _get_qpu_gate_translations(properties, properties.paradigm.nativeGateSet) + + single_qubit_gate_duration = timing_data["1Q"] + two_qubit_gate_duration = timing_data["2Q"] + average_active_reset_fidelity = timing_data["reset"] + average_T1 = timing_data["T1"] + average_T2 = timing_data["T2"] + single_qubit_rb_fidelity = fidelity_data["1Q"]["mean"] + two_qubit_rb_fidelity = fidelity_data["2Q"]["mean"] + average_readout_fidelity = fidelity_data["spam"]["mean"] + + native_gate_fidelities = [] + for native_gate in native_gates: + gate_name = native_gate + if hasattr(Gate, gate_name): + gate = getattr(Gate, gate_name) + if gate.fixed_qubit_count() != 2: + """ + The noise model applies depolarizing noise associated with the + individual qubits themselves (RB/sRB fidelities). This is a choice + of this particular model to generalize the implementation as not + all QHPs provide single-qubit gate fidelities. + """ + continue + native_gate_fidelities.append(GateFidelity(gate, two_qubit_rb_fidelity)) + single_qubit_specs = {} + two_qubit_edge_specs = {} + for ii in range(qubit_count): + qubit_spec = { + "RANDOMIZED_BENCHMARKING": single_qubit_rb_fidelity, + "SIMULTANEOUS_RANDOMIZED_BENCHMARKING": None, + "READOUT": average_readout_fidelity, + "T1": average_T1, + "T2": average_T2, + "ACTIVE_RESET": average_active_reset_fidelity, + } + single_qubit_specs[ii] = qubit_spec + + for jj in range(ii + 1, qubit_count): + two_qubit_edge_specs[(ii, jj)] = native_gate_fidelities + + return GateDeviceCalibrationData( + single_qubit_gate_duration, + two_qubit_gate_duration, + set(range(qubit_count)), + single_qubit_specs, + two_qubit_edge_specs, + ) + + +def _create_qubit_specs(qubit_properties: OneQubitProperties) -> Dict[str, int]: + T1 = qubit_properties.T1.value + T2 = qubit_properties.T2.value + qubit_fidelities = qubit_properties.oneQubitFidelity + one_qubit_fidelities = { + qubit_fidelity.fidelityType.name: qubit_fidelity.fidelity + for qubit_fidelity in qubit_fidelities + } + one_qubit_fidelities["T1"] = T1 + one_qubit_fidelities["T2"] = T2 + return one_qubit_fidelities + + +def _create_edge_specs( + properties: DeviceCapabilities, edge_properties: List[GateFidelity2Q] +) -> List[GateFidelity]: + edge_specs = [] + for edge_property in edge_properties: + gate_name = _get_qpu_gate_translations(properties, edge_property.gateName) + if hasattr(Gate, gate_name): + gate = getattr(Gate, gate_name) + edge_specs.append(GateFidelity(gate, edge_property.fidelity)) + return edge_specs + + +def _setup_basic_noise_model_strategy( + gate_calibration_data: GateDeviceCalibrationData, +) -> NoiseModel: + """ + Apply a basic noise model strategy consisting of: + - T1 Dampening + - T2 Phase Dampening + - 1 Qubit RB Depolarizing Noise + - 1 Qubit Readout Error + - 2 Qubit Gate Depolarizing Noise + """ + noise_model = NoiseModel() + gate_duration_1Q = gate_calibration_data.single_qubit_gate_duration + for qubit, data in gate_calibration_data.single_qubit_specs.items(): + # T1 dampening + T1 = data["T1"] + damping_prob = 1 - np.exp(-(gate_duration_1Q / T1)) + noise_model.add_noise(AmplitudeDamping(damping_prob), GateCriteria(qubits=qubit)) + + # T2 Phase Dampening + T2 = data["T2"] + dephasing_prob = 0.5 * (1 - np.exp(-(gate_duration_1Q / T2))) + noise_model.add_noise(PhaseDamping(dephasing_prob), GateCriteria(qubits=qubit)) + + # 1 Qubit RB Depolarizing Noise + if data.get("SIMULTANEOUS_RANDOMIZED_BENCHMARKING"): + benchmark_fidelity = data["SIMULTANEOUS_RANDOMIZED_BENCHMARKING"] + else: + benchmark_fidelity = data.get("RANDOMIZED_BENCHMARKING") + if benchmark_fidelity: + depolarizing_rate = 1 - benchmark_fidelity + noise_model.add_noise(Depolarizing(depolarizing_rate), GateCriteria(qubits=qubit)) + + # 1 Qubit Readout Error + readout_error_rate = 1 - data["READOUT"] + noise_model.add_noise(BitFlip(readout_error_rate), ObservableCriteria(qubits=qubit)) + + for edge, data in gate_calibration_data.two_qubit_edge_specs.items(): + for gate_fidelity in data: + rate = 1 - gate_fidelity.fidelity + gate = gate_fidelity.gate + noise_model.add_noise( + TwoQubitDepolarizing(rate), + GateCriteria(gate, [(edge[0], edge[1]), (edge[1], edge[0])]), + ) + + return noise_model diff --git a/src/braket/emulation/__init__.py b/src/braket/emulation/__init__.py new file mode 100644 index 000000000..e1c29ccd4 --- /dev/null +++ b/src/braket/emulation/__init__.py @@ -0,0 +1 @@ +from braket.emulation.emulator import Emulator # noqa: F40 diff --git a/src/braket/emulation/base_emulator.py b/src/braket/emulation/base_emulator.py new file mode 100644 index 000000000..0674d8663 --- /dev/null +++ b/src/braket/emulation/base_emulator.py @@ -0,0 +1,66 @@ +from __future__ import annotations + +from typing import Iterable, Union + +from braket.emulation.emulation_passes import ValidationPass +from braket.passes import BasePass, ProgramType + + +class BaseEmulator: + def __init__(self, emulator_passes: Iterable[BasePass] = None): + self._emulator_passes = emulator_passes if emulator_passes is not None else [] + + def run_passes(self, task_specification: ProgramType) -> ProgramType: + """ + This method passes the input program through the Passes contained + within this emulator. An emulator pass may simply validate a program or may + modify or entirely transform the program (to an equivalent quantum program). + + Args: + task_specification (ProgramType): The program to run the emulator passes on. + + Returns: + ProgramType: A "compiled" program of the same type as the input. + + """ + for emulator_pass in self._emulator_passes: + task_specification = emulator_pass(task_specification) + return task_specification + + def validate(self, task_specification: ProgramType) -> None: + """ + This method passes the input program through Passes that perform + only validation, without modifying the input program. + + Args: + task_specification (ProgramType): The program to validate with this + emulator's validation passes. + """ + for emulator_pass in self._emulator_passes: + if isinstance(emulator_pass, ValidationPass): + emulator_pass(task_specification) + + def add_pass(self, emulator_pass: Union[Iterable[BasePass], BasePass]) -> BaseEmulator: + """ + Append a new BasePass or a list of BasePass objects. + + Args: + emulator_pass (Union[Iterable[BasePass], BasePass]): Either a + single Pass object or a list of Pass objects that + will be used in validation and program compilation passes by this + emulator. + + Returns: + BaseEmulator: Returns an updated self. + + Raises: + TypeError: If the input is not an iterable or an Pass. + + """ + if isinstance(emulator_pass, Iterable): + self._emulator_passes.extend(emulator_pass) + elif isinstance(emulator_pass, BasePass): + self._emulator_passes.append(emulator_pass) + else: + raise TypeError("emulator_pass must be an Pass or an iterable of Pass") + return self diff --git a/src/braket/emulation/emulation_passes/__init__.py b/src/braket/emulation/emulation_passes/__init__.py new file mode 100644 index 000000000..691d2da7a --- /dev/null +++ b/src/braket/emulation/emulation_passes/__init__.py @@ -0,0 +1 @@ +from braket.emulation.emulation_passes.validation_pass import ValidationPass # noqa: F401 diff --git a/src/braket/emulation/emulation_passes/gate_device_passes/__init__.py b/src/braket/emulation/emulation_passes/gate_device_passes/__init__.py new file mode 100644 index 000000000..030791be6 --- /dev/null +++ b/src/braket/emulation/emulation_passes/gate_device_passes/__init__.py @@ -0,0 +1,12 @@ +from braket.emulation.emulation_passes.gate_device_passes.connectivity_validator import ( # noqa: F401 E501 + ConnectivityValidator, +) +from braket.emulation.emulation_passes.gate_device_passes.gate_connectivity_validator import ( # noqa: F401 E501 + GateConnectivityValidator, +) +from braket.emulation.emulation_passes.gate_device_passes.gate_validator import ( # noqa: F401 E501 + GateValidator, +) +from braket.emulation.emulation_passes.gate_device_passes.qubit_count_validator import ( # noqa: F401 E501 + QubitCountValidator, +) diff --git a/src/braket/emulation/emulation_passes/gate_device_passes/connectivity_validator.py b/src/braket/emulation/emulation_passes/gate_device_passes/connectivity_validator.py new file mode 100644 index 000000000..b858cc6a4 --- /dev/null +++ b/src/braket/emulation/emulation_passes/gate_device_passes/connectivity_validator.py @@ -0,0 +1,160 @@ +from collections.abc import Iterable +from typing import Dict, Optional, Union + +from networkx import DiGraph, complete_graph, from_dict_of_lists +from networkx.utils import graphs_equal + +from braket.circuits import Circuit +from braket.circuits.compiler_directives import StartVerbatimBox +from braket.circuits.gate import Gate +from braket.emulation.emulation_passes import ValidationPass +from braket.registers.qubit_set import QubitSet + + +class ConnectivityValidator(ValidationPass[Circuit]): + def __init__( + self, + connectivity_graph: Optional[Union[Dict[int, Iterable[int]], DiGraph]] = None, + fully_connected: bool = False, + num_qubits: Optional[int] = None, + qubit_labels: Optional[Union[Iterable[int], QubitSet]] = None, + directed: bool = True, + ): + """ + A ConnectivityValidator instance takes in a qubit connectivity graph and validates that + a circuit that uses verbatim circuits makes valid hardware qubit references in single + and two-qubit gate operations. + + Args: + connectivity_graph (Optional[Union[Dict[int, Iterable[int]], DiGraph]]): + Either a sparse matrix or DiGraph representation of the device connectivity. + Can be None if fully_connected is true. + + fully_connected (bool): If true, the all qubits in the device are connected. + + num_qubits (Optional[int]): The number of qubits in the device; if fully_connected is + True, create a complete graph with num_qubits nodes; ignored if + connectivity_graph is provided and fully_connected if False. + + qubit_labels (Optional[Union[Iterable[int], QubitSet]]): A set of qubit labels; if + fully_connected is True, the qubits_labels are used as nodes of a fully connected + topology; ignored if connectivity_graph is provided and fully_connected if False. + + directed (bool): Denotes if the connectivity graph is directed or undirected. If + the connectivity graph is undirected, this constructor attempts to fill in any + missing back edges. + + Raises: + ValueError: If the inputs do not correctly yield a connectivity graph; i.e. + fully_connected is true but neither/both num qubits and qubit labels are defined + or a valid DiGraph or dict representation of a connectivity graph is not provided. + """ + + if not ((connectivity_graph is not None) ^ fully_connected): + raise ValueError( + "Either the connectivity_graph must be provided OR fully_connected must be True\ + (not both)." + ) + + if fully_connected: + if not ((num_qubits is None) ^ (qubit_labels is None)): + raise ValueError( + "Either num_qubits or qubit_labels (NOT both) must be \ + provided if fully_connected is True." + ) + self._connectivity_graph = complete_graph( + num_qubits if num_qubits else qubit_labels, create_using=DiGraph() + ) + elif not isinstance(connectivity_graph, DiGraph): + try: + self._connectivity_graph = from_dict_of_lists( + connectivity_graph, create_using=DiGraph() + ) + except Exception as e: + raise ValueError( + f"connectivity_graph must be a valid DiGraph or a dictionary\ + mapping integers (nodes) to a list of integers (adjancency lists): {e}" + ) + else: + self._connectivity_graph = connectivity_graph + + if not directed: + for edge in self._connectivity_graph.edges: + self._connectivity_graph.add_edge(edge[1], edge[0]) + + def validate(self, program: Circuit) -> None: + """ + Verifies that any verbatim box in a circuit is runnable with respect to the + device connectivity definied by this validator. If any sub-circuit of the + input circuit is verbatim, we validate the connectivity of all gate operations + in the circuit. + + Args: + program (Circuit): The Braket circuit whose gate operations to + validate. + + Raises: + ValueError: If a hardware qubit reference does not exist in the connectivity graph. + """ + # If any of the instructions are in verbatim mode, all qubit references + # must point to hardware qubits. Otherwise, this circuit need not be validated. + if not any( + [ + isinstance(instruction.operator, StartVerbatimBox) + for instruction in program.instructions + ] + ): + return + for idx in range(len(program.instructions)): + instruction = program.instructions[idx] + if isinstance(instruction.operator, Gate): + if ( + instruction.operator.qubit_count == 2 + ): # Assuming only maximum 2-qubit native gates are supported + self._validate_instruction_connectivity(instruction.control, instruction.target) + else: + # just check that the target qubit exists in the connectivity graph + target_qubit = instruction.target[0] + if target_qubit not in self._connectivity_graph: + raise ValueError( + f"Qubit {target_qubit} does not exist in the device topology." + ) + + def _validate_instruction_connectivity( + self, control_qubits: QubitSet, target_qubits: QubitSet + ) -> None: + """ + Checks if a two-qubit instruction is valid based on this validator's connectivity + graph. + + Args: + control_qubits (QubitSet): The control qubits used in this multi-qubit + operation. + target_qubits (QubitSet): The target qubits of this operation. For many gates, + both the control and target are stored in "target_qubits", so we may + see target_qubits have length 2. + + Raises: + ValueError: If any two-qubit gate operation uses a qubit edge that does not exist + in the qubit connectivity graph. + """ + # Create edges between each of the target qubits + gate_connectivity_graph = DiGraph() + # Create an edge from each control bit to each target qubit + if len(control_qubits) == 1 and len(target_qubits) == 1: + gate_connectivity_graph.add_edge(control_qubits[0], target_qubits[0]) + elif len(control_qubits) == 0 and len(target_qubits) == 2: + gate_connectivity_graph.add_edges_from( + [(target_qubits[0], target_qubits[1]), (target_qubits[1], target_qubits[0])] + ) + else: + raise ValueError("Unrecognized qubit targetting setup for a 2 qubit gate.") + # Check that each edge exists in this validator's connectivity graph + for e in gate_connectivity_graph.edges: + if not self._connectivity_graph.has_edge(*e): + raise ValueError(f"{e[0]} is not connected to qubit {e[1]} in this device.") + + def __eq__(self, other: ValidationPass) -> bool: + return isinstance(other, ConnectivityValidator) and graphs_equal( + self._connectivity_graph, other._connectivity_graph + ) diff --git a/src/braket/emulation/emulation_passes/gate_device_passes/gate_connectivity_validator.py b/src/braket/emulation/emulation_passes/gate_device_passes/gate_connectivity_validator.py new file mode 100644 index 000000000..b236aa87c --- /dev/null +++ b/src/braket/emulation/emulation_passes/gate_device_passes/gate_connectivity_validator.py @@ -0,0 +1,130 @@ +from typing import Any, Dict, Iterable, Tuple, Union + +from networkx import DiGraph +from networkx.utils import graphs_equal + +from braket.circuits.circuit import Circuit +from braket.circuits.compiler_directives import EndVerbatimBox, StartVerbatimBox +from braket.circuits.gate import Gate +from braket.emulation.emulation_passes import ValidationPass +from braket.registers.qubit_set import QubitSet + + +class GateConnectivityValidator(ValidationPass[Circuit]): + def __init__( + self, + gate_connectivity_graph: Union[Dict[Tuple[Any, Any], Iterable[str]], DiGraph], + directed=True, + ): + super().__init__() + if isinstance(gate_connectivity_graph, dict): + self._gate_connectivity_graph = DiGraph() + for (u, v), supported_gates in gate_connectivity_graph.items(): + self._gate_connectivity_graph.add_edge(u, v, supported_gates=supported_gates) + elif isinstance(gate_connectivity_graph, DiGraph): + self._gate_connectivity_graph = gate_connectivity_graph + else: + raise TypeError( + "Gate_connectivity_graph must either be a dictionary of edges mapped to \ +supported gates lists, or a DiGraph with supported gates \ +provided as edge attributes." + ) + + if not directed: + """ + Add reverse edges and check that any supplied reverse edges have + identical supported gate sets to their corresponding forwards edge. + """ + for u, v in self._gate_connectivity_graph.edges: + back_edge = (v, u) + if back_edge not in self._gate_connectivity_graph.edges: + supported_gates = self._gate_connectivity_graph[u][v]["supported_gates"] + self._gate_connectivity_graph.add_edge( + *back_edge, supported_gates=supported_gates + ) + else: + # check that the supported gate sets are identical + if ( + self._gate_connectivity_graph[u][v]["supported_gates"] + != self._gate_connectivity_graph[v][u]["supported_gates"] + ): + raise ValueError( + f"Connectivity Graph marked as undirected\ + but edges ({u}, {v}) and ({v}, {u}) have different supported\ + gate sets." + ) + + def validate(self, program: Circuit) -> None: + """ + Verifies that any multiqubit gates used within a verbatim box are supported + by the devices gate connectivity defined by this criteria. + + Args: + program (Circuit): The circuit whose gate instructions need to be validated + against this validator's gate connectivity graph. + + Raises: + ValueError if any of the gate operations use qubits or qubit edges that don't exist + in the qubit connectivity graph or the gate operation is not supported by the edge. + """ + for idx in range(len(program.instructions)): + instruction = program.instructions[idx] + if isinstance(instruction.operator, StartVerbatimBox): + idx += 1 + while idx < len(program.instructions) and not isinstance( + program.instructions[idx].operator, EndVerbatimBox + ): + instruction = program.instructions[idx] + if isinstance(instruction.operator, Gate): + if ( + instruction.operator.qubit_count == 2 + ): # Assuming only maximum 2-qubit native gates are supported + self._validate_instruction_connectivity( + instruction.operator.name, instruction.control, instruction.target + ) + else: + # just check that the target qubit exists in the connectivity graph + target_qubit = instruction.target[0] + if target_qubit not in self._gate_connectivity_graph: + raise ValueError( + f"Qubit {target_qubit} does not exist in the device topology." + ) + idx += 1 + idx += 1 + + def _validate_instruction_connectivity( + self, gate_name: str, control_qubits: QubitSet, target_qubits: QubitSet + ) -> None: + """ + Checks if a specific is able to be applied to the control and target qubits based + on this validator's gate connectivity graph. + + Args: + gate_name (str): The name of the gate being applied. + control_qubits (QubitSet): The set of control qubits used by this gate operation. + target_qubits (QubitSet): The set of target qubits used by this gate operation. + + Raises: + ValueError if the gate operation is not possible on the qubit connectivity graph. + """ + # Create edges between each of the target qubits + if len(control_qubits) == 1 and len(target_qubits) == 1: + e = (control_qubits[0], target_qubits[0]) + elif len(control_qubits) == 0 and len(target_qubits) == 2: + e = (target_qubits[0], target_qubits[1]) + else: + raise ValueError("Unrecognized qubit targetting setup for a 2 qubit gate.") + + # Check that each edge exists in this validator's connectivity graph + if not self._gate_connectivity_graph.has_edge(*e): + raise ValueError(f"{e[0]} is not connected to {e[1]} on this device.") + supported_gates = self._gate_connectivity_graph[e[0]][e[1]]["supported_gates"] + if gate_name not in supported_gates: + raise ValueError( + f"Qubit pair ({e[0]}, {e[1]}) does not support gate {gate_name} on this device." + ) + + def __eq__(self, other: ValidationPass) -> bool: + return isinstance(other, GateConnectivityValidator) and graphs_equal( + self._gate_connectivity_graph, other._gate_connectivity_graph + ) diff --git a/src/braket/emulation/emulation_passes/gate_device_passes/gate_validator.py b/src/braket/emulation/emulation_passes/gate_device_passes/gate_validator.py new file mode 100644 index 000000000..3149b3899 --- /dev/null +++ b/src/braket/emulation/emulation_passes/gate_device_passes/gate_validator.py @@ -0,0 +1,88 @@ +from collections.abc import Iterable +from typing import Optional + +from braket.circuits import Circuit +from braket.circuits.compiler_directives import EndVerbatimBox, StartVerbatimBox +from braket.circuits.gate import Gate +from braket.circuits.translations import BRAKET_GATES +from braket.emulation.emulation_passes import ValidationPass + + +class GateValidator(ValidationPass[Circuit]): + def __init__( + self, + supported_gates: Optional[Iterable[str]] = None, + native_gates: Optional[Iterable[str]] = None, + ): + """ + Args: + supported_gates (Optional[Iterable[str]]): A list of gates supported outside of + verbatim modeby the emulator. A gate is a Braket gate name. + native_gates (Optional[Iterable[str]]): A list of gates supported inside of + verbatim mode by the emulator. + + Raises: + ValueError: If supported_gates and and native_gates are empty or any of the provided + gate are not supported by the Braket BDK. + """ + supported_gates, native_gates = (supported_gates or []), (native_gates or []) + if not len(supported_gates) and not len(native_gates): + raise ValueError("Supported gate set or native gate set must be provided.") + + try: + self._supported_gates = frozenset( + BRAKET_GATES[gate.lower()] for gate in supported_gates + ) + except KeyError as e: + raise ValueError(f"Input {str(e)} in supported_gates is not a valid Braket gate name.") + + try: + self._native_gates = frozenset(BRAKET_GATES[gate.lower()] for gate in native_gates) + except KeyError as e: + raise ValueError(f"Input {str(e)} in native_gates is not a valid Braket gate name.") + + def validate(self, program: Circuit) -> None: + """ + Checks that all non-verbatim gates used in the circuit are in this validator's + supported gate set and that all verbatim gates used in the circuit are in this + validator's native gate set. + + Args: + program (Circuit): The Braket circuit whose gates to validate. + + Raises: + ValueError: If a gate operation or verbatim gate operation is not in this validator's + supported or native gate set, respectively. + """ + idx = 0 + while idx < len(program.instructions): + instruction = program.instructions[idx] + if isinstance(instruction.operator, StartVerbatimBox): + idx += 1 + while idx < len(program.instructions) and not isinstance( + program.instructions[idx].operator, EndVerbatimBox + ): + instruction = program.instructions[idx] + if isinstance(instruction.operator, Gate): + gate = instruction.operator + if not type(gate) in self._native_gates: + raise ValueError( + f"Gate {gate.name} is not a native gate supported by this device." + ) + idx += 1 + if idx == len(program.instructions) or not isinstance( + program.instructions[idx].operator, EndVerbatimBox + ): + raise ValueError(f"No end verbatim box found at index {idx} in the circuit.") + elif isinstance(instruction.operator, Gate): + gate = instruction.operator + if not type(gate) in self._supported_gates: + raise ValueError(f"Gate {gate.name} is not supported by this device.") + idx += 1 + + def __eq__(self, other: ValidationPass) -> bool: + return ( + isinstance(other, GateValidator) + and self._supported_gates == other._supported_gates + and self._native_gates == other._native_gates + ) diff --git a/src/braket/emulation/emulation_passes/gate_device_passes/qubit_count_validator.py b/src/braket/emulation/emulation_passes/gate_device_passes/qubit_count_validator.py new file mode 100644 index 000000000..5ddf3cbbb --- /dev/null +++ b/src/braket/emulation/emulation_passes/gate_device_passes/qubit_count_validator.py @@ -0,0 +1,35 @@ +from braket.circuits import Circuit +from braket.emulation.emulation_passes import ValidationPass + + +class QubitCountValidator(ValidationPass[Circuit]): + """ + A simple validator class that checks that an input program does not use more qubits + than available on a device, as set during this validator's instantiation. + """ + + def __init__(self, qubit_count: int): + if qubit_count <= 0: + raise ValueError(f"qubit_count ({qubit_count}) must be a positive integer.") + self._qubit_count = qubit_count + + def validate(self, circuit: Circuit) -> None: + """ + Checks that the number of qubits used in this circuit does not exceed this + validator's qubit_count max. + + Args: + circuit (Circuit): The Braket circuit whose qubit count to validate. + + Raises: + ValueError: If the number of qubits used in the circuit exceeds the qubit_count. + + """ + if circuit.qubit_count > self._qubit_count: + raise ValueError( + f"Circuit must use at most {self._qubit_count} qubits, \ +but uses {circuit.qubit_count} qubits." + ) + + def __eq__(self, other: ValidationPass) -> bool: + return isinstance(other, QubitCountValidator) and self._qubit_count == other._qubit_count diff --git a/src/braket/emulation/emulation_passes/validation_pass.py b/src/braket/emulation/emulation_passes/validation_pass.py new file mode 100644 index 000000000..5a5f5b016 --- /dev/null +++ b/src/braket/emulation/emulation_passes/validation_pass.py @@ -0,0 +1,32 @@ +from __future__ import annotations + +from abc import abstractmethod + +from braket.passes.base_pass import BasePass, ProgramType + + +class ValidationPass(BasePass[ProgramType]): + @abstractmethod + def validate(self, program: ProgramType) -> None: + """ + An emulator validator is used to perform some non-modifying validation + pass on an input program. Implementations of validate should return + nothing if the input program passes validation and raise an error otherwise. + + Args: + program (ProgramType): The program to be evaluated against this criteria. + """ + raise NotImplementedError + + def run(self, program: ProgramType) -> ProgramType: + """ + Validate the input program and return the program, unmodified. + + Args: + program (ProgramType): The program to validate. + + Returns: + ProgramType: The unmodified progam passed in as input. + """ + self.validate(program) + return program diff --git a/src/braket/emulation/emulator.py b/src/braket/emulation/emulator.py new file mode 100644 index 000000000..c263d8292 --- /dev/null +++ b/src/braket/emulation/emulator.py @@ -0,0 +1,189 @@ +from __future__ import annotations + +import logging +from typing import Any, Iterable, Optional, Union + +from braket.circuits import Circuit +from braket.circuits.noise_model import NoiseModel +from braket.devices import Device +from braket.devices.local_simulator import LocalSimulator +from braket.emulation.base_emulator import BaseEmulator +from braket.ir.openqasm import Program as OpenQasmProgram +from braket.passes import BasePass, ProgramType +from braket.tasks import QuantumTask +from braket.tasks.quantum_task_batch import QuantumTaskBatch + + +class Emulator(Device, BaseEmulator): + + _DEFAULT_SIMULATOR_BACKEND = "default" + _DEFAULT_NOISY_BACKEND = "braket_dm" + + """An emulator is a simulation device that more closely resembles + the capabilities and constraints of a real device or of a specific device model.""" + + def __init__( + self, + backend: str = "default", + noise_model: Optional[NoiseModel] = None, + emulator_passes: Iterable[BasePass] = None, + **kwargs, + ): + Device.__init__(self, name=kwargs.get("name", "DeviceEmulator"), status="AVAILABLE") + BaseEmulator.__init__(self, emulator_passes) + self._noise_model = noise_model + + backend_name = self._get_local_simulator_backend(backend, noise_model) + self._backend = LocalSimulator(backend=backend_name, noise_model=noise_model) + + def _get_local_simulator_backend( + self, backend: str, noise_model: Optional[NoiseModel] = None + ) -> str: + """ + Returns the name of the backend to use with the local simulator. + + Args: + backend (str): The name of the backend requested by the customer, or default if none + were provided. + noise_model (Optional[NoiseModel]): A noise model to use with the emulator, if at all. + If a noise model is provided, the density matrix simulator is used. + + Returns: + str: The name of the backend to pass into the LocalSimulator constructor. + """ + if backend == "default": + if noise_model: + logging.info( + "Setting LocalSimulator backend to use 'braket_dm' \ + because a NoiseModel was provided." + ) + return Emulator._DEFAULT_NOISY_BACKEND + return Emulator._DEFAULT_SIMULATOR_BACKEND + return backend + + def run( + self, + task_specification: Union[ + Circuit, + OpenQasmProgram, + ], + shots: Optional[int] = 0, + inputs: Optional[dict[str, float]] = None, + *args: Any, + **kwargs: Any, + ) -> QuantumTask: + """Emulate a quantum task specification on this quantum device emulator. + A quantum task can be a circuit or an annealing problem. Emulation + involves running all emulator passes on the input program before running + the program on the emulator's backend. + + Args: + task_specification (Union[Circuit, OpenQasmProgram]): Specification of a quantum task + to run on device. + shots (Optional[int]): The number of times to run the quantum task on the device. + Default is `None`. + inputs (Optional[dict[str, float]]): Inputs to be passed along with the + IR. If IR is an OpenQASM Program, the inputs will be updated with this value. + Not all devices and IR formats support inputs. Default: {}. + *args (Any): Arbitrary arguments. + **kwargs (Any): Arbitrary keyword arguments. + + Returns: + QuantumTask: The QuantumTask tracking task execution on this device emulator. + """ + task_specification = self.run_passes(task_specification, apply_noise_model=False) + # Don't apply noise model as the local simulator will automatically apply it. + return self._backend.run(task_specification, shots, inputs, *args, **kwargs) + + def run_batch( # noqa: C901 + self, + task_specifications: Union[ + Union[Circuit, OpenQasmProgram], + list[ + Union[ + Circuit, + OpenQasmProgram, + ] + ], + ], + shots: Optional[int] = 0, + max_parallel: Optional[int] = None, + inputs: Optional[Union[dict[str, float], list[dict[str, float]]]] = None, + *args, + **kwargs, + ) -> QuantumTaskBatch: + raise NotImplementedError("Emulator.run_batch() is not implemented yet.") + + @property + def noise_model(self) -> NoiseModel: + """ + An emulator may be defined with a quantum noise model which mimics the noise + on a physical device. A quantum noise model can be defined using the + NoiseModel class. The noise model is applied to Braket Circuits before + running them on the emulator backend. + + Returns: + NoiseModel: This emulator's noise model. + """ + return self._noise_model + + @noise_model.setter + def noise_model(self, noise_model: NoiseModel) -> None: + """ + Setter method for the Emulator noise_model property. Re-instantiates + the backend with the new NoiseModel object. + + Args: + noise_model (NoiseModel): The new noise model. + """ + self._noise_model = noise_model + self._backend = LocalSimulator(backend="braket_dm", noise_model=noise_model) + + def run_passes( + self, task_specification: ProgramType, apply_noise_model: bool = True + ) -> ProgramType: + """ + Passes the input program through all Pass objects contained in this + emulator and applies the emulator's noise model, if it exists, before + returning the compiled program. + + Args: + task_specification (ProgramType): The input program to validate and + compile based on this emulator's Passes + apply_noise_model (bool): If true, apply this emulator's noise model + to the compiled program before returning the final program. + + Returns: + ProgramType: A compiled program with a noise model applied, if one + exists for this emulator and apply_noise_model is true. + """ + try: + program = super().run_passes(task_specification) + if apply_noise_model and self.noise_model: + return self._noise_model.apply(program) + return program + except Exception as e: + self._raise_exception(e) + + def validate(self, task_specification: ProgramType) -> None: + """ + Runs only Passes that are ValidationPass, i.e. all non-modifying + validation passes on the input program. + + Args: + task_specification (ProgramType): The input program to validate. + """ + try: + super().validate(task_specification) + except Exception as e: + self._raise_exception(e) + + def _raise_exception(self, exception: Exception) -> None: + """ + Wrapper for exceptions, appends the emulator's name to the exception + note. + + Args: + exception (Exception): The exception to modify and raise. + """ + raise Exception(str(exception) + f" ({self._name})") from exception diff --git a/src/braket/passes/__init__.py b/src/braket/passes/__init__.py new file mode 100644 index 000000000..1c9e6397f --- /dev/null +++ b/src/braket/passes/__init__.py @@ -0,0 +1 @@ +from braket.passes.base_pass import BasePass, ProgramType # noqa: F40 diff --git a/src/braket/passes/base_pass.py b/src/braket/passes/base_pass.py new file mode 100644 index 000000000..d5c36aea6 --- /dev/null +++ b/src/braket/passes/base_pass.py @@ -0,0 +1,26 @@ +from abc import ABC, abstractmethod +from typing import Generic, TypeVar + +ProgramType = TypeVar("ProgramType") + + +class BasePass(ABC, Generic[ProgramType]): + @abstractmethod + def run(self, program: ProgramType) -> ProgramType: + """ + Runs a pass on the provided program. + + Args: + program (ProgramType): The program to run the pass on. + + Returns: + ProgramType: The program after the pass has been applied. Same type as the input + program. + + Raises: + NotImplementedError: Method not implemented. + """ + raise NotImplementedError + + def __call__(self, program: ProgramType) -> ProgramType: + return self.run(program) diff --git a/test/unit_tests/braket/aws/test_aws_device.py b/test/unit_tests/braket/aws/test_aws_device.py index a778dfb5f..ba604cf9c 100644 --- a/test/unit_tests/braket/aws/test_aws_device.py +++ b/test/unit_tests/braket/aws/test_aws_device.py @@ -567,6 +567,17 @@ def test_device_simulator_no_aws_session(aws_session_init, aws_session): aws_session.get_device.assert_called_with(arn) +@patch("braket.aws.aws_device.AwsSession") +def test_attempt_get_emulator_with_simulators(aws_session_init, aws_session): + arn = SV1_ARN + aws_session_init.return_value = aws_session + aws_session.get_device.return_value = MOCK_GATE_MODEL_SIMULATOR + device = AwsDevice(arn) + error_message = "Creating an emulator from a Braket managed simulator is not supported." + with pytest.raises(ValueError, match=error_message): + emulator = device.emulator + + @patch("braket.aws.aws_device.AwsSession.copy_session") @patch("braket.aws.aws_device.AwsSession") @pytest.mark.parametrize( diff --git a/test/unit_tests/braket/aws/test_aws_emulation.py b/test/unit_tests/braket/aws/test_aws_emulation.py new file mode 100644 index 000000000..53d5ae53a --- /dev/null +++ b/test/unit_tests/braket/aws/test_aws_emulation.py @@ -0,0 +1,714 @@ +import json +from unittest.mock import Mock, patch + +import networkx as nx +import numpy as np +import pytest +from common_test_utils import RIGETTI_ARN, RIGETTI_REGION + +from braket.aws import AwsDevice +from braket.aws.aws_emulation import _get_qpu_gate_translations +from braket.aws.aws_noise_models import ( + GateDeviceCalibrationData, + GateFidelity, + _setup_calibration_specs, + device_noise_model, +) +from braket.circuits import Circuit, Gate +from braket.circuits.noise_model import GateCriteria, NoiseModel, ObservableCriteria +from braket.circuits.noises import ( + AmplitudeDamping, + BitFlip, + Depolarizing, + PhaseDamping, + TwoQubitDepolarizing, +) +from braket.device_schema import DeviceCapabilities +from braket.device_schema.error_mitigation.debias import Debias +from braket.device_schema.ionq import IonqDeviceCapabilities, IonqDeviceParameters +from braket.device_schema.iqm import IqmDeviceCapabilities +from braket.device_schema.rigetti import RigettiDeviceCapabilities +from braket.devices import Devices +from braket.devices.local_simulator import LocalSimulator +from braket.emulation import Emulator +from braket.emulation.emulation_passes.gate_device_passes import ( + ConnectivityValidator, + GateConnectivityValidator, + GateValidator, + QubitCountValidator, +) + +REGION = "us-west-1" + +IONQ_ARN = "arn:aws:braket:::device/qpu/ionq/Forte1" +IQM_ARN = "arn:aws:braket:::device/qpu/iqm/Garent" + + +MOCK_QPU_GATE_DURATIONS = { + RIGETTI_ARN: { + "single_qubit_gate_duration": 40e-9, + "two_qubit_gate_duration": 240e-9, + }, + IQM_ARN: {"single_qubit_gate_duration": 32e-9, "two_qubit_gate_duration": 60e-9}, +} + + +@pytest.fixture +def basic_device_capabilities(): + return DeviceCapabilities.parse_obj( + { + "service": { + "executionWindows": [ + { + "executionDay": "Everyday", + "windowStartHour": "11:00", + "windowEndHour": "12:00", + } + ], + "shotsRange": [1, 10], + }, + "action": { + "braket.ir.openqasm.program": { + "actionType": "braket.ir.openqasm.program", + "version": ["1"], + }, + "braket.ir.jaqcd.program": { + "actionType": "braket.ir.jaqcd.program", + "version": ["1"], + }, + }, + "deviceParameters": {}, + } + ) + + +MOCK_STANDARDIZED_CALIBRATION_JSON = { + "braketSchemaHeader": { + "name": "braket.device_schema.standardized_gate_model_qpu_device_properties", + "version": "1", + }, + "oneQubitProperties": { + "0": { + "T1": {"value": 0.5, "standardError": None, "unit": "S"}, + "T2": {"value": 0.2, "standardError": None, "unit": "S"}, + "oneQubitFidelity": [ + { + "fidelityType": {"name": "RANDOMIZED_BENCHMARKING", "description": None}, + "fidelity": 0.99, + "standardError": 1e-2, + }, + { + "fidelityType": { + "name": "SIMULTANEOUS_RANDOMIZED_BENCHMARKING", + "description": None, + }, + "fidelity": 0.9934, + "standardError": 0.0065, + }, + { + "fidelityType": {"name": "READOUT", "description": None}, + "fidelity": 0.958, + "standardError": None, + }, + ], + }, + "1": { + "T1": {"value": 0.97, "standardError": None, "unit": "S"}, + "T2": {"value": 0.234, "standardError": None, "unit": "S"}, + "oneQubitFidelity": [ + { + "fidelityType": {"name": "RANDOMIZED_BENCHMARKING", "description": None}, + "fidelity": 0.9983, + "standardError": 4e-5, + }, + { + "fidelityType": { + "name": "SIMULTANEOUS_RANDOMIZED_BENCHMARKING", + "description": None, + }, + "fidelity": 0.879, + "standardError": 0.00058, + }, + { + "fidelityType": {"name": "READOUT", "description": None}, + "fidelity": 0.989, + "standardError": None, + }, + ], + }, + "2": { + "T1": {"value": 0.8, "standardError": None, "unit": "S"}, + "T2": {"value": 0.4, "standardError": None, "unit": "S"}, + "oneQubitFidelity": [ + { + "fidelityType": {"name": "READOUT", "description": None}, + "fidelity": 0.958, + "standardError": None, + } + ], + }, + }, + "twoQubitProperties": { + "0-1": { + "twoQubitGateFidelity": [ + { + "direction": None, + "gateName": "CZ", + "fidelity": 0.9358, + "standardError": 0.01437, + "fidelityType": {"name": "INTERLEAVED_RANDOMIZED_BENCHMARKING"}, + }, + { + "direction": None, + "gateName": "Two_Qubit_Clifford", + "fidelity": 0.9, + "standardError": 0.0237, + "fidelityType": {"name": "INTERLEAVED_RANDOMIZED_BENCHMARKING"}, + }, + { + "direction": None, + "gateName": "CPHASE", + "fidelity": 0.9, + "standardError": 0.01437, + "fidelityType": {"name": "INTERLEAVED_RANDOMIZED_BENCHMARKING"}, + }, + ] + } + }, +} + +MOCK_STANDARDIZED_CALIBRATION_JSON_2 = MOCK_STANDARDIZED_CALIBRATION_JSON.copy() +MOCK_STANDARDIZED_CALIBRATION_JSON_2["twoQubitProperties"]["0-1"]["twoQubitGateFidelity"][2][ + "gateName" +] = "CPhaseShift" + +MOCK_RIGETTI_QPU_CAPABILITIES_1 = { + "braketSchemaHeader": { + "name": "braket.device_schema.rigetti.rigetti_device_capabilities", + "version": "1", + }, + "service": { + "executionWindows": [ + { + "executionDay": "Everyday", + "windowStartHour": "11:00", + "windowEndHour": "12:00", + } + ], + "shotsRange": [1, 10], + }, + "action": { + "braket.ir.openqasm.program": { + "actionType": "braket.ir.openqasm.program", + "version": ["1"], + "supportedOperations": ["H", "X", "CNot", "CZ", "Rx", "Ry", "YY"], + } + }, + "paradigm": { + "qubitCount": 3, + "nativeGateSet": ["cz", "prx", "cphaseshift"], + "connectivity": { + "fullyConnected": False, + "connectivityGraph": {"0": ["1", "2"], "1": ["0"], "2": ["0"]}, + }, + }, + "standardized": MOCK_STANDARDIZED_CALIBRATION_JSON_2, + "deviceParameters": {}, +} + +MOCK_IQM_QPU_CAPABILITIES_1 = { + "braketSchemaHeader": { + "name": "braket.device_schema.iqm.iqm_device_capabilities", + "version": "1", + }, + "service": { + "executionWindows": [ + { + "executionDay": "Everyday", + "windowStartHour": "11:00", + "windowEndHour": "12:00", + } + ], + "shotsRange": [1, 10], + }, + "action": { + "braket.ir.openqasm.program": { + "actionType": "braket.ir.openqasm.program", + "version": ["1"], + "supportedOperations": ["H", "CNot", "Ry", "XX", "YY"], + } + }, + "paradigm": { + "qubitCount": 4, + "nativeGateSet": ["cz", "prx", "cphaseshift"], + "connectivity": { + "fullyConnected": False, + "connectivityGraph": {"0": ["1", "2"], "2": ["3"]}, + }, + }, + "standardized": MOCK_STANDARDIZED_CALIBRATION_JSON, + "deviceParameters": {}, +} + + +@pytest.fixture +def rigetti_device_capabilities(): + return RigettiDeviceCapabilities.parse_obj(MOCK_RIGETTI_QPU_CAPABILITIES_1) + + +@pytest.fixture +def iqm_device_capabilities(): + return IqmDeviceCapabilities.parse_obj(MOCK_IQM_QPU_CAPABILITIES_1) + + +MOCK_IONQ_GATE_MODEL_CAPABILITIES_JSON_1 = { + "braketSchemaHeader": { + "name": "braket.device_schema.ionq.ionq_device_capabilities", + "version": "1", + }, + "service": { + "executionWindows": [ + { + "executionDay": "Everyday", + "windowStartHour": "11:00", + "windowEndHour": "12:00", + } + ], + "shotsRange": [1, 10], + }, + "action": { + "braket.ir.openqasm.program": { + "actionType": "braket.ir.openqasm.program", + "version": ["1"], + "supportedOperations": ["x", "y"], + } + }, + "paradigm": { + "qubitCount": 2, + "nativeGateSet": ["CZ", "CPhaseShift", "GPI"], + "connectivity": {"fullyConnected": True, "connectivityGraph": {}}, + }, + "provider": { + "braketSchemaHeader": { + "name": "braket.device_schema.ionq.ionq_provider_properties", + "version": "1", + }, + "errorMitigation": {Debias: {"minimumShots": 2500}}, + "fidelity": {"1Q": {"mean": 0.98}, "2Q": {"mean": 0.9625}, "spam": {"mean": 0.9}}, + "timing": { + "1Q": 0.000135, + "2Q": 0.0006, + "T1": 10.0, + "T2": 1.0, + "readout": 0.0003, + "reset": 2e-05, + }, + }, + "deviceParameters": json.loads(IonqDeviceParameters.schema_json()), +} + + +@pytest.fixture +def ionq_device_capabilities(): + return IonqDeviceCapabilities.parse_obj(MOCK_IONQ_GATE_MODEL_CAPABILITIES_JSON_1) + + +@pytest.fixture +def rigetti_target_noise_model(): + gate_duration_1Q = MOCK_QPU_GATE_DURATIONS[RIGETTI_ARN]["single_qubit_gate_duration"] + target_noise_model = ( + NoiseModel() + .add_noise(AmplitudeDamping(1 - np.exp(-(gate_duration_1Q / 0.5))), GateCriteria(qubits=0)) + .add_noise( + PhaseDamping(0.5 * (1 - np.exp(-(gate_duration_1Q / 0.2)))), GateCriteria(qubits=0) + ) + .add_noise(Depolarizing(1 - 0.9934), GateCriteria(qubits=0)) + .add_noise(BitFlip(1 - 0.958), ObservableCriteria(qubits=0)) + .add_noise(AmplitudeDamping(1 - np.exp(-(gate_duration_1Q / 0.97))), GateCriteria(qubits=1)) + .add_noise( + PhaseDamping(0.5 * (1 - np.exp(-(gate_duration_1Q / 0.234)))), GateCriteria(qubits=1) + ) + .add_noise(Depolarizing(1 - 0.879), GateCriteria(qubits=1)) + .add_noise(BitFlip(1 - 0.989), ObservableCriteria(qubits=1)) + .add_noise(AmplitudeDamping(1 - np.exp(-(gate_duration_1Q / 0.8))), GateCriteria(qubits=2)) + .add_noise( + PhaseDamping(0.5 * (1 - np.exp(-(gate_duration_1Q / 0.4)))), GateCriteria(qubits=2) + ) + .add_noise(BitFlip(1 - 0.958), ObservableCriteria(qubits=2)) + .add_noise(TwoQubitDepolarizing(1 - 0.9358), GateCriteria(Gate.CZ, [(1, 0), (0, 1)])) + .add_noise(TwoQubitDepolarizing(1 - 0.9), GateCriteria(Gate.CPhaseShift, [(1, 0), (0, 1)])) + ) + + return target_noise_model + + +@pytest.fixture +def iqm_target_noise_model(): + gate_duration_1Q = MOCK_QPU_GATE_DURATIONS[IQM_ARN]["single_qubit_gate_duration"] + target_noise_model = ( + NoiseModel() + .add_noise(AmplitudeDamping(1 - np.exp(-(gate_duration_1Q / 0.5))), GateCriteria(qubits=0)) + .add_noise( + PhaseDamping(0.5 * (1 - np.exp(-(gate_duration_1Q / 0.2)))), GateCriteria(qubits=0) + ) + .add_noise(Depolarizing(1 - 0.9934), GateCriteria(qubits=0)) + .add_noise(BitFlip(1 - 0.958), ObservableCriteria(qubits=0)) + .add_noise(AmplitudeDamping(1 - np.exp(-(gate_duration_1Q / 0.97))), GateCriteria(qubits=1)) + .add_noise( + PhaseDamping(0.5 * (1 - np.exp(-(gate_duration_1Q / 0.234)))), GateCriteria(qubits=1) + ) + .add_noise(Depolarizing(1 - 0.879), GateCriteria(qubits=1)) + .add_noise(BitFlip(1 - 0.989), ObservableCriteria(qubits=1)) + .add_noise(AmplitudeDamping(1 - np.exp(-(gate_duration_1Q / 0.8))), GateCriteria(qubits=2)) + .add_noise( + PhaseDamping(0.5 * (1 - np.exp(-(gate_duration_1Q / 0.4)))), GateCriteria(qubits=2) + ) + .add_noise(BitFlip(1 - 0.958), ObservableCriteria(qubits=2)) + .add_noise(TwoQubitDepolarizing(1 - 0.9358), GateCriteria(Gate.CZ, [(1, 0), (0, 1)])) + .add_noise(TwoQubitDepolarizing(1 - 0.9), GateCriteria(Gate.CPhaseShift, [(1, 0), (0, 1)])) + ) + + return target_noise_model + + +@pytest.fixture +def ionq_target_noise_model(ionq_device_capabilities): + T1 = 10.0 + T2 = 1.0 + readout = 0.9 + gate_duration_1Q = 0.000135 + single_rb = 0.98 + two_qubit_rb = 0.9625 + target_noise_model = NoiseModel() + qubit_count = ionq_device_capabilities.paradigm.qubitCount + for i in range(qubit_count): + target_noise_model = ( + target_noise_model.add_noise( + AmplitudeDamping(1 - np.exp(-(gate_duration_1Q / T1))), GateCriteria(qubits=i) + ) + .add_noise( + PhaseDamping(0.5 * (1 - np.exp(-(gate_duration_1Q / T2)))), GateCriteria(qubits=i) + ) + .add_noise(Depolarizing(1 - single_rb), GateCriteria(qubits=i)) + .add_noise(BitFlip(1 - readout), ObservableCriteria(qubits=i)) + ) + + for i in range(qubit_count): + for j in range(i, qubit_count): + if i != j: + target_noise_model = target_noise_model.add_noise( + TwoQubitDepolarizing(1 - two_qubit_rb), GateCriteria(Gate.CZ, [(i, j), (j, i)]) + ).add_noise( + TwoQubitDepolarizing(1 - two_qubit_rb), + GateCriteria(Gate.CPhaseShift, [(i, j), (j, i)]), + ) + return target_noise_model + + +@patch.dict("braket.aws.aws_noise_models._QPU_GATE_DURATIONS", MOCK_QPU_GATE_DURATIONS) +def test_standardized_noise_model(rigetti_device_capabilities, rigetti_target_noise_model): + noise_model = device_noise_model(rigetti_device_capabilities, RIGETTI_ARN) + + assert noise_model.instructions == rigetti_target_noise_model.instructions + + +@pytest.mark.parametrize( + "single_qubit_gate_duration,two_qubit_gate_duration,qubit_labels,\ + single_qubit_specs,two_qubit_edge_specs", + [ + (0.5, 0.2, {0, 1}, {2: {"h": 0.5}}, {(0, 1): GateFidelity(Gate.H, 0.5)}), + (0.5, 0.2, {0, 1}, {0: {"h": 0.5}}, {(0, 2): GateFidelity(Gate.H, 0.5)}), + ], +) +def test_invalid_gate_calibration_data( + single_qubit_gate_duration, + two_qubit_gate_duration, + qubit_labels, + single_qubit_specs, + two_qubit_edge_specs, +): + with pytest.raises(ValueError): + GateDeviceCalibrationData( + single_qubit_gate_duration, + two_qubit_gate_duration, + qubit_labels, + single_qubit_specs, + two_qubit_edge_specs, + ) + + +def test_missing_gate_durations(rigetti_device_capabilities): + with pytest.raises(ValueError): + _setup_calibration_specs(rigetti_device_capabilities, "bad_arn") + + +def test_ionq_noise_model(ionq_device_capabilities, ionq_target_noise_model): + # modify capabilities to include gate not supported by braket but included in IonQ capabilities. + ionq_device_capabilities.paradigm.nativeGateSet.append("Two_Qubit_Clifford") + noise_model = device_noise_model(ionq_device_capabilities, Devices.IonQ.Aria1) + assert noise_model.instructions == ionq_target_noise_model.instructions + + +MOCK_DEFAULT_S3_DESTINATION_FOLDER = ( + "amazon-braket-us-test-1-00000000", + "tasks", +) + + +@pytest.fixture +def mock_rigetti_qpu_device(rigetti_device_capabilities): + return { + "deviceName": "Aspen-M3", + "deviceType": "QPU", + "providerName": "Rigetti", + "deviceStatus": "OFFLINE", + "deviceCapabilities": rigetti_device_capabilities.json(), + "deviceQueueInfo": [ + {"queue": "QUANTUM_TASKS_QUEUE", "queueSize": "19", "queuePriority": "Normal"}, + {"queue": "QUANTUM_TASKS_QUEUE", "queueSize": "3", "queuePriority": "Priority"}, + {"queue": "JOBS_QUEUE", "queueSize": "0 (3 prioritized job(s) running)"}, + ], + } + + +@pytest.fixture +def mock_iqm_qpu_device(iqm_device_capabilities): + return { + "deviceName": "Harmony", + "deviceType": "QPU", + "providerName": "IQM", + "deviceStatus": "OFFLINE", + "deviceCapabilities": iqm_device_capabilities.json(), + "deviceQueueInfo": [ + {"queue": "QUANTUM_TASKS_QUEUE", "queueSize": "19", "queuePriority": "Normal"}, + {"queue": "QUANTUM_TASKS_QUEUE", "queueSize": "3", "queuePriority": "Priority"}, + {"queue": "JOBS_QUEUE", "queueSize": "0 (3 prioritized job(s) running)"}, + ], + } + + +@pytest.fixture +def mock_ionq_qpu_device(ionq_device_capabilities): + return { + "deviceName": "Aspen-M3", + "deviceType": "QPU", + "providerName": "Rigetti", + "deviceStatus": "OFFLINE", + "deviceCapabilities": ionq_device_capabilities.json(), + "deviceQueueInfo": [ + {"queue": "QUANTUM_TASKS_QUEUE", "queueSize": "19", "queuePriority": "Normal"}, + {"queue": "QUANTUM_TASKS_QUEUE", "queueSize": "3", "queuePriority": "Priority"}, + {"queue": "JOBS_QUEUE", "queueSize": "0 (3 prioritized job(s) running)"}, + ], + } + + +@pytest.fixture +def aws_session(): + _boto_session = Mock() + _boto_session.region_name = RIGETTI_REGION + _boto_session.profile_name = "test-profile" + + creds = Mock() + creds.method = "other" + _boto_session.get_credentials.return_value = creds + + _aws_session = Mock() + _aws_session.boto_session = _boto_session + _aws_session._default_bucket = MOCK_DEFAULT_S3_DESTINATION_FOLDER[0] + _aws_session.default_bucket.return_value = _aws_session._default_bucket + _aws_session._custom_default_bucket = False + _aws_session.account_id = "00000000" + _aws_session.region = RIGETTI_REGION + return _aws_session + + +@pytest.fixture( + params=[ + "arn:aws:braket:us-west-1::device/quantum-simulator/amazon/sim", + "arn:aws:braket:::device/quantum-simulator/amazon/sim", + ] +) +def arn(request): + return request.param + + +@pytest.fixture +def ionq_device(aws_session, mock_ionq_qpu_device): + def _device(): + aws_session.get_device.return_value = mock_ionq_qpu_device + aws_session.search_devices.return_value = [mock_ionq_qpu_device] + return AwsDevice(IONQ_ARN, aws_session) + + return _device() + + +@pytest.fixture +def iqm_device(aws_session, mock_iqm_qpu_device): + def _device(): + aws_session.get_device.return_value = mock_iqm_qpu_device + aws_session.search_devices.return_value = [mock_iqm_qpu_device] + return AwsDevice(IQM_ARN, aws_session) + + return _device() + + +@pytest.fixture +def rigetti_device(aws_session, mock_rigetti_qpu_device): + def _device(): + aws_session.get_device.return_value = mock_rigetti_qpu_device + aws_session.search_devices.return_value = [mock_rigetti_qpu_device] + return AwsDevice(RIGETTI_ARN, aws_session) + + return _device() + + +def test_ionq_emulator(ionq_device): + emulator = ionq_device.emulator + target_emulator_passes = [ + QubitCountValidator(ionq_device.properties.paradigm.qubitCount), + GateValidator( + supported_gates=["x", "Y"], + native_gates=["cz", "gpi", "cphaseshift"], + ), + ConnectivityValidator(nx.from_edgelist([(0, 1), (1, 0)], create_using=nx.DiGraph())), + GateConnectivityValidator( + nx.from_dict_of_dicts( + { + 0: {1: {"supported_gates": set(["CZ", "CPhaseShift", "GPi"])}}, + 1: {0: {"supported_gates": set(["CZ", "CPhaseShift", "GPi"])}}, + }, + create_using=nx.DiGraph(), + ) + ), + ] + emulator._emulator_passes == target_emulator_passes + + +@patch.dict("braket.aws.aws_noise_models._QPU_GATE_DURATIONS", MOCK_QPU_GATE_DURATIONS) +def test_rigetti_emulator(rigetti_device, rigetti_target_noise_model): + emulator = rigetti_device.emulator + assert emulator.noise_model + assert len(emulator.noise_model.instructions) == len(rigetti_target_noise_model.instructions) + assert all( + i1 == i2 + for i1, i2 in zip( + emulator.noise_model.instructions, rigetti_target_noise_model.instructions + ) + ) + + target_emulator_passes = [ + QubitCountValidator(rigetti_device.properties.paradigm.qubitCount), + GateValidator( + supported_gates=["H", "X", "CNot", "CZ", "Rx", "Ry", "YY"], + native_gates=["cz", "prx", "cphaseshift"], + ), + ConnectivityValidator( + nx.from_edgelist([(0, 1), (0, 2), (1, 0), (2, 0)], create_using=nx.DiGraph()) + ), + GateConnectivityValidator( + nx.from_dict_of_dicts( + { + 0: { + 1: {"supported_gates": set(["CZ", "CPhaseShift", "Two_Qubit_Clifford"])}, + 2: {"supported_gates": set()}, + }, + 1: {0: {"supported_gates": set(["CZ", "CPhaseShift", "Two_Qubit_Clifford"])}}, + 2: {0: {"supported_gates": set()}}, + }, + create_using=nx.DiGraph(), + ) + ), + ] + assert emulator._emulator_passes == target_emulator_passes + + +@patch.dict("braket.aws.aws_noise_models._QPU_GATE_DURATIONS", MOCK_QPU_GATE_DURATIONS) +def test_iqm_emulator(iqm_device, iqm_target_noise_model): + emulator = iqm_device.emulator + assert emulator.noise_model + assert len(emulator.noise_model.instructions) == len(iqm_target_noise_model.instructions) + assert emulator.noise_model.instructions == iqm_target_noise_model.instructions + target_emulator_passes = [ + QubitCountValidator(iqm_device.properties.paradigm.qubitCount), + GateValidator( + supported_gates=["H", "CNot", "Ry", "XX", "YY"], + native_gates=["cz", "prx", "cphaseshift"], + ), + ConnectivityValidator( + nx.from_edgelist( + [(0, 1), (0, 2), (1, 0), (2, 0), (2, 3), (3, 2)], create_using=nx.DiGraph() + ) + ), + GateConnectivityValidator( + nx.from_dict_of_dicts( + { + 0: { + 1: {"supported_gates": set(["CZ", "CPhaseShift", "Two_Qubit_Clifford"])}, + 2: {"supported_gates": set()}, + }, + 1: {0: {"supported_gates": set(["CZ", "CPhaseShift", "Two_Qubit_Clifford"])}}, + 2: {0: {"supported_gates": set()}, 3: {"supported_gates": set()}}, + 3: {2: {"supported_gates": set()}}, + }, + create_using=nx.DiGraph(), + ) + ), + ] + + for i in range(4): + assert emulator._emulator_passes[i] == target_emulator_passes[i] + + +@pytest.mark.parametrize( + "device_capabilities,gate_name,expected_result", + [ + ("basic_device_capabilities", "fake_gate", "fake_gate"), + ("rigetti_device_capabilities", "CPHASE", "CPhaseShift"), + ("ionq_device_capabilities", "GPI", "GPi"), + ("ionq_device_capabilities", ["GPI", "GPI2", "fake_gate"], ["GPi", "GPi2", "fake_gate"]), + ], +) +def test_get_gate_translations(device_capabilities, gate_name, expected_result, request): + device_capabilities_obj = request.getfixturevalue(device_capabilities) + assert _get_qpu_gate_translations(device_capabilities_obj, gate_name) == expected_result + + +@patch.dict("braket.aws.aws_noise_models._QPU_GATE_DURATIONS", MOCK_QPU_GATE_DURATIONS) +@pytest.mark.parametrize( + "circuit,is_valid", + [ + (Circuit(), True), + (Circuit().cnot(0, 1).h(2), True), + (Circuit().x(4).yy(4, 8, np.pi), True), + (Circuit().add_verbatim_box(Circuit().cz(0, 1)).h(5), False), + (Circuit().x(range(5)), False), + (Circuit().add_verbatim_box(Circuit().cz(0, 1)).rx(1, np.pi / 4), True), + (Circuit().add_verbatim_box(Circuit().cz(0, 2)), False), + (Circuit().xx(0, 1, np.pi / 4), False), + ], +) +def test_emulator_passes(circuit, is_valid, rigetti_device): + if is_valid: + rigetti_device.validate(circuit) + assert rigetti_device.run_passes(circuit, apply_noise_model=False) == circuit + else: + with pytest.raises(Exception): + rigetti_device.validate(circuit) + + +@patch.dict("braket.aws.aws_noise_models._QPU_GATE_DURATIONS", MOCK_QPU_GATE_DURATIONS) +@patch.object(LocalSimulator, "run") +def test_device_emulate(mock_run, rigetti_device): + circuit = Circuit().h(0).cnot(0, 1) + rigetti_device.emulate(circuit, shots=100) + mock_run.assert_called_once() + + +@patch.dict("braket.aws.aws_noise_models._QPU_GATE_DURATIONS", MOCK_QPU_GATE_DURATIONS) +@patch.object(AwsDevice, "_setup_emulator", return_value=Emulator()) +def test_get_emulator_multiple(mock_setup, rigetti_device): + emulator = rigetti_device.emulator + assert emulator._emulator_passes == [] + emulator = rigetti_device.emulator + mock_setup.assert_called_once() diff --git a/test/unit_tests/braket/devices/test_local_simulator.py b/test/unit_tests/braket/devices/test_local_simulator.py index 451553f02..43b86903c 100644 --- a/test/unit_tests/braket/devices/test_local_simulator.py +++ b/test/unit_tests/braket/devices/test_local_simulator.py @@ -12,6 +12,7 @@ # language governing permissions and limitations under the License. import json +import sys import textwrap import warnings from typing import Any, Optional @@ -391,20 +392,25 @@ def properties(self) -> DeviceCapabilities: return RydbergSimulatorDeviceCapabilities.parse_obj(properties) -mock_circuit_entry = Mock() -mock_program_entry = Mock() -mock_jaqcd_entry = Mock() -mock_circuit_dm_entry = Mock() -mock_circuit_entry.load.return_value = DummyCircuitSimulator -mock_program_entry.load.return_value = DummyProgramSimulator -mock_jaqcd_entry.load.return_value = DummyJaqcdSimulator -mock_circuit_dm_entry.load.return_value = DummyProgramDensityMatrixSimulator -local_simulator._simulator_devices = { - "dummy": mock_circuit_entry, - "dummy_oq3": mock_program_entry, - "dummy_jaqcd": mock_jaqcd_entry, - "dummy_oq3_dm": mock_circuit_dm_entry, -} +@pytest.fixture(autouse=True) +def _simulator_devices(request): + if request.module == sys.modules[__name__]: + mock_circuit_entry = Mock() + mock_program_entry = Mock() + mock_jaqcd_entry = Mock() + mock_circuit_dm_entry = Mock() + mock_circuit_entry.load.return_value = DummyCircuitSimulator + mock_program_entry.load.return_value = DummyProgramSimulator + mock_jaqcd_entry.load.return_value = DummyJaqcdSimulator + mock_circuit_dm_entry.load.return_value = DummyProgramDensityMatrixSimulator + local_simulator._simulator_devices = { + "dummy": mock_circuit_entry, + "dummy_oq3": mock_program_entry, + "dummy_jaqcd": mock_jaqcd_entry, + "dummy_oq3_dm": mock_circuit_dm_entry, + } + return local_simulator._simulator_devices + mock_ahs_program = AnalogHamiltonianSimulation( register=AtomArrangement(), hamiltonian=Hamiltonian() diff --git a/test/unit_tests/braket/emulation/test_connectivity_validator.py b/test/unit_tests/braket/emulation/test_connectivity_validator.py new file mode 100644 index 000000000..c79ad0106 --- /dev/null +++ b/test/unit_tests/braket/emulation/test_connectivity_validator.py @@ -0,0 +1,217 @@ +import networkx as nx +import numpy as np +import pytest +from networkx.utils import graphs_equal + +from braket.circuits import Circuit +from braket.emulation.emulation_passes.gate_device_passes import ConnectivityValidator + + +@pytest.fixture +def basic_2_node_complete_graph(): + return nx.complete_graph(2, create_using=nx.DiGraph()) + + +@pytest.fixture +def basic_noncontig_qubits_2_node_complete_graph(): + return nx.complete_graph([1, 10], create_using=nx.DiGraph()) + + +@pytest.fixture +def six_node_digraph(): + edge_set = {0: [1, 3], 1: [0, 2, 10], 2: [1, 3, 11], 10: [1, 11], 11: [2, 10]} + return nx.from_dict_of_lists(edge_set, create_using=nx.DiGraph()) + + +@pytest.mark.parametrize( + "circuit", + [ + Circuit(), + Circuit().add_verbatim_box(Circuit()), + Circuit().i(range(2)).cnot(3, 4), + Circuit().add_verbatim_box(Circuit().h(0).h(1).cnot(0, 1).cnot(1, 0)), + Circuit() + .h(range(2)) + .add_verbatim_box( + Circuit().swap(0, 1).phaseshift(1, np.pi / 4).cphaseshift01(1, 0, np.pi / 4) + ), + ], +) +def test_basic_contiguous_circuits(basic_2_node_complete_graph, circuit): + """ + ConnectivityValidator should not raise any errors when validating these circuits. + """ + ConnectivityValidator(basic_2_node_complete_graph).validate(circuit) + + +@pytest.mark.parametrize( + "circuit", + [ + Circuit(), + Circuit().add_verbatim_box(Circuit()), + Circuit().i(range(3)).cnot(3, 4).x(111), + Circuit().add_verbatim_box(Circuit().h(1).h(10).cnot(1, 10).cnot(10, 1)), + Circuit().add_verbatim_box( + Circuit().swap(1, 10).phaseshift(10, np.pi / 4).cphaseshift01(10, 1, np.pi / 4) + ), + ], +) +def test_valid_discontiguous_circuits(basic_noncontig_qubits_2_node_complete_graph, circuit): + """ + ConnectivityValidator should not raise any errors when validating these circuits. + """ + ConnectivityValidator(basic_noncontig_qubits_2_node_complete_graph).validate(circuit) + + +def test_complete_graph_instantation_with_num_qubits(): + """ + Tests that, if fully_connected is True and num_qubits are passed into the + ConnectivityValidator constructor, a fully connected graph is created. + """ + num_qubits = 5 + validator = ConnectivityValidator(num_qubits=num_qubits, fully_connected=True) + vb = Circuit() + for i in range(num_qubits): + for j in range(num_qubits): + if i != j: + vb.cnot(i, j) + else: + vb.i(i) + circuit = Circuit().add_verbatim_box(vb) + validator.validate(circuit) + assert nx.utils.graphs_equal( + validator._connectivity_graph, nx.complete_graph(num_qubits, create_using=nx.DiGraph()) + ) + + +def test_complete_graph_instantation_with_qubit_labels(): + """ + Tests that, if fully_connected is True and num_qubits are passed into the + ConnectivityValidator constructor, a fully connected graph is created. + """ + qubit_labels = [0, 1, 10, 11, 110, 111, 112, 113] + validator = ConnectivityValidator(qubit_labels=qubit_labels, fully_connected=True) + vb = Circuit() + for i in qubit_labels: + for j in qubit_labels: + if i != j: + vb.cnot(i, j) + else: + vb.i(i) + circuit = Circuit().add_verbatim_box(vb) + validator.validate(circuit) + assert nx.utils.graphs_equal( + validator._connectivity_graph, nx.complete_graph(qubit_labels, create_using=nx.DiGraph()) + ) + + +@pytest.mark.parametrize( + "circuit", + [ + Circuit().add_verbatim_box(Circuit().cnot(0, 2)), + Circuit().add_verbatim_box(Circuit().swap(1, 3)), + Circuit() + .add_verbatim_box(Circuit().cnot(1, 10).cphaseshift01(2, 11, np.pi / 4)) + .x(4) + .add_verbatim_box(Circuit().swap(2, 10)), + ], +) +def test_invalid_2_qubit_gates(six_node_digraph, circuit): + with pytest.raises(ValueError): + ConnectivityValidator(six_node_digraph).validate(circuit) + + +@pytest.mark.parametrize( + "circuit", + [ + Circuit().x(4).add_verbatim_box(Circuit().x(4)), + Circuit().x(110).add_verbatim_box(Circuit().phaseshift(4, np.pi / 4)), + Circuit() + .add_verbatim_box(Circuit().cnot(1, 10).cphaseshift01(2, 11, np.pi / 4)) + .x(4) + .add_verbatim_box(Circuit().h(111)), + ], +) +def test_invalid_1_qubit_gates(six_node_digraph, circuit): + with pytest.raises(ValueError): + ConnectivityValidator(six_node_digraph).validate(circuit) + + +def test_equality_graph_created_with_dict(six_node_digraph): + graph = {0: [1, 3], 1: [0, 2, 10], 2: [1, 3, 11], 10: [1, 11], 11: [2, 10]} + criteria_from_digraph = ConnectivityValidator(six_node_digraph) + criteria_from_dict = ConnectivityValidator(graph) + assert criteria_from_dict == criteria_from_digraph + + +@pytest.mark.parametrize( + "connectivity_graph, fully_connected, num_qubits, qubit_labels, directed", + [ + (None, True, None, None, False), + (nx.DiGraph(), True, None, None, False), + (None, True, 5, [0, 1], False), + (None, False, None, None, False), + (nx.from_edgelist([(0, 1)], create_using=nx.Graph()), False, None, None, False), + ], +) +def test_invalid_constructors( + connectivity_graph, fully_connected, num_qubits, qubit_labels, directed +): + with pytest.raises(ValueError): + ConnectivityValidator( + connectivity_graph, fully_connected, num_qubits, qubit_labels, directed + ) + + +@pytest.mark.parametrize( + "representation", + [ + {1: [0, 2, 3], 2: [3, 4], 3: [6]}, + nx.from_edgelist([(1, 0), (1, 2), (1, 3), (2, 3), (2, 4), (3, 6)], create_using=nx.DiGraph), + ], +) +def test_undirected_graph_construction(representation): + expected_digraph = nx.from_edgelist( + [ + (1, 0), + (0, 1), + (1, 2), + (2, 1), + (1, 3), + (3, 1), + (2, 3), + (3, 2), + (2, 4), + (4, 2), + (3, 6), + (6, 3), + ], + create_using=nx.DiGraph, + ) + cc = ConnectivityValidator(representation, directed=False) + assert graphs_equal(cc._connectivity_graph, expected_digraph) + + +# @pytest.fixture +# def six_node_digraph(): +# edge_set = {0: [1, 3], 1: [0, 2, 10], 2: [1, 3, 11], 10: [1, 11], 11: [2, 10]} +# return nx.from_dict_of_lists(edge_set, create_using=nx.DiGraph()) + + +@pytest.mark.parametrize( + "controls,targets,is_valid", + [ + ([0], [1], True), + ([], [0, 1], True), + ([3], [0], True), + ([0, 2], [], False), + ([0], [1, 2], False), + ], +) +def test_validate_instruction_method(controls, targets, is_valid, six_node_digraph): + gcc = ConnectivityValidator(six_node_digraph, directed=False) + if is_valid: + gcc._validate_instruction_connectivity(controls, targets) + else: + with pytest.raises(ValueError): + gcc._validate_instruction_connectivity(controls, targets) diff --git a/test/unit_tests/braket/emulation/test_emulator.py b/test/unit_tests/braket/emulation/test_emulator.py new file mode 100644 index 000000000..e532037e3 --- /dev/null +++ b/test/unit_tests/braket/emulation/test_emulator.py @@ -0,0 +1,183 @@ +import re +from unittest.mock import Mock + +import numpy as np +import pytest + +from braket.circuits import Circuit, Gate +from braket.circuits.noise_model import GateCriteria, NoiseModel +from braket.circuits.noises import BitFlip +from braket.default_simulator import DensityMatrixSimulator, StateVectorSimulator +from braket.devices import local_simulator +from braket.emulation import Emulator +from braket.emulation.emulation_passes.gate_device_passes import GateValidator, QubitCountValidator +from braket.passes import BasePass, ProgramType + + +class AlwaysFailPass(BasePass[ProgramType]): + def run(self, program: ProgramType): + raise ValueError("This pass always raises an error.") + + +@pytest.fixture +def setup_local_simulator_devices(): + mock_circuit_entry = Mock() + mock_circuit_dm_entry = Mock() + mock_circuit_entry.load.return_value = StateVectorSimulator + mock_circuit_dm_entry.load.return_value = DensityMatrixSimulator + _simulator_devices = {"default": mock_circuit_entry, "braket_dm": mock_circuit_dm_entry} + local_simulator._simulator_devices.update(_simulator_devices) + + +@pytest.fixture +def empty_emulator(setup_local_simulator_devices): + return Emulator() + + +@pytest.fixture +def basic_emulator(empty_emulator): + qubit_count_validator = QubitCountValidator(4) + return empty_emulator.add_pass(emulator_pass=[qubit_count_validator]) + + +def test_empty_emulator_validation(empty_emulator): + emulator = empty_emulator + circuit = Circuit().h(0).cnot(0, 1) + emulator.validate(circuit) + + +def test_basic_emulator(basic_emulator): + """ + Should not error out when passed a valid circuit. + """ + circuit = Circuit().cnot(0, 1) + circuit = basic_emulator.run_passes(circuit) + assert circuit == circuit + + +def test_basic_invalidate(basic_emulator): + """ + Emulator should raise an error thrown by the QubitCountValidator. + """ + circuit = Circuit().x(range(6)) + match_string = re.escape( + f"Circuit must use at most 4 qubits, \ +but uses {circuit.qubit_count} qubits. (DeviceEmulator)" + ) + with pytest.raises(Exception, match=match_string): + basic_emulator.run_passes(circuit) + + +def test_add_pass_single(empty_emulator): + emulator = empty_emulator + qubit_count_validator = QubitCountValidator(4) + emulator.add_pass(qubit_count_validator) + + assert emulator._emulator_passes == [qubit_count_validator] + + +def test_bad_add_pass(empty_emulator): + emulator = empty_emulator + with pytest.raises(TypeError): + emulator.add_pass(None) + + +def test_add_pass_multiple(setup_local_simulator_devices): + native_gate_validator = GateValidator(native_gates=["CZ", "PRx"]) + emulator = Emulator(emulator_passes=[native_gate_validator]) + qubit_count_validator = QubitCountValidator(4) + gate_validator = GateValidator(supported_gates=["H", "CNot"]) + + emulator.add_pass([qubit_count_validator, gate_validator]) + assert emulator._emulator_passes == [ + native_gate_validator, + qubit_count_validator, + gate_validator, + ] + + +def test_use_correct_backend_if_noise_model(setup_local_simulator_devices): + noise_model = NoiseModel() + emulator = Emulator(noise_model=noise_model) + assert emulator._backend.name == "DensityMatrixSimulator" + + +def test_update_noise_model(empty_emulator): + emulator = empty_emulator + assert emulator._backend.name == "StateVectorSimulator" + noise_model = NoiseModel() + noise_model.add_noise(BitFlip(0.1), GateCriteria(Gate.H())) + + emulator.noise_model = noise_model + assert emulator._backend.name == "DensityMatrixSimulator" + assert emulator._backend._noise_model == noise_model + assert emulator.noise_model == noise_model + + +def test_validation_only_pass(setup_local_simulator_devices): + qubit_count_validator = QubitCountValidator(4) + bad_pass = AlwaysFailPass() + emulator = Emulator(emulator_passes=[bad_pass, qubit_count_validator]) + + circuit = Circuit().h(range(5)) + match_string = re.escape( + f"Circuit must use at most 4 qubits, \ +but uses {circuit.qubit_count} qubits. (DeviceEmulator)" + ) + with pytest.raises(Exception, match=match_string): + emulator.validate(circuit) + + +def test_apply_noise_model(setup_local_simulator_devices): + noise_model = NoiseModel() + noise_model.add_noise(BitFlip(0.1), GateCriteria(Gate.H)) + emulator = Emulator(noise_model=noise_model) + + circuit = Circuit().h(0) + circuit = emulator.run_passes(circuit) + + noisy_circuit = Circuit().h(0).apply_gate_noise(BitFlip(0.1), Gate.H) + assert circuit == noisy_circuit + + circuit = Circuit().h(0) + circuit = emulator.run_passes(circuit, apply_noise_model=False) + + target_circ = Circuit().h(0) + assert circuit == target_circ + + +def test_noiseless_run(setup_local_simulator_devices): + qubit_count_validator = QubitCountValidator(4) + gate_validator = GateValidator(supported_gates=["H"]) + emulator = Emulator(emulator_passes=[qubit_count_validator, gate_validator]) + circuit = Circuit().h(0).state_vector() + + result = emulator.run(circuit).result() + state_vector = result.result_types[0].value + target_state_vector = np.array([1 / np.sqrt(2) + 0j, 1 / np.sqrt(2) + 0j]) + assert all(np.isclose(state_vector, target_state_vector)) + + +def test_noisy_run(setup_local_simulator_devices): + noise_model = NoiseModel() + noise_model.add_noise(BitFlip(0.1), GateCriteria(Gate.H)) + + qubit_count_validator = QubitCountValidator(4) + gate_validator = GateValidator(supported_gates=["H"]) + emulator = Emulator( + backend="braket_dm", + emulator_passes=[qubit_count_validator, gate_validator], + noise_model=noise_model, + ) + + circuit = Circuit().h(0) + open_qasm_source = """OPENQASM 3.0; +bit[1] b; +qubit[1] q; +h q[0]; +#pragma braket noise bit_flip(0.1) q[0] +b[0] = measure q[0];""".strip() + + result = emulator.run(circuit, shots=1).result() + emulation_source = result.additional_metadata.action.source.strip() + assert emulation_source == open_qasm_source diff --git a/test/unit_tests/braket/emulation/test_gate_connectivity_validator.py b/test/unit_tests/braket/emulation/test_gate_connectivity_validator.py new file mode 100644 index 000000000..ec629a4c1 --- /dev/null +++ b/test/unit_tests/braket/emulation/test_gate_connectivity_validator.py @@ -0,0 +1,312 @@ +import networkx as nx +import numpy as np +import pytest +from networkx.utils import graphs_equal + +from braket.circuits import Circuit, Gate +from braket.circuits.noises import BitFlip +from braket.emulation.emulation_passes.gate_device_passes import GateConnectivityValidator + + +@pytest.fixture +def basic_4_node_graph(): + G = nx.DiGraph() + G.add_edges_from( + [ + (0, 1, {"supported_gates": ["CNot", "CZ"]}), + (1, 2, {"supported_gates": ["Swap", "CNot"]}), + (0, 3, {"supported_gates": ["XX", "XY"]}), + ] + ) + return G + + +@pytest.fixture +def basic_discontiguous_4_node_graph(): + G = nx.DiGraph() + G.add_edges_from( + [ + (0, 5, {"supported_gates": ["CNot", "CZ"]}), + (2, 4, {"supported_gates": ["Swap", "CNot"]}), + (3, 0, {"supported_gates": ["XX", "XY"]}), + ] + ) + return G + + +@pytest.fixture +def basic_4_node_graph_as_dict(): + return { + (0, 1): ["CNot", "Swap", "CX", "XX"], + (1, 2): ["CNot", "CZ", "ISwap", "XY"], + (0, 3): ["PSwap", "CNot", "XY"], + } + + +@pytest.mark.parametrize( + "circuit", + [ + Circuit(), + Circuit().add_verbatim_box(Circuit().cnot(0, 1).h(range(4))), + Circuit() + .add_verbatim_box(Circuit().cnot(0, 1).cz(0, 1).swap(1, 2).xx(0, 3, np.pi / 2)) + .add_verbatim_box(Circuit().xy(0, 3, np.pi / 2).cnot(1, 2).cz(0, 1)), + Circuit() + .i(range(10)) + .cnot(0, 2) + .yy(8, 9, np.pi / 2) + .h(7) + .add_verbatim_box(Circuit().cnot(0, 1).cz(0, 1)) + .cnot(0, 2) + .swap(4, 6), + Circuit().add_verbatim_box( + Circuit().h(0).apply_gate_noise(BitFlip(0.1), target_gates=Gate.H) + ), + ], +) +def test_valid_basic_contiguous_circuits(basic_4_node_graph, circuit): + """ + GateConnectivityValidator should not raise any errors when validating these circuits. + """ + gate_connectivity_validator = GateConnectivityValidator(basic_4_node_graph) + gate_connectivity_validator.validate(circuit) + + +@pytest.mark.parametrize( + "circuit", + [ + Circuit(), + Circuit().add_verbatim_box(Circuit().cnot(0, 5)), + Circuit() + .add_verbatim_box(Circuit().cnot(2, 4).cz(0, 5).swap(2, 4).xx(3, 0, np.pi / 2)) + .add_verbatim_box(Circuit().xy(3, 0, np.pi / 2).cnot(2, 4).cz(0, 5)), + Circuit() + .i(range(10)) + .cnot(0, 2) + .yy(8, 9, np.pi / 2) + .h(7) + .add_verbatim_box(Circuit().cnot(0, 5).swap(2, 4)) + .cnot(0, 2) + .swap(4, 6), + ], +) +def test_valid_basic_discontiguous_circuits(basic_discontiguous_4_node_graph, circuit): + """ + GateConnectivityValidator should not raise any errors when validating these circuits. + """ + gate_connectivity_validator = GateConnectivityValidator(basic_discontiguous_4_node_graph) + gate_connectivity_validator.validate(circuit) + + +def test_directed_graph_construction_from_dict(): + """ + GateConnectivityValidator should correctly construct a graph from a dictionary + representation of the connectivity. + """ + dict_representation = { + (0, 1): ["CNot", "CZ"], + (1, 2): ["Swap", "CNot", "YY"], + (0, 2): ["XX", "XY", "CNot", "CZ"], + (2, 5): ["XX", "XY", "CNot", "CZ"], + } + digraph_representation = nx.DiGraph() + digraph_representation.add_edges_from( + [ + (0, 1, {"supported_gates": ["CNot", "CZ"]}), + (1, 2, {"supported_gates": ["Swap", "CNot", "YY"]}), + (0, 2, {"supported_gates": ["XX", "XY", "CNot", "CZ"]}), + (2, 5, {"supported_gates": ["XX", "XY", "CNot", "CZ"]}), + ] + ) + gcc = GateConnectivityValidator(dict_representation) + assert graphs_equal(gcc._gate_connectivity_graph, digraph_representation) + + +@pytest.mark.parametrize( + "circuit", + [ + Circuit(), + Circuit().add_verbatim_box(Circuit().cnot(0, 1)), + Circuit() + .swap(0, 1) + .add_verbatim_box(Circuit().iswap(2, 1).pswap(3, 0, np.pi / 2).pswap(0, 3, np.pi / 2)), + Circuit() + .cnot(2, 3) + .h(5) + .pswap(0, 6, np.pi / 2) + .add_verbatim_box(Circuit().xy(2, 1, np.pi / 2).cnot(0, 1).cnot(1, 0)) + .add_verbatim_box(Circuit().cnot(0, 3).cnot(3, 0).xy(0, 3, np.pi / 2)), + ], +) +def test_undirected_criteria_from_dict_with_valid_circuits(basic_4_node_graph_as_dict, circuit): + """ + GateConnectivityValidator should not raise any errors when validating these circuits. + """ + gate_connectivity_validator = GateConnectivityValidator( + basic_4_node_graph_as_dict, directed=False + ) + gate_connectivity_validator.validate(circuit) + + +def test_undirected_graph_construction_from_dict(): + """ + GateConnectivityValidator should correctly construct an undirected graph from a dictionary + representation of the connectivity. + """ + dict_representation = { + (0, 1): ["CNot", "CZ"], + (1, 0): ["CNot", "CZ"], + (1, 2): ["Swap", "CNot", "YY"], + (0, 2): ["XX", "XY", "CNot", "CZ"], + (2, 5): ["XX", "XY", "CNot", "CZ"], + } + digraph_representation = nx.DiGraph() + digraph_representation.add_edges_from( + [ + (0, 1, {"supported_gates": ["CNot", "CZ"]}), + (1, 2, {"supported_gates": ["Swap", "CNot", "YY"]}), + (0, 2, {"supported_gates": ["XX", "XY", "CNot", "CZ"]}), + (2, 5, {"supported_gates": ["XX", "XY", "CNot", "CZ"]}), + (1, 0, {"supported_gates": ["CNot", "CZ"]}), + (2, 1, {"supported_gates": ["Swap", "CNot", "YY"]}), + (2, 0, {"supported_gates": ["XX", "XY", "CNot", "CZ"]}), + (5, 2, {"supported_gates": ["XX", "XY", "CNot", "CZ"]}), + ] + ) + gcc = GateConnectivityValidator(dict_representation, directed=False) + assert graphs_equal(gcc._gate_connectivity_graph, digraph_representation) + + +@pytest.mark.parametrize( + "edges", + [ + [(0, 1, {"supported_gates": ["CNot, CZ"]})], + [(0, 1, {"supported_gates": ["CNot", "CZ"]}), (1, 0, {"supported_gates": ["CNot", "CZ"]})], + [ + (0, 1, {"supported_gates": ["CNot", "CZ"]}), + (1, 2, {"supported_gates": ["CNot", "CZ", "XX"]}), + (2, 3, {"supported_gates": ["CZ"]}), + (2, 1, {"supported_gates": ["CNot", "CZ", "XX"]}), + ], + [ + (0, 1, {"supported_gates": ["CNot", "CZ"]}), + (1, 2, {"supported_gates": ["CNot", "CZ"]}), + (4, 2, {"supported_gates": ["CNot", "CZ"]}), + (3, 2, {"supported_gates": ["CNot", "CZ"]}), + ], + ], +) +def test_undirected_graph_from_digraph(edges): + """ + Check that undirected topologies created from a digraph correctly add all possible + back edges to the validator's connectivity graph. + """ + directed_graph = nx.DiGraph() + directed_graph.add_edges_from(edges) + undirected_graph = directed_graph.copy() + + for edge in edges: + if (edge[1], edge[0]) not in undirected_graph.edges: + undirected_graph.add_edges_from([(edge[1], edge[0], edge[2])]) + + gcc = GateConnectivityValidator(directed_graph, directed=False) + assert graphs_equal(gcc._gate_connectivity_graph, undirected_graph) + + gcc_other = GateConnectivityValidator(undirected_graph) + assert gcc_other == gcc + + +@pytest.mark.parametrize( + "representation", + [ + {(0, 1): ["CNot", "CZ"], (1, 0): ["CZ, XX"], (2, 0): ["CNot, YY"]}, + nx.from_dict_of_dicts( + { + 0: {1: {"supported_gates": ["CNot", "CZ"]}}, + 1: {0: {"supported_gates": ["CZ", "XX"]}}, + 2: {2: {"supported_gates": ["CNot", "YY"]}}, + } + ), + ], +) +def create_undirected_graph_with_exisiting_back_edges(representation): + """ + Check that creating an undirected graph with a graph that + contains forwards and backwards edges with different constraints + is created properly. + """ + + gcc = GateConnectivityValidator(representation, directed=False) + expected_digraph_representation = nx.DiGraph() + expected_digraph_representation.add_edges_from( + [ + (0, 1, {"supported_gates": ["CNot", "CZ"]}), + (1, 0, {"supported_gates": ["CZ", "XX"]}), + (2, 0, {"supported_gates": ["CNot", "YY"]}), + (0, 2, {"supported_gates": ["CNot", "YY"]}), + ] + ) + + assert graphs_equal(gcc._gate_connectivity_graph, expected_digraph_representation) + + +@pytest.mark.parametrize( + "circuit", + [ + Circuit().add_verbatim_box(Circuit().cnot(1, 0)), + Circuit().add_verbatim_box(Circuit().h(4)), + Circuit().add_verbatim_box(Circuit().swap(1, 2).xx(0, 3, np.pi / 2).iswap(0, 1)), + Circuit().add_verbatim_box(Circuit().cnot(0, 3)), + ], +) +def test_invalid_circuits(basic_4_node_graph, circuit): + with pytest.raises(ValueError): + gate_connectivity_validator = GateConnectivityValidator(basic_4_node_graph) + gate_connectivity_validator.validate(circuit) + + +def test_invalid_connectivity_graph(): + bad_graph = nx.complete_graph(5, create_using=nx.Graph()) + with pytest.raises(TypeError): + GateConnectivityValidator(bad_graph) + + +@pytest.mark.parametrize( + "gate_name,controls,targets,is_valid", + [ + ("CZ", [0], [1], True), + ("CNot", [], [0, 1], True), + ("XY", [3], [0], True), + ("CZ", [0, 2], [], False), + ("Swap", [0], [1, 2], False), + ("ZZ", [3], [0], False), + ], +) +def test_validate_instruction_method(gate_name, controls, targets, is_valid, basic_4_node_graph): + gcc = GateConnectivityValidator(basic_4_node_graph, directed=False) + if is_valid: + gcc._validate_instruction_connectivity(gate_name, controls, targets) + else: + with pytest.raises(ValueError): + gcc._validate_instruction_connectivity(gate_name, controls, targets) + + +@pytest.mark.parametrize( + "graph", + [ + ( + nx.from_dict_of_dicts( + { + 0: {1: {"supported_gates": ["cnot", "cz"]}}, + 1: {0: {"supported_gates": ["cz", "cnot", "xx"]}}, + }, + create_using=nx.DiGraph(), + ) + ), + ({(0, 1): ["cnot", "cz"], (1, 0): ["cz", "cnot", "xx"]}), + ({(0, 1): ["xx", "yy"], (1, 0): ["yy", "xx"]}), + ], +) +def test_invalid_undirected_graph(graph): + with pytest.raises(ValueError): + GateConnectivityValidator(graph, directed=False) diff --git a/test/unit_tests/braket/emulation/test_gate_validator.py b/test/unit_tests/braket/emulation/test_gate_validator.py new file mode 100644 index 000000000..a6de6a509 --- /dev/null +++ b/test/unit_tests/braket/emulation/test_gate_validator.py @@ -0,0 +1,178 @@ +import numpy as np +import pytest + +from braket.circuits import Circuit, Gate, Instruction +from braket.circuits.compiler_directives import StartVerbatimBox +from braket.circuits.noises import BitFlip +from braket.emulation.emulation_passes.gate_device_passes import GateValidator + + +@pytest.fixture +def basic_gate_set(): + return (["h", "cnot"], ["cz", "prx"]) + + +@pytest.fixture +def mock_qpu_gates(): + supported_gates = [ + "ccnot", + "cnot", + "cphaseshift", + "cswap", + "swap", + "iswap", + "pswap", + "ecr", + "cy", + "cz", + "xy", + "zz", + "h", + "i", + "phaseshift", + "rx", + "ry", + "v", + "vi", + "x", + "y", + "z", + ] + + native_gates = ["cz", "prx"] + return (supported_gates, native_gates) + + +@pytest.mark.parametrize( + "circuit", + [ + Circuit(), + Circuit() + .h(range(4)) + .cnot(0, 5) + .pswap(0, 1, np.pi / 4) + .xy(0, 1, 0.5) + .cphaseshift(0, 1, np.pi / 4), + Circuit() + .swap(0, 1) + .rx(0, 0.5) + .v(1) + .h(2) + .add_verbatim_box(Circuit().cz(0, 1).cz(0, 6).prx(0, np.pi / 4, np.pi / 5)) + .z(3), + Circuit() + .add_verbatim_box(Circuit().cz(0, 2).prx(0, 0.5, 0.5)) + .add_verbatim_box(Circuit().cz(0, 4).cz(3, 6)), + Circuit().h(0).add_verbatim_box(Circuit()), + Circuit() + .add_verbatim_box( + Circuit().prx(0, np.pi / 4, np.pi / 4).apply_gate_noise(BitFlip(0.1), Gate.PRx) + ) + .v(1) + .apply_gate_noise(BitFlip(0.1), Gate.V), + ], +) +def test_valid_circuits(mock_qpu_gates, circuit): + """ + GateValidator should not raise any errors when validating these circuits. + """ + GateValidator(mock_qpu_gates[0], mock_qpu_gates[1]).validate(circuit) + + +def test_only_supported_gates(): + supported_gates = ["h", "cnot", "rx", "xx", "y"] + validator = GateValidator(supported_gates=supported_gates) + circuit = Circuit().h(0).cnot(0, 1).rx(4, np.pi / 4).xx(2, 3, np.pi / 4).y(7) + validator.validate(circuit) + + +def test_verbatim_circuit_only_supported_gates(): + supported_gates = ["h", "cnot", "rx", "xx", "y"] + validator = GateValidator(supported_gates=supported_gates) + circuit = Circuit().add_verbatim_box(Circuit().h(0)) + + with pytest.raises(ValueError): + validator.validate(circuit) + + +def test_only_native_gates(): + native_gates = ["h", "cnot", "rx", "xx", "y"] + validator = GateValidator(native_gates=native_gates) + vb = Circuit().h(0).cnot(0, 1).rx(4, np.pi / 4).xx(2, 3, np.pi / 4).y(7) + circuit = Circuit().add_verbatim_box(vb) + validator.validate(circuit) + + +def test_non_verbatim_circuit_only_native_gates(): + native_gates = ["h", "cnot", "rx", "xx", "y"] + validator = GateValidator(native_gates=native_gates) + vb = Circuit().h(0).cnot(0, 1).rx(4, np.pi / 4).xx(2, 3, np.pi / 4).y(7) + circuit = Circuit().add_verbatim_box(vb) + circuit.i(0) + with pytest.raises(ValueError): + validator.validate(circuit) + + +@pytest.mark.parametrize( + "supported_gates,native_gates,error_message", + [ + ([], [], "Supported gate set or native gate set must be provided."), + (["CX"], [], "Input 'cx' in supported_gates is not a valid Braket gate name."), + ([], ["CX"], "Input 'cx' in native_gates is not a valid Braket gate name."), + ( + ["Toffoli"], + ["CX"], + "Input 'toffoli' in supported_gates is not a valid Braket gate name.", + ), + ], +) +def test_invalid_instantiation(supported_gates, native_gates, error_message): + with pytest.raises(ValueError, match=error_message): + GateValidator(supported_gates, native_gates) + + +@pytest.mark.parametrize( + "circuit", + [ + Circuit().z(0), + Circuit().h(0).cnot(0, 1).cz(0, 1), + Circuit().add_verbatim_box(Circuit().h(2)), + Circuit().cphaseshift01(0, 1, np.pi / 4).h(0).cnot(0, 1), + Circuit() + .h(0) + .add_verbatim_box(Circuit().cz(1, 2).prx(range(5), np.pi / 4, np.pi / 2).cz(2, 6)) + .prx(range(4), np.pi / 4, np.pi / 6), + Circuit().add_instruction(Instruction(StartVerbatimBox())), + ], +) +def test_invalid_circuits(basic_gate_set, circuit): + """ + GateValidator should raise errors when validating these circuits. + """ + with pytest.raises(ValueError): + GateValidator(basic_gate_set[0], basic_gate_set[1]).validate(circuit) + + +@pytest.mark.parametrize( + "gate_set_1, gate_set_2", + [ + (["h"], ["h"]), + (["cnot", "h"], ["h", "cnot"]), + (["phaseshift", "cnot", "rx", "ry"], ["ry", "rx", "cnot", "phaseshift"]), + ], +) +def test_equality(gate_set_1, gate_set_2): + assert GateValidator(gate_set_1) == GateValidator(gate_set_2) + + +@pytest.mark.parametrize( + "gate_set_1, gate_set_2", + [ + (["h"], ["x"]), + (["cnot"], ["h", "cnot"]), + (["cnot", "h"], ["h"]), + (["phaseshift", "cnot", "ms", "ry"], ["ry", "rx", "cnot", "ms"]), + ], +) +def test_inequality(gate_set_1, gate_set_2): + assert GateValidator(gate_set_1) != GateValidator(gate_set_2) diff --git a/test/unit_tests/braket/emulation/test_qubit_count_validator.py b/test/unit_tests/braket/emulation/test_qubit_count_validator.py new file mode 100644 index 000000000..effb21367 --- /dev/null +++ b/test/unit_tests/braket/emulation/test_qubit_count_validator.py @@ -0,0 +1,55 @@ +import numpy as np +import pytest + +from braket.circuits import Circuit +from braket.emulation.emulation_passes.gate_device_passes import QubitCountValidator + + +@pytest.mark.parametrize( + "qubit_count,circuit", + [ + (1, Circuit()), + (10, Circuit().add_verbatim_box(Circuit())), + (1, Circuit().z(0)), + (1, Circuit().z(3).x(3)), + (2, Circuit().cnot(0, 1).swap(1, 0)), + (2, Circuit().z(0).add_verbatim_box(Circuit().cnot(0, 4)).yy(0, 4, np.pi / 4)), + (50, Circuit().i(range(50)).measure(range(50))), + ], +) +def test_valid_circuits(qubit_count, circuit): + """ + QubitCountValidator should not raise any errors when validating these circuits. + """ + QubitCountValidator(qubit_count=qubit_count).__call__(circuit) + + +@pytest.mark.parametrize("qubit_count", [0, -1]) +def test_invalid_instantiation(qubit_count): + with pytest.raises(ValueError): + QubitCountValidator(qubit_count) + + +@pytest.mark.parametrize( + "qubit_count,circuit", + [ + (1, Circuit().cnot(0, 1)), + (2, Circuit().cnot(0, 1).x(2)), + (50, Circuit().i(range(50)).measure(range(50)).measure(50)), + ], +) +def test_invalid_circuits(qubit_count, circuit): + with pytest.raises( + ValueError, + match=f"Circuit must use at most {qubit_count} qubits, \ +but uses {circuit.qubit_count} qubits.", + ): + QubitCountValidator(qubit_count).validate(circuit) + + +def test_equality(): + qcc_1 = QubitCountValidator(1) + qcc_2 = QubitCountValidator(2) + + assert qcc_1 != qcc_2 + assert qcc_1 == QubitCountValidator(1)