Add fast transpilation method to BaseExperiment

Author: wshanks

qiskit_experiments/calibration_management/base_calibration_experiment.py

@@ -21,13 +21,6 @@
 from qiskit import QuantumCircuit
 from qiskit.providers.options import Options
 from qiskit.pulse import ScheduleBlock
-from qiskit.transpiler import StagedPassManager, PassManager, Layout, CouplingMap
-from qiskit.transpiler.passes import (
-    EnlargeWithAncilla,
-    FullAncillaAllocation,
-    ApplyLayout,
-    SetLayout,
-)
 
 from qiskit_experiments.calibration_management.calibrations import Calibrations
 from qiskit_experiments.calibration_management.update_library import BaseUpdater
@@ -198,20 +191,6 @@ def _default_experiment_options(cls) -> Options:
         options.update_options(result_index=-1, group="default")
         return options
 
-    @classmethod
-    def _default_transpile_options(cls) -> Options:
-        """Return empty default transpile options as optimization_level is not used."""
-        return Options()
-
-    def set_transpile_options(self, **fields):
-        r"""Add a warning message.
-
-        .. note::
-            If your experiment has overridden `_transpiled_circuits` and needs
-            transpile options then please also override `set_transpile_options`.
-        """
-        warnings.warn(f"Transpile options are not used in {self.__class__.__name__ }.")
-
     def update_calibrations(self, experiment_data: ExperimentData):
         """Update parameter values in the :class:`.Calibrations` instance.
 
@@ -295,42 +274,12 @@ def _transpiled_circuits(self) -> List[QuantumCircuit]:
         Returns:
             A list of transpiled circuits.
         """
-        transpiled = []
-        for circ in self.circuits():
-            circ = self._map_to_physical_qubits(circ)
+        transpiled = super()._transpiled_circuits()
+        for circ in transpiled:
             self._attach_calibrations(circ)
 
-            transpiled.append(circ)
-
         return transpiled
 
-    def _map_to_physical_qubits(self, circuit: QuantumCircuit) -> QuantumCircuit:
-        """Map program qubits to physical qubits.
-
-        Args:
-            circuit: The quantum circuit to map to device qubits.
-
-        Returns:
-            A quantum circuit that has the same number of qubits as the backend and where
-            the physical qubits of the experiment have been properly mapped.
-        """
-        initial_layout = Layout.from_intlist(list(self.physical_qubits), *circuit.qregs)
-
-        coupling_map = self._backend_data.coupling_map
-        if coupling_map is not None:
-            coupling_map = CouplingMap(self._backend_data.coupling_map)
-
-        layout = PassManager(
-            [
-                SetLayout(initial_layout),
-                FullAncillaAllocation(coupling_map),
-                EnlargeWithAncilla(),
-                ApplyLayout(),
-            ]
-        )
-
-        return StagedPassManager(["layout"], layout=layout).run(circuit)
-
     @abstractmethod
     def _attach_calibrations(self, circuit: QuantumCircuit):
         """Attach the calibrations to the quantum circuit.

qiskit_experiments/framework/base_experiment.py

@@ -13,8 +13,10 @@
 Base Experiment class.
 """
 
-from abc import ABC, abstractmethod
 import copy
+import importlib.metadata
+import logging
+from abc import ABC, abstractmethod
 from collections import OrderedDict
 from typing import Sequence, Optional, Tuple, List, Dict, Union
 
@@ -25,12 +27,16 @@
 from qiskit.providers.options import Options
 from qiskit_experiments.framework import BackendData
 from qiskit_experiments.framework.store_init_args import StoreInitArgs
+from qiskit_experiments.framework.transpilation import check_transpilation_needed, map_qubits
 from qiskit_experiments.framework.base_analysis import BaseAnalysis
 from qiskit_experiments.framework.experiment_data import ExperimentData
 from qiskit_experiments.framework.configs import ExperimentConfig
 from qiskit_experiments.database_service import Qubit
 
 
+LOGGER = logging.getLogger(__name__)
+
+
 class BaseExperiment(ABC, StoreInitArgs):
     """Abstract base class for experiments."""
 
@@ -374,8 +380,35 @@ def _transpiled_circuits(self) -> List[QuantumCircuit]:
         This function can be overridden to define custom transpilation.
         """
         transpile_opts = copy.copy(self.transpile_options.__dict__)
-        transpile_opts["initial_layout"] = list(self.physical_qubits)
-        transpiled = transpile(self.circuits(), self.backend, **transpile_opts)
+
+        virtual_circuits = self.circuits()
+        mapped_circuits = [map_qubits(c, self.physical_qubits) for c in virtual_circuits]
+
+        if "full_transpile" not in transpile_opts:
+            LOGGER.debug(
+                "Performing full transpile because base transpile options "
+                "were overwritten and full_transpile was not specified."
+            )
+            full_transpile = True
+        else:
+            full_transpile = transpile_opts.pop("full_transpile", False)
+        if not full_transpile and set(transpile_opts) - set(
+            BaseExperiment._default_transpile_options()
+        ):
+            # If an experiment specifies transpile options, it needs to go
+            # through transpile()
+            full_transpile = True
+            LOGGER.debug(
+                "Performing full transpile because non-default transpile options are specified."
+            )
+
+        if not full_transpile:
+            full_transpile = check_transpilation_needed(mapped_circuits, self.backend)
+
+        if full_transpile:
+            transpiled = transpile(mapped_circuits, self.backend, **transpile_opts)
+        else:
+            transpiled = mapped_circuits
 
         return transpiled
 
@@ -418,11 +451,39 @@ def set_experiment_options(self, **fields):
 
     @classmethod
     def _default_transpile_options(cls) -> Options:
-        """Default transpiler options for transpilation of circuits"""
+        """Default transpiler options for transpilation of circuits
+
+        Transpile Options:
+            optimization_level (int): Optimization level to pass to
+                :func:`qiskit.transpile`.
+            num_processes (int): Number of processes to use during
+                transpilation on Qiskit >= 1.0.
+            full_transpile (bool): If ``True``,
+                ``BaseExperiment._transpiled_circuits`` (called by
+                :meth:`BaseExperiment.run` if not overridden by a subclass)
+                will call :func:`qiskit.transpile` on the output of
+                :meth:`BaseExperiment.circuits` before executing the circuits.
+                If ``False``, ``BaseExperiment._transpiled_circuits`` will
+                reindex the qubits in the output of
+                :meth:`BaseExperiment.circuits` using the experiments'
+                :meth:`BaseExperiment.physical_qubits`.  Then it will check if
+                the circuit operations are all defined in the
+                :class:`qiskit.transpiler.Target` of the experiment's backend
+                or in the indiivdual circuit calibrations. If not, it will use
+                :class:`qiskit.transpiler.passes.BasisTranslator` to map the
+                circuit instructions to the backend. Additionally,
+                the :class:`qiskit.transpiler.passes.PulseGates` transpiler
+                pass will be run if the :class:`qiskit.transpiler.Target`
+                contains any custom pulse gate calibrations.
+
+        """
         # Experiment subclasses can override this method if they need
         # to set specific default transpiler options to transpile the
         # experiment circuits.
-        return Options(optimization_level=0)
+        opts = Options(optimization_level=0, full_transpile=False)
+        if importlib.metadata.version("qiskit").partition(".")[0] != "0":
+            opts["num_processes"] = 1
+        return opts
 
     @property
     def transpile_options(self) -> Options:

qiskit_experiments/framework/experiment_data.py

@@ -866,11 +866,10 @@ def _add_job_data(
                 LOG.warning("Job was cancelled before completion [Job ID: %s]", jid)
                 return jid, False
             if status == JobStatus.ERROR:
-                LOG.error(
-                    "Job data not added for errored job [Job ID: %s]\nError message: %s",
-                    jid,
-                    job.error_message(),
-                )
+                msg = f"Job data not added for errored job [Job ID: {jid}]"
+                if hasattr(job, "error_message"):
+                    msg += f"\nError message: {job.error_message()}"
+                LOG.error(msg)
                 return jid, False
             LOG.warning("Adding data from job failed [Job ID: %s]", job.job_id())
             raise ex

qiskit_experiments/framework/transpilation.py

@@ -0,0 +1,149 @@
+# This code is part of Qiskit.
+#
+# (C) Copyright IBM 2021.
+#
+# This code is licensed under the Apache License, Version 2.0. You may
+# obtain a copy of this license in the LICENSE.txt file in the root directory
+# of this source tree or at http://www.apache.org/licenses/LICENSE-2.0.
+#
+# Any modifications or derivative works of this code must retain this
+# copyright notice, and modified files need to carry a notice indicating
+# that they have been altered from the originals.
+"""
+Functions for preparing circuits for execution
+"""
+
+from __future__ import annotations
+
+from collections.abc import Sequence
+
+from qiskit import QuantumCircuit, QuantumRegister
+from qiskit.exceptions import QiskitError
+from qiskit.providers import Backend
+from qiskit.pulse.calibration_entries import CalibrationPublisher
+from qiskit.transpiler import Target
+
+
+def map_qubits(
+    circuit: QuantumCircuit,
+    physical_qubits: Sequence[int],
+    n_qubits: int | None = None,
+) -> QuantumCircuit:
+    """Generate a new version of a circuit with new qubit indices
+
+    This function iterates through the instructions of ``circuit`` and copies
+    them into a new circuit with qubit indices replaced according to the
+    entries in ``physical_qubits``. So qubit 0's instructions are applied to
+    ``physical_qubits[0]`` and qubit 1's to ``physical_qubits[1]``, etc.
+
+    This function behaves similarly to passing ``initial_layout`` to
+    :func:`qiskit.transpile` but does not use a Qiskit
+    :class:`~qiskit.transpiler.PassManager` and does not fill the circuit with
+    ancillas.
+
+    Args:
+        circuit: The :class:`~qiskit.QuantumCircuit` to re-index.
+        physical_qubits: The list of new indices for ``circuit``'s qubit indices.
+        n_qubits: Optional qubit size to use for the output circuit. If
+            ``None``, then the maximum of ``physical_qubits`` will be used.
+
+    Returns:
+        The quantum circuit with new qubit indices
+    """
+    if len(physical_qubits) != circuit.num_qubits:
+        raise QiskitError(
+            f"Circuit to map has {circuit.num_qubits} qubits, but "
+            f"{len(physical_qubits)} physical qubits specified for mapping."
+        )
+
+    # if all(p == r for p, r in zip(physical_qubits, range(circuit.num_qubits))):
+    #     # No mapping necessary
+    #     return circuit
+
+    circ_size = n_qubits if n_qubits is not None else (max(physical_qubits) + 1)
+    p_qregs = QuantumRegister(circ_size)
+    p_circ = QuantumCircuit(
+        p_qregs,
+        *circuit.cregs,
+        name=circuit.name,
+        metadata=circuit.metadata,
+        global_phase=circuit.global_phase,
+    )
+    p_circ.compose(
+        circuit,
+        qubits=physical_qubits,
+        inplace=True,
+        copy=False,
+    )
+    return p_circ
+
+
+def _has_calibration(target: Target, name: str, qubits: tuple[int, ...]) -> bool:
+    """Wrapper to work around bug in Target.has_calibration"""
+    try:
+        has_cal = target.has_calibration(name, qubits)
+    except AttributeError:
+        has_cal = False
+
+    return has_cal
+
+
+def check_transpilation_needed(
+    circuits: Sequence[QuantumCircuit],
+    backend: Backend,
+) -> bool:
+    """Test if circuits are already compatible with backend
+
+    This function checks if circuits are able to be executed on ``backend``
+    without transpilation. It loops through the circuits to check if any gate
+    instructions are not included in the backend's
+    :class:`~qiskit.transpiler.Target`.  The :class:`~qiskit.transpiler.Target`
+    is also checked for custom pulse gate calibrations for circuit's
+    instructions.  If all gates are included in the target and there are no
+    custom calibrations, the function returns ``False`` indicating that
+    transpilation is not needed.
+
+    This function returns ``True`` if the version of ``backend`` is less than
+    2.
+
+    The motivation for this function is that when no transpilation is necessary
+    it is faster to check the circuits in this way than to run
+    :func:`~qiskit.transpile` and have it do nothing.
+
+    Args:
+        circuits: The circuits to prepare for the backend.
+        backend: The backend for which the circuits should be prepared.
+
+    Returns:
+        ``True`` if transpilation is needed. Otherwise, ``False``.
+    """
+    transpilation_needed = False
+
+    if getattr(backend, "version", 0) <= 1:
+        # Fall back to transpilation for BackendV1
+        return True
+
+    target = backend.target
+
+    for circ in circuits:
+        for inst in circ.data:
+            if inst.operation.name == "barrier":
+                continue
+            qubits = tuple(circ.find_bit(q).index for q in inst.qubits)
+            if not target.instruction_supported(inst.operation.name, qubits):
+                transpilation_needed = True
+                break
+            if not circ.has_calibration_for(inst) and _has_calibration(
+                target, inst.operation.name, qubits
+            ):
+                cal = target.get_calibration(inst.operation.name, qubits, *inst.operation.params)
+                if (
+                    cal.metadata.get("publisher", CalibrationPublisher.QISKIT)
+                    != CalibrationPublisher.BACKEND_PROVIDER
+                ):
+                    transpilation_needed = True
+                    break
+        if transpilation_needed:
+            break
+
+    return transpilation_needed

qiskit_experiments/test/mock_iq_backend.py

@@ -135,10 +135,10 @@ def run(self, run_input, **options):
 
         self._compute_outcome_probabilities(run_input)
 
-        if run_input[0].num_qubits != 2:
+        if run_input[0].num_qubits != 1:
             raise DataProcessorError(f"{self.__class__.__name__} is a two qubit mock device.")
 
-        prev_outcome, state_strings = "00", self._get_state_strings(2)
+        prev_outcome, state_strings = "0", self._get_state_strings(1)
 
         # Setup the list of dicts where each dict corresponds to a circuit.
         sorted_memory = [{"memory": [], "metadata": circ.metadata} for circ in run_input]
@@ -155,7 +155,7 @@ def run(self, run_input, **options):
 
         for idx, circ in enumerate(run_input):
             counts = {}
-            for key1, key2 in zip(["00", "01", "10", "11"], ["0x0", "0x1", "0x2", "0x3"]):
+            for key1, key2 in zip(["0", "1"], ["0x0", "0x1"]):
                 counts[key1] = sorted_memory[idx]["memory"].count(key2)
             run_result = {
                 "shots": shots,
@@ -215,8 +215,8 @@ def _compute_outcome_probabilities(self, circuits: List[QuantumCircuit]):
             prob_1 = np.sin(angle / 2) ** 2
             prob_0 = 1 - prob_1
 
-            self._precomputed_probabilities[(idx, "00")] = [prob_0, prob_1, 0, 0]
-            self._precomputed_probabilities[(idx, "01")] = [prob_1, prob_0, 0, 0]
+            self._precomputed_probabilities[(idx, "0")] = [prob_0, prob_1]
+            self._precomputed_probabilities[(idx, "1")] = [prob_1, prob_0]
 
 
 class MockIQBackend(FakeOpenPulse2QV2):

test/data_processing/test_restless_experiment.py

@@ -82,7 +82,7 @@ def test_end_to_end_restless_standard_processor(self, pi_ratio):
 
         amp_exp = FineXAmplitude([0], backend)
         # standard data processor.
-        standard_processor = DataProcessor("counts", [Probability("01")])
+        standard_processor = DataProcessor("counts", [Probability("1")])
         amp_exp.analysis.set_options(data_processor=standard_processor)
         # enable a restless measurement setting.
         amp_exp.enable_restless(rep_delay=1e-6, override_processor_by_restless=False)