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.

qiskit/pulse/calibration_entries.py

@@ -16,7 +16,7 @@
 from enum import IntEnum
 from typing import Callable, List, Union, Optional, Sequence, Any
 
-from qiskit.pulse.exceptions import PulseError
+from qiskit.pulse.exceptions import PulseError, UnassignedDurationError
 from qiskit.pulse.schedule import Schedule, ScheduleBlock
 from qiskit.qobj.converters import QobjToInstructionConverter
 from qiskit.qobj.pulse_qobj import PulseQobjInstruction
@@ -64,6 +64,15 @@ def get_schedule(self, *args, **kwargs) -> Union[Schedule, ScheduleBlock]:
         """
         pass
 
+    @abstractmethod
+    def get_duration(self) -> int:
+        """Get duration of schedule.
+
+        Returns:
+            Integer value representing a duration of schedule in units of dt.
+        """
+        pass
+
 
 class ScheduleDef(CalibrationEntry):
     """In-memory Qiskit Pulse representation.
@@ -143,6 +152,12 @@ def get_schedule(self, *args, **kwargs) -> Union[Schedule, ScheduleBlock]:
                 pass
         return self._definition.assign_parameters(value_dict, inplace=False)
 
+    def get_duration(self) -> int:
+        try:
+            return self._definition.duration
+        except UnassignedDurationError:
+            return None
+
     def __eq__(self, other):
         # This delegates equality check to Schedule or ScheduleBlock.
         return self._definition == other._definition
@@ -186,6 +201,10 @@ def get_schedule(self, *args, **kwargs) -> Union[Schedule, ScheduleBlock]:
 
         return self._definition(**to_bind.arguments)
 
+    def get_duration(self) -> int:
+        # Duration is undetermined until funciton is called with full arguments.
+        return None
+
     def __eq__(self, other):
         # We cannot evaluate function equality without parsing python AST.
         # This simply compares wether they are the same object.
@@ -249,6 +268,29 @@ def get_schedule(self, *args, **kwargs) -> Union[Schedule, ScheduleBlock]:
             self._build_schedule()
         return super().get_schedule(*args, **kwargs)
 
+    def get_duration(self) -> int:
+        if self._definition:
+            return self._definition.duration
+
+        # Parse Qobj instruction and find max instruction end time
+        t_max = 0
+        for inst in self._source:
+            if inst.name == "parametric_pulse":
+                # Play instruction
+                duration = inst.parameters.get("duration", 0)
+            elif inst.name not in self._converter.get_supported_instructions():
+                # Play instruction with Waveform
+                try:
+                    # Get samples from the pulse library dictionary
+                    duration = len(self._converter._pulse_library[inst.name])
+                except KeyError:
+                    continue
+            else:
+                # Acquire, Delay instruction
+                duration = getattr(inst, "duration", 0)
+            t_max = max(t_max, inst.t0 + duration)
+        return t_max
+
     def __eq__(self, other):
         if isinstance(other, PulseQobjDef):
             # If both objects are Qobj just check Qobj equality.

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,16 @@ 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 target is None:
+            target = Target()
+        if inst_map is not None:
+            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 +78,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.instruction_supported(operation_name=node_op.name, qargs=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 +89,22 @@ 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)
+        inst_property = self.target[node_op.name][tuple(qubits)]
+        if not node_op.params:
+            return inst_property.calibration
+        try:
+            # CircuitInstruction doesn't preserve parameter name after parameter binding.
+            # Thus schedule cannot generate bind dictionary.
+            # Use CalibraionEntry to utilize inspected signature object.
+            calibration_entry = inst_property._calibration
+            return calibration_entry.get_schedule(*node_op.params)
+        except AttributeError as ex:
+            raise TranspilerError(
+                f"Calibraton for {node_op.name} of {qubits} is not a CalibraryEntry instance. "
+                f"Mapping from parameter values {node_op.params} to parameter objects "
+                f"in the schedule cannot be identified."
+            ) from ex

qiskit/transpiler/target.py

@@ -28,8 +28,10 @@
 import rustworkx as rx
 
 from qiskit.circuit.parameter import Parameter
+from qiskit.circuit.gate import Gate
+from qiskit.circuit.library.standard_gates import get_standard_gate_name_mapping
 from qiskit.pulse.instruction_schedule_map import InstructionScheduleMap
-from qiskit.pulse.calibration_entries import CalibrationEntry
+from qiskit.pulse.calibration_entries import CalibrationEntry, ScheduleDef
 from qiskit.pulse.schedule import Schedule, ScheduleBlock
 from qiskit.transpiler.coupling import CouplingMap
 from qiskit.transpiler.exceptions import TranspilerError
@@ -79,13 +81,17 @@ def __init__(
     @property
     def calibration(self):
         """The pulse representation of the instruction."""
-        if isinstance(self._calibration, CalibrationEntry):
-            return self._calibration.get_schedule()
-        return self._calibration
+        return self._calibration.get_schedule()
 
     @calibration.setter
     def calibration(self, calibration: Union[Schedule, ScheduleBlock, CalibrationEntry]):
-        self._calibration = calibration
+        if not isinstance(calibration, CalibrationEntry):
+            # Convert into entry to use signature generation mechanism for parameter assignment.
+            new_entry = ScheduleDef()
+            new_entry.define(calibration)
+        else:
+            new_entry = calibration
+        self._calibration = new_entry
 
     def __repr__(self):
         return (
@@ -416,7 +422,9 @@ def update_from_instruction_schedule_map(self, inst_map, inst_name_map=None, err
         Args:
             inst_map (InstructionScheduleMap): The instruction
             inst_name_map (dict): An optional dictionary that maps any
-                instruction name in ``inst_map`` to an instruction object
+                instruction name in ``inst_map`` to an instruction object.
+                If not provided, instruction is pulled from the standard Qiskit gates,
+                and finally custom gate instnace is created with schedule name.
             error_dict (dict): A dictionary of errors of the form::
 
                 {gate_name: {qarg: error}}
@@ -429,48 +437,53 @@ def update_from_instruction_schedule_map(self, inst_map, inst_name_map=None, err
             a when updating the ``Target`` the error value will be pulled from
             this dictionary. If one is not found in ``error_dict`` then
             ``None`` will be used.
-
-        Raises:
-            ValueError: If ``inst_map`` contains new instructions and
-                ``inst_name_map`` isn't specified
-            KeyError: If a ``inst_map`` contains a qarg for an instruction
-                that's not in the target
         """
-        for inst in inst_map.instructions:
+        for inst_name in inst_map.instructions:
             out_props = {}
-            for qarg in inst_map.qubits_with_instruction(inst):
-                sched = inst_map.get(inst, qarg)
-                val = InstructionProperties(calibration=sched)
+            for qarg in inst_map.qubits_with_instruction(inst_name):
                 try:
                     qarg = tuple(qarg)
                 except TypeError:
                     qarg = (qarg,)
-                if inst in self._gate_map:
-                    if self.dt is not None:
-                        val.duration = sched.duration * self.dt
-                    else:
-                        val.duration = None
-                    if error_dict is not None:
-                        error_inst = error_dict.get(inst)
-                        if error_inst:
-                            error = error_inst.get(qarg)
-                            val.error = error
-                        else:
-                            val.error = None
+                entry = inst_map._get_calibration_entry(inst_name, qarg)
+                val = InstructionProperties(calibration=entry)
+                if self.dt is not None:
+                    val.duration = entry.get_duration() * self.dt
+                else:
+                    val.duration = None
+                if inst_name in self._gate_map and error_dict is not None:
+                    error_inst = error_dict.get(inst_name)
+                    if error_inst:
+                        error = error_inst.get(qarg)
+                        val.error = error
                     else:
                         val.error = None
+                else:
+                    val.error = None
                 out_props[qarg] = val
-            if inst not in self._gate_map:
-                if inst_name_map is not None:
-                    self.add_instruction(inst_name_map[inst], out_props, name=inst)
+            if inst_name not in self._gate_map:
+                if inst_name_map is None:
+                    inst_name_map = get_standard_gate_name_mapping()
+                if inst_name in inst_name_map:
+                    inst_obj = inst_name_map[inst_name]
                 else:
-                    raise ValueError(
-                        "An inst_name_map kwarg must be specified to add new "
-                        "instructions from an InstructionScheduleMap"
+                    # Custom gate object, which doesn't belong to standard Qiskit gates.
+                    inst_obj = Gate(
+                        name=inst_name,
+                        num_qubits=len(qarg),
+                        params=list(map(Parameter, entry.get_signature().parameters.keys())),
                     )
+                normalized_props = {}
+                for qargs, prop in out_props.items():
+                    if len(qargs) != inst_obj.num_qubits:
+                        continue
+                    normalized_props[qargs] = prop
+                self.add_instruction(inst_obj, normalized_props, name=inst_name)
             else:
                 for qarg, prop in out_props.items():
-                    self.update_instruction_properties(inst, qarg, prop)
+                    if qarg not in self._gate_map[inst_name]:
+                        continue
+                    self.update_instruction_properties(inst_name, qarg, prop)
 
     @property
     def qargs(self):
@@ -532,8 +545,9 @@ def instruction_schedule_map(self):
         out_inst_schedule_map = InstructionScheduleMap()
         for instruction, qargs in self._gate_map.items():
             for qarg, properties in qargs.items():
-                if properties is not None and properties.calibration is not None:
-                    out_inst_schedule_map.add(instruction, qarg, properties.calibration)
+                if properties is not None and properties._calibration is not None:
+                    # Do not parse schedule. Build instmap with CalibrationEntry instances.
+                    out_inst_schedule_map._add(instruction, qarg, properties._calibration)
         self._instruction_schedule_map = out_inst_schedule_map
         return out_inst_schedule_map
 

releasenotes/notes/update-pulse-gate-pass-for-target-ebfb0ec9571f058e.yaml

@@ -0,0 +1,23 @@
+---
+features:
+  - |
+    A new method :meth:`.CalibrationEntry.get_duration` has been added to
+    calibration entries. This method returns duration of calibration
+    in units of dt when available, otherwise returns None.
+upgrade:
+  - |
+    :meth:`.Target.update_from_instruction_schedule_map` no longer raises
+    KeyError nor ValueError when qubits are missing in the target instruction
+    or inst_name_map is not provided for undefined instruction.
+    In the former case, it just ignores the inst map definition for undefined qubits.
+    In the latter case, gate mapping is pulled from the standard Qiskit gates
+    and finally custom :class:`.Gate` object is defined from the schedule name.
+  - |
+    :class:`PulseGates` transpiler pass has been upgraded to respect
+    ``inst_map`` when ``target`` is also provided, rather than ignoring it.
+    The pass now uses the target data to find custom calibration, but it
+    copies instruction schedule map into it before transforming the DAG circuit.
+fixes:
+  - |
+    A bug that custom gates in the instruction schedule map is ignored
+    when transpiling a circuit with V2 backend has been fixed.

test/python/pulse/test_calibration_entries.py

@@ -40,10 +40,12 @@ class TestSchedule(QiskitTestCase):
 
     def test_add_schedule(self):
         """Basic test pulse Schedule format."""
+        ref_duration = 10
+
         program = Schedule()
         program.insert(
             0,
-            Play(Constant(duration=10, amp=0.1, angle=0.0), DriveChannel(0)),
+            Play(Constant(duration=ref_duration, amp=0.1, angle=0.0), DriveChannel(0)),
             inplace=True,
         )
 
@@ -58,11 +60,15 @@ def test_add_schedule(self):
         schedule_ref = program
         self.assertEqual(schedule_to_test, schedule_ref)
 
+        self.assertEqual(entry.get_duration(), ref_duration)
+
     def test_add_block(self):
         """Basic test pulse Schedule format."""
+        ref_duration = 10
+
         program = ScheduleBlock()
         program.append(
-            Play(Constant(duration=10, amp=0.1, angle=0.0), DriveChannel(0)),
+            Play(Constant(duration=ref_duration, amp=0.1, angle=0.0), DriveChannel(0)),
             inplace=True,
         )
 
@@ -77,6 +83,8 @@ def test_add_block(self):
         schedule_ref = program
         self.assertEqual(schedule_to_test, schedule_ref)
 
+        self.assertEqual(entry.get_duration(), ref_duration)
+
     def test_parameterized_schedule(self):
         """Test adding and managing parameterized schedule."""
         param1 = Parameter("P1")
@@ -99,6 +107,9 @@ def test_parameterized_schedule(self):
         schedule_ref = program.assign_parameters({param1: 10, param2: 0.1}, inplace=False)
         self.assertEqual(schedule_to_test, schedule_ref)
 
+        # Undetermined
+        self.assertEqual(entry.get_duration(), None)
+
     def test_parameterized_schedule_with_user_args(self):
         """Test adding schedule with user signature.
 
@@ -207,6 +218,8 @@ def factory():
         schedule_ref = program
         self.assertEqual(schedule_to_test, schedule_ref)
 
+        self.assertEqual(entry.get_duration(), None)
+
     def test_add_callable_with_argument(self):
         """Basic test callable format."""
 
@@ -308,6 +321,9 @@ def test_add_qobj(self):
         entry = PulseQobjDef(converter=self.converter, name="my_gate")
         entry.define(serialized_program)
 
+        self.assertEqual(entry.get_duration(), 25)
+        self.assertIsNone(entry._definition)  # Check if schedule is not parsed
+
         signature_to_test = list(entry.get_signature().parameters.keys())
         signature_ref = []
         self.assertListEqual(signature_to_test, signature_ref)