Add pulse gate pass

qiskit/compiler/transpiler.py

@@ -25,21 +25,23 @@
 from qiskit.providers.backend import Backend
 from qiskit.providers.models import BackendProperties
 from qiskit.providers.models.backendproperties import Gate
-from qiskit.pulse import Schedule
+from qiskit.pulse import Schedule, InstructionScheduleMap
+from qiskit.pulse.instruction_schedule_map import CalibrationPublisher
+from qiskit.pulse.exceptions import PulseError
 from qiskit.tools.parallel import parallel_map
 from qiskit.transpiler import Layout, CouplingMap, PropertySet, PassManager
 from qiskit.transpiler.basepasses import BasePass
 from qiskit.transpiler.exceptions import TranspilerError
 from qiskit.transpiler.instruction_durations import InstructionDurations, InstructionDurationsType
 from qiskit.transpiler.passes import ApplyLayout
 from qiskit.transpiler.passmanager_config import PassManagerConfig
-from qiskit.transpiler.timing_constraints import TimingConstraints
 from qiskit.transpiler.preset_passmanagers import (
     level_0_pass_manager,
     level_1_pass_manager,
     level_2_pass_manager,
     level_3_pass_manager,
 )
+from qiskit.transpiler.timing_constraints import TimingConstraints
 
 logger = logging.getLogger(__name__)
 
@@ -48,6 +50,7 @@ def transpile(
     circuits: Union[QuantumCircuit, List[QuantumCircuit]],
     backend: Optional[Union[Backend, BaseBackend]] = None,
     basis_gates: Optional[List[str]] = None,
+    inst_map: Optional[List[InstructionScheduleMap]] = None,
     coupling_map: Optional[Union[CouplingMap, List[List[int]]]] = None,
     backend_properties: Optional[BackendProperties] = None,
     initial_layout: Optional[Union[Layout, Dict, List]] = None,
@@ -85,6 +88,12 @@ def transpile(
                 circuit may be run on any backend as long as it is compatible.
         basis_gates: List of basis gate names to unroll to
             (e.g: ``['u1', 'u2', 'u3', 'cx']``). If ``None``, do not unroll.
+        inst_map: Mapping of unrolled gates to pulse schedules. If this is not provided,
+            transpiler tries to get from the backend. If any user defined calibration
+            is found in the map and this is used in a circuit, transpiler attaches
+            the custom gate definition to the circuit. This enables one to flexibly
+            override the low-level instruction implementation. This feature is available
+            iff the backend supports the pulse gate experiment.
         coupling_map: Coupling map (perhaps custom) to target in mapping.
             Multiple formats are supported:
 
@@ -270,6 +279,7 @@ def callback_func(**kwargs):
         circuits,
         backend,
         basis_gates,
+        inst_map,
         coupling_map,
         backend_properties,
         initial_layout,
@@ -451,6 +461,7 @@ def _parse_transpile_args(
     circuits,
     backend,
     basis_gates,
+    inst_map,
     coupling_map,
     backend_properties,
     initial_layout,
@@ -488,6 +499,7 @@ def _parse_transpile_args(
     num_circuits = len(circuits)
 
     basis_gates = _parse_basis_gates(basis_gates, backend, circuits)
+    inst_map = _parse_inst_map(inst_map, backend, num_circuits)
     faulty_qubits_map = _parse_faulty_qubits_map(backend, num_circuits)
     coupling_map = _parse_coupling_map(coupling_map, backend, num_circuits)
     backend_properties = _parse_backend_properties(backend_properties, backend, num_circuits)
@@ -513,6 +525,7 @@ def _parse_transpile_args(
     list_transpile_args = []
     for args in zip(
         basis_gates,
+        inst_map,
         coupling_map,
         backend_properties,
         initial_layout,
@@ -533,23 +546,24 @@ def _parse_transpile_args(
         transpile_args = {
             "pass_manager_config": PassManagerConfig(
                 basis_gates=args[0],
-                coupling_map=args[1],
-                backend_properties=args[2],
-                initial_layout=args[3],
-                layout_method=args[4],
-                routing_method=args[5],
-                translation_method=args[6],
-                scheduling_method=args[7],
-                instruction_durations=args[8],
-                approximation_degree=args[9],
-                timing_constraints=args[10],
-                seed_transpiler=args[11],
+                inst_map=args[1],
+                coupling_map=args[2],
+                backend_properties=args[3],
+                initial_layout=args[4],
+                layout_method=args[5],
+                routing_method=args[6],
+                translation_method=args[7],
+                scheduling_method=args[8],
+                instruction_durations=args[9],
+                approximation_degree=args[10],
+                timing_constraints=args[11],
+                seed_transpiler=args[12],
             ),
-            "optimization_level": args[12],
-            "output_name": args[13],
-            "callback": args[14],
-            "backend_num_qubits": args[15],
-            "faulty_qubits_map": args[16],
+            "optimization_level": args[13],
+            "output_name": args[14],
+            "callback": args[15],
+            "backend_num_qubits": args[16],
+            "faulty_qubits_map": args[17],
         }
         list_transpile_args.append(transpile_args)
 
@@ -603,6 +617,19 @@ def _parse_basis_gates(basis_gates, backend, circuits):
     return basis_gates
 
 
+def _parse_inst_map(inst_map, backend, num_circuits):
+    # try getting inst_map from user, else backend
+    if inst_map is None:
+        if getattr(backend, "defaults", None):
+            inst_map = getattr(backend.defaults(), "instruction_schedule_map", None)
+
+    # inst_maps could be None, or single entry
+    if inst_map is None or isinstance(inst_map, InstructionScheduleMap):
+        inst_map = [inst_map] * num_circuits
+
+    return inst_map
+
+
 def _parse_coupling_map(coupling_map, backend, num_circuits):
     # try getting coupling_map from user, else backend
     if coupling_map is None:

qiskit/providers/models/pulsedefaults.py

@@ -15,7 +15,7 @@
 import copy
 from typing import Any, Dict, List
 
-from qiskit.pulse.instruction_schedule_map import InstructionScheduleMap
+from qiskit.pulse.instruction_schedule_map import InstructionScheduleMap, CalibrationPublisher
 from qiskit.pulse.schedule import Schedule
 from qiskit.qobj import PulseLibraryItem, PulseQobjInstruction
 from qiskit.qobj.converters import QobjToInstructionConverter
@@ -205,6 +205,7 @@ def __init__(
         for inst in cmd_def:
             pulse_insts = [self.converter(inst) for inst in inst.sequence]
             schedule = Schedule(*pulse_insts, name=inst.name)
+            schedule.metadata["publisher"] = CalibrationPublisher.BackendProvider
             self.instruction_schedule_map.add(inst.name, inst.qubits, schedule)
 
         if meas_kernel is not None:

qiskit/pulse/instruction_schedule_map.py

@@ -29,6 +29,7 @@
 import inspect
 import warnings
 from collections import defaultdict
+from enum import IntEnum
 from typing import Callable, Iterable, List, Tuple, Union, Optional, NamedTuple
 
 from qiskit.circuit.instruction import Instruction
@@ -42,6 +43,14 @@
 )
 
 
+class CalibrationPublisher(IntEnum):
+    """Defines who defined schedule entry."""
+
+    BackendProvider = 0
+    Qiskit = 1
+    ExperimentService = 2
+
+
 class InstructionScheduleMap:
     """Mapping from :py:class:`~qiskit.circuit.QuantumCircuit`
     :py:class:`qiskit.circuit.Instruction` names and qubits to
@@ -301,6 +310,10 @@ def add(
                 "callable that outputs a schedule."
             )
 
+        # add metadata
+        if "publisher" not in schedule.metadata:
+            schedule.metadata["publisher"] = CalibrationPublisher.Qiskit
+
         self._map[instruction][qubits] = Generator(schedule, signature)
         self._qubit_instructions[qubits].add(instruction)
 

qiskit/transpiler/passes/__init__.py

@@ -76,6 +76,14 @@
    CrosstalkAdaptiveSchedule
    TemplateOptimization
 
+Calibration
+=============
+
+.. autosummary::
+   :toctree: ../stubs/
+
+    PulseGates
+
 Scheduling
 =============
 
@@ -187,6 +195,9 @@
 # synthesis
 from .synthesis import UnitarySynthesis
 
+# calibration
+from .calibration import PulseGates
+
 # circuit scheduling
 from .scheduling import TimeUnitConversion
 from .scheduling import ALAPSchedule

qiskit/transpiler/passes/calibration/__init__.py

@@ -0,0 +1,15 @@
+# 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.
+
+"""Module containing transpiler calibration passes."""
+
+from .pulse_gates import PulseGates

qiskit/transpiler/passes/calibration/pulse_gates.py

@@ -0,0 +1,77 @@
+# 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.
+
+"""Add user provided calibrations."""
+
+from qiskit.dagcircuit import DAGCircuit
+from qiskit.pulse.instruction_schedule_map import InstructionScheduleMap, CalibrationPublisher
+from qiskit.transpiler.basepasses import TransformationPass
+
+
+class PulseGates(TransformationPass):
+    """Pulse gate adding pass.
+
+    This pass adds gate calibrations to a quantum circuit.
+    In QASM3 [1], the gate calibration can be encapsulated as ``defcal`` entry that declares
+    the hardware implementation of ``gate`` in specified grammar, such as OpenPulse.
+
+    This pass checks each DAG circuit node and acquires a corresponding schedule from
+    the instruction schedule map object provided by the target backend.
+    This mapping object returns a schedule with "publisher" metadata which is an integer Enum
+    value representing who created the gate schedule.
+    If the gate schedule is provided by a client, this pass attaches this to
+    the DAG circuit as a ``defcal`` entry.
+
+    This pass allows users to easily override quantum circuit with custom gate definition
+    without directly dealing with those schedules.
+
+    References
+        * [1] OpenQASM 3: A broader and deeper quantum assembly language
+          https://arxiv.org/abs/2104.14722
+    """
+
+    def __init__(
+            self,
+            inst_map: InstructionScheduleMap,
+    ):
+        """Create new pass.
+
+        Args:
+            inst_map: Instruction schedule map that user may override.
+        """
+        super().__init__()
+        self.inst_map = inst_map
+
+    def run(self, dag: DAGCircuit):
+        """Run the TimeUnitAnalysis pass on `dag`.
+
+        Args:
+            dag (DAGCircuit): DAG to attach calibrations.
+
+        Returns:
+            DAGCircuit: DAG with calibration definitions.
+        """
+        already_defined = set()
+        for node in dag.gate_nodes():
+            qubits = tuple(dag.qubits.index(q) for q in node.qargs)
+            key = [node.op.name, qubits, *node.op.params]
+            # check if calibration is already defined, or already defined
+            if not dag.has_calibration_for(node) and key not in already_defined:
+                sched = self.inst_map.get(*key)
+                publisher = sched.metadata.get("publisher", CalibrationPublisher.Qiskit)
+                if publisher != CalibrationPublisher.BackendProvider:
+                    # likely user provided calibration, add schedule
+                    dag.add_calibration(gate=node.op, qubits=qubits, schedule=sched)
+                else:
+                    # to avoid next check, this calibration is backend default
+                    already_defined.add(key)
+        return dag

qiskit/transpiler/passmanager_config.py

@@ -20,6 +20,7 @@ def __init__(
         self,
         initial_layout=None,
         basis_gates=None,
+        inst_map=None,
         coupling_map=None,
         layout_method=None,
         routing_method=None,
@@ -37,6 +38,7 @@ def __init__(
             initial_layout (Layout): Initial position of virtual qubits on
                 physical qubits.
             basis_gates (list): List of basis gate names to unroll to.
+            inst_map (InstructionScheduleMap): Mapping object that maps gate to schedule.
             coupling_map (CouplingMap): Directed graph represented a coupling
                 map.
             layout_method (str): the pass to use for choosing initial qubit
@@ -59,6 +61,7 @@ def __init__(
         """
         self.initial_layout = initial_layout
         self.basis_gates = basis_gates
+        self.inst_map = inst_map
         self.coupling_map = coupling_map
         self.layout_method = layout_method
         self.routing_method = routing_method

qiskit/transpiler/preset_passmanagers/level0.py

@@ -47,6 +47,7 @@
 from qiskit.transpiler.passes import ASAPSchedule
 from qiskit.transpiler.passes import AlignMeasures
 from qiskit.transpiler.passes import ValidatePulseGates
+from qiskit.transpiler.passes import PulseGates
 from qiskit.transpiler.passes import Error
 
 from qiskit.transpiler import TranspilerError
@@ -76,6 +77,7 @@ def level_0_pass_manager(pass_manager_config: PassManagerConfig) -> PassManager:
         TranspilerError: if the passmanager config is invalid.
     """
     basis_gates = pass_manager_config.basis_gates
+    inst_map = pass_manager_config.inst_map
     coupling_map = pass_manager_config.coupling_map
     initial_layout = pass_manager_config.initial_layout
     layout_method = pass_manager_config.layout_method or "trivial"
@@ -162,7 +164,10 @@ def _direction_condition(property_set):
 
     _direction = [GateDirection(coupling_map)]
 
-    # 7. Unify all durations (either SI, or convert to dt if known)
+    # 7. Add calibrations for pulse gates
+    _calibrations = PulseGates(inst_map=inst_map)
+
+    # 8. Unify all durations (either SI, or convert to dt if known)
     # Schedule the circuit only when scheduling_method is supplied
     _scheduling = [TimeUnitConversion(instruction_durations)]
     if scheduling_method:
@@ -173,7 +178,7 @@ def _direction_condition(property_set):
         else:
             raise TranspilerError("Invalid scheduling method %s." % scheduling_method)
 
-    # 8. Call measure alignment. Should come after scheduling.
+    # 9. Call measure alignment. Should come after scheduling.
     _alignments = [
         ValidatePulseGates(
             granularity=timing_constraints.granularity, min_length=timing_constraints.min_length
@@ -195,6 +200,8 @@ def _direction_condition(property_set):
         pm0.append(_direction_check)
         pm0.append(_direction, condition=_direction_condition)
         pm0.append(_unroll)
+    if inst_map and inst_map.instructions:
+        pm0.append(_calibrations)
     pm0.append(_scheduling)
     pm0.append(_alignments)
     return pm0