Update pulse gate transpiler pass to use target. (Qiskit#9587)

* Update PulseGate pass to use Target internally.

When inst_map is provided, it copies schedules there into target instance. This fixes a bug that custom schedules in the inst_map are ignored when transpiling circuit with V2 backend. To support this behavior, internal machinery of Target is updated so that a target instance can update itself only with inst_map without raising any error. Also InstructionProperties.calibration now only stores CalibrationEntry instances. When Schedule or ScheduleBlock are provided as a calibration, it converts schedule into CalibrationEntry instance.

* Remove fix note

* Remove get_duration

* Update the logic to get instruction object

* Update target immediately when inst map is available

* Update tests

* Fix edge case.

IBM backend still provide ugate calibrations in CmdDef and they are loaded in the instmap. If we update target with the instmap, these gates are accidentally registered in the target, and they may be used in the following 1q decomposition. To prevent this, update_from_instruction_schedule_map method is updated.

* cleanup release note

* Minor review suggestions

* More strict gate uniformity check when create from schedules.

* Added note for calibration behavior

* More documentation for CalibrationEntry

* Add logic to prevent unintentional backend mutation with instmap.

* fix lint

qiskit/compiler/transpiler.py

@@ -13,6 +13,7 @@
 # pylint: disable=invalid-sequence-index
 
 """Circuit transpile function"""
+import copy
 import io
 from itertools import cycle
 import logging
@@ -665,6 +666,10 @@ def _parse_transpile_args(
     callback = _parse_callback(callback, num_circuits)
     durations = _parse_instruction_durations(backend, instruction_durations, dt, circuits)
     timing_constraints = _parse_timing_constraints(backend, timing_constraints, num_circuits)
+    if inst_map is not None and inst_map.has_custom_gate() and target is not None:
+        # Do not mutate backend target
+        target = copy.deepcopy(target)
+        target.update_from_instruction_schedule_map(inst_map)
     if scheduling_method and any(d is None for d in durations):
         raise TranspilerError(
             "Transpiling a circuit with a scheduling method"

qiskit/providers/models/pulsedefaults.py

@@ -205,7 +205,7 @@ def __init__(
 
         for inst in cmd_def:
             entry = PulseQobjDef(converter=self.converter, name=inst.name)
-            entry.define(inst.sequence)
+            entry.define(inst.sequence, user_provided=False)
             self.instruction_schedule_map._add(
                 instruction_name=inst.name,
                 qubits=tuple(inst.qubits),

qiskit/pulse/calibration_entries.py

@@ -31,14 +31,41 @@ class CalibrationPublisher(IntEnum):
 
 
 class CalibrationEntry(metaclass=ABCMeta):
-    """A metaclass of a calibration entry."""
+    """A metaclass of a calibration entry.
+
+    This class defines a standard model of Qiskit pulse program that is
+    agnostic to the underlying in-memory representation.
+
+    This entry distinguishes whether this is provided by end-users or a backend
+    by :attr:`.user_provided` attribute which may be provided when
+    the actual calibration data is provided to the entry with by :meth:`define`.
+
+    Note that a custom entry provided by an end-user may appear in the wire-format
+    as an inline calibration, e.g. :code:`defcal` of the QASM3,
+    that may update the backend instruction set architecture for execution.
+
+    .. note::
+
+        This and built-in subclasses are expected to be private without stable user-facing API.
+        The purpose of this class is to wrap different
+        in-memory pulse program representations in Qiskit, so that it can provide
+        the standard data model and API which are primarily used by the transpiler ecosystem.
+        It is assumed that end-users will never directly instantiate this class,
+        but :class:`.Target` or :class:`.InstructionScheduleMap` internally use this data model
+        to avoid implementing a complicated branching logic to
+        manage different calibration data formats.
+
+    """
 
     @abstractmethod
-    def define(self, definition: Any):
+    def define(self, definition: Any, user_provided: bool):
         """Attach definition to the calibration entry.
 
         Args:
             definition: Definition of this entry.
+            user_provided: If this entry is defined by user.
+                If the flag is set, this calibration may appear in the wire format
+                as an inline calibration, to override the backend instruction set architecture.
         """
         pass
 
@@ -55,6 +82,10 @@ def get_signature(self) -> inspect.Signature:
     def get_schedule(self, *args, **kwargs) -> Union[Schedule, ScheduleBlock]:
         """Generate schedule from entry definition.
 
+        If the pulse program is templated with :class:`.Parameter` objects,
+        you can provide corresponding parameter values for this method
+        to get a particular pulse program with assigned parameters.
+
         Args:
             args: Command parameters.
             kwargs: Command keyword parameters.
@@ -64,13 +95,23 @@ def get_schedule(self, *args, **kwargs) -> Union[Schedule, ScheduleBlock]:
         """
         pass
 
+    @property
+    @abstractmethod
+    def user_provided(self) -> bool:
+        """Return if this entry is user defined."""
+        pass
+
 
 class ScheduleDef(CalibrationEntry):
     """In-memory Qiskit Pulse representation.
 
     A pulse schedule must provide signature with the .parameters attribute.
     This entry can be parameterized by a Qiskit Parameter object.
     The .get_schedule method returns a parameter-assigned pulse program.
+
+    .. see_also::
+        :class:`.CalibrationEntry` for the purpose of this class.
+
     """
 
     def __init__(self, arguments: Optional[Sequence[str]] = None):
@@ -90,6 +131,11 @@ def __init__(self, arguments: Optional[Sequence[str]] = None):
 
         self._definition = None
         self._signature = None
+        self._user_provided = None
+
+    @property
+    def user_provided(self) -> bool:
+        return self._user_provided
 
     def _parse_argument(self):
         """Generate signature from program and user provided argument names."""
@@ -120,35 +166,48 @@ def _parse_argument(self):
         )
         self._signature = signature
 
-    def define(self, definition: Union[Schedule, ScheduleBlock]):
+    def define(
+        self,
+        definition: Union[Schedule, ScheduleBlock],
+        user_provided: bool = True,
+    ):
         self._definition = definition
-        # add metadata
-        if "publisher" not in definition.metadata:
-            definition.metadata["publisher"] = CalibrationPublisher.QISKIT
         self._parse_argument()
+        self._user_provided = user_provided
 
     def get_signature(self) -> inspect.Signature:
         return self._signature
 
     def get_schedule(self, *args, **kwargs) -> Union[Schedule, ScheduleBlock]:
         if not args and not kwargs:
-            return self._definition
-        try:
-            to_bind = self.get_signature().bind_partial(*args, **kwargs)
-        except TypeError as ex:
-            raise PulseError("Assigned parameter doesn't match with schedule parameters.") from ex
-        value_dict = {}
-        for param in self._definition.parameters:
-            # Schedule allows partial bind. This results in parameterized Schedule.
+            out = self._definition
+        else:
             try:
-                value_dict[param] = to_bind.arguments[param.name]
-            except KeyError:
-                pass
-        return self._definition.assign_parameters(value_dict, inplace=False)
+                to_bind = self.get_signature().bind_partial(*args, **kwargs)
+            except TypeError as ex:
+                raise PulseError(
+                    "Assigned parameter doesn't match with schedule parameters."
+                ) from ex
+            value_dict = {}
+            for param in self._definition.parameters:
+                # Schedule allows partial bind. This results in parameterized Schedule.
+                try:
+                    value_dict[param] = to_bind.arguments[param.name]
+                except KeyError:
+                    pass
+            out = self._definition.assign_parameters(value_dict, inplace=False)
+        if "publisher" not in out.metadata:
+            if self.user_provided:
+                out.metadata["publisher"] = CalibrationPublisher.QISKIT
+            else:
+                out.metadata["publisher"] = CalibrationPublisher.BACKEND_PROVIDER
+        return out
 
     def __eq__(self, other):
         # This delegates equality check to Schedule or ScheduleBlock.
-        return self._definition == other._definition
+        if hasattr(other, "_definition"):
+            return self._definition == other._definition
+        return False
 
     def __str__(self):
         out = f"Schedule {self._definition.name}"
@@ -165,16 +224,30 @@ class CallableDef(CalibrationEntry):
     provide the signature. This entry is parameterized by the function signature
     and .get_schedule method returns a non-parameterized pulse program
     by consuming the provided arguments and keyword arguments.
+
+    .. see_also::
+        :class:`.CalibrationEntry` for the purpose of this class.
+
     """
 
     def __init__(self):
         """Define an empty entry."""
         self._definition = None
         self._signature = None
+        self._user_provided = None
 
-    def define(self, definition: Callable):
+    @property
+    def user_provided(self) -> bool:
+        return self._user_provided
+
+    def define(
+        self,
+        definition: Callable,
+        user_provided: bool = True,
+    ):
         self._definition = definition
         self._signature = inspect.signature(definition)
+        self._user_provided = user_provided
 
     def get_signature(self) -> inspect.Signature:
         return self._signature
@@ -186,17 +259,20 @@ def get_schedule(self, *args, **kwargs) -> Union[Schedule, ScheduleBlock]:
             to_bind.apply_defaults()
         except TypeError as ex:
             raise PulseError("Assigned parameter doesn't match with function signature.") from ex
-
-        schedule = self._definition(**to_bind.arguments)
-        # add metadata
-        if "publisher" not in schedule.metadata:
-            schedule.metadata["publisher"] = CalibrationPublisher.QISKIT
-        return schedule
+        out = self._definition(**to_bind.arguments)
+        if "publisher" not in out.metadata:
+            if self.user_provided:
+                out.metadata["publisher"] = CalibrationPublisher.QISKIT
+            else:
+                out.metadata["publisher"] = CalibrationPublisher.BACKEND_PROVIDER
+        return out
 
     def __eq__(self, other):
         # We cannot evaluate function equality without parsing python AST.
-        # This simply compares wether they are the same object.
-        return self._definition is other._definition
+        # This simply compares weather they are the same object.
+        if hasattr(other, "_definition"):
+            return self._definition == other._definition
+        return False
 
     def __str__(self):
         params_str = ", ".join(self.get_signature().parameters.keys())
@@ -210,6 +286,10 @@ class PulseQobjDef(ScheduleDef):
     the provided qobj converter. Because the Qobj JSON doesn't provide signature,
     conversion process occurs when the signature is requested for the first time
     and the generated pulse program is cached for performance.
+
+    .. see_also::
+        :class:`.CalibrationEntry` for the purpose of this class.
+
     """
 
     def __init__(
@@ -237,14 +317,17 @@ def _build_schedule(self):
         for qobj_inst in self._source:
             for qiskit_inst in self._converter._get_sequences(qobj_inst):
                 schedule.insert(qobj_inst.t0, qiskit_inst, inplace=True)
-        schedule.metadata["publisher"] = CalibrationPublisher.BACKEND_PROVIDER
-
         self._definition = schedule
         self._parse_argument()
 
-    def define(self, definition: List[PulseQobjInstruction]):
+    def define(
+        self,
+        definition: List[PulseQobjInstruction],
+        user_provided: bool = False,
+    ):
         # This doesn't generate signature immediately, because of lazy schedule build.
         self._source = definition
+        self._user_provided = user_provided
 
     def get_signature(self) -> inspect.Signature:
         if self._definition is None:
@@ -261,9 +344,11 @@ def __eq__(self, other):
             # If both objects are Qobj just check Qobj equality.
             return self._source == other._source
         if isinstance(other, ScheduleDef) and self._definition is None:
-            # To compare with other scheudle def, this also generates schedule object from qobj.
+            # To compare with other schedule def, this also generates schedule object from qobj.
             self._build_schedule()
-        return self._definition == other._definition
+        if hasattr(other, "_definition"):
+            return self._definition == other._definition
+        return False
 
     def __str__(self):
         if self._definition is None:

qiskit/pulse/instruction_schedule_map.py

@@ -39,8 +39,8 @@
     CalibrationEntry,
     ScheduleDef,
     CallableDef,
-    PulseQobjDef,
     # for backward compatibility
+    PulseQobjDef,
     CalibrationPublisher,
 )
 from qiskit.pulse.exceptions import PulseError
@@ -77,7 +77,7 @@ def has_custom_gate(self) -> bool:
         """Return ``True`` if the map has user provided instruction."""
         for qubit_inst in self._map.values():
             for entry in qubit_inst.values():
-                if not isinstance(entry, PulseQobjDef):
+                if entry.user_provided:
                     return True
         return False
 
@@ -264,7 +264,7 @@ def add(
                 "Supplied schedule must be one of the Schedule, ScheduleBlock or a "
                 "callable that outputs a schedule."
             )
-        entry.define(schedule)
+        entry.define(schedule, user_provided=True)
         self._add(instruction, qubits, entry)
 
     def _add(

qiskit/transpiler/passes/calibration/pulse_gate.py

@@ -15,12 +15,10 @@
 from typing import List, Union
 
 from qiskit.circuit import Instruction as CircuitInst
-from qiskit.pulse import (
-    Schedule,
-    ScheduleBlock,
-)
+from qiskit.pulse import Schedule, ScheduleBlock
 from qiskit.pulse.instruction_schedule_map import InstructionScheduleMap
 from qiskit.transpiler.target import Target
+from qiskit.transpiler.exceptions import TranspilerError
 
 from .base_builder import CalibrationBuilder
 
@@ -59,13 +57,18 @@ def __init__(
         Args:
             inst_map: Instruction schedule map that user may override.
             target: The :class:`~.Target` representing the target backend, if both
-                ``inst_map`` and this are specified then this argument will take
-                precedence and ``inst_map`` will be ignored.
+                ``inst_map`` and this are specified then it updates instructions
+                in the ``target`` with ``inst_map``.
         """
         super().__init__()
-        self.inst_map = inst_map
-        if target:
-            self.inst_map = target.instruction_schedule_map()
+
+        if inst_map is None and target is None:
+            raise TranspilerError("inst_map and target cannot be None simulataneously.")
+
+        if target is None:
+            target = Target()
+            target.update_from_instruction_schedule_map(inst_map)
+        self.target = target
 
     def supported(self, node_op: CircuitInst, qubits: List) -> bool:
         """Determine if a given node supports the calibration.
@@ -77,7 +80,7 @@ def supported(self, node_op: CircuitInst, qubits: List) -> bool:
         Returns:
             Return ``True`` is calibration can be provided.
         """
-        return self.inst_map.has(instruction=node_op.name, qubits=qubits)
+        return self.target.has_calibration(node_op.name, tuple(qubits))
 
     def get_calibration(self, node_op: CircuitInst, qubits: List) -> Union[Schedule, ScheduleBlock]:
         """Gets the calibrated schedule for the given instruction and qubits.
@@ -88,5 +91,8 @@ def get_calibration(self, node_op: CircuitInst, qubits: List) -> Union[Schedule,
 
         Returns:
             Return Schedule of target gate instruction.
+
+        Raises:
+            TranspilerError: When node is parameterized and calibration is raw schedule object.
         """
-        return self.inst_map.get(node_op.name, qubits, *node_op.params)
+        return self.target.get_calibration(node_op.name, tuple(qubits), *node_op.params)