Lazy pulse qobj loading for large backend

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)

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._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
-                    else:
-                        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
     target.add_instruction(
         Delay(Parameter("t")), {(bit,): None for bit in range(target.num_qubits)}
     )

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):

qiskit/providers/models/pulsedefaults.py

@@ -12,11 +12,9 @@
 
 
 """Model and schema for pulse defaults."""
-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
 
@@ -152,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)
 
 
@@ -200,13 +201,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(converter=self.converter, name=inst.name)
+            entry.define(inst.sequence)
+            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
@@ -267,15 +271,25 @@ def from_dict(cls, data):
         Returns:
             PulseDefaults: The PulseDefaults from the input dictionary.
         """
-        in_data = copy.copy(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"))
+        schema = {
+            "pulse_library": PulseLibraryItem,
+            "cmd_def": Command,
+            "meas_kernel": MeasurementKernel,
+            "discriminator": Discriminator,
+        }
+
+        # Pulse defaults 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 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):