From 90a5932dc5ba263ca2200a9f1f5d5412683b9590 Mon Sep 17 00:00:00 2001
From: Naoki Kanazawa <nkanazawa1989@gmail.com>
Date: Thu, 22 Sep 2022 12:50:18 +0900
Subject: [PATCH 01/10] Update instmap to support lazy qobj conversion

---
 qiskit/providers/models/pulsedefaults.py    |  22 +-
 qiskit/pulse/instruction_schedule_map.py    | 382 ++++++++++++++++----
 qiskit/qobj/converters/pulse_instruction.py |  34 +-
 3 files changed, 355 insertions(+), 83 deletions(-)

diff --git a/qiskit/providers/models/pulsedefaults.py b/qiskit/providers/models/pulsedefaults.py
index e5d73f80ccfa..8cb5b44a6f9c 100644
--- a/qiskit/providers/models/pulsedefaults.py
+++ b/qiskit/providers/models/pulsedefaults.py
@@ -15,8 +15,7 @@
 import copy
 from typing import Any, Dict, List
 
-from qiskit.pulse.instruction_schedule_map import InstructionScheduleMap, CalibrationPublisher
-from qiskit.pulse.schedule import Schedule
+from qiskit.pulse.instruction_schedule_map import InstructionScheduleMap, PulseQobjDef
 from qiskit.qobj import PulseLibraryItem, PulseQobjInstruction
 from qiskit.qobj.converters import QobjToInstructionConverter
 
@@ -108,7 +107,7 @@ def __init__(self, name: str, qubits=None, sequence=None, **kwargs):
         Args:
             name (str): The name of the command
             qubits: The qubits for the command
-            sequence (PulseQobjInstruction): The sequence for the Command
+            sequence (List[PulseQobjInstruction]): The sequence for the Command
             kwargs: Optional additional fields
         """
         self._data = {}
@@ -200,13 +199,16 @@ def __init__(
         self.pulse_library = pulse_library
         self.cmd_def = cmd_def
         self.instruction_schedule_map = InstructionScheduleMap()
-
         self.converter = QobjToInstructionConverter(pulse_library)
+
         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.BACKEND_PROVIDER
-            self.instruction_schedule_map.add(inst.name, inst.qubits, schedule)
+            entry = PulseQobjDef(
+                name=inst.name,
+                qubits=inst.qubits,
+                converter=self.converter,
+            )
+            entry.define(inst.sequence)
+            self.instruction_schedule_map._add(entry)
 
         if meas_kernel is not None:
             self.meas_kernel = meas_kernel
@@ -267,7 +269,9 @@ def from_dict(cls, data):
         Returns:
             PulseDefaults: The PulseDefaults from the input dictionary.
         """
-        in_data = copy.copy(data)
+        # Pulse defaults data is nested dictionary.
+        # Some elements of the inner dictionaries are poped, and thus do deepcopy here.
+        in_data = copy.deepcopy(data)
         in_data["pulse_library"] = [
             PulseLibraryItem.from_dict(x) for x in in_data.pop("pulse_library")
         ]
diff --git a/qiskit/pulse/instruction_schedule_map.py b/qiskit/pulse/instruction_schedule_map.py
index 3ddea95f4ea5..f0a23159fdae 100644
--- a/qiskit/pulse/instruction_schedule_map.py
+++ b/qiskit/pulse/instruction_schedule_map.py
@@ -26,13 +26,16 @@
     inst_map = backend.defaults().instruction_schedule_map
 
 """
+from abc import ABCMeta, abstractmethod
 import inspect
 import functools
 import warnings
 from collections import defaultdict
 from enum import IntEnum
-from typing import Callable, Iterable, List, Tuple, Union, Optional, NamedTuple
+from typing import Callable, Iterable, List, Tuple, Union, Optional, NamedTuple, Sequence, Any
 
+from qiskit.qobj.pulse_qobj import PulseQobjInstruction
+from qiskit.qobj.converters import QobjToInstructionConverter
 from qiskit.circuit.instruction import Instruction
 from qiskit.circuit.parameterexpression import ParameterExpression
 from qiskit.pulse.exceptions import PulseError
@@ -52,6 +55,278 @@ class CalibrationPublisher(IntEnum):
     EXPERIMENT_SERVICE = 2
 
 
+class CalibrationEntry(metaclass=ABCMeta):
+    """A calibration entry."""
+
+    @abstractmethod
+    def define(self, definition: Any):
+        """Attach definition to the calibration entry.
+
+        Args:
+            definition: Definition of this entry.
+        """
+        pass
+
+    @abstractmethod
+    def get_name(self) -> str:
+        """Return instruction name of this entry.
+
+        Returns:
+            Instruction name.
+        """
+        pass
+
+    @abstractmethod
+    def get_qubits(self) -> Tuple[int, ...]:
+        """Return qubit index tuple of this entry.
+
+        Returns:
+            Qubit index tuple.
+        """
+        pass
+
+    @abstractmethod
+    def get_signature(self) -> inspect.Signature:
+        """Return signature object associated with entry definition.
+
+        Returns:
+            Signature object.
+        """
+        pass
+
+    @abstractmethod
+    def get_schedule(self, *args, **kwargs) -> Union[Schedule, ScheduleBlock]:
+        """Generate schedule from entry definition.
+
+        Args:
+            args: Command parameters.
+            kwargs: Command keyword parameters.
+
+        Returns:
+            Pulse schedule with assigned parameters.
+        """
+        pass
+
+    def __str__(self):
+        qubits_str = ", ".join(map(str, self.get_qubits()))
+        params_str = ", ".join(self.get_signature().parameters.keys())
+
+        out = f"{self.get_name()} gate of qubit {qubits_str}"
+        if params_str:
+            out += f" (parameters = {params_str})"
+        return out
+
+
+class ScheduleDef(CalibrationEntry):
+    """A calibration entry provided by in-memory Pulse representation."""
+
+    def __init__(
+        self,
+        name: str,
+        qubits: Sequence[int],
+        arguments: Optional[Sequence[str]] = None,
+    ):
+        """Define an empty entry.
+
+        Args:
+            name: Instruction name.
+            qubits: Physical qubit index.
+            arguments: User provided argument names for this entry, if parameterized.
+        """
+        self._name = name
+        self._qubits = tuple(qubits)
+        self._user_arguments = arguments
+
+        self._definition = None
+        self._signature = None
+
+    def _parse_argument(self):
+        """Generate signature from program and user provided argument names."""
+        # This doesn't assume multiple parameters with the same name
+        # Parameters with the same name are treated identically
+        all_argnames = set(map(lambda x: x.name, self._definition.parameters))
+
+        if self._user_arguments:
+            if set(self._user_arguments) != all_argnames:
+                raise PulseError(
+                    "Specified arguments don't match with schedule parameters. "
+                    f"{self._user_arguments} != {self._definition.parameters}."
+                )
+            argnames = list(self._user_arguments)
+        else:
+            argnames = sorted(all_argnames)
+
+        params = []
+        for argname in argnames:
+            param = inspect.Parameter(
+                argname,
+                kind=inspect.Parameter.POSITIONAL_OR_KEYWORD,
+            )
+            params.append(param)
+        signature = inspect.Signature(
+            parameters=params,
+            return_annotation=type(self._definition),
+        )
+        self._signature = signature
+
+    def define(self, definition: Union[Schedule, ScheduleBlock]):
+        self._definition = definition
+        self._parse_argument()
+
+    def get_name(self) -> str:
+        return self._name
+
+    def get_qubits(self) -> Tuple[int, ...]:
+        return self._qubits
+
+    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.
+            try:
+                value_dict[param] = to_bind.arguments[param.name]
+            except KeyError:
+                pass
+        return self._definition.assign_parameters(value_dict, inplace=False)
+
+    def __eq__(self, other):
+        if self._name != other._name:
+            return False
+        if self._qubits != other._qubits:
+            return False
+        # This delegates equality check to Schedule or ScheduleBlock.
+        return self._definition == other._definition
+
+
+class CallableDef(CalibrationEntry):
+    """A calibration entry provided by python callback function."""
+
+    def __init__(
+        self,
+        name: str,
+        qubits: Sequence[int],
+    ):
+        """Define an empty entry.
+
+        Args:
+            name: Instruction name.
+            qubits: Physical qubit index.
+        """
+        self._name = name
+        self._qubits = tuple(qubits)
+        self._definition = None
+        self._signature = None
+
+    def define(self, definition: Callable):
+        self._definition = definition
+        self._signature = inspect.signature(definition)
+
+    def get_name(self) -> str:
+        return self._name
+
+    def get_qubits(self) -> Tuple[int, ...]:
+        return self._qubits
+
+    def get_signature(self) -> inspect.Signature:
+        return self._signature
+
+    def get_schedule(self, *args, **kwargs) -> Union[Schedule, ScheduleBlock]:
+        try:
+            # Python function doesn't allow partial bind, but default value can exist.
+            to_bind = self._signature.bind(*args, **kwargs)
+            to_bind.apply_defaults()
+        except TypeError as ex:
+            raise PulseError("Assigned parameter doesn't match with function signature.") from ex
+
+        return self._definition(**to_bind.arguments)
+
+    def __eq__(self, other):
+        if self._name != other._name:
+            return False
+        if self._qubits != other._qubits:
+            return False
+        # We cannot evaluate function equality without parsing python AST.
+        # This simply compares wether they are the same object.
+        return self._definition is other._definition
+
+
+class PulseQobjDef(ScheduleDef):
+    """A calibration entry provided by Qobj instruction sequence."""
+
+    def __init__(
+        self,
+        name: str,
+        qubits: Sequence[int],
+        arguments: Optional[Sequence[str]] = None,
+        converter: Optional[QobjToInstructionConverter] = None,
+    ):
+        """Define an empty entry.
+
+        Args:
+            name: Instruction name.
+            qubits: Physical qubit index.
+            arguments: User provided argument names for this entry, if parameterized.
+            converter: Optional. Qobj to Qiskit converter.
+        """
+        super().__init__(name=name, qubits=qubits, arguments=arguments)
+
+        self._converter = converter or QobjToInstructionConverter()
+        self._source = None
+
+    def _build_schedule(self):
+        """Build pulse schedule from cmd-def sequence."""
+        schedule = Schedule(name=self.get_name())
+        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]):
+        # This doesn't generate signature immediately, because of lazy schedule build.
+        self._source = definition
+
+    def get_name(self) -> str:
+        return self._name
+
+    def get_qubits(self) -> Tuple[int, ...]:
+        return self._qubits
+
+    def get_signature(self) -> inspect.Signature:
+        if self._definition is None:
+            self._build_schedule()
+        return super().get_signature()
+
+    def get_schedule(self, *args, **kwargs) -> Union[Schedule, ScheduleBlock]:
+        if self._definition is None:
+            self._build_schedule()
+        return super().get_schedule(*args, **kwargs)
+
+    def __eq__(self, other):
+        if self._name != other._name:
+            return False
+        if self._qubits != other._qubits:
+            return False
+        if isinstance(other, PulseQobjDef):
+            # 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.
+            self._build_schedule()
+        return self._definition == other._definition
+
+
 class InstructionScheduleMap:
     """Mapping from :py:class:`~qiskit.circuit.QuantumCircuit`
     :py:class:`qiskit.circuit.Instruction` names and qubits to
@@ -73,7 +348,7 @@ def __init__(self):
         # Do not use lambda function for nested defaultdict, i.e. lambda: defaultdict(Generator).
         # This crashes qiskit parallel. Note that parallel framework passes args as
         # pickled object, however lambda function cannot be pickled.
-        self._map = defaultdict(functools.partial(defaultdict, Generator))
+        self._map = defaultdict(functools.partial(defaultdict, CalibrationEntry))
 
         # A backwards mapping from qubit to supported instructions
         self._qubit_instructions = defaultdict(set)
@@ -81,10 +356,8 @@ def __init__(self):
     def has_custom_gate(self) -> bool:
         """Return ``True`` if the map has user provided instruction."""
         for qubit_inst in self._map.values():
-            for generator in qubit_inst.values():
-                metadata = getattr(generator.function, "metadata", {})
-                publisher = metadata.get("publisher", CalibrationPublisher.QISKIT)
-                if publisher != CalibrationPublisher.BACKEND_PROVIDER:
+            for entry in qubit_inst.values():
+                if not isinstance(entry, PulseQobjDef):
                     return True
         return False
 
@@ -204,40 +477,8 @@ def get(
         """
         instruction = _get_instruction_string(instruction)
         self.assert_has(instruction, qubits)
-        generator = self._map[instruction][_to_tuple(qubits)]
-
-        _error_message = (
-            f"*params={params}, **kwparams={kwparams} do not match with "
-            f"the schedule generator signature {generator.signature}."
-        )
 
-        function = generator.function
-        if callable(function):
-            try:
-                # callables require full binding, but default values can exist.
-                binds = generator.signature.bind(*params, **kwparams)
-                binds.apply_defaults()
-            except TypeError as ex:
-                raise PulseError(_error_message) from ex
-            return function(**binds.arguments)
-
-        try:
-            # schedules allow partial binding
-            binds = generator.signature.bind_partial(*params, **kwparams)
-        except TypeError as ex:
-            raise PulseError(_error_message) from ex
-
-        if len(binds.arguments) > 0:
-            value_dict = dict()
-            for param in function.parameters:
-                try:
-                    value_dict[param] = binds.arguments[param.name]
-                except KeyError:
-                    pass
-
-            return function.assign_parameters(value_dict, inplace=False)
-        else:
-            return function
+        return self._map[instruction][_to_tuple(qubits)].get_schedule(*params, **kwparams)
 
     def add(
         self,
@@ -269,44 +510,45 @@ def add(
 
         # generate signature
         if isinstance(schedule, (Schedule, ScheduleBlock)):
-            ordered_names = sorted(list({par.name for par in schedule.parameters}))
-            if arguments:
-                if set(arguments) != set(ordered_names):
-                    raise PulseError(
-                        "Arguments does not match with schedule parameters. "
-                        f"{set(arguments)} != {schedule.parameters}."
-                    )
-                ordered_names = arguments
-
-            parameters = []
-            for argname in ordered_names:
-                param_signature = inspect.Parameter(
-                    argname,
-                    kind=inspect.Parameter.POSITIONAL_OR_KEYWORD,
-                )
-                parameters.append(param_signature)
-            signature = inspect.Signature(parameters=parameters, return_annotation=type(schedule))
-
+            entry = ScheduleDef(instruction, qubits, arguments)
+            # add metadata
+            if "publisher" not in schedule.metadata:
+                schedule.metadata["publisher"] = CalibrationPublisher.QISKIT
+            entry.define(schedule)
         elif callable(schedule):
             if arguments:
                 warnings.warn(
-                    "Arguments are overridden by the callback function signature. "
+                    "Arguments are overruled by the callback function signature. "
                     "Input `arguments` are ignored.",
                     UserWarning,
                 )
-            signature = inspect.signature(schedule)
-
+            entry = CallableDef(instruction, qubits)
+            entry.define(schedule)
         else:
             raise PulseError(
                 "Supplied schedule must be one of the Schedule, ScheduleBlock or a "
                 "callable that outputs a schedule."
             )
+        self._add(entry)
+
+    def _add(self, entry: CalibrationEntry):
+        """A method to resister calibration entry.
 
-        # add metadata
-        if hasattr(schedule, "metadata") and "publisher" not in schedule.metadata:
-            schedule.metadata["publisher"] = CalibrationPublisher.QISKIT
+        .. note::
 
-        self._map[instruction][qubits] = Generator(schedule, signature)
+            This is internal fast-path function, and caller must ensure
+            the entry is properly formatted. This function may be used by other programs
+            that load backend calibrations to create Qiskit representation of it.
+
+        Args:
+            entry: Calibration entry to register.
+
+        :meta public:
+        """
+        instruction = entry.get_name()
+        qubits = entry.get_qubits()
+
+        self._map[instruction][qubits] = entry
         self._qubit_instructions[qubits].add(instruction)
 
     def remove(
@@ -321,12 +563,14 @@ def remove(
         instruction = _get_instruction_string(instruction)
         qubits = _to_tuple(qubits)
         self.assert_has(instruction, qubits)
-        self._map[instruction].pop(qubits)
-        self._qubit_instructions[qubits].remove(instruction)
+
+        del self._map[instruction][qubits]
         if not self._map[instruction]:
-            self._map.pop(instruction)
+            del self._map[instruction]
+
+        self._qubit_instructions[qubits].remove(instruction)
         if not self._qubit_instructions[qubits]:
-            self._qubit_instructions.pop(qubits)
+            del self._qubit_instructions[qubits]
 
     def pop(
         self,
@@ -367,7 +611,7 @@ def get_parameters(
         instruction = _get_instruction_string(instruction)
 
         self.assert_has(instruction, qubits)
-        signature = self._map[instruction][_to_tuple(qubits)].signature
+        signature = self._map[instruction][_to_tuple(qubits)].get_signature()
         return tuple(signature.parameters.keys())
 
     def __str__(self):
diff --git a/qiskit/qobj/converters/pulse_instruction.py b/qiskit/qobj/converters/pulse_instruction.py
index 5715223336e3..502021429295 100644
--- a/qiskit/qobj/converters/pulse_instruction.py
+++ b/qiskit/qobj/converters/pulse_instruction.py
@@ -617,16 +617,40 @@ def __call__(self, instruction: PulseQobjInstruction) -> Schedule:
         Returns:
             Scheduled Qiskit Pulse instruction in Schedule format.
         """
+        schedule = Schedule()
+        for inst in self._get_sequences(instruction):
+            schedule.insert(instruction.t0, inst, inplace=True)
+        return schedule
+
+    def _get_sequences(
+        self,
+        instruction: PulseQobjInstruction,
+    ) -> Iterator[instructions.Instruction]:
+        """A method to iterate over pulse instructions without creating Schedule.
+
+        .. note::
+
+            This is internal fast-path function, and callers other than this converter class
+            might directly use this method to generate schedule from multiple
+            Qobj instructions. Because __call__ always returns a schedule with the time offset
+            parsed instruction, composing multiple Qobj instructions to create
+            a gate schedule is somewhat inefficient due to composing overhead of schedules.
+            Directly combining instructions with this method is much performant.
+
+        Args:
+            instruction: Instruction data in Qobj format.
+
+        Yields:
+            Qiskit Pulse instructions.
+
+        :meta public:
+        """
         try:
             method = getattr(self, f"_convert_{instruction.name}")
         except AttributeError:
             method = self._convert_generic
 
-        t0 = instruction.t0
-        schedule = Schedule()
-        for inst in method(instruction):
-            schedule.insert(t0, inst, inplace=True)
-        return schedule
+        yield from method(instruction)
 
     def get_supported_instructions(self) -> List[str]:
         """Retrun a list of supported instructions."""

From 7e3f741499ed78ed043d4db37a058f745d9d8cad Mon Sep 17 00:00:00 2001
From: Naoki Kanazawa <nkanazawa1989@gmail.com>
Date: Wed, 12 Oct 2022 19:18:31 +0900
Subject: [PATCH 02/10] Avoid deepcopy and mutating the source dict

---
 qiskit/providers/models/pulsedefaults.py | 42 +++++++++++++++---------
 qiskit/qobj/pulse_qobj.py                | 42 ++++++++++++++++--------
 2 files changed, 54 insertions(+), 30 deletions(-)

diff --git a/qiskit/providers/models/pulsedefaults.py b/qiskit/providers/models/pulsedefaults.py
index 8cb5b44a6f9c..1c1d48c184fd 100644
--- a/qiskit/providers/models/pulsedefaults.py
+++ b/qiskit/providers/models/pulsedefaults.py
@@ -12,7 +12,6 @@
 
 
 """Model and schema for pulse defaults."""
-import copy
 from typing import Any, Dict, List
 
 from qiskit.pulse.instruction_schedule_map import InstructionScheduleMap, PulseQobjDef
@@ -151,11 +150,14 @@ def from_dict(cls, data):
             qiskit.providers.model.Command: The ``Command`` from the input
                 dictionary.
         """
-        in_data = copy.copy(data)
-        if "sequence" in in_data:
-            in_data["sequence"] = [
-                PulseQobjInstruction.from_dict(x) for x in in_data.pop("sequence")
-            ]
+        # Pulse command data is nested dictionary.
+        # To avoid deepcopy and avoid mutating the source object, create new dict here.
+        in_data = {}
+        for key, value in data.items():
+            if key == "sequence":
+                in_data[key] = list(map(PulseQobjInstruction.from_dict, value))
+            else:
+                in_data[key] = value
         return cls(**in_data)
 
 
@@ -269,17 +271,25 @@ def from_dict(cls, data):
         Returns:
             PulseDefaults: The PulseDefaults from the input dictionary.
         """
+        schema = {
+            "pulse_library": PulseLibraryItem,
+            "cmd_def": Command,
+            "meas_kernel": MeasurementKernel,
+            "discriminator": Discriminator,
+        }
+
         # Pulse defaults data is nested dictionary.
-        # Some elements of the inner dictionaries are poped, and thus do deepcopy here.
-        in_data = copy.deepcopy(data)
-        in_data["pulse_library"] = [
-            PulseLibraryItem.from_dict(x) for x in in_data.pop("pulse_library")
-        ]
-        in_data["cmd_def"] = [Command.from_dict(x) for x in in_data.pop("cmd_def")]
-        if "meas_kernel" in in_data:
-            in_data["meas_kernel"] = MeasurementKernel.from_dict(in_data.pop("meas_kernel"))
-        if "discriminator" in in_data:
-            in_data["discriminator"] = Discriminator.from_dict(in_data.pop("discriminator"))
+        # To avoid deepcopy and avoid mutating the source object, create new dict here.
+        in_data = {}
+        for key, value in data.items():
+            if key in schema:
+                if isinstance(value, list):
+                    in_data[key] = list(map(schema[key].from_dict, value))
+                else:
+                    in_data[key] = schema[key].from_dict(value)
+            else:
+                in_data[key] = value
+
         return cls(**in_data)
 
     def __str__(self):
diff --git a/qiskit/qobj/pulse_qobj.py b/qiskit/qobj/pulse_qobj.py
index 56cec9fe9cfc..f14fa193c5a5 100644
--- a/qiskit/qobj/pulse_qobj.py
+++ b/qiskit/qobj/pulse_qobj.py
@@ -226,20 +226,34 @@ def from_dict(cls, data):
         Returns:
             PulseQobjInstruction: The object from the input dictionary.
         """
-        t0 = data.pop("t0")
-        name = data.pop("name")
-        if "kernels" in data:
-            kernels = data.pop("kernels")
-            kernel_obj = [QobjMeasurementOption.from_dict(x) for x in kernels]
-            data["kernels"] = kernel_obj
-        if "discriminators" in data:
-            discriminators = data.pop("discriminators")
-            discriminators_obj = [QobjMeasurementOption.from_dict(x) for x in discriminators]
-            data["discriminators"] = discriminators_obj
-        if "parameters" in data and "amp" in data["parameters"]:
-            data["parameters"]["amp"] = _to_complex(data["parameters"]["amp"])
-
-        return cls(name, t0, **data)
+        schema = {
+            "discriminators": QobjMeasurementOption,
+            "kernels": QobjMeasurementOption,
+        }
+        skip = ["t0", "name"]
+
+        # Pulse instruction data is nested dictionary.
+        # To avoid deepcopy and avoid mutating the source object, create new dict here.
+        in_data = {}
+        for key, value in data.items():
+            if key in skip:
+                continue
+            if key == "parameters":
+                # This is flat dictionary of parametric pulse parameters
+                formatted_value = value.copy()
+                if "amp" in formatted_value:
+                    formatted_value["amp"] = _to_complex(formatted_value["amp"])
+                in_data[key] = formatted_value
+                continue
+            if key in schema:
+                if isinstance(value, list):
+                    in_data[key] = list(map(schema[key].from_dict, value))
+                else:
+                    in_data[key] = schema[key].from_dict(value)
+            else:
+                in_data[key] = value
+
+        return cls(data["name"], data["t0"], **in_data)
 
     def __eq__(self, other):
         if isinstance(other, PulseQobjInstruction):

From ba5024800ec272f6808f51ac7d4f5cc2efc0a656 Mon Sep 17 00:00:00 2001
From: Naoki Kanazawa <nkanazawa1989@gmail.com>
Date: Thu, 13 Oct 2022 21:47:36 +0900
Subject: [PATCH 03/10] Support V2 lazy load

---
 qiskit/providers/models/pulsedefaults.py |  12 +-
 qiskit/pulse/instruction_schedule_map.py | 136 +++++++----------------
 qiskit/transpiler/target.py              |  26 +++--
 3 files changed, 64 insertions(+), 110 deletions(-)

diff --git a/qiskit/providers/models/pulsedefaults.py b/qiskit/providers/models/pulsedefaults.py
index 1c1d48c184fd..a91646d9db82 100644
--- a/qiskit/providers/models/pulsedefaults.py
+++ b/qiskit/providers/models/pulsedefaults.py
@@ -204,13 +204,13 @@ def __init__(
         self.converter = QobjToInstructionConverter(pulse_library)
 
         for inst in cmd_def:
-            entry = PulseQobjDef(
-                name=inst.name,
-                qubits=inst.qubits,
-                converter=self.converter,
-            )
+            entry = PulseQobjDef(converter=self.converter, name=inst.name)
             entry.define(inst.sequence)
-            self.instruction_schedule_map._add(entry)
+            self.instruction_schedule_map._add(
+                instruction_name=inst.name,
+                qubits=tuple(inst.qubits),
+                entry=entry,
+            )
 
         if meas_kernel is not None:
             self.meas_kernel = meas_kernel
diff --git a/qiskit/pulse/instruction_schedule_map.py b/qiskit/pulse/instruction_schedule_map.py
index f0a23159fdae..de0a80b84282 100644
--- a/qiskit/pulse/instruction_schedule_map.py
+++ b/qiskit/pulse/instruction_schedule_map.py
@@ -67,24 +67,6 @@ def define(self, definition: Any):
         """
         pass
 
-    @abstractmethod
-    def get_name(self) -> str:
-        """Return instruction name of this entry.
-
-        Returns:
-            Instruction name.
-        """
-        pass
-
-    @abstractmethod
-    def get_qubits(self) -> Tuple[int, ...]:
-        """Return qubit index tuple of this entry.
-
-        Returns:
-            Qubit index tuple.
-        """
-        pass
-
     @abstractmethod
     def get_signature(self) -> inspect.Signature:
         """Return signature object associated with entry definition.
@@ -107,34 +89,16 @@ def get_schedule(self, *args, **kwargs) -> Union[Schedule, ScheduleBlock]:
         """
         pass
 
-    def __str__(self):
-        qubits_str = ", ".join(map(str, self.get_qubits()))
-        params_str = ", ".join(self.get_signature().parameters.keys())
-
-        out = f"{self.get_name()} gate of qubit {qubits_str}"
-        if params_str:
-            out += f" (parameters = {params_str})"
-        return out
-
 
 class ScheduleDef(CalibrationEntry):
     """A calibration entry provided by in-memory Pulse representation."""
 
-    def __init__(
-        self,
-        name: str,
-        qubits: Sequence[int],
-        arguments: Optional[Sequence[str]] = None,
-    ):
+    def __init__(self, arguments: Optional[Sequence[str]] = None):
         """Define an empty entry.
 
         Args:
-            name: Instruction name.
-            qubits: Physical qubit index.
             arguments: User provided argument names for this entry, if parameterized.
         """
-        self._name = name
-        self._qubits = tuple(qubits)
         self._user_arguments = arguments
 
         self._definition = None
@@ -173,12 +137,6 @@ def define(self, definition: Union[Schedule, ScheduleBlock]):
         self._definition = definition
         self._parse_argument()
 
-    def get_name(self) -> str:
-        return self._name
-
-    def get_qubits(self) -> Tuple[int, ...]:
-        return self._qubits
-
     def get_signature(self) -> inspect.Signature:
         return self._signature
 
@@ -199,30 +157,22 @@ def get_schedule(self, *args, **kwargs) -> Union[Schedule, ScheduleBlock]:
         return self._definition.assign_parameters(value_dict, inplace=False)
 
     def __eq__(self, other):
-        if self._name != other._name:
-            return False
-        if self._qubits != other._qubits:
-            return False
         # This delegates equality check to Schedule or ScheduleBlock.
         return self._definition == other._definition
 
+    def __str__(self):
+        out = f"Schedule {self._definition.name}"
+        params_str = ", ".join(self.get_signature().parameters.keys())
+        if params_str:
+            out += f"({params_str})"
+        return out
+
 
 class CallableDef(CalibrationEntry):
     """A calibration entry provided by python callback function."""
 
-    def __init__(
-        self,
-        name: str,
-        qubits: Sequence[int],
-    ):
-        """Define an empty entry.
-
-        Args:
-            name: Instruction name.
-            qubits: Physical qubit index.
-        """
-        self._name = name
-        self._qubits = tuple(qubits)
+    def __init__(self):
+        """Define an empty entry."""
         self._definition = None
         self._signature = None
 
@@ -230,12 +180,6 @@ def define(self, definition: Callable):
         self._definition = definition
         self._signature = inspect.signature(definition)
 
-    def get_name(self) -> str:
-        return self._name
-
-    def get_qubits(self) -> Tuple[int, ...]:
-        return self._qubits
-
     def get_signature(self) -> inspect.Signature:
         return self._signature
 
@@ -250,41 +194,40 @@ def get_schedule(self, *args, **kwargs) -> Union[Schedule, ScheduleBlock]:
         return self._definition(**to_bind.arguments)
 
     def __eq__(self, other):
-        if self._name != other._name:
-            return False
-        if self._qubits != other._qubits:
-            return False
         # We cannot evaluate function equality without parsing python AST.
         # This simply compares wether they are the same object.
         return self._definition is other._definition
 
+    def __str__(self):
+        params_str = ", ".join(self.get_signature().parameters.keys())
+        return f"Callable {self._definition.__name__}({params_str})"
+
 
 class PulseQobjDef(ScheduleDef):
     """A calibration entry provided by Qobj instruction sequence."""
 
     def __init__(
         self,
-        name: str,
-        qubits: Sequence[int],
         arguments: Optional[Sequence[str]] = None,
         converter: Optional[QobjToInstructionConverter] = None,
+        name: Optional[str] = None,
     ):
         """Define an empty entry.
 
         Args:
-            name: Instruction name.
-            qubits: Physical qubit index.
             arguments: User provided argument names for this entry, if parameterized.
             converter: Optional. Qobj to Qiskit converter.
+            name: Name of schedule.
         """
-        super().__init__(name=name, qubits=qubits, arguments=arguments)
+        super().__init__(arguments=arguments)
 
         self._converter = converter or QobjToInstructionConverter()
+        self._name = name
         self._source = None
 
     def _build_schedule(self):
         """Build pulse schedule from cmd-def sequence."""
-        schedule = Schedule(name=self.get_name())
+        schedule = Schedule(name=self._name)
         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)
@@ -297,12 +240,6 @@ def define(self, definition: List[PulseQobjInstruction]):
         # This doesn't generate signature immediately, because of lazy schedule build.
         self._source = definition
 
-    def get_name(self) -> str:
-        return self._name
-
-    def get_qubits(self) -> Tuple[int, ...]:
-        return self._qubits
-
     def get_signature(self) -> inspect.Signature:
         if self._definition is None:
             self._build_schedule()
@@ -314,10 +251,6 @@ def get_schedule(self, *args, **kwargs) -> Union[Schedule, ScheduleBlock]:
         return super().get_schedule(*args, **kwargs)
 
     def __eq__(self, other):
-        if self._name != other._name:
-            return False
-        if self._qubits != other._qubits:
-            return False
         if isinstance(other, PulseQobjDef):
             # If both objects are Qobj just check Qobj equality.
             return self._source == other._source
@@ -326,6 +259,12 @@ def __eq__(self, other):
             self._build_schedule()
         return self._definition == other._definition
 
+    def __str__(self):
+        if self._definition is None:
+            # Avoid parsing schedule for pretty print.
+            return "PulseQobj"
+        return super().__str__()
+
 
 class InstructionScheduleMap:
     """Mapping from :py:class:`~qiskit.circuit.QuantumCircuit`
@@ -510,11 +449,10 @@ def add(
 
         # generate signature
         if isinstance(schedule, (Schedule, ScheduleBlock)):
-            entry = ScheduleDef(instruction, qubits, arguments)
+            entry = ScheduleDef(arguments)
             # add metadata
             if "publisher" not in schedule.metadata:
                 schedule.metadata["publisher"] = CalibrationPublisher.QISKIT
-            entry.define(schedule)
         elif callable(schedule):
             if arguments:
                 warnings.warn(
@@ -522,16 +460,21 @@ def add(
                     "Input `arguments` are ignored.",
                     UserWarning,
                 )
-            entry = CallableDef(instruction, qubits)
-            entry.define(schedule)
+            entry = CallableDef()
         else:
             raise PulseError(
                 "Supplied schedule must be one of the Schedule, ScheduleBlock or a "
                 "callable that outputs a schedule."
             )
-        self._add(entry)
+        entry.define(schedule)
+        self._add(instruction, qubits, entry)
 
-    def _add(self, entry: CalibrationEntry):
+    def _add(
+        self,
+        instruction_name: str,
+        qubits: Tuple[int, ...],
+        entry: CalibrationEntry,
+    ):
         """A method to resister calibration entry.
 
         .. note::
@@ -541,15 +484,14 @@ def _add(self, entry: CalibrationEntry):
             that load backend calibrations to create Qiskit representation of it.
 
         Args:
+            instruction_name: Name of instruction.
+            qubits: List of qubits that this calibration is applied.
             entry: Calibration entry to register.
 
         :meta public:
         """
-        instruction = entry.get_name()
-        qubits = entry.get_qubits()
-
-        self._map[instruction][qubits] = entry
-        self._qubit_instructions[qubits].add(instruction)
+        self._map[instruction_name][qubits] = entry
+        self._qubit_instructions[qubits].add(instruction_name)
 
     def remove(
         self, instruction: Union[str, Instruction], qubits: Union[int, Iterable[int]]
diff --git a/qiskit/transpiler/target.py b/qiskit/transpiler/target.py
index 0590c972eb86..634e709c64d6 100644
--- a/qiskit/transpiler/target.py
+++ b/qiskit/transpiler/target.py
@@ -17,6 +17,7 @@
 from a backend
 """
 
+from typing import Union
 from collections.abc import Mapping
 from collections import defaultdict
 import datetime
@@ -27,7 +28,8 @@
 import rustworkx as rx
 
 from qiskit.circuit.parameter import Parameter
-from qiskit.pulse.instruction_schedule_map import InstructionScheduleMap
+from qiskit.pulse.instruction_schedule_map import InstructionScheduleMap, CalibrationEntry
+from qiskit.pulse.schedule import Schedule, ScheduleBlock
 from qiskit.transpiler.coupling import CouplingMap
 from qiskit.transpiler.exceptions import TranspilerError
 from qiskit.transpiler.instruction_durations import InstructionDurations
@@ -51,13 +53,13 @@ class InstructionProperties:
     custom attributes for those custom/additional properties by the backend.
     """
 
-    __slots__ = ("duration", "error", "calibration")
+    __slots__ = ("duration", "error", "_calibration")
 
     def __init__(
         self,
         duration: float = None,
         error: float = None,
-        calibration=None,
+        calibration: Union[Schedule, ScheduleBlock, CalibrationEntry] = None,
     ):
         """Create a new ``InstructionProperties`` object
 
@@ -66,17 +68,27 @@ def __init__(
                 specified set of qubits
             error: The average error rate for the instruction on the specified
                 set of qubits.
-            calibration (Union["qiskit.pulse.Schedule", "qiskit.pulse.ScheduleBlock"]): The pulse
-                representation of the instruction
+            calibration: The pulse representation of the instruction.
         """
         self.duration = duration
         self.error = error
-        self.calibration = calibration
+        self._calibration = calibration
+
+    @property
+    def calibration(self):
+        """The pulse representation of the instruction."""
+        if isinstance(self._calibration, CalibrationEntry):
+            return self._calibration.get_schedule()
+        return self._calibration
+
+    @calibration.setter
+    def calibration(self, calibration: Union[Schedule, ScheduleBlock, CalibrationEntry]):
+        self._calibration = calibration
 
     def __repr__(self):
         return (
             f"InstructionProperties(duration={self.duration}, error={self.error}"
-            f", calibration={self.calibration})"
+            f", calibration={self._calibration})"
         )
 
 

From bd3b0ecf064c196c578d9f6a00468b3353d080c7 Mon Sep 17 00:00:00 2001
From: Naoki Kanazawa <nkanazawa1989@gmail.com>
Date: Fri, 14 Oct 2022 01:06:44 +0900
Subject: [PATCH 04/10] Update Gate.from_dict

---
 qiskit/providers/models/backendproperties.py | 11 ++++++-----
 1 file changed, 6 insertions(+), 5 deletions(-)

diff --git a/qiskit/providers/models/backendproperties.py b/qiskit/providers/models/backendproperties.py
index 46e30c79fe5f..ea28ae956dc6 100644
--- a/qiskit/providers/models/backendproperties.py
+++ b/qiskit/providers/models/backendproperties.py
@@ -128,11 +128,12 @@ def from_dict(cls, data):
         Returns:
             Gate: The Nduv from the input dictionary.
         """
-        in_data = copy.copy(data)
-        nduvs = []
-        for nduv in in_data.pop("parameters"):
-            nduvs.append(Nduv.from_dict(nduv))
-        in_data["parameters"] = nduvs
+        in_data = {}
+        for key, value in data.items():
+            if key == "parameters":
+                in_data[key] = list(map(Nduv.from_dict, value))
+            else:
+                in_data[key] = value
         return cls(**in_data)
 
     def to_dict(self):

From 3a20f0fdf94a0206cf72c340fa7afdff53dc9214 Mon Sep 17 00:00:00 2001
From: Naoki Kanazawa <nkanazawa1989@gmail.com>
Date: Sat, 7 Jan 2023 16:16:52 +0900
Subject: [PATCH 05/10] Move CalibrationEntry and CalibrationPublisher to
 dedicated file for future deprecation of instruction schedule map.

---
 qiskit/pulse/calibration_entries.py           | 258 ++++++++++++++++++
 qiskit/pulse/instruction_schedule_map.py      | 240 +---------------
 .../passes/calibration/base_builder.py        |   2 +-
 qiskit/transpiler/target.py                   |   3 +-
 4 files changed, 270 insertions(+), 233 deletions(-)
 create mode 100644 qiskit/pulse/calibration_entries.py

diff --git a/qiskit/pulse/calibration_entries.py b/qiskit/pulse/calibration_entries.py
new file mode 100644
index 000000000000..2e4366aefce5
--- /dev/null
+++ b/qiskit/pulse/calibration_entries.py
@@ -0,0 +1,258 @@
+# This code is part of Qiskit.
+#
+# (C) Copyright IBM 2023.
+#
+# 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.
+
+"""Internal format of calibration data in target."""
+import inspect
+from abc import ABCMeta, abstractmethod
+from enum import IntEnum
+from typing import Callable, List, Union, Optional, Sequence, Any
+
+from qiskit.pulse.exceptions import PulseError
+from qiskit.pulse.schedule import Schedule, ScheduleBlock
+from qiskit.qobj.converters import QobjToInstructionConverter
+from qiskit.qobj.pulse_qobj import PulseQobjInstruction
+
+
+class CalibrationPublisher(IntEnum):
+    """Defines who defined schedule entry."""
+
+    BACKEND_PROVIDER = 0
+    QISKIT = 1
+    EXPERIMENT_SERVICE = 2
+
+
+class CalibrationEntry(metaclass=ABCMeta):
+    """A metaclass of a calibration entry."""
+
+    @abstractmethod
+    def define(self, definition: Any):
+        """Attach definition to the calibration entry.
+
+        Args:
+            definition: Definition of this entry.
+        """
+        pass
+
+    @abstractmethod
+    def get_signature(self) -> inspect.Signature:
+        """Return signature object associated with entry definition.
+
+        Returns:
+            Signature object.
+        """
+        pass
+
+    @abstractmethod
+    def get_schedule(self, *args, **kwargs) -> Union[Schedule, ScheduleBlock]:
+        """Generate schedule from entry definition.
+
+        Args:
+            args: Command parameters.
+            kwargs: Command keyword parameters.
+
+        Returns:
+            Pulse schedule with assigned parameters.
+        """
+        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.
+    """
+
+    def __init__(self, arguments: Optional[Sequence[str]] = None):
+        """Define an empty entry.
+
+        Args:
+            arguments: User provided argument names for this entry, if parameterized.
+        """
+        self._user_arguments = arguments
+
+        self._definition = None
+        self._signature = None
+
+    def _parse_argument(self):
+        """Generate signature from program and user provided argument names."""
+        # This doesn't assume multiple parameters with the same name
+        # Parameters with the same name are treated identically
+        all_argnames = set(map(lambda x: x.name, self._definition.parameters))
+
+        if self._user_arguments:
+            if set(self._user_arguments) != all_argnames:
+                raise PulseError(
+                    "Specified arguments don't match with schedule parameters. "
+                    f"{self._user_arguments} != {self._definition.parameters}."
+                )
+            argnames = list(self._user_arguments)
+        else:
+            argnames = sorted(all_argnames)
+
+        params = []
+        for argname in argnames:
+            param = inspect.Parameter(
+                argname,
+                kind=inspect.Parameter.POSITIONAL_OR_KEYWORD,
+            )
+            params.append(param)
+        signature = inspect.Signature(
+            parameters=params,
+            return_annotation=type(self._definition),
+        )
+        self._signature = signature
+
+    def define(self, definition: Union[Schedule, ScheduleBlock]):
+        self._definition = definition
+        self._parse_argument()
+
+    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.
+            try:
+                value_dict[param] = to_bind.arguments[param.name]
+            except KeyError:
+                pass
+        return self._definition.assign_parameters(value_dict, inplace=False)
+
+    def __eq__(self, other):
+        # This delegates equality check to Schedule or ScheduleBlock.
+        return self._definition == other._definition
+
+    def __str__(self):
+        out = f"Schedule {self._definition.name}"
+        params_str = ", ".join(self.get_signature().parameters.keys())
+        if params_str:
+            out += f"({params_str})"
+        return out
+
+
+class CallableDef(CalibrationEntry):
+    """Python callback function that generates Qiskit Pulse program.
+
+    A callable is inspected by the python built-in inspection module and
+    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.
+    """
+
+    def __init__(self):
+        """Define an empty entry."""
+        self._definition = None
+        self._signature = None
+
+    def define(self, definition: Callable):
+        self._definition = definition
+        self._signature = inspect.signature(definition)
+
+    def get_signature(self) -> inspect.Signature:
+        return self._signature
+
+    def get_schedule(self, *args, **kwargs) -> Union[Schedule, ScheduleBlock]:
+        try:
+            # Python function doesn't allow partial bind, but default value can exist.
+            to_bind = self._signature.bind(*args, **kwargs)
+            to_bind.apply_defaults()
+        except TypeError as ex:
+            raise PulseError("Assigned parameter doesn't match with function signature.") from ex
+
+        return self._definition(**to_bind.arguments)
+
+    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
+
+    def __str__(self):
+        params_str = ", ".join(self.get_signature().parameters.keys())
+        return f"Callable {self._definition.__name__}({params_str})"
+
+
+class PulseQobjDef(ScheduleDef):
+    """Qobj JSON serialized format instruction sequence.
+
+    A JSON serialized program can be converted into Qiskit Pulse program with
+    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.
+    """
+
+    def __init__(
+        self,
+        arguments: Optional[Sequence[str]] = None,
+        converter: Optional[QobjToInstructionConverter] = None,
+        name: Optional[str] = None,
+    ):
+        """Define an empty entry.
+
+        Args:
+            arguments: User provided argument names for this entry, if parameterized.
+            converter: Optional. Qobj to Qiskit converter.
+            name: Name of schedule.
+        """
+        super().__init__(arguments=arguments)
+
+        self._converter = converter or QobjToInstructionConverter()
+        self._name = name
+        self._source = None
+
+    def _build_schedule(self):
+        """Build pulse schedule from cmd-def sequence."""
+        schedule = Schedule(name=self._name)
+        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]):
+        # This doesn't generate signature immediately, because of lazy schedule build.
+        self._source = definition
+
+    def get_signature(self) -> inspect.Signature:
+        if self._definition is None:
+            self._build_schedule()
+        return super().get_signature()
+
+    def get_schedule(self, *args, **kwargs) -> Union[Schedule, ScheduleBlock]:
+        if self._definition is None:
+            self._build_schedule()
+        return super().get_schedule(*args, **kwargs)
+
+    def __eq__(self, other):
+        if isinstance(other, PulseQobjDef):
+            # 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.
+            self._build_schedule()
+        return self._definition == other._definition
+
+    def __str__(self):
+        if self._definition is None:
+            # Avoid parsing schedule for pretty print.
+            return "PulseQobj"
+        return super().__str__()
diff --git a/qiskit/pulse/instruction_schedule_map.py b/qiskit/pulse/instruction_schedule_map.py
index de0a80b84282..df1599e11327 100644
--- a/qiskit/pulse/instruction_schedule_map.py
+++ b/qiskit/pulse/instruction_schedule_map.py
@@ -26,245 +26,23 @@
     inst_map = backend.defaults().instruction_schedule_map
 
 """
-from abc import ABCMeta, abstractmethod
-import inspect
 import functools
 import warnings
 from collections import defaultdict
-from enum import IntEnum
-from typing import Callable, Iterable, List, Tuple, Union, Optional, NamedTuple, Sequence, Any
+from typing import Callable, Iterable, List, Tuple, Union, Optional
 
-from qiskit.qobj.pulse_qobj import PulseQobjInstruction
-from qiskit.qobj.converters import QobjToInstructionConverter
 from qiskit.circuit.instruction import Instruction
 from qiskit.circuit.parameterexpression import ParameterExpression
+from qiskit.pulse.calibration_entries import (
+    CalibrationPublisher,
+    CalibrationEntry,
+    ScheduleDef,
+    CallableDef,
+    PulseQobjDef,
+)
 from qiskit.pulse.exceptions import PulseError
 from qiskit.pulse.schedule import Schedule, ScheduleBlock
 
-Generator = NamedTuple(
-    "Generator",
-    [("function", Union[Callable, Schedule, ScheduleBlock]), ("signature", inspect.Signature)],
-)
-
-
-class CalibrationPublisher(IntEnum):
-    """Defines who defined schedule entry."""
-
-    BACKEND_PROVIDER = 0
-    QISKIT = 1
-    EXPERIMENT_SERVICE = 2
-
-
-class CalibrationEntry(metaclass=ABCMeta):
-    """A calibration entry."""
-
-    @abstractmethod
-    def define(self, definition: Any):
-        """Attach definition to the calibration entry.
-
-        Args:
-            definition: Definition of this entry.
-        """
-        pass
-
-    @abstractmethod
-    def get_signature(self) -> inspect.Signature:
-        """Return signature object associated with entry definition.
-
-        Returns:
-            Signature object.
-        """
-        pass
-
-    @abstractmethod
-    def get_schedule(self, *args, **kwargs) -> Union[Schedule, ScheduleBlock]:
-        """Generate schedule from entry definition.
-
-        Args:
-            args: Command parameters.
-            kwargs: Command keyword parameters.
-
-        Returns:
-            Pulse schedule with assigned parameters.
-        """
-        pass
-
-
-class ScheduleDef(CalibrationEntry):
-    """A calibration entry provided by in-memory Pulse representation."""
-
-    def __init__(self, arguments: Optional[Sequence[str]] = None):
-        """Define an empty entry.
-
-        Args:
-            arguments: User provided argument names for this entry, if parameterized.
-        """
-        self._user_arguments = arguments
-
-        self._definition = None
-        self._signature = None
-
-    def _parse_argument(self):
-        """Generate signature from program and user provided argument names."""
-        # This doesn't assume multiple parameters with the same name
-        # Parameters with the same name are treated identically
-        all_argnames = set(map(lambda x: x.name, self._definition.parameters))
-
-        if self._user_arguments:
-            if set(self._user_arguments) != all_argnames:
-                raise PulseError(
-                    "Specified arguments don't match with schedule parameters. "
-                    f"{self._user_arguments} != {self._definition.parameters}."
-                )
-            argnames = list(self._user_arguments)
-        else:
-            argnames = sorted(all_argnames)
-
-        params = []
-        for argname in argnames:
-            param = inspect.Parameter(
-                argname,
-                kind=inspect.Parameter.POSITIONAL_OR_KEYWORD,
-            )
-            params.append(param)
-        signature = inspect.Signature(
-            parameters=params,
-            return_annotation=type(self._definition),
-        )
-        self._signature = signature
-
-    def define(self, definition: Union[Schedule, ScheduleBlock]):
-        self._definition = definition
-        self._parse_argument()
-
-    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.
-            try:
-                value_dict[param] = to_bind.arguments[param.name]
-            except KeyError:
-                pass
-        return self._definition.assign_parameters(value_dict, inplace=False)
-
-    def __eq__(self, other):
-        # This delegates equality check to Schedule or ScheduleBlock.
-        return self._definition == other._definition
-
-    def __str__(self):
-        out = f"Schedule {self._definition.name}"
-        params_str = ", ".join(self.get_signature().parameters.keys())
-        if params_str:
-            out += f"({params_str})"
-        return out
-
-
-class CallableDef(CalibrationEntry):
-    """A calibration entry provided by python callback function."""
-
-    def __init__(self):
-        """Define an empty entry."""
-        self._definition = None
-        self._signature = None
-
-    def define(self, definition: Callable):
-        self._definition = definition
-        self._signature = inspect.signature(definition)
-
-    def get_signature(self) -> inspect.Signature:
-        return self._signature
-
-    def get_schedule(self, *args, **kwargs) -> Union[Schedule, ScheduleBlock]:
-        try:
-            # Python function doesn't allow partial bind, but default value can exist.
-            to_bind = self._signature.bind(*args, **kwargs)
-            to_bind.apply_defaults()
-        except TypeError as ex:
-            raise PulseError("Assigned parameter doesn't match with function signature.") from ex
-
-        return self._definition(**to_bind.arguments)
-
-    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
-
-    def __str__(self):
-        params_str = ", ".join(self.get_signature().parameters.keys())
-        return f"Callable {self._definition.__name__}({params_str})"
-
-
-class PulseQobjDef(ScheduleDef):
-    """A calibration entry provided by Qobj instruction sequence."""
-
-    def __init__(
-        self,
-        arguments: Optional[Sequence[str]] = None,
-        converter: Optional[QobjToInstructionConverter] = None,
-        name: Optional[str] = None,
-    ):
-        """Define an empty entry.
-
-        Args:
-            arguments: User provided argument names for this entry, if parameterized.
-            converter: Optional. Qobj to Qiskit converter.
-            name: Name of schedule.
-        """
-        super().__init__(arguments=arguments)
-
-        self._converter = converter or QobjToInstructionConverter()
-        self._name = name
-        self._source = None
-
-    def _build_schedule(self):
-        """Build pulse schedule from cmd-def sequence."""
-        schedule = Schedule(name=self._name)
-        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]):
-        # This doesn't generate signature immediately, because of lazy schedule build.
-        self._source = definition
-
-    def get_signature(self) -> inspect.Signature:
-        if self._definition is None:
-            self._build_schedule()
-        return super().get_signature()
-
-    def get_schedule(self, *args, **kwargs) -> Union[Schedule, ScheduleBlock]:
-        if self._definition is None:
-            self._build_schedule()
-        return super().get_schedule(*args, **kwargs)
-
-    def __eq__(self, other):
-        if isinstance(other, PulseQobjDef):
-            # 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.
-            self._build_schedule()
-        return self._definition == other._definition
-
-    def __str__(self):
-        if self._definition is None:
-            # Avoid parsing schedule for pretty print.
-            return "PulseQobj"
-        return super().__str__()
-
 
 class InstructionScheduleMap:
     """Mapping from :py:class:`~qiskit.circuit.QuantumCircuit`
@@ -284,7 +62,7 @@ def __init__(self):
         """Initialize a circuit instruction to schedule mapper instance."""
         # The processed and reformatted circuit instruction definitions
 
-        # Do not use lambda function for nested defaultdict, i.e. lambda: defaultdict(Generator).
+        # Do not use lambda function for nested defaultdict, i.e. lambda: defaultdict(CalibrationEntry).
         # This crashes qiskit parallel. Note that parallel framework passes args as
         # pickled object, however lambda function cannot be pickled.
         self._map = defaultdict(functools.partial(defaultdict, CalibrationEntry))
diff --git a/qiskit/transpiler/passes/calibration/base_builder.py b/qiskit/transpiler/passes/calibration/base_builder.py
index 488cc1dc2e65..6d49c23abb31 100644
--- a/qiskit/transpiler/passes/calibration/base_builder.py
+++ b/qiskit/transpiler/passes/calibration/base_builder.py
@@ -18,7 +18,7 @@
 from qiskit.circuit import Instruction as CircuitInst
 from qiskit.dagcircuit import DAGCircuit
 from qiskit.pulse import Schedule, ScheduleBlock
-from qiskit.pulse.instruction_schedule_map import CalibrationPublisher
+from qiskit.pulse.calibration_entries import CalibrationPublisher
 from qiskit.transpiler.basepasses import TransformationPass
 
 from .exceptions import CalibrationNotAvailable
diff --git a/qiskit/transpiler/target.py b/qiskit/transpiler/target.py
index 634e709c64d6..a3872fa26b0d 100644
--- a/qiskit/transpiler/target.py
+++ b/qiskit/transpiler/target.py
@@ -28,7 +28,8 @@
 import rustworkx as rx
 
 from qiskit.circuit.parameter import Parameter
-from qiskit.pulse.instruction_schedule_map import InstructionScheduleMap, CalibrationEntry
+from qiskit.pulse.instruction_schedule_map import InstructionScheduleMap
+from qiskit.pulse.calibration_entries import CalibrationEntry
 from qiskit.pulse.schedule import Schedule, ScheduleBlock
 from qiskit.transpiler.coupling import CouplingMap
 from qiskit.transpiler.exceptions import TranspilerError

From 8c82a021c23a2002670d56c24f9d8e03e39db3e5 Mon Sep 17 00:00:00 2001
From: Naoki Kanazawa <nkanazawa1989@gmail.com>
Date: Sat, 7 Jan 2023 16:21:27 +0900
Subject: [PATCH 06/10] revert typehint change

---
 qiskit/providers/models/pulsedefaults.py | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/qiskit/providers/models/pulsedefaults.py b/qiskit/providers/models/pulsedefaults.py
index a91646d9db82..c65aeb256396 100644
--- a/qiskit/providers/models/pulsedefaults.py
+++ b/qiskit/providers/models/pulsedefaults.py
@@ -106,7 +106,7 @@ def __init__(self, name: str, qubits=None, sequence=None, **kwargs):
         Args:
             name (str): The name of the command
             qubits: The qubits for the command
-            sequence (List[PulseQobjInstruction]): The sequence for the Command
+            sequence (PulseQobjInstruction): The sequence for the Command
             kwargs: Optional additional fields
         """
         self._data = {}

From e8cf02eb59598c2798005d70f02544b8302b3424 Mon Sep 17 00:00:00 2001
From: Naoki Kanazawa <nkanazawa1989@gmail.com>
Date: Sun, 8 Jan 2023 12:35:05 +0900
Subject: [PATCH 07/10] add test for cal entries

---
 qiskit/pulse/calibration_entries.py           |   2 +-
 test/python/pulse/test_calibration_entries.py | 436 ++++++++++++++++++
 2 files changed, 437 insertions(+), 1 deletion(-)
 create mode 100644 test/python/pulse/test_calibration_entries.py

diff --git a/qiskit/pulse/calibration_entries.py b/qiskit/pulse/calibration_entries.py
index 2e4366aefce5..c2f991936e10 100644
--- a/qiskit/pulse/calibration_entries.py
+++ b/qiskit/pulse/calibration_entries.py
@@ -213,7 +213,7 @@ def __init__(
         """
         super().__init__(arguments=arguments)
 
-        self._converter = converter or QobjToInstructionConverter()
+        self._converter = converter or QobjToInstructionConverter(pulse_library=[])
         self._name = name
         self._source = None
 
diff --git a/test/python/pulse/test_calibration_entries.py b/test/python/pulse/test_calibration_entries.py
new file mode 100644
index 000000000000..3259d3b8a6aa
--- /dev/null
+++ b/test/python/pulse/test_calibration_entries.py
@@ -0,0 +1,436 @@
+# This code is part of Qiskit.
+#
+# (C) Copyright IBM 2023.
+#
+# 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.
+
+"""Test for calibration entries."""
+
+import numpy as np
+
+from qiskit.circuit.parameter import Parameter
+from qiskit.pulse import (
+    Schedule,
+    ScheduleBlock,
+    Play,
+    ShiftPhase,
+    Constant,
+    Waveform,
+    DriveChannel,
+)
+from qiskit.pulse.calibration_entries import (
+    ScheduleDef,
+    CallableDef,
+    PulseQobjDef,
+)
+from qiskit.pulse.exceptions import PulseError
+from qiskit.qobj.converters.pulse_instruction import QobjToInstructionConverter
+from qiskit.qobj.pulse_qobj import PulseLibraryItem, PulseQobjInstruction
+from qiskit.test import QiskitTestCase
+
+
+class TestSchedule(QiskitTestCase):
+    """Test case for the ScheduleDef."""
+
+    def test_add_schedule(self):
+        """Basic test pulse Schedule format."""
+        program = Schedule()
+        program.insert(
+            0,
+            Play(Constant(duration=10, amp=0.1, angle=0.0), DriveChannel(0)),
+            inplace=True,
+        )
+
+        entry = ScheduleDef()
+        entry.define(program)
+
+        signature_to_test = list(entry.get_signature().parameters.keys())
+        signature_ref = []
+        self.assertListEqual(signature_to_test, signature_ref)
+
+        schedule_to_test = entry.get_schedule()
+        schedule_ref = program
+        self.assertEqual(schedule_to_test, schedule_ref)
+
+    def test_add_block(self):
+        """Basic test pulse Schedule format."""
+        program = ScheduleBlock()
+        program.append(
+            Play(Constant(duration=10, amp=0.1, angle=0.0), DriveChannel(0)),
+            inplace=True,
+        )
+
+        entry = ScheduleDef()
+        entry.define(program)
+
+        signature_to_test = list(entry.get_signature().parameters.keys())
+        signature_ref = []
+        self.assertListEqual(signature_to_test, signature_ref)
+
+        schedule_to_test = entry.get_schedule()
+        schedule_ref = program
+        self.assertEqual(schedule_to_test, schedule_ref)
+
+    def test_parameterized_schedule(self):
+        """Test adding and managing parameterized schedule."""
+        param1 = Parameter("P1")
+        param2 = Parameter("P2")
+
+        program = ScheduleBlock()
+        program.append(
+            Play(Constant(duration=param1, amp=param2, angle=0.0), DriveChannel(0)),
+            inplace=True,
+        )
+
+        entry = ScheduleDef()
+        entry.define(program)
+
+        signature_to_test = list(entry.get_signature().parameters.keys())
+        signature_ref = ["P1", "P2"]
+        self.assertListEqual(signature_to_test, signature_ref)
+
+        schedule_to_test = entry.get_schedule(P1=10, P2=0.1)
+        schedule_ref = program.assign_parameters({param1: 10, param2: 0.1}, inplace=False)
+        self.assertEqual(schedule_to_test, schedule_ref)
+
+    def test_parameterized_schedule_with_user_args(self):
+        """Test adding schedule with user signature.
+
+        Bind parameters to a pulse schedule but expecting non-lexicographical order.
+        """
+        theta = Parameter("theta")
+        lam = Parameter("lam")
+        phi = Parameter("phi")
+
+        program = ScheduleBlock()
+        program.append(
+            Play(Constant(duration=10, amp=phi, angle=0.0), DriveChannel(0)),
+            inplace=True,
+        )
+        program.append(
+            Play(Constant(duration=10, amp=theta, angle=0.0), DriveChannel(0)),
+            inplace=True,
+        )
+        program.append(
+            Play(Constant(duration=10, amp=lam, angle=0.0), DriveChannel(0)),
+            inplace=True,
+        )
+
+        entry = ScheduleDef(arguments=["theta", "lam", "phi"])
+        entry.define(program)
+
+        signature_to_test = list(entry.get_signature().parameters.keys())
+        signature_ref = ["theta", "lam", "phi"]
+        self.assertListEqual(signature_to_test, signature_ref)
+
+        # Do not specify kwargs. This is order sensitive.
+        schedule_to_test = entry.get_schedule(0.1, 0.2, 0.3)
+        schedule_ref = program.assign_parameters(
+            {theta: 0.1, lam: 0.2, phi: 0.3},
+            inplace=False,
+        )
+        self.assertEqual(schedule_to_test, schedule_ref)
+
+    def test_parameterized_schedule_with_wrong_signature(self):
+        """Test raising PulseError when signature doesn't match."""
+        param1 = Parameter("P1")
+
+        program = ScheduleBlock()
+        program.append(
+            Play(Constant(duration=10, amp=param1, angle=0.0), DriveChannel(0)),
+            inplace=True,
+        )
+
+        entry = ScheduleDef(arguments=["This_is_wrong_param_name"])
+
+        with self.assertRaises(PulseError):
+            entry.define(program)
+
+    def test_equality(self):
+        """Test equality evaluation between the schedule entries."""
+        program1 = Schedule()
+        program1.insert(
+            0,
+            Play(Constant(duration=10, amp=0.1, angle=0.0), DriveChannel(0)),
+            inplace=True,
+        )
+
+        program2 = Schedule()
+        program2.insert(
+            0,
+            Play(Constant(duration=10, amp=0.2, angle=0.0), DriveChannel(0)),
+            inplace=True,
+        )
+
+        entry1 = ScheduleDef()
+        entry1.define(program1)
+
+        entry2 = ScheduleDef()
+        entry2.define(program2)
+
+        entry3 = ScheduleDef()
+        entry3.define(program1)
+
+        self.assertEqual(entry1, entry3)
+        self.assertNotEqual(entry1, entry2)
+
+
+class TestCallable(QiskitTestCase):
+    """Test case for the CallableDef."""
+
+    def test_add_callable(self):
+        """Basic test callable format."""
+        program = Schedule()
+        program.insert(
+            0,
+            Play(Constant(duration=10, amp=0.1, angle=0.0), DriveChannel(0)),
+            inplace=True,
+        )
+
+        def factory():
+            return program
+
+        entry = CallableDef()
+        entry.define(factory)
+
+        signature_to_test = list(entry.get_signature().parameters.keys())
+        signature_ref = []
+        self.assertListEqual(signature_to_test, signature_ref)
+
+        schedule_to_test = entry.get_schedule()
+        schedule_ref = program
+        self.assertEqual(schedule_to_test, schedule_ref)
+
+    def test_add_callable_with_argument(self):
+        """Basic test callable format."""
+
+        def factory(var1, var2):
+            program = Schedule()
+            if var1 > 0:
+                program.insert(
+                    0,
+                    Play(Constant(duration=var2, amp=var1, angle=0.0), DriveChannel(0)),
+                    inplace=True,
+                )
+            else:
+                program.insert(
+                    0,
+                    Play(Constant(duration=var2, amp=np.abs(var1), angle=np.pi), DriveChannel(0)),
+                    inplace=True,
+                )
+            return program
+
+        entry = CallableDef()
+        entry.define(factory)
+
+        signature_to_test = list(entry.get_signature().parameters.keys())
+        signature_ref = ["var1", "var2"]
+        self.assertListEqual(signature_to_test, signature_ref)
+
+        schedule_to_test = entry.get_schedule(0.1, 10)
+        schedule_ref = Schedule()
+        schedule_ref.insert(
+            0,
+            Play(Constant(duration=10, amp=0.1, angle=0.0), DriveChannel(0)),
+            inplace=True,
+        )
+        self.assertEqual(schedule_to_test, schedule_ref)
+
+        schedule_to_test = entry.get_schedule(-0.1, 10)
+        schedule_ref = Schedule()
+        schedule_ref.insert(
+            0,
+            Play(Constant(duration=10, amp=0.1, angle=np.pi), DriveChannel(0)),
+            inplace=True,
+        )
+        self.assertEqual(schedule_to_test, schedule_ref)
+
+    def test_equality(self):
+        """Test equality evaluation between the callable entries.
+
+        This does NOT compare the code. Just object equality.
+        """
+
+        def factory1():
+            return Schedule()
+
+        def factory2():
+            return Schedule()
+
+        entry1 = CallableDef()
+        entry1.define(factory1)
+
+        entry2 = CallableDef()
+        entry2.define(factory2)
+
+        entry3 = CallableDef()
+        entry3.define(factory1)
+
+        self.assertEqual(entry1, entry3)
+        self.assertNotEqual(entry1, entry2)
+
+
+class TestPulseQobj(QiskitTestCase):
+    """Test case for the PulseQobjDef."""
+
+    def setUp(self):
+        super().setUp()
+        self.converter = QobjToInstructionConverter(
+            pulse_library=[
+                PulseLibraryItem(name="waveform", samples=[0.3, 0.1, 0.2, 0.2, 0.3]),
+            ]
+        )
+
+    def test_add_qobj(self):
+        """Basic test PulseQobj format."""
+        serialized_program = [
+            PulseQobjInstruction(
+                name="parametric_pulse",
+                t0=0,
+                ch="d0",
+                label="TestPulse",
+                pulse_shape="constant",
+                parameters={"amp": 0.1 + 0j, "duration": 10},
+            ),
+            PulseQobjInstruction(
+                name="waveform",
+                t0=20,
+                ch="d0",
+            ),
+        ]
+
+        entry = PulseQobjDef(converter=self.converter, name="my_gate")
+        entry.define(serialized_program)
+
+        signature_to_test = list(entry.get_signature().parameters.keys())
+        signature_ref = []
+        self.assertListEqual(signature_to_test, signature_ref)
+
+        schedule_to_test = entry.get_schedule()
+        schedule_ref = Schedule()
+        schedule_ref.insert(
+            0,
+            Play(Constant(duration=10, amp=0.1, angle=0.0), DriveChannel(0)),
+            inplace=True,
+        )
+        schedule_ref.insert(
+            20,
+            Play(Waveform([0.3, 0.1, 0.2, 0.2, 0.3]), DriveChannel(0)),
+            inplace=True,
+        )
+        self.assertEqual(schedule_to_test, schedule_ref)
+
+    def test_parameterized_qobj(self):
+        """Test adding and managing parameterized qobj.
+
+        Note that pulse parameter cannot be parameterized by convention.
+        """
+        serialized_program = [
+            PulseQobjInstruction(
+                name="parametric_pulse",
+                t0=0,
+                ch="d0",
+                label="TestPulse",
+                pulse_shape="constant",
+                parameters={"amp": 0.1, "duration": 10},
+            ),
+            PulseQobjInstruction(
+                name="fc",
+                t0=0,
+                ch="d0",
+                phase="P1",
+            ),
+        ]
+
+        entry = PulseQobjDef(converter=self.converter, name="my_gate")
+        entry.define(serialized_program)
+
+        signature_to_test = list(entry.get_signature().parameters.keys())
+        signature_ref = ["P1"]
+        self.assertListEqual(signature_to_test, signature_ref)
+
+        schedule_to_test = entry.get_schedule(P1=1.57)
+        schedule_ref = Schedule()
+        schedule_ref.insert(
+            0,
+            Play(Constant(duration=10, amp=0.1, angle=0.0), DriveChannel(0)),
+            inplace=True,
+        )
+        schedule_ref.insert(
+            0,
+            ShiftPhase(1.57, DriveChannel(0)),
+            inplace=True,
+        )
+        self.assertEqual(schedule_to_test, schedule_ref)
+
+    def test_equality(self):
+        """Test equality evaluation between the pulse qobj entries."""
+        serialized_program1 = [
+            PulseQobjInstruction(
+                name="parametric_pulse",
+                t0=0,
+                ch="d0",
+                label="TestPulse",
+                pulse_shape="constant",
+                parameters={"amp": 0.1, "duration": 10},
+            )
+        ]
+
+        serialized_program2 = [
+            PulseQobjInstruction(
+                name="parametric_pulse",
+                t0=0,
+                ch="d0",
+                label="TestPulse",
+                pulse_shape="constant",
+                parameters={"amp": 0.2, "duration": 10},
+            )
+        ]
+
+        entry1 = PulseQobjDef(name="my_gate1")
+        entry1.define(serialized_program1)
+
+        entry2 = PulseQobjDef(name="my_gate2")
+        entry2.define(serialized_program2)
+
+        entry3 = PulseQobjDef(name="my_gate3")
+        entry3.define(serialized_program1)
+
+        self.assertEqual(entry1, entry3)
+        self.assertNotEqual(entry1, entry2)
+
+    def test_equality_with_schedule(self):
+        """Test equality, but other is schedule entry.
+
+        Because the pulse qobj entry is a subclass of the schedule entry,
+        these instances can be compared by the generated definition, i.e. Schedule.
+        """
+        serialized_program = [
+            PulseQobjInstruction(
+                name="parametric_pulse",
+                t0=0,
+                ch="d0",
+                label="TestPulse",
+                pulse_shape="constant",
+                parameters={"amp": 0.1, "duration": 10},
+            )
+        ]
+        entry1 = PulseQobjDef(name="qobj_entry")
+        entry1.define(serialized_program)
+
+        program = Schedule()
+        program.insert(
+            0,
+            Play(Constant(duration=10, amp=0.1, angle=0.0), DriveChannel(0)),
+            inplace=True,
+        )
+        entry2 = ScheduleDef()
+        entry2.define(program)
+
+        self.assertEqual(entry1, entry2)

From 95a49607c63a2b950683bab63f344d3ecda42f34 Mon Sep 17 00:00:00 2001
From: Naoki Kanazawa <nkanazawa1989@gmail.com>
Date: Sun, 8 Jan 2023 13:23:38 +0900
Subject: [PATCH 08/10] add test for parallel transpile and fix converter

---
 .../fake_provider/utils/backend_converter.py  | 18 +++++----
 test/python/compiler/test_transpiler.py       | 39 ++++++++++++++++++-
 2 files changed, 48 insertions(+), 9 deletions(-)

diff --git a/qiskit/providers/fake_provider/utils/backend_converter.py b/qiskit/providers/fake_provider/utils/backend_converter.py
index c58881c1257c..9105f291b1dc 100644
--- a/qiskit/providers/fake_provider/utils/backend_converter.py
+++ b/qiskit/providers/fake_provider/utils/backend_converter.py
@@ -113,17 +113,19 @@ def convert_to_target(conf_dict: dict, props_dict: dict = None, defs_dict: dict
         inst_map = pulse_defs.instruction_schedule_map
         for inst in inst_map.instructions:
             for qarg in inst_map.qubits_with_instruction(inst):
-                sched = inst_map.get(inst, qarg)
+                try:
+                    qargs = tuple(qarg)
+                except TypeError:
+                    qargs = (qarg,)
+                # Do NOT call .get method. This parses Qpbj immediately.
+                # This operation is computationally expensive and should be bypassed.
+                calibration_entry = inst_map._map[inst][qargs]
                 if inst in target:
-                    try:
-                        qarg = tuple(qarg)
-                    except TypeError:
-                        qarg = (qarg,)
                     if inst == "measure":
-                        for qubit in qarg:
-                            target[inst][(qubit,)].calibration = sched
+                        for qubit in qargs:
+                            target[inst][(qubit,)].calibration = calibration_entry
                     else:
-                        target[inst][qarg].calibration = sched
+                        target[inst][qargs].calibration = calibration_entry
     target.add_instruction(
         Delay(Parameter("t")), {(bit,): None for bit in range(target.num_qubits)}
     )
diff --git a/test/python/compiler/test_transpiler.py b/test/python/compiler/test_transpiler.py
index 0c984b3875d7..8ede5702ae60 100644
--- a/test/python/compiler/test_transpiler.py
+++ b/test/python/compiler/test_transpiler.py
@@ -53,7 +53,7 @@
     FakeMumbaiV2,
 )
 from qiskit.transpiler import Layout, CouplingMap
-from qiskit.transpiler import PassManager
+from qiskit.transpiler import PassManager, TransformationPass
 from qiskit.transpiler.target import Target
 from qiskit.transpiler.exceptions import TranspilerError
 from qiskit.transpiler.passes import BarrierBeforeFinalMeasurements, GateDirection
@@ -1862,3 +1862,40 @@ def test_parallel_dispatch(self, opt_level):
         for count in counts:
             self.assertTrue(math.isclose(count["0000000000000000"], 500, rel_tol=0.1))
             self.assertTrue(math.isclose(count["0111111111111111"], 500, rel_tol=0.1))
+
+    def test_parallel_dispatch_lazy_cal_loading(self):
+        """Test adding calibration by lazy loading in parallel environment."""
+
+        class TestAddCalibration(TransformationPass):
+            """A fake pass to test lazy pulse qobj loading in parallel environment."""
+
+            def __init__(self, target):
+                """Instantiate with target."""
+                super().__init__()
+                self.target = target
+
+            def run(self, dag):
+                """Run test pass that adds calibration of SX gate of qubit 0."""
+                dag.add_calibration(
+                    "sx",
+                    qubits=(0,),
+                    schedule=self.target["sx"][(0,)].calibration,  # PulseQobj is parsed here
+                )
+                return dag
+
+        backend = FakeMumbaiV2()
+
+        # This target has PulseQobj entries that provides a serialized schedule data
+        pass_ = TestAddCalibration(backend.target)
+        pm = PassManager(passes=[pass_])
+        self.assertIsNone(backend.target["sx"][(0,)]._calibration._definition)
+
+        qc = QuantumCircuit(1)
+        qc.sx(0)
+        qc_copied = [qc for _ in range(10)]
+
+        qcs_cal_added = pm.run(qc_copied)
+        ref_cal = backend.target["sx"][(0,)].calibration
+        for qc_test in qcs_cal_added:
+            added_cal = qc_test.calibrations["sx"][((0,), tuple())]
+            self.assertEqual(added_cal, ref_cal)

From 880a252ff549d3ad588947f295d19b3278d89213 Mon Sep 17 00:00:00 2001
From: Naoki Kanazawa <nkanazawa1989@gmail.com>
Date: Thu, 12 Jan 2023 11:31:50 +0900
Subject: [PATCH 09/10] add API for lazy get

---
 qiskit/providers/backend_compat.py            | 20 +++++++-------
 .../fake_provider/utils/backend_converter.py  |  4 +--
 qiskit/pulse/instruction_schedule_map.py      | 26 ++++++++++++++++---
 3 files changed, 36 insertions(+), 14 deletions(-)

diff --git a/qiskit/providers/backend_compat.py b/qiskit/providers/backend_compat.py
index 60889ff22c70..6adb19c86ab2 100644
--- a/qiskit/providers/backend_compat.py
+++ b/qiskit/providers/backend_compat.py
@@ -119,17 +119,19 @@ def convert_to_target(
         inst_map = defaults.instruction_schedule_map
         for inst in inst_map.instructions:
             for qarg in inst_map.qubits_with_instruction(inst):
-                sched = inst_map.get(inst, qarg)
+                try:
+                    qargs = tuple(qarg)
+                except TypeError:
+                    qargs = (qarg,)
+                # Do NOT call .get method. This parses Qpbj immediately.
+                # This operation is computationally expensive and should be bypassed.
+                calibration_entry = inst_map._get_calibration_entry(inst, qargs)
                 if inst in target:
-                    try:
-                        qarg = tuple(qarg)
-                    except TypeError:
-                        qarg = (qarg,)
                     if inst == "measure":
-                        for qubit in qarg:
-                            target[inst][(qubit,)].calibration = sched
-                    elif qarg in target[inst]:
-                        target[inst][qarg].calibration = sched
+                        for qubit in qargs:
+                            target[inst][(qubit,)].calibration = calibration_entry
+                    elif qargs in target[inst]:
+                        target[inst][qargs].calibration = calibration_entry
     combined_global_ops = set()
     if configuration.basis_gates:
         combined_global_ops.update(configuration.basis_gates)
diff --git a/qiskit/providers/fake_provider/utils/backend_converter.py b/qiskit/providers/fake_provider/utils/backend_converter.py
index 9105f291b1dc..c6aeeb426fa8 100644
--- a/qiskit/providers/fake_provider/utils/backend_converter.py
+++ b/qiskit/providers/fake_provider/utils/backend_converter.py
@@ -119,12 +119,12 @@ def convert_to_target(conf_dict: dict, props_dict: dict = None, defs_dict: dict
                     qargs = (qarg,)
                 # Do NOT call .get method. This parses Qpbj immediately.
                 # This operation is computationally expensive and should be bypassed.
-                calibration_entry = inst_map._map[inst][qargs]
+                calibration_entry = inst_map._get_calibration_entry(inst, qargs)
                 if inst in target:
                     if inst == "measure":
                         for qubit in qargs:
                             target[inst][(qubit,)].calibration = calibration_entry
-                    else:
+                    elif qargs in target[inst]:
                         target[inst][qargs].calibration = calibration_entry
     target.add_instruction(
         Delay(Parameter("t")), {(bit,): None for bit in range(target.num_qubits)}
diff --git a/qiskit/pulse/instruction_schedule_map.py b/qiskit/pulse/instruction_schedule_map.py
index df1599e11327..dec2b0aec9f4 100644
--- a/qiskit/pulse/instruction_schedule_map.py
+++ b/qiskit/pulse/instruction_schedule_map.py
@@ -188,14 +188,34 @@ def get(
 
         Returns:
             The Schedule defined for the input.
+        """
+        return self._get_calibration_entry(instruction, qubits).get_schedule(*params, **kwparams)
 
-        Raises:
-            PulseError: When invalid parameters are specified.
+    def _get_calibration_entry(
+        self,
+        instruction: Union[str, Instruction],
+        qubits: Union[int, Iterable[int]],
+    ) -> CalibrationEntry:
+        """Return the :class:`.CalibrationEntry` without generating schedule.
+
+        When calibration entry is un-parsed Pulse Qobj, this returns calibration
+        without parsing it. :meth:`CalibrationEntry.get_schedule` method
+        must be manually called with assigned parameters to get corresponding pulse schedule.
+
+        This method is expected be directly used internally by the V2 backend converter
+        for faster loading of the backend calibrations.
+
+        Args:
+            instruction: Name of the instruction or the instruction itself.
+            qubits: The qubits for the instruction.
+
+        Returns:
+            The calibration entry.
         """
         instruction = _get_instruction_string(instruction)
         self.assert_has(instruction, qubits)
 
-        return self._map[instruction][_to_tuple(qubits)].get_schedule(*params, **kwparams)
+        return self._map[instruction][_to_tuple(qubits)]
 
     def add(
         self,

From 56b41ce8b3213c9e144d581b2ba02fd835258832 Mon Sep 17 00:00:00 2001
From: Naoki Kanazawa <nkanazawa1989@gmail.com>
Date: Fri, 13 Jan 2023 06:06:31 +0900
Subject: [PATCH 10/10] add release note

---
 .../load-backend-fast-9030885adcd9248f.yaml   | 20 +++++++++++++++++++
 1 file changed, 20 insertions(+)
 create mode 100644 releasenotes/notes/load-backend-fast-9030885adcd9248f.yaml

diff --git a/releasenotes/notes/load-backend-fast-9030885adcd9248f.yaml b/releasenotes/notes/load-backend-fast-9030885adcd9248f.yaml
new file mode 100644
index 000000000000..bb16cf3acba9
--- /dev/null
+++ b/releasenotes/notes/load-backend-fast-9030885adcd9248f.yaml
@@ -0,0 +1,20 @@
+---
+features:
+  - |
+    :class:`.InstructionScheduleMap` has been updated to store backend calibration data
+    in the format of PulseQobj JSON and invokes conversion when the data is accessed
+    for the first time, i.e. lazy conversion is implemented.
+    This internal logic update drastically improves the performance of loading backend
+    especially with many calibration entries.
+  - |
+    New module :mod:`qiskit.pulse.calibration_entries` has been added. This
+    contains several wrapper classes for different pulse schedule representations.
+
+    * :class:`~qiskit.pulse.calibration_entries.ScheduleDef`
+    * :class:`~qiskit.pulse.calibration_entries.CallableDef`
+    * :class:`~qiskit.pulse.calibration_entries.PulseQobjDef`
+
+    These classes implement get_schedule and get_signature method
+    that returns pulse schedule and parameter names to assign, respectively.
+    These classes are internally managed by the instruction schedule map or backend target,
+    and thus they will not appear in the user programs.