diff --git a/qiskit_ibm_runtime/estimator.py b/qiskit_ibm_runtime/estimator.py index d385f98c5..c55b58857 100644 --- a/qiskit_ibm_runtime/estimator.py +++ b/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, diff --git a/qiskit_ibm_runtime/ibm_backend.py b/qiskit_ibm_runtime/ibm_backend.py index 22ea6443a..864d2a945 100644 --- a/qiskit_ibm_runtime/ibm_backend.py +++ b/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.""" diff --git a/qiskit_ibm_runtime/sampler.py b/qiskit_ibm_runtime/sampler.py index e0dd62bd2..c20e5ba27 100644 --- a/qiskit_ibm_runtime/sampler.py +++ b/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, diff --git a/test/unit/test_ibm_primitives.py b/test/unit/test_ibm_primitives.py index ee2c426af..18bf99c86 100644 --- a/test/unit/test_ibm_primitives.py +++ b/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): diff --git a/test/utils.py b/test/utils.py index bab2e785e..86d600189 100644 --- a/test/utils.py +++ b/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