11909933211

Committed 19 Nov 2024 08:11AM UTC coverage: 88.933% (+0.01%) from 88.92%

Build # 11909933211

Build Type

push

github

Committed by

web-flow

Commit Message

Fix post-oxidization change in `ConsolidateBlocks` behavior  (#13450) (#13453)

* Set decomposer to None if kak_gates not found during ConsolidateBlocks initialization.

* Add test that reproduces issue

* Re-add rzz to list of kak gates

* Add path for rzx gate.

(cherry picked from commit 94cea4190)

Co-authored-by: Elena Peña Tapia <57907331+ElePT@users.noreply.github.com>

Run Details

4 of 5 new or added lines in 1 file covered. (80.0%)

7 existing lines in 2 files now uncovered.

79069 of 88909 relevant lines covered (88.93%)

355823.01 hits per line

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

96.36

/qiskit/transpiler/passes/optimization/consolidate_blocks.py

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

"""Replace each block of consecutive gates by a single Unitary node."""
from __future__ import annotations
from math import pi

from qiskit.synthesis.two_qubit import TwoQubitBasisDecomposer
from qiskit.circuit.library.standard_gates import CXGate, CZGate, iSwapGate, ECRGate, RXXGate

from qiskit.transpiler.basepasses import TransformationPass
from qiskit.transpiler.passmanager import PassManager
from qiskit._accelerate.consolidate_blocks import consolidate_blocks

from .collect_1q_runs import Collect1qRuns
from .collect_2q_blocks import Collect2qBlocks

KAK_GATE_NAMES = {
    "cx": CXGate(),
    "cz": CZGate(),
    "iswap": iSwapGate(),
    "ecr": ECRGate(),
    "rxx": RXXGate(pi / 2),
}


class ConsolidateBlocks(TransformationPass):
    """Replace each block of consecutive gates by a single Unitary node.

    Pass to consolidate sequences of uninterrupted gates acting on
    the same qubits into a Unitary node, to be resynthesized later,
    to a potentially more optimal subcircuit.

    Notes:
        This pass assumes that the 'blocks_list' property that it reads is
        given such that blocks are in topological order. The blocks are
        collected by a previous pass, such as `Collect2qBlocks`.
    """

    def __init__(
        self,
        kak_basis_gate=None,
        force_consolidate=False,
        basis_gates=None,
        approximation_degree=1.0,
        target=None,
    ):
        """ConsolidateBlocks initializer.

        If ``kak_basis_gate`` is not ``None`` it will be used as the basis gate for KAK decomposition.
        Otherwise, if ``basis_gates`` is not ``None`` a basis gate will be chosen from this list.
        Otherwise, the basis gate will be :class:`.CXGate`.

        Args:
            kak_basis_gate (Gate): Basis gate for KAK decomposition.
            force_consolidate (bool): Force block consolidation.
            basis_gates (List(str)): Basis gates from which to choose a KAK gate.
            approximation_degree (float): a float between :math:`[0.0, 1.0]`. Lower approximates more.
            target (Target): The target object for the compilation target backend.
        """
        super().__init__()
        self.basis_gates = None
        self.target = target
        if basis_gates is not None:
            self.basis_gates = set(basis_gates)
        self.force_consolidate = force_consolidate
        if kak_basis_gate is not None:
            self.decomposer = TwoQubitBasisDecomposer(kak_basis_gate)
        elif basis_gates is not None:
            kak_gates = KAK_GATE_NAMES.keys() & (basis_gates or [])
            if kak_gates:
                self.decomposer = TwoQubitBasisDecomposer(
                    KAK_GATE_NAMES[kak_gates.pop()], basis_fidelity=approximation_degree or 1.0
                )
            elif "rzx" in basis_gates:
                self.decomposer = TwoQubitBasisDecomposer(
                    CXGate(), basis_fidelity=approximation_degree or 1.0
                )
            else:
                self.decomposer = None
        else:
            self.decomposer = TwoQubitBasisDecomposer(CXGate())

    def run(self, dag):
        """Run the ConsolidateBlocks pass on `dag`.

        Iterate over each block and replace it with an equivalent Unitary
        on the same wires.
        """
        if self.decomposer is None:
            return dag

        blocks = self.property_set["block_list"]
        if blocks is not None:
            blocks = [[node._node_id for node in block] for block in blocks]
        runs = self.property_set["run_list"]
        if runs is not None:
            runs = [[node._node_id for node in run] for run in runs]

        consolidate_blocks(
            dag,
            self.decomposer._inner_decomposer,
            self.force_consolidate,
            target=self.target,
            basis_gates=self.basis_gates,
            blocks=blocks,
            runs=runs,
        )
        dag = self._handle_control_flow_ops(dag)

        # Clear collected blocks and runs as they are no longer valid after consolidation
        if "run_list" in self.property_set:
            del self.property_set["run_list"]
        if "block_list" in self.property_set:
            del self.property_set["block_list"]

        return dag

    def _handle_control_flow_ops(self, dag):
        """
        This is similar to transpiler/passes/utils/control_flow.py except that the
        collect blocks is redone for the control flow blocks.
        """

        pass_manager = PassManager()
        if "run_list" in self.property_set:
            pass_manager.append(Collect1qRuns())
            pass_manager.append(Collect2qBlocks())

        pass_manager.append(self)
        control_flow_nodes = dag.control_flow_op_nodes()
        if control_flow_nodes is not None:
            for node in control_flow_nodes:
                dag.substitute_node(
                    node,
                    node.op.replace_blocks(pass_manager.run(block) for block in node.op.blocks),
                    propagate_condition=False,
                )
        return dag

1	# This code is part of Qiskit.
2	#
3	# (C) Copyright IBM 2017, 2019.
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	"""Replace each block of consecutive gates by a single Unitary node."""
14	from __future__ import annotations	1✔
15	from math import pi	1✔
16
17	from qiskit.synthesis.two_qubit import TwoQubitBasisDecomposer	1✔
18	from qiskit.circuit.library.standard_gates import CXGate, CZGate, iSwapGate, ECRGate, RXXGate	1✔
19
20	from qiskit.transpiler.basepasses import TransformationPass	1✔
21	from qiskit.transpiler.passmanager import PassManager	1✔
22	from qiskit._accelerate.consolidate_blocks import consolidate_blocks	1✔
23
24	from .collect_1q_runs import Collect1qRuns	1✔
25	from .collect_2q_blocks import Collect2qBlocks	1✔
26
27	KAK_GATE_NAMES = {	1✔
28	"cx": CXGate(),
29	"cz": CZGate(),
30	"iswap": iSwapGate(),
31	"ecr": ECRGate(),
32	"rxx": RXXGate(pi / 2),
33	}
34
35
36	class ConsolidateBlocks(TransformationPass):	1✔
37	"""Replace each block of consecutive gates by a single Unitary node.
38
39	Pass to consolidate sequences of uninterrupted gates acting on
40	the same qubits into a Unitary node, to be resynthesized later,
41	to a potentially more optimal subcircuit.
42
43	Notes:
44	This pass assumes that the 'blocks_list' property that it reads is
45	given such that blocks are in topological order. The blocks are
46	collected by a previous pass, such as `Collect2qBlocks`.
47	"""
48
49	def __init__(	1✔
50	self,
51	kak_basis_gate=None,
52	force_consolidate=False,
53	basis_gates=None,
54	approximation_degree=1.0,
55	target=None,
56	):
57	"""ConsolidateBlocks initializer.
58
59	If ``kak_basis_gate`` is not ``None`` it will be used as the basis gate for KAK decomposition.
60	Otherwise, if ``basis_gates`` is not ``None`` a basis gate will be chosen from this list.
61	Otherwise, the basis gate will be :class:`.CXGate`.
62
63	Args:
64	kak_basis_gate (Gate): Basis gate for KAK decomposition.
65	force_consolidate (bool): Force block consolidation.
66	basis_gates (List(str)): Basis gates from which to choose a KAK gate.
67	approximation_degree (float): a float between :math:`[0.0, 1.0]`. Lower approximates more.
68	target (Target): The target object for the compilation target backend.
69	"""
70	super().__init__()	1✔
71	self.basis_gates = None	1✔
72	self.target = target	1✔
73	if basis_gates is not None:	1✔
74	self.basis_gates = set(basis_gates)	1✔
75	self.force_consolidate = force_consolidate	1✔
76	if kak_basis_gate is not None:	1✔
UNCOV 77	self.decomposer = TwoQubitBasisDecomposer(kak_basis_gate)	×
78	elif basis_gates is not None:	1✔
79	kak_gates = KAK_GATE_NAMES.keys() & (basis_gates or [])	1✔
80	if kak_gates:	1✔
81	self.decomposer = TwoQubitBasisDecomposer(	1✔
82	KAK_GATE_NAMES[kak_gates.pop()], basis_fidelity=approximation_degree or 1.0
83	)
84	elif "rzx" in basis_gates:	1✔
NEW 85	self.decomposer = TwoQubitBasisDecomposer(	×
86	CXGate(), basis_fidelity=approximation_degree or 1.0
87	)
88	else:
89	self.decomposer = None	1✔
90	else:
91	self.decomposer = TwoQubitBasisDecomposer(CXGate())	1✔
92
93	def run(self, dag):	1✔
94	"""Run the ConsolidateBlocks pass on `dag`.
95
96	Iterate over each block and replace it with an equivalent Unitary
97	on the same wires.
98	"""
99	if self.decomposer is None:	1✔
100	return dag	1✔
101
102	blocks = self.property_set["block_list"]	1✔
103	if blocks is not None:	1✔
104	blocks = [[node._node_id for node in block] for block in blocks]	1✔
105	runs = self.property_set["run_list"]	1✔
106	if runs is not None:	1✔
107	runs = [[node._node_id for node in run] for run in runs]	1✔
108
109	consolidate_blocks(	1✔
110	dag,
111	self.decomposer._inner_decomposer,
112	self.force_consolidate,
113	target=self.target,
114	basis_gates=self.basis_gates,
115	blocks=blocks,
116	runs=runs,
117	)
118	dag = self._handle_control_flow_ops(dag)	1✔
119
120	# Clear collected blocks and runs as they are no longer valid after consolidation
121	if "run_list" in self.property_set:	1✔
122	del self.property_set["run_list"]	1✔
123	if "block_list" in self.property_set:	1✔
124	del self.property_set["block_list"]	1✔
125
126	return dag	1✔
127
128	def _handle_control_flow_ops(self, dag):	1✔
129	"""
130	This is similar to transpiler/passes/utils/control_flow.py except that the
131	collect blocks is redone for the control flow blocks.
132	"""
133
134	pass_manager = PassManager()	1✔
135	if "run_list" in self.property_set:	1✔
136	pass_manager.append(Collect1qRuns())	1✔
137	pass_manager.append(Collect2qBlocks())	1✔
138
139	pass_manager.append(self)	1✔
140	control_flow_nodes = dag.control_flow_op_nodes()	1✔
141	if control_flow_nodes is not None:	1✔
142	for node in control_flow_nodes:	1✔
143	dag.substitute_node(	1✔
144	node,
145	node.op.replace_blocks(pass_manager.run(block) for block in node.op.blocks),
146	propagate_condition=False,
147	)
148	return dag	1✔

Qiskit / qiskit / 11909933211

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