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 c58881c1257c..c6aeeb426fa8 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._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)} ) 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): diff --git a/qiskit/providers/models/pulsedefaults.py b/qiskit/providers/models/pulsedefaults.py index e5d73f80ccfa..c65aeb256396 100644 --- a/qiskit/providers/models/pulsedefaults.py +++ b/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): diff --git a/qiskit/pulse/calibration_entries.py b/qiskit/pulse/calibration_entries.py new file mode 100644 index 000000000000..c2f991936e10 --- /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(pulse_library=[]) + 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 3ddea95f4ea5..dec2b0aec9f4 100644 --- a/qiskit/pulse/instruction_schedule_map.py +++ b/qiskit/pulse/instruction_schedule_map.py @@ -26,31 +26,23 @@ inst_map = backend.defaults().instruction_schedule_map """ -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 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 InstructionScheduleMap: """Mapping from :py:class:`~qiskit.circuit.QuantumCircuit` @@ -70,10 +62,10 @@ 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, Generator)) + self._map = defaultdict(functools.partial(defaultdict, CalibrationEntry)) # A backwards mapping from qubit to supported instructions self._qubit_instructions = defaultdict(set) @@ -81,10 +73,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 @@ -198,46 +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) - 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)] def add( self, @@ -269,45 +247,49 @@ 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(arguments) + # add metadata + if "publisher" not in schedule.metadata: + schedule.metadata["publisher"] = CalibrationPublisher.QISKIT 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() else: raise PulseError( "Supplied schedule must be one of the Schedule, ScheduleBlock or a " "callable that outputs a schedule." ) + entry.define(schedule) + self._add(instruction, qubits, entry) - # add metadata - if hasattr(schedule, "metadata") and "publisher" not in schedule.metadata: - schedule.metadata["publisher"] = CalibrationPublisher.QISKIT + def _add( + self, + instruction_name: str, + qubits: Tuple[int, ...], + entry: CalibrationEntry, + ): + """A method to resister calibration entry. + + .. note:: + + 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: + instruction_name: Name of instruction. + qubits: List of qubits that this calibration is applied. + entry: Calibration entry to register. - self._map[instruction][qubits] = Generator(schedule, signature) - self._qubit_instructions[qubits].add(instruction) + :meta public: + """ + 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]] @@ -321,12 +303,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 +351,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.""" 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): 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 0590c972eb86..a3872fa26b0d 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 @@ -28,6 +29,8 @@ from qiskit.circuit.parameter import Parameter 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 qiskit.transpiler.instruction_durations import InstructionDurations @@ -51,13 +54,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 +69,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})" ) 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. 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) 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)