diff --git a/qiskit/pulse/calibration_entries.py b/qiskit/pulse/calibration_entries.py index 78b8598f6055..09feddad4ccb 100644 --- a/qiskit/pulse/calibration_entries.py +++ b/qiskit/pulse/calibration_entries.py @@ -16,7 +16,7 @@ from enum import IntEnum from typing import Callable, List, Union, Optional, Sequence, Any -from qiskit.pulse.exceptions import PulseError +from qiskit.pulse.exceptions import PulseError, UnassignedDurationError from qiskit.pulse.schedule import Schedule, ScheduleBlock from qiskit.qobj.converters import QobjToInstructionConverter from qiskit.qobj.pulse_qobj import PulseQobjInstruction @@ -64,6 +64,15 @@ def get_schedule(self, *args, **kwargs) -> Union[Schedule, ScheduleBlock]: """ pass + @abstractmethod + def get_duration(self) -> int: + """Get duration of schedule. + + Returns: + Integer value representing a duration of schedule in units of dt. + """ + pass + class ScheduleDef(CalibrationEntry): """In-memory Qiskit Pulse representation. @@ -143,6 +152,12 @@ def get_schedule(self, *args, **kwargs) -> Union[Schedule, ScheduleBlock]: pass return self._definition.assign_parameters(value_dict, inplace=False) + def get_duration(self) -> int: + try: + return self._definition.duration + except UnassignedDurationError: + return None + def __eq__(self, other): # This delegates equality check to Schedule or ScheduleBlock. return self._definition == other._definition @@ -186,6 +201,10 @@ def get_schedule(self, *args, **kwargs) -> Union[Schedule, ScheduleBlock]: return self._definition(**to_bind.arguments) + def get_duration(self) -> int: + # Duration is undetermined until funciton is called with full arguments. + return None + def __eq__(self, other): # We cannot evaluate function equality without parsing python AST. # This simply compares wether they are the same object. @@ -249,6 +268,29 @@ def get_schedule(self, *args, **kwargs) -> Union[Schedule, ScheduleBlock]: self._build_schedule() return super().get_schedule(*args, **kwargs) + def get_duration(self) -> int: + if self._definition: + return self._definition.duration + + # Parse Qobj instruction and find max instruction end time + t_max = 0 + for inst in self._source: + if inst.name == "parametric_pulse": + # Play instruction + duration = inst.parameters.get("duration", 0) + elif inst.name not in self._converter.get_supported_instructions(): + # Play instruction with Waveform + try: + # Get samples from the pulse library dictionary + duration = len(self._converter._pulse_library[inst.name]) + except KeyError: + continue + else: + # Acquire, Delay instruction + duration = getattr(inst, "duration", 0) + t_max = max(t_max, inst.t0 + duration) + return t_max + def __eq__(self, other): if isinstance(other, PulseQobjDef): # If both objects are Qobj just check Qobj equality. diff --git a/qiskit/transpiler/passes/calibration/pulse_gate.py b/qiskit/transpiler/passes/calibration/pulse_gate.py index 5ed960253b62..f6f2f675b5f3 100644 --- a/qiskit/transpiler/passes/calibration/pulse_gate.py +++ b/qiskit/transpiler/passes/calibration/pulse_gate.py @@ -15,12 +15,10 @@ from typing import List, Union from qiskit.circuit import Instruction as CircuitInst -from qiskit.pulse import ( - Schedule, - ScheduleBlock, -) +from qiskit.pulse import Schedule, ScheduleBlock from qiskit.pulse.instruction_schedule_map import InstructionScheduleMap from qiskit.transpiler.target import Target +from qiskit.transpiler.exceptions import TranspilerError from .base_builder import CalibrationBuilder @@ -59,13 +57,16 @@ def __init__( Args: inst_map: Instruction schedule map that user may override. target: The :class:`~.Target` representing the target backend, if both - ``inst_map`` and this are specified then this argument will take - precedence and ``inst_map`` will be ignored. + ``inst_map`` and this are specified then it updates instructions + in the ``target`` with ``inst_map``. """ super().__init__() - self.inst_map = inst_map - if target: - self.inst_map = target.instruction_schedule_map() + + if target is None: + target = Target() + if inst_map is not None: + target.update_from_instruction_schedule_map(inst_map) + self.target = target def supported(self, node_op: CircuitInst, qubits: List) -> bool: """Determine if a given node supports the calibration. @@ -77,7 +78,7 @@ def supported(self, node_op: CircuitInst, qubits: List) -> bool: Returns: Return ``True`` is calibration can be provided. """ - return self.inst_map.has(instruction=node_op.name, qubits=qubits) + return self.target.instruction_supported(operation_name=node_op.name, qargs=qubits) def get_calibration(self, node_op: CircuitInst, qubits: List) -> Union[Schedule, ScheduleBlock]: """Gets the calibrated schedule for the given instruction and qubits. @@ -88,5 +89,22 @@ def get_calibration(self, node_op: CircuitInst, qubits: List) -> Union[Schedule, Returns: Return Schedule of target gate instruction. + + Raises: + TranspilerError: When node is parameterized and calibration is raw schedule object. """ - return self.inst_map.get(node_op.name, qubits, *node_op.params) + inst_property = self.target[node_op.name][tuple(qubits)] + if not node_op.params: + return inst_property.calibration + try: + # CircuitInstruction doesn't preserve parameter name after parameter binding. + # Thus schedule cannot generate bind dictionary. + # Use CalibraionEntry to utilize inspected signature object. + calibration_entry = inst_property._calibration + return calibration_entry.get_schedule(*node_op.params) + except AttributeError as ex: + raise TranspilerError( + f"Calibraton for {node_op.name} of {qubits} is not a CalibraryEntry instance. " + f"Mapping from parameter values {node_op.params} to parameter objects " + f"in the schedule cannot be identified." + ) from ex diff --git a/qiskit/transpiler/target.py b/qiskit/transpiler/target.py index 77381747d86d..12ddba4afe69 100644 --- a/qiskit/transpiler/target.py +++ b/qiskit/transpiler/target.py @@ -28,8 +28,10 @@ import rustworkx as rx from qiskit.circuit.parameter import Parameter +from qiskit.circuit.gate import Gate +from qiskit.circuit.library.standard_gates import get_standard_gate_name_mapping from qiskit.pulse.instruction_schedule_map import InstructionScheduleMap -from qiskit.pulse.calibration_entries import CalibrationEntry +from qiskit.pulse.calibration_entries import CalibrationEntry, ScheduleDef from qiskit.pulse.schedule import Schedule, ScheduleBlock from qiskit.transpiler.coupling import CouplingMap from qiskit.transpiler.exceptions import TranspilerError @@ -79,13 +81,17 @@ def __init__( @property def calibration(self): """The pulse representation of the instruction.""" - if isinstance(self._calibration, CalibrationEntry): - return self._calibration.get_schedule() - return self._calibration + return self._calibration.get_schedule() @calibration.setter def calibration(self, calibration: Union[Schedule, ScheduleBlock, CalibrationEntry]): - self._calibration = calibration + if not isinstance(calibration, CalibrationEntry): + # Convert into entry to use signature generation mechanism for parameter assignment. + new_entry = ScheduleDef() + new_entry.define(calibration) + else: + new_entry = calibration + self._calibration = new_entry def __repr__(self): return ( @@ -416,7 +422,9 @@ def update_from_instruction_schedule_map(self, inst_map, inst_name_map=None, err Args: inst_map (InstructionScheduleMap): The instruction inst_name_map (dict): An optional dictionary that maps any - instruction name in ``inst_map`` to an instruction object + instruction name in ``inst_map`` to an instruction object. + If not provided, instruction is pulled from the standard Qiskit gates, + and finally custom gate instnace is created with schedule name. error_dict (dict): A dictionary of errors of the form:: {gate_name: {qarg: error}} @@ -429,48 +437,53 @@ def update_from_instruction_schedule_map(self, inst_map, inst_name_map=None, err a when updating the ``Target`` the error value will be pulled from this dictionary. If one is not found in ``error_dict`` then ``None`` will be used. - - Raises: - ValueError: If ``inst_map`` contains new instructions and - ``inst_name_map`` isn't specified - KeyError: If a ``inst_map`` contains a qarg for an instruction - that's not in the target """ - for inst in inst_map.instructions: + for inst_name in inst_map.instructions: out_props = {} - for qarg in inst_map.qubits_with_instruction(inst): - sched = inst_map.get(inst, qarg) - val = InstructionProperties(calibration=sched) + for qarg in inst_map.qubits_with_instruction(inst_name): try: qarg = tuple(qarg) except TypeError: qarg = (qarg,) - if inst in self._gate_map: - if self.dt is not None: - val.duration = sched.duration * self.dt - else: - val.duration = None - if error_dict is not None: - error_inst = error_dict.get(inst) - if error_inst: - error = error_inst.get(qarg) - val.error = error - else: - val.error = None + entry = inst_map._get_calibration_entry(inst_name, qarg) + val = InstructionProperties(calibration=entry) + if self.dt is not None: + val.duration = entry.get_duration() * self.dt + else: + val.duration = None + if inst_name in self._gate_map and error_dict is not None: + error_inst = error_dict.get(inst_name) + if error_inst: + error = error_inst.get(qarg) + val.error = error else: val.error = None + else: + val.error = None out_props[qarg] = val - if inst not in self._gate_map: - if inst_name_map is not None: - self.add_instruction(inst_name_map[inst], out_props, name=inst) + if inst_name not in self._gate_map: + if inst_name_map is None: + inst_name_map = get_standard_gate_name_mapping() + if inst_name in inst_name_map: + inst_obj = inst_name_map[inst_name] else: - raise ValueError( - "An inst_name_map kwarg must be specified to add new " - "instructions from an InstructionScheduleMap" + # Custom gate object, which doesn't belong to standard Qiskit gates. + inst_obj = Gate( + name=inst_name, + num_qubits=len(qarg), + params=list(map(Parameter, entry.get_signature().parameters.keys())), ) + normalized_props = {} + for qargs, prop in out_props.items(): + if len(qargs) != inst_obj.num_qubits: + continue + normalized_props[qargs] = prop + self.add_instruction(inst_obj, normalized_props, name=inst_name) else: for qarg, prop in out_props.items(): - self.update_instruction_properties(inst, qarg, prop) + if qarg not in self._gate_map[inst_name]: + continue + self.update_instruction_properties(inst_name, qarg, prop) @property def qargs(self): @@ -532,8 +545,9 @@ def instruction_schedule_map(self): out_inst_schedule_map = InstructionScheduleMap() for instruction, qargs in self._gate_map.items(): for qarg, properties in qargs.items(): - if properties is not None and properties.calibration is not None: - out_inst_schedule_map.add(instruction, qarg, properties.calibration) + if properties is not None and properties._calibration is not None: + # Do not parse schedule. Build instmap with CalibrationEntry instances. + out_inst_schedule_map._add(instruction, qarg, properties._calibration) self._instruction_schedule_map = out_inst_schedule_map return out_inst_schedule_map diff --git a/releasenotes/notes/update-pulse-gate-pass-for-target-ebfb0ec9571f058e.yaml b/releasenotes/notes/update-pulse-gate-pass-for-target-ebfb0ec9571f058e.yaml new file mode 100644 index 000000000000..8281f85ad6b1 --- /dev/null +++ b/releasenotes/notes/update-pulse-gate-pass-for-target-ebfb0ec9571f058e.yaml @@ -0,0 +1,23 @@ +--- +features: + - | + A new method :meth:`.CalibrationEntry.get_duration` has been added to + calibration entries. This method returns duration of calibration + in units of dt when available, otherwise returns None. +upgrade: + - | + :meth:`.Target.update_from_instruction_schedule_map` no longer raises + KeyError nor ValueError when qubits are missing in the target instruction + or inst_name_map is not provided for undefined instruction. + In the former case, it just ignores the inst map definition for undefined qubits. + In the latter case, gate mapping is pulled from the standard Qiskit gates + and finally custom :class:`.Gate` object is defined from the schedule name. + - | + :class:`PulseGates` transpiler pass has been upgraded to respect + ``inst_map`` when ``target`` is also provided, rather than ignoring it. + The pass now uses the target data to find custom calibration, but it + copies instruction schedule map into it before transforming the DAG circuit. +fixes: + - | + A bug that custom gates in the instruction schedule map is ignored + when transpiling a circuit with V2 backend has been fixed. diff --git a/test/python/pulse/test_calibration_entries.py b/test/python/pulse/test_calibration_entries.py index 3259d3b8a6aa..aaf33926e56e 100644 --- a/test/python/pulse/test_calibration_entries.py +++ b/test/python/pulse/test_calibration_entries.py @@ -40,10 +40,12 @@ class TestSchedule(QiskitTestCase): def test_add_schedule(self): """Basic test pulse Schedule format.""" + ref_duration = 10 + program = Schedule() program.insert( 0, - Play(Constant(duration=10, amp=0.1, angle=0.0), DriveChannel(0)), + Play(Constant(duration=ref_duration, amp=0.1, angle=0.0), DriveChannel(0)), inplace=True, ) @@ -58,11 +60,15 @@ def test_add_schedule(self): schedule_ref = program self.assertEqual(schedule_to_test, schedule_ref) + self.assertEqual(entry.get_duration(), ref_duration) + def test_add_block(self): """Basic test pulse Schedule format.""" + ref_duration = 10 + program = ScheduleBlock() program.append( - Play(Constant(duration=10, amp=0.1, angle=0.0), DriveChannel(0)), + Play(Constant(duration=ref_duration, amp=0.1, angle=0.0), DriveChannel(0)), inplace=True, ) @@ -77,6 +83,8 @@ def test_add_block(self): schedule_ref = program self.assertEqual(schedule_to_test, schedule_ref) + self.assertEqual(entry.get_duration(), ref_duration) + def test_parameterized_schedule(self): """Test adding and managing parameterized schedule.""" param1 = Parameter("P1") @@ -99,6 +107,9 @@ def test_parameterized_schedule(self): schedule_ref = program.assign_parameters({param1: 10, param2: 0.1}, inplace=False) self.assertEqual(schedule_to_test, schedule_ref) + # Undetermined + self.assertEqual(entry.get_duration(), None) + def test_parameterized_schedule_with_user_args(self): """Test adding schedule with user signature. @@ -207,6 +218,8 @@ def factory(): schedule_ref = program self.assertEqual(schedule_to_test, schedule_ref) + self.assertEqual(entry.get_duration(), None) + def test_add_callable_with_argument(self): """Basic test callable format.""" @@ -308,6 +321,9 @@ def test_add_qobj(self): entry = PulseQobjDef(converter=self.converter, name="my_gate") entry.define(serialized_program) + self.assertEqual(entry.get_duration(), 25) + self.assertIsNone(entry._definition) # Check if schedule is not parsed + signature_to_test = list(entry.get_signature().parameters.keys()) signature_ref = [] self.assertListEqual(signature_to_test, signature_ref) diff --git a/test/python/transpiler/test_pulse_gate_pass.py b/test/python/transpiler/test_pulse_gate_pass.py index beb08d5e66fc..2ddec714c7db 100644 --- a/test/python/transpiler/test_pulse_gate_pass.py +++ b/test/python/transpiler/test_pulse_gate_pass.py @@ -264,3 +264,88 @@ def test_transpile_with_different_qubit(self): transpiled_qc = transpile(qc, backend, initial_layout=[3]) self.assertDictEqual(transpiled_qc.calibrations, {}) + + def test_transpile_with_both_instmap_and_empty_target(self): + """Test when instmap and target are both provided + and only instmap contains custom schedules. + + Test case from Qiskit/qiskit-terra/#9489 + """ + instmap = FakeAthens().defaults().instruction_schedule_map + instmap.add("sx", (0,), self.custom_sx_q0) + instmap.add("sx", (1,), self.custom_sx_q1) + + # This doesn't have custom schedule definition + target = FakeAthensV2().target + + qc = circuit.QuantumCircuit(2) + qc.sx(0) + qc.x(0) + qc.rz(0, 0) + qc.sx(1) + qc.measure_all() + + transpiled_qc = transpile( + qc, + basis_gates=["sx", "rz", "x"], + inst_map=instmap, + target=target, + initial_layout=[0, 1], + ) + ref_calibration = { + "sx": { + ((0,), ()): self.custom_sx_q0, + ((1,), ()): self.custom_sx_q1, + } + } + self.assertDictEqual(transpiled_qc.calibrations, ref_calibration) + + def test_transpile_with_instmap_with_v2backend(self): + """Test when instmap is provided with V2 backend. + + Test case from Qiskit/qiskit-terra/#9489 + """ + instmap = FakeAthens().defaults().instruction_schedule_map + instmap.add("sx", (0,), self.custom_sx_q0) + instmap.add("sx", (1,), self.custom_sx_q1) + + qc = circuit.QuantumCircuit(2) + qc.sx(0) + qc.x(0) + qc.rz(0, 0) + qc.sx(1) + qc.measure_all() + + transpiled_qc = transpile(qc, FakeAthensV2(), inst_map=instmap, initial_layout=[0, 1]) + ref_calibration = { + "sx": { + ((0,), ()): self.custom_sx_q0, + ((1,), ()): self.custom_sx_q1, + } + } + self.assertDictEqual(transpiled_qc.calibrations, ref_calibration) + + def test_transpile_with_instmap_with_v2backend_with_custom_gate(self): + """Test when instmap is provided with V2 backend. + + In this test case, instmap contains a custom gate which doesn't belong to + Qiskit standard gate. Target must define a custom gete on the fly + to reflect user-provided instmap. + + Test case from Qiskit/qiskit-terra/#9489 + """ + instmap = FakeAthens().defaults().instruction_schedule_map + instmap.add("rabi12", (0,), self.custom_sx_q0) + + gate = circuit.Gate("rabi12", 1, []) + qc = circuit.QuantumCircuit(1) + qc.append(gate, [0]) + qc.measure_all() + + transpiled_qc = transpile(qc, FakeAthensV2(), inst_map=instmap, initial_layout=[0]) + ref_calibration = { + "rabi12": { + ((0,), ()): self.custom_sx_q0, + } + } + self.assertDictEqual(transpiled_qc.calibrations, ref_calibration) diff --git a/test/python/transpiler/test_target.py b/test/python/transpiler/test_target.py index 66a7bf4901de..b3a708da4ffd 100644 --- a/test/python/transpiler/test_target.py +++ b/test/python/transpiler/test_target.py @@ -1176,16 +1176,17 @@ def test_update_from_instruction_schedule_map_new_instruction_no_name_map(self): inst_map = InstructionScheduleMap() inst_map.add("sx", 0, self.custom_sx_q0) inst_map.add("sx", 1, self.custom_sx_q1) - with self.assertRaises(ValueError): - target.update_from_instruction_schedule_map(inst_map) + target.update_from_instruction_schedule_map(inst_map) + self.assertEqual(target["sx"][(0,)].calibration, self.custom_sx_q0) + self.assertEqual(target["sx"][(1,)].calibration, self.custom_sx_q1) def test_update_from_instruction_schedule_map_new_qarg_raises(self): inst_map = InstructionScheduleMap() inst_map.add("sx", 0, self.custom_sx_q0) inst_map.add("sx", 1, self.custom_sx_q1) inst_map.add("sx", 2, self.custom_sx_q1) - with self.assertRaises(KeyError): - self.pulse_target.update_from_instruction_schedule_map(inst_map) + self.pulse_target.update_from_instruction_schedule_map(inst_map) + self.assertFalse(self.pulse_target.instruction_supported("sx", (2,))) def test_update_from_instruction_schedule_map_with_dt_set(self): inst_map = InstructionScheduleMap()