diff --git a/qiskit/compiler/transpiler.py b/qiskit/compiler/transpiler.py index b0369d9561f0..c4d78b89d0a8 100644 --- a/qiskit/compiler/transpiler.py +++ b/qiskit/compiler/transpiler.py @@ -13,6 +13,7 @@ # pylint: disable=invalid-sequence-index """Circuit transpile function""" +import copy import io from itertools import cycle import logging @@ -663,6 +664,10 @@ def _parse_transpile_args( callback = _parse_callback(callback, num_circuits) durations = _parse_instruction_durations(backend, instruction_durations, dt, circuits) timing_constraints = _parse_timing_constraints(backend, timing_constraints, num_circuits) + if inst_map is not None and inst_map.has_custom_gate() and target is not None: + # Do not mutate backend target + target = copy.deepcopy(target) + target.update_from_instruction_schedule_map(inst_map) if scheduling_method and any(d is None for d in durations): raise TranspilerError( "Transpiling a circuit with a scheduling method" 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..6564dab1e193 100644 --- a/qiskit/pulse/calibration_entries.py +++ b/qiskit/pulse/calibration_entries.py @@ -31,14 +31,41 @@ class CalibrationPublisher(IntEnum): class CalibrationEntry(metaclass=ABCMeta): - """A metaclass of a calibration entry.""" + """A metaclass of a calibration entry. + + This class defines a standard model of Qiskit pulse program that is + agnostic to the underlying in-memory representation. + + This entry distinguishes whether this is provided by end-users or a backend + by :attr:`.user_provided` attribute which may be provided when + the actual calibration data is provided to the entry with by :meth:`define`. + + Note that a custom entry provided by an end-user may appear in the wire-format + as an inline calibration, e.g. :code:`defcal` of the QASM3, + that may update the backend instruction set architecture for execution. + + .. note:: + + This and built-in subclasses are expected to be private without stable user-facing API. + The purpose of this class is to wrap different + in-memory pulse program representations in Qiskit, so that it can provide + the standard data model and API which are primarily used by the transpiler ecosystem. + It is assumed that end-users will never directly instantiate this class, + but :class:`.Target` or :class:`.InstructionScheduleMap` internally use this data model + to avoid implementing a complicated branching logic to + manage different calibration data formats. + + """ @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. + If the flag is set, this calibration may appear in the wire format + as an inline calibration, to override the backend instruction set architecture. """ pass @@ -55,6 +82,10 @@ def get_signature(self) -> inspect.Signature: def get_schedule(self, *args, **kwargs) -> Union[Schedule, ScheduleBlock]: """Generate schedule from entry definition. + If the pulse program is templated with :class:`.Parameter` objects, + you can provide corresponding parameter values for this method + to get a particular pulse program with assigned parameters. + Args: args: Command parameters. kwargs: Command keyword parameters. @@ -64,6 +95,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. @@ -71,6 +108,10 @@ class ScheduleDef(CalibrationEntry): 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. + + .. see_also:: + :class:`.CalibrationEntry` for the purpose of this class. + """ def __init__(self, arguments: Optional[Sequence[str]] = None): @@ -90,6 +131,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 +166,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}" @@ -165,16 +224,30 @@ class CallableDef(CalibrationEntry): 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. + + .. see_also:: + :class:`.CalibrationEntry` for the purpose of this class. + """ def __init__(self): """Define an empty entry.""" self._definition = None self._signature = None + self._user_provided = None - def define(self, definition: Callable): + @property + def user_provided(self) -> bool: + return self._user_provided + + 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 +259,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()) @@ -210,6 +286,10 @@ class PulseQobjDef(ScheduleDef): 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. + + .. see_also:: + :class:`.CalibrationEntry` for the purpose of this class. + """ def __init__( @@ -237,14 +317,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 +344,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 5ed960253b62..9bfd3c544779 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,18 @@ 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 inst_map is None and target is None: + raise TranspilerError("inst_map and target cannot be None simulataneously.") + + if target is None: + target = Target() + 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 +80,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.has_calibration(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. @@ -88,5 +91,8 @@ 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) + 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 4a0359d5a479..27b9f8fa57af 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,9 @@ 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 from qiskit.pulse.calibration_entries import CalibrationEntry, ScheduleDef from qiskit.pulse.schedule import Schedule, ScheduleBlock @@ -36,6 +39,7 @@ from qiskit.transpiler.instruction_durations import InstructionDurations from qiskit.transpiler.timing_constraints import TimingConstraints from qiskit.utils.deprecation import deprecate_arguments +from qiskit.exceptions import QiskitError # import QubitProperties here to provide convenience alias for building a # full target @@ -80,7 +84,32 @@ def __init__( @property def calibration(self): - """The pulse representation of the instruction.""" + """The pulse representation of the instruction. + + .. note:: + + This attribute always returns a Qiskit pulse program, but it is internally + wrapped by the :class:`.CalibrationEntry` to manage unbound parameters + and to uniformly handle different data representation, + for example, un-parsed Pulse Qobj JSON that a backend provider may provide. + + This value can be overridden through the property setter in following manner. + When you set either :class:`.Schedule` or :class:`.ScheduleBlock` this is + always treated as a user-defined (custom) calibration and + the transpiler may automatically attach the calibration data to the output circuit. + This calibration data may appear in the wire format as an inline calibration, + which may further update the backend standard instruction set architecture. + + If you are a backend provider who provides a default calibration data + that is not needed to be attached to the transpiled quantum circuit, + you can directly set :class:`.CalibrationEntry` instance to this attribute, + in which you should set :code:`user_provided=False` when you define + calibration data for the entry. End users can still intentionally utilize + the calibration data, for example, to run pulse-level simulation of the circuit. + However, such entry doesn't appear in the wire format, and backend must + use own definition to compile the circuit down to the execution format. + + """ if self._calibration is None: return None return self._calibration.get_schedule() @@ -89,7 +118,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 @@ -417,13 +446,15 @@ def update_from_instruction_schedule_map(self, inst_map, inst_name_map=None, err """Update the target from an instruction schedule map. If the input instruction schedule map contains new instructions not in - the target they will be added. However if it contains additional qargs + the target they will be added. However, if it contains additional qargs for an existing instruction in the target it will error. 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}} @@ -436,48 +467,94 @@ 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: + get_calibration = getattr(inst_map, "_get_calibration_entry") + + # Expand name mapping with custom gate name provided by user. + qiskit_inst_name_map = get_standard_gate_name_mapping() + if inst_name_map is not None: + qiskit_inst_name_map.update(inst_name_map) + + for inst_name in inst_map.instructions: + # Prepare dictionary of instruction properties out_props = {} - for qarg in inst_map.qubits_with_instruction(inst): - sched = inst_map.get(inst, qarg) - val = InstructionProperties(calibration=sched) + for qargs in inst_map.qubits_with_instruction(inst_name): try: - qarg = tuple(qarg) + qargs = tuple(qargs) except TypeError: - qarg = (qarg,) - if inst in self._gate_map: + qargs = (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: - val.duration = sched.duration * self.dt + duration = entry.get_schedule().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 - 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) + duration = None + props = InstructionProperties( + duration=duration, + calibration=entry, + ) else: - raise ValueError( - "An inst_name_map kwarg must be specified to add new " - "instructions from an InstructionScheduleMap" + 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 + # Prepare Qiskit Gate object assigned to the entries + if inst_name not in self._gate_map: + # Entry not found: Add new instruction + if inst_name in qiskit_inst_name_map: + # Remove qargs with length that doesn't match with instruction qubit number + inst_obj = qiskit_inst_name_map[inst_name] + 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: + # Check qubit length parameter name uniformity. + qlen = set() + param_names = set() + for qargs in inst_map.qubits_with_instruction(inst_name): + if isinstance(qargs, int): + qargs = (qargs,) + qlen.add(len(qargs)) + cal = getattr(out_props[tuple(qargs)], "_calibration") + param_names.add(tuple(cal.get_signature().parameters.keys())) + if len(qlen) > 1 or len(param_names) > 1: + raise QiskitError( + f"Schedules for {inst_name} are defined non-uniformly for " + f"multiple qubit lengths {qlen}, " + f"or different parameter names {param_names}. " + "Provide these schedules with inst_name_map or define them with " + "different names for different gate parameters." + ) + inst_obj = Gate( + name=inst_name, + num_qubits=next(iter(qlen)), + params=list(map(Parameter, next(iter(param_names)))), ) + self.add_instruction(inst_obj, out_props, name=inst_name) else: - for qarg, prop in out_props.items(): - self.update_instruction_properties(inst, qarg, prop) + # Entry found: Update "existing" instructions. + for qargs, prop in out_props.items(): + if qargs not in self._gate_map[inst_name]: + continue + self.update_instruction_properties(inst_name, qargs, prop) @property def qargs(self): @@ -773,6 +850,55 @@ def check_obj_params(parameters, obj): ) return False + def has_calibration( + 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.has_calibration(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/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..0504772659cb --- /dev/null +++ b/releasenotes/notes/update-pulse-gate-pass-for-target-ebfb0ec9571f058e.yaml @@ -0,0 +1,21 @@ +--- +features: + - | + A new method :meth:`.CalibrationEntry.user_provided` has been added to + calibration entries. This method can be called to check whether the entry + is defined by an end user or backend. + - | + New method :meth:`.Target.calibration_supported` and + :meth:`.Target.get_calibration` have been added to provide + convenient access to the calibration of instruction. + The getter method can be called with parameter args and kwargs, and + it returns a pulse schedule built with parameters when the calibration + is templated with parameters. +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. diff --git a/test/python/transpiler/test_pulse_gate_pass.py b/test/python/transpiler/test_pulse_gate_pass.py index beb08d5e66fc..b5f8c03c4711 100644 --- a/test/python/transpiler/test_pulse_gate_pass.py +++ b/test/python/transpiler/test_pulse_gate_pass.py @@ -12,39 +12,43 @@ """Transpiler pulse gate pass testing.""" +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 class TestPulseGate(QiskitTestCase): """Integration test of pulse gate pass with custom backend.""" def setUp(self): super().setUp() + self.sched_param = circuit.Parameter("P0") + with pulse.build(name="sx_q0") as custom_sx_q0: pulse.play(pulse.Constant(100, 0.1), pulse.DriveChannel(0)) - self.custom_sx_q0 = custom_sx_q0 with pulse.build(name="sx_q1") as custom_sx_q1: pulse.play(pulse.Constant(100, 0.2), pulse.DriveChannel(1)) - self.custom_sx_q1 = custom_sx_q1 - self.sched_param = circuit.Parameter("P0") + with pulse.build(name="cx_q01") as custom_cx_q01: + pulse.play(pulse.Constant(100, 0.4), pulse.ControlChannel(0)) + self.custom_cx_q01 = custom_cx_q01 with pulse.build(name="my_gate_q0") as my_gate_q0: pulse.shift_phase(self.sched_param, pulse.DriveChannel(0)) pulse.play(pulse.Constant(120, 0.1), pulse.DriveChannel(0)) - self.my_gate_q0 = my_gate_q0 with pulse.build(name="my_gate_q1") as my_gate_q1: pulse.shift_phase(self.sched_param, pulse.DriveChannel(1)) pulse.play(pulse.Constant(120, 0.2), pulse.DriveChannel(1)) - self.my_gate_q1 = my_gate_q1 def test_transpile_with_bare_backend(self): @@ -264,3 +268,140 @@ def test_transpile_with_different_qubit(self): transpiled_qc = transpile(qc, backend, initial_layout=[3]) self.assertDictEqual(transpiled_qc.calibrations, {}) + + @ddt.data(0, 1, 2, 3) + def test_transpile_with_both_instmap_and_empty_target(self, opt_level): + """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) + instmap.add("cx", (0, 1), self.custom_cx_q01) + + # This doesn't have custom schedule definition + target = FakeAthensV2().target + + qc = circuit.QuantumCircuit(2) + qc.append(random_unitary(4, seed=123), [0, 1]) + qc.measure_all() + + transpiled_qc = transpile( + qc, + optimization_level=opt_level, + basis_gates=["sx", "rz", "x", "cx"], + inst_map=instmap, + target=target, + initial_layout=[0, 1], + ) + ref_calibration = { + "sx": { + ((0,), ()): self.custom_sx_q0, + ((1,), ()): self.custom_sx_q1, + }, + "cx": { + ((0, 1), ()): self.custom_cx_q01, + }, + } + self.assertDictEqual(transpiled_qc.calibrations, ref_calibration) + + @ddt.data(0, 1, 2, 3) + def test_transpile_with_instmap_with_v2backend(self, opt_level): + """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) + instmap.add("cx", (0, 1), self.custom_cx_q01) + + qc = circuit.QuantumCircuit(2) + qc.append(random_unitary(4, seed=123), [0, 1]) + qc.measure_all() + + transpiled_qc = transpile( + qc, + FakeAthensV2(), + optimization_level=opt_level, + inst_map=instmap, + initial_layout=[0, 1], + ) + ref_calibration = { + "sx": { + ((0,), ()): self.custom_sx_q0, + ((1,), ()): self.custom_sx_q1, + }, + "cx": { + ((0, 1), ()): self.custom_cx_q01, + }, + } + self.assertDictEqual(transpiled_qc.calibrations, ref_calibration) + + @ddt.data(0, 1, 2, 3) + def test_transpile_with_instmap_with_v2backend_with_custom_gate(self, opt_level): + """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 + """ + 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,), rabi12) + + gate = circuit.Gate("rabi12", 1, []) + qc = circuit.QuantumCircuit(1) + qc.append(gate, [0]) + qc.measure_all() + + transpiled_qc = transpile( + qc, + FakeAthensV2(), + optimization_level=opt_level, + inst_map=instmap, + initial_layout=[0], + ) + ref_calibration = { + "rabi12": { + ((0,), ()): rabi12, + } + } + self.assertDictEqual(transpiled_qc.calibrations, ref_calibration) + + def test_transpile_with_instmap_not_mutate_backend(self): + """Do not override default backend target when transpile with inst map. + + Providing an instmap for the transpile arguments may override target, + which might be pulled from the provided backend instance. + This should not override the source object since the same backend may + be used for future transpile without intention of instruction overriding. + """ + backend = FakeAthensV2() + original_sx0 = backend.target["sx"][(0,)].calibration + + instmap = FakeAthens().defaults().instruction_schedule_map + instmap.add("sx", (0,), self.custom_sx_q0) + + qc = circuit.QuantumCircuit(1) + qc.sx(0) + qc.measure_all() + + transpiled_qc = transpile( + qc, + FakeAthensV2(), + inst_map=instmap, + initial_layout=[0], + ) + self.assertTrue(transpiled_qc.has_calibration_for(transpiled_qc.data[0])) + + self.assertEqual( + backend.target["sx"][(0,)].calibration, + original_sx0, + ) diff --git a/test/python/transpiler/test_target.py b/test/python/transpiler/test_target.py index e3980eff1814..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 @@ -1167,8 +1167,10 @@ def test_update_from_instruction_schedule_map_update_schedule(self): inst_map.add("sx", 1, custom_sx) self.pulse_target.update_from_instruction_schedule_map(inst_map, {"sx": SXGate()}) self.assertEqual(inst_map, self.pulse_target.instruction_schedule_map()) - self.assertIsNone(self.pulse_target["sx"][(0,)].duration) - self.assertIsNone(self.pulse_target["sx"][(0,)].error) + # Calibration doesn't change for q0 + self.assertEqual(self.pulse_target["sx"][(0,)].duration, 35.5e-9) + self.assertEqual(self.pulse_target["sx"][(0,)].error, 0.000413) + # Calibration is updated for q1 without error dict and gate time self.assertIsNone(self.pulse_target["sx"][(1,)].duration) self.assertIsNone(self.pulse_target["sx"][(1,)].error) @@ -1177,16 +1179,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() @@ -1200,7 +1203,11 @@ def test_update_from_instruction_schedule_map_with_dt_set(self): self.assertEqual(inst_map, self.pulse_target.instruction_schedule_map()) self.assertEqual(self.pulse_target["sx"][(1,)].duration, 1000.0) self.assertIsNone(self.pulse_target["sx"][(1,)].error) - self.assertIsNone(self.pulse_target["sx"][(0,)].error) + # This is an edge case. + # System dt is read-only property and changing it will break all underlying calibrations. + # duration of sx0 returns previous value since calibration doesn't change. + self.assertEqual(self.pulse_target["sx"][(0,)].duration, 35.5e-9) + self.assertEqual(self.pulse_target["sx"][(0,)].error, 0.000413) def test_update_from_instruction_schedule_map_with_error_dict(self): inst_map = InstructionScheduleMap() @@ -1216,7 +1223,7 @@ def test_update_from_instruction_schedule_map_with_error_dict(self): inst_map, {"sx": SXGate()}, error_dict=error_dict ) self.assertEqual(self.pulse_target["sx"][(1,)].error, 1.0) - self.assertIsNone(self.pulse_target["sx"][(0,)].error) + self.assertEqual(self.pulse_target["sx"][(0,)].error, 0.000413) def test_timing_constraints(self): generated_constraints = self.pulse_target.timing_constraints() @@ -1266,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):