Check for faulty qubits and edges

qiskit_ibm_runtime/estimator.py

@@ -305,6 +305,8 @@ def _run(  # pylint: disable=arguments-differ
         }
 
         combined = Options._merge_options(self._options, kwargs.get("_user_kwargs", {}))
+
+        backend_obj: Optional[IBMBackend] = None
         if self._session.backend():
             backend_obj = self._session.service.backend(self._session.backend())
             combined = set_default_error_levels(
@@ -319,6 +321,10 @@ def _run(  # pylint: disable=arguments-differ
         logger.info("Submitting job using options %s", combined)
         inputs.update(Options._get_program_inputs(combined))
 
+        if backend_obj and combined["transpilation"]["skip_transpilation"]:
+            for circ in circuits:
+                backend_obj.check_faulty(circ)
+
         return self._session.run(
             program_id=self._PROGRAM_ID,
             inputs=inputs,

qiskit_ibm_runtime/ibm_backend.py

@@ -17,6 +17,7 @@
 from typing import Iterable, Union, Optional, Any, List
 from datetime import datetime as python_datetime
 
+from qiskit import QuantumCircuit
 from qiskit.qobj.utils import MeasLevel, MeasReturnType
 from qiskit.providers.backend import BackendV2 as Backend
 from qiskit.providers.options import Options
@@ -502,6 +503,42 @@ def run(self, *args: Any, **kwargs: Any) -> None:
             "IBMBackend.run() is not supported in the Qiskit Runtime environment."
         )
 
+    def check_faulty(self, circuit: QuantumCircuit) -> None:
+        """Check if the input circuit uses faulty qubits or edges.
+
+        Args:
+            circuit: Circuit to check.
+
+        Raises:
+            ValueError: If an instruction operating on a faulty qubit or edge is found.
+        """
+        if not self.properties():
+            return
+
+        faulty_qubits = self.properties().faulty_qubits()
+        faulty_gates = self.properties().faulty_gates()
+        faulty_edges = [
+            tuple(gate.qubits) for gate in faulty_gates if len(gate.qubits) > 1
+        ]
+
+        for instr in circuit.data:
+            if instr.operation.name == "barrier":
+                continue
+            qubit_indices = tuple(circuit.find_bit(x).index for x in instr.qubits)
+
+            for circ_qubit in qubit_indices:
+                if circ_qubit in faulty_qubits:
+                    raise ValueError(
+                        f"Circuit {circuit.name} contains instruction "
+                        f"{instr} operating on a faulty qubit {circ_qubit}."
+                    )
+
+            if len(qubit_indices) == 2 and qubit_indices in faulty_edges:
+                raise ValueError(
+                    f"Circuit {circuit.name} contains instruction "
+                    f"{instr} operating on a faulty edge {qubit_indices}"
+                )
+
 
 class IBMRetiredBackend(IBMBackend):
     """Backend class interfacing with an IBM Quantum device no longer available."""

qiskit_ibm_runtime/sampler.py

@@ -259,6 +259,8 @@ def _run(  # pylint: disable=arguments-differ
             "parameter_values": parameter_values,
         }
         combined = Options._merge_options(self._options, kwargs.get("_user_kwargs", {}))
+
+        backend_obj: Optional[IBMBackend] = None
         if self._session.backend():
             backend_obj = self._session.service.backend(self._session.backend())
             combined = set_default_error_levels(
@@ -273,6 +275,10 @@ def _run(  # pylint: disable=arguments-differ
         logger.info("Submitting job using options %s", combined)
         inputs.update(Options._get_program_inputs(combined))
 
+        if backend_obj and combined["transpilation"]["skip_transpilation"]:
+            for circ in circuits:
+                backend_obj.check_faulty(circ)
+
         return self._session.run(
             program_id=self._PROGRAM_ID,
             inputs=inputs,

test/unit/test_ibm_primitives.py

@@ -21,11 +21,13 @@
 from typing import Dict
 import unittest
 
+from qiskit import transpile
 from qiskit.circuit import QuantumCircuit
 from qiskit.circuit.library import RealAmplitudes
 from qiskit.test.reference_circuits import ReferenceCircuits
 from qiskit.quantum_info import SparsePauliOp
 from qiskit.primitives.utils import _circuit_key
+from qiskit.providers.fake_provider import FakeManila
 
 from qiskit_ibm_runtime import (
     Sampler,
@@ -39,7 +41,12 @@
 from qiskit_ibm_runtime.utils.utils import _hash
 
 from ..ibm_test_case import IBMTestCase
-from ..utils import dict_paritally_equal, flat_dict_partially_equal, dict_keys_equal
+from ..utils import (
+    dict_paritally_equal,
+    flat_dict_partially_equal,
+    dict_keys_equal,
+    create_faulty_backend,
+)
 from .mock.fake_runtime_service import FakeRuntimeService
 
 
@@ -628,6 +635,178 @@ def test_default_error_levels(self):
                 )
                 self.assertEqual(inputs["resilience_settings"]["level"], 0)
 
+    def test_raise_faulty_qubits(self):
+        """Test faulty qubits is raised."""
+        fake_backend = FakeManila()
+        num_qubits = fake_backend.configuration().num_qubits
+        circ = QuantumCircuit(num_qubits, num_qubits)
+        for i in range(num_qubits):
+            circ.x(i)
+        transpiled = transpile(circ, backend=fake_backend)
+        observable = SparsePauliOp("Z" * num_qubits)
+
+        faulty_qubit = 4
+        ibm_backend = create_faulty_backend(fake_backend, faulty_qubit=faulty_qubit)
+        service = MagicMock()
+        service.backend.return_value = ibm_backend
+        session = Session(service=service, backend=fake_backend.name)
+        sampler = Sampler(session=session)
+        estimator = Estimator(session=session)
+
+        with self.assertRaises(ValueError) as err:
+            sampler.run(transpiled, skip_transpilation=True)
+        self.assertIn(f"faulty qubit {faulty_qubit}", str(err.exception))
+
+        with self.assertRaises(ValueError) as err:
+            estimator.run(transpiled, observable, skip_transpilation=True)
+        self.assertIn(f"faulty qubit {faulty_qubit}", str(err.exception))
+
+    def test_raise_faulty_qubits_many(self):
+        """Test faulty qubits is raised if one circuit uses it."""
+        fake_backend = FakeManila()
+        num_qubits = fake_backend.configuration().num_qubits
+
+        circ1 = QuantumCircuit(1, 1)
+        circ1.x(0)
+        circ2 = QuantumCircuit(num_qubits, num_qubits)
+        for i in range(num_qubits):
+            circ2.x(i)
+        transpiled = transpile([circ1, circ2], backend=fake_backend)
+        observable = SparsePauliOp("Z" * num_qubits)
+
+        faulty_qubit = 4
+        ibm_backend = create_faulty_backend(fake_backend, faulty_qubit=faulty_qubit)
+        service = MagicMock()
+        service.backend.return_value = ibm_backend
+        session = Session(service=service, backend=fake_backend.name)
+        sampler = Sampler(session=session)
+        estimator = Estimator(session=session)
+
+        with self.assertRaises(ValueError) as err:
+            sampler.run(transpiled, skip_transpilation=True)
+        self.assertIn(f"faulty qubit {faulty_qubit}", str(err.exception))
+
+        with self.assertRaises(ValueError) as err:
+            estimator.run(transpiled, [observable, observable], skip_transpilation=True)
+        self.assertIn(f"faulty qubit {faulty_qubit}", str(err.exception))
+
+    def test_raise_faulty_edge(self):
+        """Test faulty edge is raised."""
+        fake_backend = FakeManila()
+        num_qubits = fake_backend.configuration().num_qubits
+        circ = QuantumCircuit(num_qubits, num_qubits)
+        for i in range(num_qubits - 2):
+            circ.cx(i, i + 1)
+        transpiled = transpile(circ, backend=fake_backend)
+        observable = SparsePauliOp("Z" * num_qubits)
+
+        edge_qubits = [0, 1]
+        ibm_backend = create_faulty_backend(
+            fake_backend, faulty_edge=("cx", edge_qubits)
+        )
+        service = MagicMock()
+        service.backend.return_value = ibm_backend
+        session = Session(service=service, backend=fake_backend.name)
+        sampler = Sampler(session=session)
+        estimator = Estimator(session=session)
+
+        with self.assertRaises(ValueError) as err:
+            sampler.run(transpiled, skip_transpilation=True)
+        self.assertIn("cx", str(err.exception))
+        self.assertIn(f"faulty edge {tuple(edge_qubits)}", str(err.exception))
+
+        with self.assertRaises(ValueError) as err:
+            estimator.run(transpiled, observable, skip_transpilation=True)
+        self.assertIn("cx", str(err.exception))
+        self.assertIn(f"faulty edge {tuple(edge_qubits)}", str(err.exception))
+
+    def test_faulty_qubit_not_used(self):
+        """Test faulty qubit is not raise if not used."""
+        fake_backend = FakeManila()
+        circ = QuantumCircuit(2, 2)
+        for i in range(2):
+            circ.x(i)
+        transpiled = transpile(circ, backend=fake_backend, initial_layout=[0, 1])
+        observable = SparsePauliOp("Z" * fake_backend.configuration().num_qubits)
+
+        faulty_qubit = 4
+        ibm_backend = create_faulty_backend(fake_backend, faulty_qubit=faulty_qubit)
+
+        service = MagicMock()
+        service.backend.return_value = ibm_backend
+        session = Session(service=service, backend=fake_backend.name)
+        sampler = Sampler(session=session)
+        estimator = Estimator(session=session)
+
+        with patch.object(Session, "run") as mock_run:
+            sampler.run(transpiled, skip_transpilation=True)
+        mock_run.assert_called_once()
+
+        with patch.object(Session, "run") as mock_run:
+            estimator.run(transpiled, observable, skip_transpilation=True)
+        mock_run.assert_called_once()
+
+    def test_faulty_edge_not_used(self):
+        """Test faulty edge is not raised if not used."""
+        fake_backend = FakeManila()
+        coupling_map = fake_backend.configuration().coupling_map
+
+        circ = QuantumCircuit(2, 2)
+        circ.cx(0, 1)
+
+        transpiled = transpile(
+            circ, backend=fake_backend, initial_layout=coupling_map[0]
+        )
+        observable = SparsePauliOp("Z" * fake_backend.configuration().num_qubits)
+
+        edge_qubits = coupling_map[-1]
+        ibm_backend = create_faulty_backend(
+            fake_backend, faulty_edge=("cx", edge_qubits)
+        )
+
+        service = MagicMock()
+        service.backend.return_value = ibm_backend
+        session = Session(service=service, backend=fake_backend.name)
+        sampler = Sampler(session=session)
+        estimator = Estimator(session=session)
+
+        with patch.object(Session, "run") as mock_run:
+            sampler.run(transpiled, skip_transpilation=True)
+        mock_run.assert_called_once()
+
+        with patch.object(Session, "run") as mock_run:
+            estimator.run(transpiled, observable, skip_transpilation=True)
+        mock_run.assert_called_once()
+
+    def test_no_raise_skip_transpilation(self):
+        """Test faulty qubits and edges are not raise if not skipping."""
+        fake_backend = FakeManila()
+        num_qubits = fake_backend.configuration().num_qubits
+        circ = QuantumCircuit(num_qubits, num_qubits)
+        for i in range(num_qubits - 2):
+            circ.cx(i, i + 1)
+        transpiled = transpile(circ, backend=fake_backend)
+        observable = SparsePauliOp("Z" * num_qubits)
+
+        edge_qubits = [0, 1]
+        ibm_backend = create_faulty_backend(
+            fake_backend, faulty_qubit=0, faulty_edge=("cx", edge_qubits)
+        )
+
+        service = MagicMock()
+        service.backend.return_value = ibm_backend
+        session = Session(service=service, backend=fake_backend.name)
+        sampler = Sampler(session=session)
+        estimator = Estimator(session=session)
+
+        with patch.object(Session, "run") as mock_run:
+            sampler.run(transpiled)
+        mock_run.assert_called_once()
+
+        with patch.object(Session, "run") as mock_run:
+            estimator.run(transpiled, observable)
+        mock_run.assert_called_once()
+
     def _update_dict(self, dict1, dict2):
         for key, val in dict1.items():
             if isinstance(val, dict):

test/utils.py

@@ -17,11 +17,14 @@
 import time
 import unittest
 from unittest import mock
-from typing import Dict
+from typing import Dict, Optional
+from datetime import datetime
 
 from qiskit.circuit import QuantumCircuit
 from qiskit.providers.jobstatus import JOB_FINAL_STATES, JobStatus
 from qiskit.providers.exceptions import QiskitBackendNotFoundError
+from qiskit.providers.models import BackendStatus, BackendProperties
+from qiskit.providers.backend import Backend
 from qiskit_ibm_runtime.hub_group_project import HubGroupProject
 from qiskit_ibm_runtime import QiskitRuntimeService
 from qiskit_ibm_runtime.ibm_backend import IBMBackend
@@ -194,3 +197,57 @@ def dict_keys_equal(dict1: dict, dict2: dict) -> bool:
                 return False
 
     return True
+
+
+def create_faulty_backend(
+    model_backend: Backend,
+    faulty_qubit: Optional[int] = None,
+    faulty_edge: Optional[tuple] = None,
+) -> IBMBackend:
+    """Create an IBMBackend that has faulty qubits and/or edges.
+
+    Args:
+        model_backend: Fake backend to model after.
+        faulty_qubit: Faulty qubit.
+        faulty_edge: Faulty edge, a tuple of (gate, qubits)
+
+    Returns:
+        An IBMBackend with faulty qubits/edges.
+    """
+
+    properties = model_backend.properties().to_dict()
+
+    if faulty_qubit:
+        properties["qubits"][faulty_qubit].append(
+            {"date": datetime.now(), "name": "operational", "unit": "", "value": 0}
+        )
+
+    if faulty_edge:
+        gate, qubits = faulty_edge
+        for gate_obj in properties["gates"]:
+            if gate_obj["gate"] == gate and gate_obj["qubits"] == qubits:
+                gate_obj["parameters"].append(
+                    {
+                        "date": datetime.now(),
+                        "name": "operational",
+                        "unit": "",
+                        "value": 0,
+                    }
+                )
+
+    out_backend = IBMBackend(
+        configuration=model_backend.configuration(),
+        service=mock.MagicMock(),
+        api_client=None,
+        instance=None,
+    )
+
+    out_backend.status = lambda: BackendStatus(  # type: ignore[assignment]
+        backend_name="foo",
+        backend_version="1.0",
+        operational=True,
+        pending_jobs=0,
+        status_msg="",
+    )
+    out_backend.properties = lambda: BackendProperties.from_dict(properties)  # type: ignore
+    return out_backend