diff --git a/qiskit/providers/models/pulsedefaults.py b/qiskit/providers/models/pulsedefaults.py index c65aeb256396..2ee1594571e1 100644 --- a/qiskit/providers/models/pulsedefaults.py +++ b/qiskit/providers/models/pulsedefaults.py @@ -205,7 +205,7 @@ def __init__( for inst in cmd_def: entry = PulseQobjDef(converter=self.converter, name=inst.name) - entry.define(inst.sequence) + entry.define(inst.sequence, user_provided=False) self.instruction_schedule_map._add( instruction_name=inst.name, qubits=tuple(inst.qubits), diff --git a/qiskit/pulse/calibration_entries.py b/qiskit/pulse/calibration_entries.py index 29f32b89f04d..1aa64b8779b7 100644 --- a/qiskit/pulse/calibration_entries.py +++ b/qiskit/pulse/calibration_entries.py @@ -34,11 +34,12 @@ class CalibrationEntry(metaclass=ABCMeta): """A metaclass of a calibration entry.""" @abstractmethod - def define(self, definition: Any): + def define(self, definition: Any, user_provided: bool): """Attach definition to the calibration entry. Args: definition: Definition of this entry. + user_provided: If this entry is defined by user. """ pass @@ -64,6 +65,12 @@ def get_schedule(self, *args, **kwargs) -> Union[Schedule, ScheduleBlock]: """ pass + @property + @abstractmethod + def user_provided(self) -> bool: + """Return if this entry is user defined.""" + pass + class ScheduleDef(CalibrationEntry): """In-memory Qiskit Pulse representation. @@ -90,6 +97,11 @@ def __init__(self, arguments: Optional[Sequence[str]] = None): self._definition = None self._signature = None + self._user_provided = None + + @property + def user_provided(self) -> bool: + return self._user_provided def _parse_argument(self): """Generate signature from program and user provided argument names.""" @@ -120,35 +132,48 @@ def _parse_argument(self): ) self._signature = signature - def define(self, definition: Union[Schedule, ScheduleBlock]): + def define( + self, + definition: Union[Schedule, ScheduleBlock], + user_provided: bool = True, + ): self._definition = definition - # add metadata - if "publisher" not in definition.metadata: - definition.metadata["publisher"] = CalibrationPublisher.QISKIT self._parse_argument() + self._user_provided = user_provided def get_signature(self) -> inspect.Signature: return self._signature def get_schedule(self, *args, **kwargs) -> Union[Schedule, ScheduleBlock]: if not args and not kwargs: - return self._definition - try: - to_bind = self.get_signature().bind_partial(*args, **kwargs) - except TypeError as ex: - raise PulseError("Assigned parameter doesn't match with schedule parameters.") from ex - value_dict = {} - for param in self._definition.parameters: - # Schedule allows partial bind. This results in parameterized Schedule. + out = self._definition + else: try: - value_dict[param] = to_bind.arguments[param.name] - except KeyError: - pass - return self._definition.assign_parameters(value_dict, inplace=False) + to_bind = self.get_signature().bind_partial(*args, **kwargs) + except TypeError as ex: + raise PulseError( + "Assigned parameter doesn't match with schedule parameters." + ) from ex + value_dict = {} + for param in self._definition.parameters: + # Schedule allows partial bind. This results in parameterized Schedule. + try: + value_dict[param] = to_bind.arguments[param.name] + except KeyError: + pass + out = self._definition.assign_parameters(value_dict, inplace=False) + if "publisher" not in out.metadata: + if self.user_provided: + out.metadata["publisher"] = CalibrationPublisher.QISKIT + else: + out.metadata["publisher"] = CalibrationPublisher.BACKEND_PROVIDER + return out def __eq__(self, other): # This delegates equality check to Schedule or ScheduleBlock. - return self._definition == other._definition + if hasattr(other, "_definition"): + return self._definition == other._definition + return False def __str__(self): out = f"Schedule {self._definition.name}" @@ -171,10 +196,20 @@ def __init__(self): """Define an empty entry.""" self._definition = None self._signature = None + self._user_provided = None + + @property + def user_provided(self) -> bool: + return self._user_provided - def define(self, definition: Callable): + def define( + self, + definition: Callable, + user_provided: bool = True, + ): self._definition = definition self._signature = inspect.signature(definition) + self._user_provided = user_provided def get_signature(self) -> inspect.Signature: return self._signature @@ -186,17 +221,20 @@ def get_schedule(self, *args, **kwargs) -> Union[Schedule, ScheduleBlock]: to_bind.apply_defaults() except TypeError as ex: raise PulseError("Assigned parameter doesn't match with function signature.") from ex - - schedule = self._definition(**to_bind.arguments) - # add metadata - if "publisher" not in schedule.metadata: - schedule.metadata["publisher"] = CalibrationPublisher.QISKIT - return schedule + out = self._definition(**to_bind.arguments) + if "publisher" not in out.metadata: + if self.user_provided: + out.metadata["publisher"] = CalibrationPublisher.QISKIT + else: + out.metadata["publisher"] = CalibrationPublisher.BACKEND_PROVIDER + return out def __eq__(self, other): # We cannot evaluate function equality without parsing python AST. - # This simply compares wether they are the same object. - return self._definition is other._definition + # This simply compares weather they are the same object. + if hasattr(other, "_definition"): + return self._definition == other._definition + return False def __str__(self): params_str = ", ".join(self.get_signature().parameters.keys()) @@ -237,14 +275,17 @@ def _build_schedule(self): for qobj_inst in self._source: for qiskit_inst in self._converter._get_sequences(qobj_inst): schedule.insert(qobj_inst.t0, qiskit_inst, inplace=True) - schedule.metadata["publisher"] = CalibrationPublisher.BACKEND_PROVIDER - self._definition = schedule self._parse_argument() - def define(self, definition: List[PulseQobjInstruction]): + def define( + self, + definition: List[PulseQobjInstruction], + user_provided: bool = False, + ): # This doesn't generate signature immediately, because of lazy schedule build. self._source = definition + self._user_provided = user_provided def get_signature(self) -> inspect.Signature: if self._definition is None: @@ -261,9 +302,11 @@ def __eq__(self, other): # If both objects are Qobj just check Qobj equality. return self._source == other._source if isinstance(other, ScheduleDef) and self._definition is None: - # To compare with other scheudle def, this also generates schedule object from qobj. + # To compare with other schedule def, this also generates schedule object from qobj. self._build_schedule() - return self._definition == other._definition + if hasattr(other, "_definition"): + return self._definition == other._definition + return False def __str__(self): if self._definition is None: diff --git a/qiskit/pulse/instruction_schedule_map.py b/qiskit/pulse/instruction_schedule_map.py index c5d144d11ecd..7fe22cef1e94 100644 --- a/qiskit/pulse/instruction_schedule_map.py +++ b/qiskit/pulse/instruction_schedule_map.py @@ -39,8 +39,8 @@ CalibrationEntry, ScheduleDef, CallableDef, - PulseQobjDef, # for backward compatibility + PulseQobjDef, CalibrationPublisher, ) from qiskit.pulse.exceptions import PulseError @@ -77,7 +77,7 @@ def has_custom_gate(self) -> bool: """Return ``True`` if the map has user provided instruction.""" for qubit_inst in self._map.values(): for entry in qubit_inst.values(): - if not isinstance(entry, PulseQobjDef): + if entry.user_provided: return True return False @@ -264,7 +264,7 @@ def add( "Supplied schedule must be one of the Schedule, ScheduleBlock or a " "callable that outputs a schedule." ) - entry.define(schedule) + entry.define(schedule, user_provided=True) self._add(instruction, qubits, entry) def _add( diff --git a/qiskit/transpiler/passes/calibration/pulse_gate.py b/qiskit/transpiler/passes/calibration/pulse_gate.py index ca81316754e9..71536ed89fd5 100644 --- a/qiskit/transpiler/passes/calibration/pulse_gate.py +++ b/qiskit/transpiler/passes/calibration/pulse_gate.py @@ -80,7 +80,7 @@ def supported(self, node_op: CircuitInst, qubits: List) -> bool: Returns: Return ``True`` is calibration can be provided. """ - return self.target.instruction_supported(operation_name=node_op.name, qargs=qubits) + return self.target.calibration_supported(node_op.name, tuple(qubits)) def get_calibration(self, node_op: CircuitInst, qubits: List) -> Union[Schedule, ScheduleBlock]: """Gets the calibrated schedule for the given instruction and qubits. @@ -95,18 +95,4 @@ def get_calibration(self, node_op: CircuitInst, qubits: List) -> Union[Schedule, Raises: TranspilerError: When node is parameterized and calibration is raw schedule object. """ - inst_property = self.target[node_op.name][tuple(qubits)] - if not node_op.params: - return inst_property.calibration - try: - # CircuitInstruction doesn't preserve parameter name after parameter binding. - # Thus schedule cannot generate bind dictionary. - # Use CalibraionEntry to utilize inspected signature object. - calibration_entry = inst_property._calibration - return calibration_entry.get_schedule(*node_op.params) - except AttributeError as ex: - raise TranspilerError( - f"Calibraton for {node_op.name} of {qubits} is not a CalibraryEntry instance. " - f"Mapping from parameter values {node_op.params} to parameter objects " - f"in the schedule cannot be identified." - ) from ex + return self.target.get_calibration(node_op.name, tuple(qubits), *node_op.params) diff --git a/qiskit/transpiler/target.py b/qiskit/transpiler/target.py index a3b51e12dd20..be4ae6e51c15 100644 --- a/qiskit/transpiler/target.py +++ b/qiskit/transpiler/target.py @@ -17,7 +17,7 @@ from a backend """ import warnings -from typing import Union +from typing import Tuple, Union from collections.abc import Mapping from collections import defaultdict import datetime @@ -28,6 +28,7 @@ import rustworkx as rx from qiskit.circuit.parameter import Parameter +from qiskit.circuit.parameterexpression import ParameterValueType from qiskit.circuit.gate import Gate from qiskit.circuit.library.standard_gates import get_standard_gate_name_mapping from qiskit.pulse.instruction_schedule_map import InstructionScheduleMap @@ -91,7 +92,7 @@ def calibration(self): def calibration(self, calibration: Union[Schedule, ScheduleBlock, CalibrationEntry]): if isinstance(calibration, (Schedule, ScheduleBlock)): new_entry = ScheduleDef() - new_entry.define(calibration) + new_entry.define(calibration, user_provided=True) else: new_entry = calibration self._calibration = new_entry @@ -441,6 +442,8 @@ def update_from_instruction_schedule_map(self, inst_map, inst_name_map=None, err this dictionary. If one is not found in ``error_dict`` then ``None`` will be used. """ + get_calibration = getattr(inst_map, "_get_calibration_entry") + for inst_name in inst_map.instructions: out_props = {} for qargs in inst_map.qubits_with_instruction(inst_name): @@ -448,29 +451,36 @@ def update_from_instruction_schedule_map(self, inst_map, inst_name_map=None, err qargs = tuple(qargs) except TypeError: qargs = (qargs,) - entry = inst_map._get_calibration_entry(inst_name, qargs) - if inst_name in self._gate_map and qargs in self._gate_map[inst_name]: - default_entry = self._gate_map[inst_name][qargs]._calibration - if entry == default_entry: - # Skip parsing existing entry, e.g. backend calibrated schedule. - continue - if self.dt is not None: - duration = entry.get_schedule().duration * self.dt - else: - duration = None - if inst_name in self._gate_map and error_dict is not None: - error_inst = error_dict.get(inst_name) - if error_inst: - error = error_inst.get(qargs) + try: + props = self._gate_map[inst_name][qargs] + except (KeyError, TypeError): + props = None + + entry = get_calibration(inst_name, qargs) + if entry.user_provided and getattr(props, "_calibration", None) != entry: + # It only copies user-provided calibration from the inst map. + # Backend defined entry must already exist in Target. + if self.dt is not None: + duration = entry.get_schedule().duration * self.dt else: - error = None + duration = None + props = InstructionProperties( + duration=duration, + calibration=entry, + ) else: - error = None - out_props[qargs] = InstructionProperties( - duration=duration, - error=error, - calibration=entry, - ) + if props is None: + # Edge case. Calibration is backend defined, but this is not + # registered in the backend target. Ignore this entry. + continue + try: + # Update gate error if provided. + props.error = error_dict[inst_name][qargs] + except (KeyError, TypeError): + pass + out_props[qargs] = props + if not out_props: + continue if inst_name not in self._gate_map: if inst_name_map is None: inst_name_map = get_standard_gate_name_mapping() @@ -797,6 +807,55 @@ def check_obj_params(parameters, obj): ) return False + def calibration_supported( + self, + operation_name: str, + qargs: Tuple[int, ...], + ) -> bool: + """Return whether the instruction (operation + qubits) defines a calibration. + + Args: + operation_name: The name of the operation for the instruction. + qargs: The tuple of qubit indices for the instruction. + + Returns: + Returns ``True`` if the calibration is supported and ``False`` if it isn't. + """ + qargs = tuple(qargs) + if operation_name not in self._gate_map: + return False + if qargs not in self._gate_map[operation_name]: + return False + return getattr(self._gate_map[operation_name][qargs], "_calibration") is not None + + def get_calibration( + self, + operation_name: str, + qargs: Tuple[int, ...], + *args: ParameterValueType, + **kwargs: ParameterValueType, + ) -> Union[Schedule, ScheduleBlock]: + """Get calibrated pulse schedule for the instruction. + + If calibration is templated with parameters, one can also provide those values + to build a schedule with assigned parameters. + + Args: + operation_name: The name of the operation for the instruction. + qargs: The tuple of qubit indices for the instruction. + args: Parameter values to build schedule if any. + kwargs: Parameter values with name to build schedule if any. + + Returns: + Calibrated pulse schedule of corresponding instruction. + """ + if not self.calibration_supported(operation_name, qargs): + raise KeyError( + f"Calibration of instruction {operation_name} for qubit {qargs} is not defined." + ) + cal_entry = getattr(self._gate_map[operation_name][qargs], "_calibration") + return cal_entry.get_schedule(*args, **kwargs) + @property def operation_names(self): """Get the operation names in the target.""" diff --git a/test/python/transpiler/test_pulse_gate_pass.py b/test/python/transpiler/test_pulse_gate_pass.py index 70bab9f11460..0ada1e03a04e 100644 --- a/test/python/transpiler/test_pulse_gate_pass.py +++ b/test/python/transpiler/test_pulse_gate_pass.py @@ -15,9 +15,9 @@ import ddt from qiskit import pulse, circuit, transpile -from qiskit.test import QiskitTestCase from qiskit.providers.fake_provider import FakeAthens, FakeAthensV2 from qiskit.quantum_info.random import random_unitary +from qiskit.test import QiskitTestCase @ddt.ddt @@ -350,8 +350,11 @@ def test_transpile_with_instmap_with_v2backend_with_custom_gate(self, opt_level) Test case from Qiskit/qiskit-terra/#9489 """ + with pulse.build(name="custom") as rabi12: + pulse.play(pulse.Constant(100, 0.4), pulse.DriveChannel(0)) + instmap = FakeAthens().defaults().instruction_schedule_map - instmap.add("rabi12", (0,), self.custom_sx_q0) + instmap.add("rabi12", (0,), rabi12) gate = circuit.Gate("rabi12", 1, []) qc = circuit.QuantumCircuit(1) @@ -367,7 +370,7 @@ def test_transpile_with_instmap_with_v2backend_with_custom_gate(self, opt_level) ) ref_calibration = { "rabi12": { - ((0,), ()): self.custom_sx_q0, + ((0,), ()): rabi12, } } self.assertDictEqual(transpiled_qc.calibrations, ref_calibration) diff --git a/test/python/transpiler/test_target.py b/test/python/transpiler/test_target.py index 942f37a0dfe7..980daef3d9bd 100644 --- a/test/python/transpiler/test_target.py +++ b/test/python/transpiler/test_target.py @@ -35,7 +35,7 @@ from qiskit.circuit.parameter import Parameter from qiskit import pulse from qiskit.pulse.instruction_schedule_map import InstructionScheduleMap -from qiskit.pulse.calibration_entries import CalibrationPublisher +from qiskit.pulse.calibration_entries import CalibrationPublisher, ScheduleDef from qiskit.transpiler.coupling import CouplingMap from qiskit.transpiler.instruction_durations import InstructionDurations from qiskit.transpiler.timing_constraints import TimingConstraints @@ -1273,6 +1273,33 @@ def test_get_empty_target_calibration(self): self.assertIsNone(target["x"][(0,)].calibration) + def test_loading_legacy_ugate_instmap(self): + # This is typical IBM backend situation. + # IBM provider used to have u1, u2, u3 in the basis gates and + # these have been replaced with sx and rz. + # However, IBM provider still provides calibration of these u gates, + # and the inst map loads them as backend calibrations. + # Target is implicitly updated with inst map when it is set in transpile. + # If u gates are not excluded, they may appear in the transpiled circuit. + # These gates are no longer supported by hardware. + entry = ScheduleDef() + entry.define(pulse.Schedule(name="fake_u3"), user_provided=False) # backend provided + instmap = InstructionScheduleMap() + instmap._add("u3", (0,), entry) + + # Today's standard IBM backend target with sx, rz basis + target = Target() + target.add_instruction(SXGate(), {(0,): InstructionProperties()}) + target.add_instruction(RZGate(Parameter("θ")), {(0,): InstructionProperties()}) + target.add_instruction(Measure(), {(0,): InstructionProperties()}) + names_before = set(target.operation_names) + + target.update_from_instruction_schedule_map(instmap) + names_after = set(target.operation_names) + + # Otherwise u3 and sx-rz basis conflict in 1q decomposition. + self.assertSetEqual(names_before, names_after) + class TestGlobalVariableWidthOperations(QiskitTestCase): def setUp(self):