6206652911

Committed 16 Sep 2023 08:10AM UTC coverage: 87.26% (+0.02%) from 87.245%

Build # 6206652911

Build Type

push

github

Committed by

web-flow

Commit Message

Allow dynamical decoupling pass to work on circuits with pulse gates (#10834) (#10848)

The pass was accessing the circuit calibrations incorrectly, resulting in
an error when one of the gates in the dynamical decoupling sequence had
a pulse gate calibration.

(cherry picked from commit 4e49a5694)

Co-authored-by: Will Shanks <willshanks@us.ibm.com>

Run Details

11 of 11 new or added lines in 1 file covered. (100.0%)

74131 of 84954 relevant lines covered (87.26%)

60858.93 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

93.67

/qiskit/transpiler/passes/scheduling/padding/dynamical_decoupling.py

# This code is part of Qiskit.
#
# (C) Copyright IBM 2021.
#
# 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.

"""Dynamical Decoupling insertion pass."""
from __future__ import annotations

import logging
import numpy as np

from qiskit.circuit import Gate, ParameterExpression, Qubit
from qiskit.circuit.delay import Delay
from qiskit.circuit.library.standard_gates import IGate, UGate, U3Gate
from qiskit.circuit.reset import Reset
from qiskit.dagcircuit import DAGCircuit, DAGNode, DAGInNode, DAGOpNode
from qiskit.quantum_info.operators.predicates import matrix_equal
from qiskit.quantum_info.synthesis import OneQubitEulerDecomposer
from qiskit.transpiler.exceptions import TranspilerError
from qiskit.transpiler.instruction_durations import InstructionDurations
from qiskit.transpiler.passes.optimization import Optimize1qGates
from qiskit.transpiler.target import Target

from .base_padding import BasePadding

logger = logging.getLogger(__name__)


class PadDynamicalDecoupling(BasePadding):
    """Dynamical decoupling insertion pass.

    This pass works on a scheduled, physical circuit. It scans the circuit for
    idle periods of time (i.e. those containing delay instructions) and inserts
    a DD sequence of gates in those spots. These gates amount to the identity,
    so do not alter the logical action of the circuit, but have the effect of
    mitigating decoherence in those idle periods.

    As a special case, the pass allows a length-1 sequence (e.g. [XGate()]).
    In this case the DD insertion happens only when the gate inverse can be
    absorbed into a neighboring gate in the circuit (so we would still be
    replacing Delay with something that is equivalent to the identity).
    This can be used, for instance, as a Hahn echo.

    This pass ensures that the inserted sequence preserves the circuit exactly
    (including global phase).

    .. plot::
       :include-source:

        import numpy as np
        from qiskit.circuit import QuantumCircuit
        from qiskit.circuit.library import XGate
        from qiskit.transpiler import PassManager, InstructionDurations
        from qiskit.transpiler.passes import ALAPScheduleAnalysis, PadDynamicalDecoupling
        from qiskit.visualization import timeline_drawer
        circ = QuantumCircuit(4)
        circ.h(0)
        circ.cx(0, 1)
        circ.cx(1, 2)
        circ.cx(2, 3)
        circ.measure_all()
        durations = InstructionDurations(
            [("h", 0, 50), ("cx", [0, 1], 700), ("reset", None, 10),
             ("cx", [1, 2], 200), ("cx", [2, 3], 300),
             ("x", None, 50), ("measure", None, 1000)]
        )

        # balanced X-X sequence on all qubits
        dd_sequence = [XGate(), XGate()]
        pm = PassManager([ALAPScheduleAnalysis(durations),
                          PadDynamicalDecoupling(durations, dd_sequence)])
        circ_dd = pm.run(circ)
        timeline_drawer(circ_dd)

        # Uhrig sequence on qubit 0
        n = 8
        dd_sequence = [XGate()] * n
        def uhrig_pulse_location(k):
            return np.sin(np.pi * (k + 1) / (2 * n + 2)) ** 2
        spacing = []
        for k in range(n):
            spacing.append(uhrig_pulse_location(k) - sum(spacing))
        spacing.append(1 - sum(spacing))
        pm = PassManager(
            [
                ALAPScheduleAnalysis(durations),
                PadDynamicalDecoupling(durations, dd_sequence, qubits=[0], spacing=spacing),
            ]
        )
        circ_dd = pm.run(circ)
        timeline_drawer(circ_dd)

    .. note::

        You may need to call alignment pass before running dynamical decoupling to guarantee
        your circuit satisfies acquisition alignment constraints.
    """

    def __init__(
        self,
        durations: InstructionDurations = None,
        dd_sequence: list[Gate] = None,
        qubits: list[int] | None = None,
        spacing: list[float] | None = None,
        skip_reset_qubits: bool = True,
        pulse_alignment: int = 1,
        extra_slack_distribution: str = "middle",
        target: Target = None,
    ):
        """Dynamical decoupling initializer.

        Args:
            durations: Durations of instructions to be used in scheduling.
            dd_sequence: Sequence of gates to apply in idle spots.
            qubits: Physical qubits on which to apply DD.
                If None, all qubits will undergo DD (when possible).
            spacing: A list of spacings between the DD gates.
                The available slack will be divided according to this.
                The list length must be one more than the length of dd_sequence,
                and the elements must sum to 1. If None, a balanced spacing
                will be used [d/2, d, d, ..., d, d, d/2].
            skip_reset_qubits: If True, does not insert DD on idle periods that
                immediately follow initialized/reset qubits
                (as qubits in the ground state are less susceptile to decoherence).
            pulse_alignment: The hardware constraints for gate timing allocation.
                This is usually provided from ``backend.configuration().timing_constraints``.
                If provided, the delay length, i.e. ``spacing``, is implicitly adjusted to
                satisfy this constraint.
            extra_slack_distribution: The option to control the behavior of DD sequence generation.
                The duration of the DD sequence should be identical to an idle time in the
                scheduled quantum circuit, however, the delay in between gates comprising the sequence
                should be integer number in units of dt, and it might be further truncated
                when ``pulse_alignment`` is specified. This sometimes results in the duration of
                the created sequence being shorter than the idle time
                that you want to fill with the sequence, i.e. `extra slack`.
                This option takes following values.

                    - "middle": Put the extra slack to the interval at the middle of the sequence.
                    - "edges": Divide the extra slack as evenly as possible into
                      intervals at beginning and end of the sequence.
            target: The :class:`~.Target` representing the target backend, if both
                  ``durations`` and this are specified then this argument will take
                  precedence and ``durations`` will be ignored.

        Raises:
            TranspilerError: When invalid DD sequence is specified.
            TranspilerError: When pulse gate with the duration which is
                non-multiple of the alignment constraint value is found.
            TypeError: If ``dd_sequence`` is not specified
        """
        super().__init__(target=target)
        self._durations = durations
        if dd_sequence is None:
            raise TypeError("required argument 'dd_sequence' is not specified")
        self._dd_sequence = dd_sequence
        self._qubits = qubits
        self._skip_reset_qubits = skip_reset_qubits
        self._alignment = pulse_alignment
        self._spacing = spacing
        self._extra_slack_distribution = extra_slack_distribution

        self._no_dd_qubits: set[int] = set()
        self._dd_sequence_lengths: dict[Qubit, list[int]] = {}
        self._sequence_phase = 0
        if target is not None:
            self._durations = target.durations()
            for gate in dd_sequence:
                if gate.name not in target.operation_names:
                    raise TranspilerError(
                        f"{gate.name} in dd_sequence is not supported in the target"
                    )

    def _pre_runhook(self, dag: DAGCircuit):
        super()._pre_runhook(dag)

        num_pulses = len(self._dd_sequence)

        # Check if physical circuit is given
        if len(dag.qregs) != 1 or dag.qregs.get("q", None) is None:
            raise TranspilerError("DD runs on physical circuits only.")

        # Set default spacing otherwise validate user input
        if self._spacing is None:
            mid = 1 / num_pulses
            end = mid / 2
            self._spacing = [end] + [mid] * (num_pulses - 1) + [end]
        else:
            if sum(self._spacing) != 1 or any(a < 0 for a in self._spacing):
                raise TranspilerError(
                    "The spacings must be given in terms of fractions "
                    "of the slack period and sum to 1."
                )

        # Check if DD sequence is identity
        if num_pulses != 1:
            if num_pulses % 2 != 0:
                raise TranspilerError("DD sequence must contain an even number of gates (or 1).")
            noop = np.eye(2)
            for gate in self._dd_sequence:
                noop = noop.dot(gate.to_matrix())
            if not matrix_equal(noop, IGate().to_matrix(), ignore_phase=True):
                raise TranspilerError("The DD sequence does not make an identity operation.")
            self._sequence_phase = np.angle(noop[0][0])

        # Compute no DD qubits on which any gate in dd_sequence is not supported in the target
        for qarg, _ in enumerate(dag.qubits):
            for gate in self._dd_sequence:
                if not self.__gate_supported(gate, qarg):
                    self._no_dd_qubits.add(qarg)
                    logger.debug(
                        "No DD on qubit %d as gate %s is not supported on it", qarg, gate.name
                    )
                    break
        # Precompute qubit-wise DD sequence length for performance
        for physical_index, qubit in enumerate(dag.qubits):
            if not self.__is_dd_qubit(physical_index):
                continue

            sequence_lengths = []
            for gate in self._dd_sequence:
                try:
                    # Check calibration.
                    params = self._resolve_params(gate)
                    gate_length = dag.calibrations[gate.name][((physical_index,), params)].duration
                    if gate_length % self._alignment != 0:
                        # This is necessary to implement lightweight scheduling logic for this pass.
                        # Usually the pulse alignment constraint and pulse data chunk size take
                        # the same value, however, we can intentionally violate this pattern
                        # at the gate level. For example, we can create a schedule consisting of
                        # a pi-pulse of 32 dt followed by a post buffer, i.e. delay, of 4 dt
                        # on the device with 16 dt constraint. Note that the pi-pulse length
                        # is multiple of 16 dt but the gate length of 36 is not multiple of it.
                        # Such pulse gate should be excluded.
                        raise TranspilerError(
                            f"Pulse gate {gate.name} with length non-multiple of {self._alignment} "
                            f"is not acceptable in {self.__class__.__name__} pass."
                        )
                except KeyError:
                    gate_length = self._durations.get(gate, physical_index)
                sequence_lengths.append(gate_length)
                # Update gate duration. This is necessary for current timeline drawer, i.e. scheduled.
                gate.duration = gate_length
            self._dd_sequence_lengths[qubit] = sequence_lengths

    def __gate_supported(self, gate: Gate, qarg: int) -> bool:
        """A gate is supported on the qubit (qarg) or not."""
        if self.target is None or self.target.instruction_supported(gate.name, qargs=(qarg,)):
            return True
        return False

    def __is_dd_qubit(self, qubit_index: int) -> bool:
        """DD can be inserted in the qubit or not."""
        if (qubit_index in self._no_dd_qubits) or (
            self._qubits and qubit_index not in self._qubits
        ):
            return False
        return True

    def _pad(
        self,
        dag: DAGCircuit,
        qubit: Qubit,
        t_start: int,
        t_end: int,
        next_node: DAGNode,
        prev_node: DAGNode,
    ):
        # This routine takes care of the pulse alignment constraint for the DD sequence.
        # Note that the alignment constraint acts on the t0 of the DAGOpNode.
        # Now this constrained scheduling problem is simplified to the problem of
        # finding a delay amount which is a multiple of the constraint value by assuming
        # that the duration of every DAGOpNode is also a multiple of the constraint value.
        #
        # For example, given the constraint value of 16 and XY4 with 160 dt gates.
        # Here we assume current interval is 992 dt.
        #
        # relative spacing := [0.125, 0.25, 0.25, 0.25, 0.125]
        # slack = 992 dt - 4 x 160 dt = 352 dt
        #
        # unconstraind sequence: 44dt-X1-88dt-Y2-88dt-X3-88dt-Y4-44dt
        # constraind sequence  : 32dt-X1-80dt-Y2-80dt-X3-80dt-Y4-32dt + extra slack 48 dt
        #
        # Now we evenly split extra slack into start and end of the sequence.
        # The distributed slack should be multiple of 16.
        # Start = +16, End += 32
        #
        # final sequence       : 48dt-X1-80dt-Y2-80dt-X3-80dt-Y4-64dt / in total 992 dt
        #
        # Now we verify t0 of every node starts from multiple of 16 dt.
        #
        # X1:  48 dt (3 x 16 dt)
        # Y2:  48 dt + 160 dt + 80 dt = 288 dt (18 x 16 dt)
        # Y3: 288 dt + 160 dt + 80 dt = 528 dt (33 x 16 dt)
        # Y4: 368 dt + 160 dt + 80 dt = 768 dt (48 x 16 dt)
        #
        # As you can see, constraints on t0 are all satisfied without explicit scheduling.
        time_interval = t_end - t_start
        if time_interval % self._alignment != 0:
            raise TranspilerError(
                f"Time interval {time_interval} is not divisible by alignment {self._alignment} "
                f"between DAGNode {prev_node.name} on qargs {prev_node.qargs} and {next_node.name} "
                f"on qargs {next_node.qargs}."
            )

        if not self.__is_dd_qubit(dag.qubits.index(qubit)):
            # Target physical qubit is not the target of this DD sequence.
            self._apply_scheduled_op(dag, t_start, Delay(time_interval, dag.unit), qubit)
            return

        if self._skip_reset_qubits and (
            isinstance(prev_node, DAGInNode) or isinstance(prev_node.op, Reset)
        ):
            # Previous node is the start edge or reset, i.e. qubit is ground state.
            self._apply_scheduled_op(dag, t_start, Delay(time_interval, dag.unit), qubit)
            return

        slack = time_interval - np.sum(self._dd_sequence_lengths[qubit])
        sequence_gphase = self._sequence_phase

        if slack <= 0:
            # Interval too short.
            self._apply_scheduled_op(dag, t_start, Delay(time_interval, dag.unit), qubit)
            return

        if len(self._dd_sequence) == 1:
            # Special case of using a single gate for DD
            u_inv = self._dd_sequence[0].inverse().to_matrix()
            theta, phi, lam, phase = OneQubitEulerDecomposer().angles_and_phase(u_inv)
            if isinstance(next_node, DAGOpNode) and isinstance(next_node.op, (UGate, U3Gate)):
                # Absorb the inverse into the successor (from left in circuit)
                theta_r, phi_r, lam_r = next_node.op.params
                next_node.op.params = Optimize1qGates.compose_u3(
                    theta_r, phi_r, lam_r, theta, phi, lam
                )
                sequence_gphase += phase
            elif isinstance(prev_node, DAGOpNode) and isinstance(prev_node.op, (UGate, U3Gate)):
                # Absorb the inverse into the predecessor (from right in circuit)
                theta_l, phi_l, lam_l = prev_node.op.params
                prev_node.op.params = Optimize1qGates.compose_u3(
                    theta, phi, lam, theta_l, phi_l, lam_l
                )
                sequence_gphase += phase
            else:
                # Don't do anything if there's no single-qubit gate to absorb the inverse
                self._apply_scheduled_op(dag, t_start, Delay(time_interval, dag.unit), qubit)
                return

        def _constrained_length(values):
            return self._alignment * np.floor(values / self._alignment)

        # (1) Compute DD intervals satisfying the constraint
        taus = _constrained_length(slack * np.asarray(self._spacing))
        extra_slack = slack - np.sum(taus)

        # (2) Distribute extra slack
        if self._extra_slack_distribution == "middle":
            mid_ind = int((len(taus) - 1) / 2)
            to_middle = _constrained_length(extra_slack)
            taus[mid_ind] += to_middle
            if extra_slack - to_middle:
                # If to_middle is not a multiple value of the pulse alignment,
                # it is truncated to the nearlest multiple value and
                # the rest of slack is added to the end.
                taus[-1] += extra_slack - to_middle
        elif self._extra_slack_distribution == "edges":
            to_begin_edge = _constrained_length(extra_slack / 2)
            taus[0] += to_begin_edge
            taus[-1] += extra_slack - to_begin_edge
        else:
            raise TranspilerError(
                f"Option extra_slack_distribution = {self._extra_slack_distribution} is invalid."
            )

        # (3) Construct DD sequence with delays
        num_elements = max(len(self._dd_sequence), len(taus))
        idle_after = t_start
        for dd_ind in range(num_elements):
            if dd_ind < len(taus):
                tau = taus[dd_ind]
                if tau > 0:
                    self._apply_scheduled_op(dag, idle_after, Delay(tau, dag.unit), qubit)
                    idle_after += tau
            if dd_ind < len(self._dd_sequence):
                gate = self._dd_sequence[dd_ind]
                gate_length = self._dd_sequence_lengths[qubit][dd_ind]
                self._apply_scheduled_op(dag, idle_after, gate, qubit)
                idle_after += gate_length

        dag.global_phase = self._mod_2pi(dag.global_phase + sequence_gphase)

    @staticmethod
    def _resolve_params(gate: Gate) -> tuple:
        """Return gate params with any bound parameters replaced with floats"""
        params = []
        for p in gate.params:
            if isinstance(p, ParameterExpression) and not p.parameters:
                params.append(float(p))
            else:
                params.append(p)
        return tuple(params)

    @staticmethod
    def _mod_2pi(angle: float, atol: float = 0):
        """Wrap angle into interval [-π,π). If within atol of the endpoint, clamp to -π"""
        wrapped = (angle + np.pi) % (2 * np.pi) - np.pi
        if abs(wrapped - np.pi) < atol:
            wrapped = -np.pi
        return wrapped

1	# This code is part of Qiskit.
2	#
3	# (C) Copyright IBM 2021.
4	#
5	# This code is licensed under the Apache License, Version 2.0. You may
6	# obtain a copy of this license in the LICENSE.txt file in the root directory
7	# of this source tree or at http://www.apache.org/licenses/LICENSE-2.0.
8	#
9	# Any modifications or derivative works of this code must retain this
10	# copyright notice, and modified files need to carry a notice indicating
11	# that they have been altered from the originals.
12
13	"""Dynamical Decoupling insertion pass."""	1✔
14	from __future__ import annotations	1✔
15
16	import logging	1✔
17	import numpy as np	1✔
18
19	from qiskit.circuit import Gate, ParameterExpression, Qubit	1✔
20	from qiskit.circuit.delay import Delay	1✔
21	from qiskit.circuit.library.standard_gates import IGate, UGate, U3Gate	1✔
22	from qiskit.circuit.reset import Reset	1✔
23	from qiskit.dagcircuit import DAGCircuit, DAGNode, DAGInNode, DAGOpNode	1✔
24	from qiskit.quantum_info.operators.predicates import matrix_equal	1✔
25	from qiskit.quantum_info.synthesis import OneQubitEulerDecomposer	1✔
26	from qiskit.transpiler.exceptions import TranspilerError	1✔
27	from qiskit.transpiler.instruction_durations import InstructionDurations	1✔
28	from qiskit.transpiler.passes.optimization import Optimize1qGates	1✔
29	from qiskit.transpiler.target import Target	1✔
30
31	from .base_padding import BasePadding	1✔
32
33	logger = logging.getLogger(__name__)	1✔
34
35
36	class PadDynamicalDecoupling(BasePadding):	1✔
37	"""Dynamical decoupling insertion pass.	1✔
38
39	This pass works on a scheduled, physical circuit. It scans the circuit for
40	idle periods of time (i.e. those containing delay instructions) and inserts
41	a DD sequence of gates in those spots. These gates amount to the identity,
42	so do not alter the logical action of the circuit, but have the effect of
43	mitigating decoherence in those idle periods.
44
45	As a special case, the pass allows a length-1 sequence (e.g. [XGate()]).
46	In this case the DD insertion happens only when the gate inverse can be
47	absorbed into a neighboring gate in the circuit (so we would still be
48	replacing Delay with something that is equivalent to the identity).
49	This can be used, for instance, as a Hahn echo.
50
51	This pass ensures that the inserted sequence preserves the circuit exactly
52	(including global phase).
53
54	.. plot::
55	:include-source:
56
57	import numpy as np
58	from qiskit.circuit import QuantumCircuit
59	from qiskit.circuit.library import XGate
60	from qiskit.transpiler import PassManager, InstructionDurations
61	from qiskit.transpiler.passes import ALAPScheduleAnalysis, PadDynamicalDecoupling
62	from qiskit.visualization import timeline_drawer
63	circ = QuantumCircuit(4)
64	circ.h(0)
65	circ.cx(0, 1)
66	circ.cx(1, 2)
67	circ.cx(2, 3)
68	circ.measure_all()
69	durations = InstructionDurations(
70	[("h", 0, 50), ("cx", [0, 1], 700), ("reset", None, 10),
71	("cx", [1, 2], 200), ("cx", [2, 3], 300),
72	("x", None, 50), ("measure", None, 1000)]
73	)
74
75	# balanced X-X sequence on all qubits
76	dd_sequence = [XGate(), XGate()]
77	pm = PassManager([ALAPScheduleAnalysis(durations),
78	PadDynamicalDecoupling(durations, dd_sequence)])
79	circ_dd = pm.run(circ)
80	timeline_drawer(circ_dd)
81
82	# Uhrig sequence on qubit 0
83	n = 8
84	dd_sequence = [XGate()] * n
85	def uhrig_pulse_location(k):
86	return np.sin(np.pi * (k + 1) / (2 * n + 2)) ** 2
87	spacing = []
88	for k in range(n):
89	spacing.append(uhrig_pulse_location(k) - sum(spacing))
90	spacing.append(1 - sum(spacing))
91	pm = PassManager(
92	[
93	ALAPScheduleAnalysis(durations),
94	PadDynamicalDecoupling(durations, dd_sequence, qubits=[0], spacing=spacing),
95	]
96	)
97	circ_dd = pm.run(circ)
98	timeline_drawer(circ_dd)
99
100	.. note::
101
102	You may need to call alignment pass before running dynamical decoupling to guarantee
103	your circuit satisfies acquisition alignment constraints.
104	"""
105
106	def __init__(	1✔
107	self,
108	durations: InstructionDurations = None,
109	dd_sequence: list[Gate] = None,
110	qubits: list[int] \| None = None,
111	spacing: list[float] \| None = None,
112	skip_reset_qubits: bool = True,
113	pulse_alignment: int = 1,
114	extra_slack_distribution: str = "middle",
115	target: Target = None,
116	):
117	"""Dynamical decoupling initializer.
118
119	Args:
120	durations: Durations of instructions to be used in scheduling.
121	dd_sequence: Sequence of gates to apply in idle spots.
122	qubits: Physical qubits on which to apply DD.
123	If None, all qubits will undergo DD (when possible).
124	spacing: A list of spacings between the DD gates.
125	The available slack will be divided according to this.
126	The list length must be one more than the length of dd_sequence,
127	and the elements must sum to 1. If None, a balanced spacing
128	will be used [d/2, d, d, ..., d, d, d/2].
129	skip_reset_qubits: If True, does not insert DD on idle periods that
130	immediately follow initialized/reset qubits
131	(as qubits in the ground state are less susceptile to decoherence).
132	pulse_alignment: The hardware constraints for gate timing allocation.
133	This is usually provided from ``backend.configuration().timing_constraints``.
134	If provided, the delay length, i.e. ``spacing``, is implicitly adjusted to
135	satisfy this constraint.
136	extra_slack_distribution: The option to control the behavior of DD sequence generation.
137	The duration of the DD sequence should be identical to an idle time in the
138	scheduled quantum circuit, however, the delay in between gates comprising the sequence
139	should be integer number in units of dt, and it might be further truncated
140	when ``pulse_alignment`` is specified. This sometimes results in the duration of
141	the created sequence being shorter than the idle time
142	that you want to fill with the sequence, i.e. `extra slack`.
143	This option takes following values.
144
145	- "middle": Put the extra slack to the interval at the middle of the sequence.
146	- "edges": Divide the extra slack as evenly as possible into
147	intervals at beginning and end of the sequence.
148	target: The :class:`~.Target` representing the target backend, if both
149	``durations`` and this are specified then this argument will take
150	precedence and ``durations`` will be ignored.
151
152	Raises:
153	TranspilerError: When invalid DD sequence is specified.
154	TranspilerError: When pulse gate with the duration which is
155	non-multiple of the alignment constraint value is found.
156	TypeError: If ``dd_sequence`` is not specified
157	"""
158	super().__init__(target=target)	1✔
159	self._durations = durations	1✔
160	if dd_sequence is None:	1✔
161	raise TypeError("required argument 'dd_sequence' is not specified")	×
162	self._dd_sequence = dd_sequence	1✔
163	self._qubits = qubits	1✔
164	self._skip_reset_qubits = skip_reset_qubits	1✔
165	self._alignment = pulse_alignment	1✔
166	self._spacing = spacing	1✔
167	self._extra_slack_distribution = extra_slack_distribution	1✔
168
169	self._no_dd_qubits: set[int] = set()	1✔
170	self._dd_sequence_lengths: dict[Qubit, list[int]] = {}	1✔
171	self._sequence_phase = 0	1✔
172	if target is not None:	1✔
173	self._durations = target.durations()	1✔
174	for gate in dd_sequence:	1✔
175	if gate.name not in target.operation_names:	1✔
176	raise TranspilerError(	1✔
177	f"{gate.name} in dd_sequence is not supported in the target"
178	)
179
180	def _pre_runhook(self, dag: DAGCircuit):	1✔
181	super()._pre_runhook(dag)	1✔
182
183	num_pulses = len(self._dd_sequence)	1✔
184
185	# Check if physical circuit is given
186	if len(dag.qregs) != 1 or dag.qregs.get("q", None) is None:	1✔
187	raise TranspilerError("DD runs on physical circuits only.")	×
188
189	# Set default spacing otherwise validate user input
190	if self._spacing is None:	1✔
191	mid = 1 / num_pulses	1✔
192	end = mid / 2	1✔
193	self._spacing = [end] + [mid] * (num_pulses - 1) + [end]	1✔
194	else:
195	if sum(self._spacing) != 1 or any(a < 0 for a in self._spacing):	1✔
196	raise TranspilerError(	×
197	"The spacings must be given in terms of fractions "
198	"of the slack period and sum to 1."
199	)
200
201	# Check if DD sequence is identity
202	if num_pulses != 1:	1✔
203	if num_pulses % 2 != 0:	1✔
204	raise TranspilerError("DD sequence must contain an even number of gates (or 1).")	×
205	noop = np.eye(2)	1✔
206	for gate in self._dd_sequence:	1✔
207	noop = noop.dot(gate.to_matrix())	1✔
208	if not matrix_equal(noop, IGate().to_matrix(), ignore_phase=True):	1✔
209	raise TranspilerError("The DD sequence does not make an identity operation.")	1✔
210	self._sequence_phase = np.angle(noop[0][0])	1✔
211
212	# Compute no DD qubits on which any gate in dd_sequence is not supported in the target
213	for qarg, _ in enumerate(dag.qubits):	1✔
214	for gate in self._dd_sequence:	1✔
215	if not self.__gate_supported(gate, qarg):	1✔
216	self._no_dd_qubits.add(qarg)	1✔
217	logger.debug(	1✔
218	"No DD on qubit %d as gate %s is not supported on it", qarg, gate.name
219	)
220	break	1✔
221	# Precompute qubit-wise DD sequence length for performance
222	for physical_index, qubit in enumerate(dag.qubits):	1✔
223	if not self.__is_dd_qubit(physical_index):	1✔
224	continue	1✔
225
226	sequence_lengths = []	1✔
227	for gate in self._dd_sequence:	1✔
228	try:	1✔
229	# Check calibration.
230	params = self._resolve_params(gate)	1✔
231	gate_length = dag.calibrations[gate.name][((physical_index,), params)].duration	1✔
232	if gate_length % self._alignment != 0:	1✔
233	# This is necessary to implement lightweight scheduling logic for this pass.
234	# Usually the pulse alignment constraint and pulse data chunk size take
235	# the same value, however, we can intentionally violate this pattern
236	# at the gate level. For example, we can create a schedule consisting of
237	# a pi-pulse of 32 dt followed by a post buffer, i.e. delay, of 4 dt
238	# on the device with 16 dt constraint. Note that the pi-pulse length
239	# is multiple of 16 dt but the gate length of 36 is not multiple of it.
240	# Such pulse gate should be excluded.
241	raise TranspilerError(	×
242	f"Pulse gate {gate.name} with length non-multiple of {self._alignment} "
243	f"is not acceptable in {self.__class__.__name__} pass."
244	)
245	except KeyError:	1✔
246	gate_length = self._durations.get(gate, physical_index)	1✔
247	sequence_lengths.append(gate_length)	1✔
248	# Update gate duration. This is necessary for current timeline drawer, i.e. scheduled.
249	gate.duration = gate_length	1✔
250	self._dd_sequence_lengths[qubit] = sequence_lengths	1✔
251
252	def __gate_supported(self, gate: Gate, qarg: int) -> bool:	1✔
253	"""A gate is supported on the qubit (qarg) or not."""
254	if self.target is None or self.target.instruction_supported(gate.name, qargs=(qarg,)):	1✔
255	return True	1✔
256	return False	1✔
257
258	def __is_dd_qubit(self, qubit_index: int) -> bool:	1✔
259	"""DD can be inserted in the qubit or not."""
260	if (qubit_index in self._no_dd_qubits) or (	1✔
261	self._qubits and qubit_index not in self._qubits
262	):
263	return False	1✔
264	return True	1✔
265
266	def _pad(	1✔
267	self,
268	dag: DAGCircuit,
269	qubit: Qubit,
270	t_start: int,
271	t_end: int,
272	next_node: DAGNode,
273	prev_node: DAGNode,
274	):
275	# This routine takes care of the pulse alignment constraint for the DD sequence.
276	# Note that the alignment constraint acts on the t0 of the DAGOpNode.
277	# Now this constrained scheduling problem is simplified to the problem of
278	# finding a delay amount which is a multiple of the constraint value by assuming
279	# that the duration of every DAGOpNode is also a multiple of the constraint value.
280	#
281	# For example, given the constraint value of 16 and XY4 with 160 dt gates.
282	# Here we assume current interval is 992 dt.
283	#
284	# relative spacing := [0.125, 0.25, 0.25, 0.25, 0.125]
285	# slack = 992 dt - 4 x 160 dt = 352 dt
286	#
287	# unconstraind sequence: 44dt-X1-88dt-Y2-88dt-X3-88dt-Y4-44dt
288	# constraind sequence : 32dt-X1-80dt-Y2-80dt-X3-80dt-Y4-32dt + extra slack 48 dt
289	#
290	# Now we evenly split extra slack into start and end of the sequence.
291	# The distributed slack should be multiple of 16.
292	# Start = +16, End += 32
293	#
294	# final sequence : 48dt-X1-80dt-Y2-80dt-X3-80dt-Y4-64dt / in total 992 dt
295	#
296	# Now we verify t0 of every node starts from multiple of 16 dt.
297	#
298	# X1: 48 dt (3 x 16 dt)
299	# Y2: 48 dt + 160 dt + 80 dt = 288 dt (18 x 16 dt)
300	# Y3: 288 dt + 160 dt + 80 dt = 528 dt (33 x 16 dt)
301	# Y4: 368 dt + 160 dt + 80 dt = 768 dt (48 x 16 dt)
302	#
303	# As you can see, constraints on t0 are all satisfied without explicit scheduling.
304	time_interval = t_end - t_start	1✔
305	if time_interval % self._alignment != 0:	1✔
306	raise TranspilerError(	×
307	f"Time interval {time_interval} is not divisible by alignment {self._alignment} "
308	f"between DAGNode {prev_node.name} on qargs {prev_node.qargs} and {next_node.name} "
309	f"on qargs {next_node.qargs}."
310	)
311
312	if not self.__is_dd_qubit(dag.qubits.index(qubit)):	1✔
313	# Target physical qubit is not the target of this DD sequence.
314	self._apply_scheduled_op(dag, t_start, Delay(time_interval, dag.unit), qubit)	1✔
315	return	1✔
316
317	if self._skip_reset_qubits and (	1✔
318	isinstance(prev_node, DAGInNode) or isinstance(prev_node.op, Reset)
319	):
320	# Previous node is the start edge or reset, i.e. qubit is ground state.
321	self._apply_scheduled_op(dag, t_start, Delay(time_interval, dag.unit), qubit)	1✔
322	return	1✔
323
324	slack = time_interval - np.sum(self._dd_sequence_lengths[qubit])	1✔
325	sequence_gphase = self._sequence_phase	1✔
326
327	if slack <= 0:	1✔
328	# Interval too short.
329	self._apply_scheduled_op(dag, t_start, Delay(time_interval, dag.unit), qubit)	1✔
330	return	1✔
331
332	if len(self._dd_sequence) == 1:	1✔
333	# Special case of using a single gate for DD
334	u_inv = self._dd_sequence[0].inverse().to_matrix()	1✔
335	theta, phi, lam, phase = OneQubitEulerDecomposer().angles_and_phase(u_inv)	1✔
336	if isinstance(next_node, DAGOpNode) and isinstance(next_node.op, (UGate, U3Gate)):	1✔
337	# Absorb the inverse into the successor (from left in circuit)
338	theta_r, phi_r, lam_r = next_node.op.params	1✔
339	next_node.op.params = Optimize1qGates.compose_u3(	1✔
340	theta_r, phi_r, lam_r, theta, phi, lam
341	)
342	sequence_gphase += phase	1✔
343	elif isinstance(prev_node, DAGOpNode) and isinstance(prev_node.op, (UGate, U3Gate)):	1✔
344	# Absorb the inverse into the predecessor (from right in circuit)
345	theta_l, phi_l, lam_l = prev_node.op.params	1✔
346	prev_node.op.params = Optimize1qGates.compose_u3(	1✔
347	theta, phi, lam, theta_l, phi_l, lam_l
348	)
349	sequence_gphase += phase	1✔
350	else:
351	# Don't do anything if there's no single-qubit gate to absorb the inverse
352	self._apply_scheduled_op(dag, t_start, Delay(time_interval, dag.unit), qubit)	1✔
353	return	1✔
354
355	def _constrained_length(values):	1✔
356	return self._alignment * np.floor(values / self._alignment)	1✔
357
358	# (1) Compute DD intervals satisfying the constraint
359	taus = _constrained_length(slack * np.asarray(self._spacing))	1✔
360	extra_slack = slack - np.sum(taus)	1✔
361
362	# (2) Distribute extra slack
363	if self._extra_slack_distribution == "middle":	1✔
364	mid_ind = int((len(taus) - 1) / 2)	1✔
365	to_middle = _constrained_length(extra_slack)	1✔
366	taus[mid_ind] += to_middle	1✔
367	if extra_slack - to_middle:	1✔
368	# If to_middle is not a multiple value of the pulse alignment,
369	# it is truncated to the nearlest multiple value and
370	# the rest of slack is added to the end.
371	taus[-1] += extra_slack - to_middle	×
372	elif self._extra_slack_distribution == "edges":	1✔
373	to_begin_edge = _constrained_length(extra_slack / 2)	1✔
374	taus[0] += to_begin_edge	1✔
375	taus[-1] += extra_slack - to_begin_edge	1✔
376	else:
377	raise TranspilerError(	×
378	f"Option extra_slack_distribution = {self._extra_slack_distribution} is invalid."
379	)
380
381	# (3) Construct DD sequence with delays
382	num_elements = max(len(self._dd_sequence), len(taus))	1✔
383	idle_after = t_start	1✔
384	for dd_ind in range(num_elements):	1✔
385	if dd_ind < len(taus):	1✔
386	tau = taus[dd_ind]	1✔
387	if tau > 0:	1✔
388	self._apply_scheduled_op(dag, idle_after, Delay(tau, dag.unit), qubit)	1✔
389	idle_after += tau	1✔
390	if dd_ind < len(self._dd_sequence):	1✔
391	gate = self._dd_sequence[dd_ind]	1✔
392	gate_length = self._dd_sequence_lengths[qubit][dd_ind]	1✔
393	self._apply_scheduled_op(dag, idle_after, gate, qubit)	1✔
394	idle_after += gate_length	1✔
395
396	dag.global_phase = self._mod_2pi(dag.global_phase + sequence_gphase)	1✔
397
398	@staticmethod	1✔
399	def _resolve_params(gate: Gate) -> tuple:	1✔
400	"""Return gate params with any bound parameters replaced with floats"""
401	params = []	1✔
402	for p in gate.params:	1✔
403	if isinstance(p, ParameterExpression) and not p.parameters:	1✔
404	params.append(float(p))	×
405	else:
406	params.append(p)	1✔
407	return tuple(params)	1✔
408
409	@staticmethod	1✔
410	def _mod_2pi(angle: float, atol: float = 0):	1✔
411	"""Wrap angle into interval [-π,π). If within atol of the endpoint, clamp to -π"""
412	wrapped = (angle + np.pi) % (2 * np.pi) - np.pi	1✔
413	if abs(wrapped - np.pi) < atol:	1✔
414	wrapped = -np.pi	×
415	return wrapped	1✔

Qiskit / qiskit / 6206652911

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous