feature: Introduce Validating Emulators with Noise Models to the SDK

src/braket/aws/aws_device.py

@@ -27,6 +27,13 @@
 
 from braket.ahs.analog_hamiltonian_simulation import AnalogHamiltonianSimulation
 from braket.annealing.problem import Problem
+from braket.aws.aws_emulator_helpers import (
+    create_connectivity_criterion,
+    create_gate_connectivity_criterion,
+    create_gate_criterion,
+    create_qubit_count_criterion,
+)
+from braket.aws.aws_noise_models import create_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 +46,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.emulators import Emulator
+from braket.emulators.emulator_passes import ProgramType
 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.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 +866,116 @@ 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 an 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 on an 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.
+        """
+        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 simulated).
+
+        Returns:
+            Emulator: An emulator with a noise model, compilation passes, and validation passes
+            based on this device's properites.
+        """
+        emulator_noise_model = create_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(create_qubit_count_criterion(self.properties))
+        self._emulator.add_pass(create_gate_criterion(self.properties))
+        self._emulator.add_pass(create_connectivity_criterion(self.properties, self.topology_graph))
+        self._emulator.add_pass(
+            create_gate_connectivity_criterion(self.properties, self.topology_graph)
+        )
+        return self._emulator
+
+    def validate(
+        self,
+        task_specification: Circuit,
+    ) -> None:
+        """
+        Runs all non-modifying emulator passes on the input program and raises an
+        error if any device-specific criterion are not met by the program. If the
+        program meets all criterion, returns.
+
+        Args:
+            task_specification (Circuit): The quantum program to emulate against
+                this AwsDevice device properties.
+
+        """
+        self.emulator.run_validation_passes(task_specification)
+        return
+
+    def run_emulator_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 = task_specification.copy()
+        return self.emulator.run_program_passes(task_specification, apply_noise_model)
+
+    def emulate(
+        self,
+        task_specification: Circuit,
+        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 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 (Circuit): 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 = task_specification.copy()
+        return self.emulator.run(task_specification, shots, inputs)

src/braket/aws/aws_emulator_helpers.py

@@ -0,0 +1,201 @@
+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.emulators.emulator_passes import (
+    ConnectivityCriterion,
+    GateConnectivityCriterion,
+    GateCriterion,
+    QubitCountCriterion,
+)
+
+
+def create_qubit_count_criterion(properties: DeviceCapabilities) -> QubitCountCriterion:
+    """
+    Create a QubitCountCriterion 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:
+        QubitCountCriterion: An eulator 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 QubitCountCriterion(qubit_count)
+
+
+def create_gate_criterion(properties: DeviceCapabilities) -> GateCriterion:
+    supported_gates = properties.action[DeviceActionType.OPENQASM].supportedOperations
+    """
+    Create a GateCriterion 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:
+        GateCriterion: An emulator pass that checks that a circuit only uses supported gates and
+        verbatim circuits only use native gates.
+    """
+
+    if isinstance(properties, IqmDeviceCapabilities):
+        try:
+            supported_gates.remove("start_verbatim_box")
+            supported_gates.remove("end_verbatim_box")
+        except ValueError:
+            pass
+
+    native_gates = properties.paradigm.nativeGateSet
+
+    return GateCriterion(supported_gates=supported_gates, native_gates=native_gates)
+
+
+@singledispatch
+def create_connectivity_criterion(
+    properties: DeviceCapabilities, connectivity_graph: DiGraph
+) -> ConnectivityCriterion:
+    """
+    Creates a ConnectivityCriterion pass which validates that multi-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:
+        ConnectivityCriterion: An emulator pass that checks that a circuit only applies two-qubit
+        gates to connected qubits on the device.
+    """
+    connectivity_criterion = ConnectivityCriterion(connectivity_graph)
+    return connectivity_criterion
+
+
+@create_connectivity_criterion.register(IqmDeviceCapabilities)
+def _(properties: IqmDeviceCapabilities, connectivity_graph: DiGraph) -> ConnectivityCriterion:
+    """
+    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 ConnectivityCriterion for an IQM device.
+    """
+    connectivity_graph = connectivity_graph.copy()
+    for edge in connectivity_graph.edges:
+        connectivity_graph.add_edge(edge[1], edge[0])
+    return ConnectivityCriterion(connectivity_graph)
+
+
+@singledispatch
+def create_gate_connectivity_criterion(
+    properties: DeviceCapabilities, connectivity_graph: DiGraph
+) -> GateConnectivityCriterion:
+    raise NotImplementedError
+
+
+@create_gate_connectivity_criterion.register(IqmDeviceCapabilities)
+@create_gate_connectivity_criterion.register(RigettiDeviceCapabilities)
+def _(
+    properties: RigettiDeviceCapabilities, connectivity_graph: DiGraph
+) -> GateConnectivityCriterion:
+    """
+    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)
+        if not edge_property:
+            gate_connectivity_graph[u][v]["supported_gates"] = set()
+            continue
+        edge_supported_gates = get_qpu_gate_translation(
+            properties, [property.gateName for property in edge_property.twoQubitGateFidelity]
+        )
+        gate_connectivity_graph[u][v]["supported_gates"] = set(edge_supported_gates)
+
+    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 GateConnectivityCriterion(gate_connectivity_graph)
+
+
+@create_gate_connectivity_criterion.register(IonqDeviceCapabilities)
+def _(properties: IonqDeviceCapabilities, connectivity_graph: DiGraph) -> GateConnectivityCriterion:
+    """
+    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_translation(properties, properties.paradigm.nativeGateSet)
+
+    for edge in gate_connectivity_graph.edges:
+        gate_connectivity_graph[edge[0]][edge[1]]["supported_gates"] = set(native_gates)
+
+    return GateConnectivityCriterion(gate_connectivity_graph)
+
+
+def get_qpu_gate_translation(
+    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)
+
+    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)