Add native switch to run method (#189)

* Add native switch to run method

qiskit_braket_provider/providers/adapter.py

@@ -15,7 +15,11 @@
     Instruction,
     measure,
 )
-from braket.device_schema import DeviceActionType, OpenQASMDeviceActionProperties
+from braket.device_schema import (
+    DeviceActionType,
+    DeviceCapabilities,
+    OpenQASMDeviceActionProperties,
+)
 from braket.device_schema.ionq import IonqDeviceCapabilities
 from braket.device_schema.iqm import IqmDeviceCapabilities
 from braket.device_schema.oqc import OqcDeviceCapabilities
@@ -192,6 +196,48 @@
 }
 
 
+def native_gate_connectivity(
+    properties: DeviceCapabilities,
+) -> Optional[list[list[int]]]:
+    """Returns the connectivity natively supported by a Braket device from its properties
+
+    Args:
+        properties (DeviceCapabilities): The device properties of the Braket device.
+
+    Returns:
+        Optional[list[list[int]]]: A list of connected qubit pairs or `None` if the device is fully
+            connected.
+    """
+    device_connectivity = properties.paradigm.connectivity
+    connectivity = (
+        [
+            [int(x), int(y)]
+            for x, neighborhood in device_connectivity.connectivityGraph.items()
+            for y in neighborhood
+        ]
+        if not device_connectivity.fullyConnected
+        else None
+    )
+    return connectivity
+
+
+def native_gate_set(properties: DeviceCapabilities) -> set[str]:
+    """Returns the gate set natively supported by a Braket device from its properties
+
+    Args:
+        properties (DeviceCapabilities): The device properties of the Braket device.
+
+    Returns:
+        set[str]: The names of qiskit gates natively supported by the Braket device.
+    """
+    native_list = properties.paradigm.nativeGateSet
+    return {
+        _BRAKET_TO_QISKIT_NAMES[op.lower()]
+        for op in native_list
+        if op.lower() in _BRAKET_TO_QISKIT_NAMES
+    }
+
+
 def gateset_from_properties(properties: OpenQASMDeviceActionProperties) -> set[str]:
     """Returns the gateset supported by a Braket device with the given properties
 

qiskit_braket_provider/providers/braket_backend.py

@@ -23,6 +23,8 @@
     aws_device_to_target,
     gateset_from_properties,
     local_simulator_to_target,
+    native_gate_connectivity,
+    native_gate_set,
     to_braket,
 )
 from .braket_quantum_task import BraketQuantumTask
@@ -52,9 +54,12 @@ def _validate_meas_level(self, meas_level: Union[enum.Enum, int]):
                 f"results, received meas_level={meas_level}."
             )
 
-    def _get_gateset(self) -> Optional[set[str]]:
-        action = self._device.properties.action.get(DeviceActionType.OPENQASM)
-        return gateset_from_properties(action) if action else None
+    def _get_gateset(self, native=False) -> Optional[set[str]]:
+        if native:
+            return native_gate_set(self._device.properties)
+        else:
+            action = self._device.properties.action.get(DeviceActionType.OPENQASM)
+            return gateset_from_properties(action) if action else None
 
 
 class BraketLocalBackend(BraketBackend):
@@ -314,7 +319,7 @@ def acquire_channel(self, qubit: int):
     def control_channel(self, qubits: Iterable[int]):
         raise NotImplementedError(f"Control channel is not supported by {self.name}.")
 
-    def run(self, run_input, **options):
+    def run(self, run_input, verbatim: bool = False, native: bool = False, **options):
         if isinstance(run_input, QuantumCircuit):
             circuits = [run_input]
         elif isinstance(run_input, list):
@@ -326,9 +331,17 @@ def run(self, run_input, **options):
             self._validate_meas_level(options["meas_level"])
             del options["meas_level"]
 
-        verbatim = options.pop("verbatim", False)
-        gateset = self._get_gateset() if not verbatim else None
-        braket_circuits = [to_braket(circ, gateset, verbatim) for circ in circuits]
+        gateset = self._get_gateset(native) if not verbatim else None
+        connectivity = (
+            native_gate_connectivity(self._device.properties) if native else None
+        )
+
+        braket_circuits = [
+            to_braket(
+                circ, basis_gates=gateset, verbatim=verbatim, connectivity=connectivity
+            )
+            for circ in circuits
+        ]
 
         batch_task: AwsQuantumTaskBatch = self._device.run_batch(
             braket_circuits, **options

tests/providers/test_braket_backend.py

@@ -424,6 +424,43 @@ def test_native_circuits_with_measurements_can_be_run_in_verbatim_mode(self):
 
         self.assertEqual(sum(result.get_counts().values()), 10)
 
+    @patch("qiskit_braket_provider.providers.braket_backend.to_braket")
+    def test_native_transpilation(self, mock_to_braket):
+        """Tests running circuit with native mode"""
+        mock_device = Mock()
+        mock_device.properties = RIGETTI_MOCK_GATE_MODEL_QPU_CAPABILITIES
+        mock_device.properties.paradigm.connectivity.connectivityGraph = {
+            "0": ["1"],
+            "1": ["0", "2"],
+            "2": ["1"],
+        }
+        mock_device.properties.paradigm.nativeGateSet = ["rx", "rz", "cnot"]
+
+        mock_batch = Mock()
+        mock_batch.tasks = [Mock(id="abcd1234")]
+        mock_device.run_batch.return_value = mock_batch
+
+        circuit = QuantumCircuit(3)
+        circuit.h(0)
+        circuit.cx(0, 1)
+        circuit.cx(0, 2)
+
+        backend = AWSBraketBackend(device=mock_device)
+
+        backend.run(circuit, native=True)
+        assert mock_to_braket.call_args.kwargs["basis_gates"] == {"rx", "rz", "cx"}
+        assert mock_to_braket.call_args.kwargs["connectivity"] == [
+            [0, 1],
+            [1, 0],
+            [1, 2],
+            [2, 1],
+        ]
+
+        backend.run(circuit, verbatim=True)
+        assert mock_to_braket.call_args.kwargs["basis_gates"] is None
+        assert mock_to_braket.call_args.kwargs["verbatim"] is True
+        assert mock_to_braket.call_args.kwargs["connectivity"] is None
+
     @patch("qiskit_braket_provider.providers.braket_provider.AwsDevice")
     def test_queue_depth(self, mocked_device):
         """Tests queue depth."""