13540158629

Committed 26 Feb 2025 09:01AM UTC coverage: 87.854% (-0.7%) from 88.599%

Build # 13540158629

Build Type

Pull #12814

github

Committed by

web-flow

Commit Message

Merge f02a7b9b8 into 7169f6db0

Pull Request Pull Request #12814: Light Cone Transpiler Pass

Coverage Stats

79 of 81 new or added lines in 2 files covered. (97.53%)

2576 existing lines in 133 files now uncovered.

77200 of 87873 relevant lines covered (87.85%)

339322.89 hits per line

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

95.24

/qiskit/transpiler/passes/optimization/optimize_annotated.py

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

"""Optimize annotated operations on a circuit."""

from typing import Optional, List, Tuple, Union

from qiskit.circuit.controlflow import CONTROL_FLOW_OP_NAMES
from qiskit.converters import circuit_to_dag, dag_to_circuit
from qiskit.circuit.annotated_operation import AnnotatedOperation, _canonicalize_modifiers
from qiskit.circuit import (
    QuantumCircuit,
    Instruction,
    EquivalenceLibrary,
    ControlledGate,
    Operation,
    ControlFlowOp,
)
from qiskit.transpiler.basepasses import TransformationPass
from qiskit.transpiler.passes.utils import control_flow
from qiskit.transpiler.target import Target
from qiskit.dagcircuit import DAGCircuit
from qiskit.transpiler.exceptions import TranspilerError


class OptimizeAnnotated(TransformationPass):
    """Optimization pass on circuits with annotated operations.

    Implemented optimizations:

    * For each annotated operation, converting the list of its modifiers to a canonical form.
      For example, consecutively applying ``inverse()``, ``control(2)`` and ``inverse()``
      is equivalent to applying ``control(2)``.

    * Removing annotations when possible.
      For example, ``AnnotatedOperation(SwapGate(), [InverseModifier(), InverseModifier()])``
      is equivalent to ``SwapGate()``.

    * Recursively combining annotations.
      For example, if ``g1 = AnnotatedOperation(SwapGate(), InverseModifier())`` and
      ``g2 = AnnotatedOperation(g1, ControlModifier(2))``, then ``g2`` can be replaced with
      ``AnnotatedOperation(SwapGate(), [InverseModifier(), ControlModifier(2)])``.

    * Applies conjugate reduction to annotated operations. As an example,
      ``control - [P -- Q -- P^{-1}]`` can be rewritten as ``P -- control - [Q] -- P^{-1}``,
      that is, only the middle part needs to be controlled. This also works for inverse
      and power modifiers.

    """

    def __init__(
        self,
        target: Optional[Target] = None,
        equivalence_library: Optional[EquivalenceLibrary] = None,
        basis_gates: Optional[List[str]] = None,
        recurse: bool = True,
        do_conjugate_reduction: bool = True,
    ):
        """
        OptimizeAnnotated initializer.

        Args:
            target: Optional, the backend target to use for this pass.
            equivalence_library: The equivalence library used
                (instructions in this library will not be optimized by this pass).
            basis_gates: Optional, target basis names to unroll to, e.g. `['u3', 'cx']`
                (instructions in this list will not be optimized by this pass).
                Ignored if ``target`` is also specified.
            recurse: By default, when either ``target`` or ``basis_gates`` is specified,
                the pass recursively descends into gate definitions (and the recursion is
                not applied when neither is specified since such objects do not need to
                be synthesized). Setting this value to ``False`` precludes the recursion in
                every case.
            do_conjugate_reduction: controls whether conjugate reduction should be performed.
        """
        super().__init__()

        self._target = target
        self._equiv_lib = equivalence_library
        self._basis_gates = basis_gates
        self._do_conjugate_reduction = do_conjugate_reduction

        self._top_level_only = not recurse or (self._basis_gates is None and self._target is None)

        if not self._top_level_only and self._target is None:
            basic_insts = {"measure", "reset", "barrier", "snapshot", "delay", "store"}
            self._device_insts = basic_insts | set(self._basis_gates)

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

        Args:
            dag: input dag.

        Returns:
            Output dag with higher-level operations optimized.

        Raises:
            TranspilerError: when something goes wrong.

        """
        dag, _ = self._run_inner(dag)
        return dag

    def _run_inner(self, dag) -> Tuple[DAGCircuit, bool]:
        """
        Optimizes annotated operations.
        Returns True if did something.
        """
        # Fast return
        if self._top_level_only:
            op_names = dag.count_ops(recurse=False)
            if "annotated" not in op_names and not CONTROL_FLOW_OP_NAMES.intersection(op_names):
                return dag, False

        # Handle control-flow
        for node in dag.op_nodes():
            if isinstance(node.op, ControlFlowOp):
                dag.substitute_node(
                    node,
                    control_flow.map_blocks(self.run, node.op),
                )

        # First, optimize every node in the DAG.
        dag, opt1 = self._canonicalize(dag)

        opt2 = False
        if not self._top_level_only:
            # Second, recursively descend into definitions.
            # Note that it is important to recurse only after the optimization methods have been run,
            # as they may remove annotated gates.
            dag, opt2 = self._recurse(dag)

        opt3 = False
        if not self._top_level_only and self._do_conjugate_reduction:
            dag, opt3 = self._conjugate_reduction(dag)

        return dag, opt1 or opt2 or opt3

    def _canonicalize(self, dag) -> Tuple[DAGCircuit, bool]:
        """
        Combines recursive annotated operations and canonicalizes modifiers.
        Returns True if did something.
        """

        did_something = False
        for node in dag.op_nodes(op=AnnotatedOperation):
            modifiers = []
            cur = node.op
            while isinstance(cur, AnnotatedOperation):
                modifiers.extend(cur.modifiers)
                cur = cur.base_op
            canonical_modifiers = _canonicalize_modifiers(modifiers)
            if len(canonical_modifiers) > 0:
                # this is still an annotated operation
                node.op.base_op = cur
                node.op.modifiers = canonical_modifiers
            else:
                # no need for annotated operations
                dag.substitute_node(node, cur)
            did_something = True
        return dag, did_something

    def _conjugate_decomposition(
        self, dag: DAGCircuit
    ) -> Union[Tuple[DAGCircuit, DAGCircuit, DAGCircuit], None]:
        """
        Decomposes a circuit ``A`` into 3 sub-circuits ``P``, ``Q``, ``R`` such that
        ``A = P -- Q -- R`` and ``R = P^{-1}``.

        This is accomplished by iteratively finding inverse nodes at the front and at the back of the
        circuit.
        """

        front_block = []  # nodes collected from the front of the circuit (aka P)
        back_block = []  # nodes collected from the back of the circuit (aka R)

        # Stores in- and out- degree for each node. These degrees are computed at the start of this
        # function and are updated when nodes are collected into front_block or into back_block.
        in_degree = {}
        out_degree = {}

        # We use dicts to track for each qubit a DAG node at the front of the circuit that involves
        # this qubit and a DAG node at the end of the circuit that involves this qubit (when exist).
        # Note that for the DAGCircuit structure for each qubit there can be at most one such front
        # and such back node.
        # This allows for an efficient way to find an inverse pair of gates (one from the front and
        # one from the back of the circuit).
        # A qubit that was never examined does not appear in these dicts, and a qubit that was examined
        # but currently is not involved at the front (resp. at the back) of the circuit has the value of
        # None.
        front_node_for_qubit = {}
        back_node_for_qubit = {}

        # Keep the set of nodes that have been moved either to front_block or to back_block
        processed_nodes = set()

        # Keep the set of qubits that are involved in nodes at the front or at the back of the circuit.
        # When looking for inverse pairs of gates we will only iterate over these qubits.
        active_qubits = set()

        # Keep pairs of nodes for which the inverse check was performed and the nodes
        # were found to be not inverse to each other (memoization).
        checked_node_pairs = set()

        # compute in- and out- degree for every node
        # also update information for nodes at the start and at the end of the circuit
        for node in dag.op_nodes():
            preds = list(dag.op_predecessors(node))
            in_degree[node] = len(preds)
            if len(preds) == 0:
                for q in node.qargs:
                    front_node_for_qubit[q] = node
                    active_qubits.add(q)
            succs = list(dag.op_successors(node))
            out_degree[node] = len(succs)
            if len(succs) == 0:
                for q in node.qargs:
                    back_node_for_qubit[q] = node
                    active_qubits.add(q)

        # iterate while there is a possibility to find more inverse pairs
        while len(active_qubits) > 0:
            to_check = active_qubits.copy()
            active_qubits.clear()

            # For each qubit q, check whether the gate at the front of the circuit that involves q
            # and the gate at the end of the circuit that involves q are inverse
            for q in to_check:

                if (front_node := front_node_for_qubit.get(q, None)) is None:
                    continue
                if (back_node := back_node_for_qubit.get(q, None)) is None:
                    continue

                # front_node or back_node could be already collected when considering other qubits
                if front_node in processed_nodes or back_node in processed_nodes:
                    continue

                # it is possible that the same node is both at the front and at the back,
                # it should not be collected
                if front_node == back_node:
                    continue

                # have been checked before
                if (front_node, back_node) in checked_node_pairs:
                    continue

                # fast check based on the arguments
                if front_node.qargs != back_node.qargs or front_node.cargs != back_node.cargs:
                    continue

                # in the future we want to include a more precise check whether a pair
                # of nodes are inverse
                if front_node.op == back_node.op.inverse():
                    # update front_node_for_qubit and back_node_for_qubit
                    for q in front_node.qargs:
                        front_node_for_qubit[q] = None
                    for q in back_node.qargs:
                        back_node_for_qubit[q] = None

                    # see which other nodes become at the front and update information
                    for node in dag.op_successors(front_node):
                        if node not in processed_nodes:
                            in_degree[node] -= 1
                            if in_degree[node] == 0:
                                for q in node.qargs:
                                    front_node_for_qubit[q] = node
                                    active_qubits.add(q)

                    # see which other nodes become at the back and update information
                    for node in dag.op_predecessors(back_node):
                        if node not in processed_nodes:
                            out_degree[node] -= 1
                            if out_degree[node] == 0:
                                for q in node.qargs:
                                    back_node_for_qubit[q] = node
                                    active_qubits.add(q)

                    # collect and mark as processed
                    front_block.append(front_node)
                    back_block.append(back_node)
                    processed_nodes.add(front_node)
                    processed_nodes.add(back_node)

                else:
                    checked_node_pairs.add((front_node, back_node))

        # if nothing is found, return None
        if len(front_block) == 0:
            return None

        # create the output DAGs
        front_circuit = dag.copy_empty_like()
        middle_circuit = dag.copy_empty_like()
        back_circuit = dag.copy_empty_like()
        front_circuit.global_phase = 0
        back_circuit.global_phase = 0

        for node in front_block:
            front_circuit.apply_operation_back(node.op, node.qargs, node.cargs)

        for node in back_block:
            back_circuit.apply_operation_front(node.op, node.qargs, node.cargs)

        for node in dag.op_nodes():
            if node not in processed_nodes:
                middle_circuit.apply_operation_back(node.op, node.qargs, node.cargs)

        return front_circuit, middle_circuit, back_circuit

    def _conjugate_reduce_op(
        self, op: AnnotatedOperation, base_decomposition: Tuple[DAGCircuit, DAGCircuit, DAGCircuit]
    ) -> Operation:
        """
        We are given an annotated-operation ``op = M [ B ]`` (where ``B`` is the base operation and
        ``M`` is the list of modifiers) and the "conjugate decomposition" of the definition of ``B``,
        i.e. ``B = P * Q * R``, with ``R = P^{-1}`` (with ``P``, ``Q`` and ``R`` represented as
        ``DAGCircuit`` objects).

        Let ``IQ`` denote a new custom instruction with definitions ``Q``.

        We return the operation ``op_new`` which a new custom instruction with definition
        ``P * A * R``, where ``A`` is a new annotated-operation with modifiers ``M`` and
        base gate ``IQ``.
        """
        p_dag, q_dag, r_dag = base_decomposition

        q_instr = Instruction(
            name="iq", num_qubits=op.base_op.num_qubits, num_clbits=op.base_op.num_clbits, params=[]
        )
        q_instr.definition = dag_to_circuit(q_dag)

        op_new = Instruction(
            "optimized", num_qubits=op.num_qubits, num_clbits=op.num_clbits, params=[]
        )
        num_control_qubits = op.num_qubits - op.base_op.num_qubits

        circ = QuantumCircuit(op.num_qubits, op.num_clbits)
        qubits = circ.qubits
        circ.compose(
            dag_to_circuit(p_dag), qubits[num_control_qubits : op.num_qubits], inplace=True
        )
        circ.append(
            AnnotatedOperation(base_op=q_instr, modifiers=op.modifiers), range(op.num_qubits)
        )
        circ.compose(
            dag_to_circuit(r_dag), qubits[num_control_qubits : op.num_qubits], inplace=True
        )
        op_new.definition = circ
        return op_new

    def _conjugate_reduction(self, dag) -> Tuple[DAGCircuit, bool]:
        """
        Looks for annotated operations whose base operation has a nontrivial conjugate decomposition.
        In such cases, the modifiers of the annotated operation can be moved to the "middle" part of
        the decomposition.

        Returns the modified DAG and whether it did something.
        """
        did_something = False
        for node in dag.op_nodes(op=AnnotatedOperation):
            base_op = node.op.base_op
            if not self._skip_definition(base_op):
                base_dag = circuit_to_dag(base_op.definition, copy_operations=False)
                base_decomposition = self._conjugate_decomposition(base_dag)
                if base_decomposition is not None:
                    new_op = self._conjugate_reduce_op(node.op, base_decomposition)
                    dag.substitute_node(node, new_op)
                    did_something = True
        return dag, did_something

    def _skip_definition(self, op: Operation) -> bool:
        """
        Returns True if we should not recurse into a gate's definition.
        """
        # Similar to HighLevelSynthesis transpiler pass, we do not recurse into a gate's
        # `definition` for a gate that is supported by the target or in equivalence library.

        controlled_gate_open_ctrl = isinstance(op, ControlledGate) and op._open_ctrl
        if not controlled_gate_open_ctrl:
            inst_supported = (
                self._target.instruction_supported(operation_name=op.name)
                if self._target is not None
                else op.name in self._device_insts
            )
            if inst_supported or (self._equiv_lib is not None and self._equiv_lib.has_entry(op)):
                return True
        return False

    def _recursively_process_definitions(self, op: Operation) -> bool:
        """
        Recursively applies optimizations to op's definition (or to op.base_op's
        definition if op is an annotated operation).
        Returns True if did something.
        """

        # If op is an annotated operation, we descend into its base_op
        if isinstance(op, AnnotatedOperation):
            return self._recursively_process_definitions(op.base_op)

        if self._skip_definition(op):
            return False

        try:
            # extract definition
            definition = op.definition
        except TypeError as err:
            raise TranspilerError(
                f"OptimizeAnnotated was unable to extract definition for {op.name}: {err}"
            ) from err
        except AttributeError:
            # definition is None
            definition = None

        if definition is None:
            raise TranspilerError(f"OptimizeAnnotated was unable to optimize {op}.")

        definition_dag = circuit_to_dag(definition, copy_operations=False)
        definition_dag, opt = self._run_inner(definition_dag)

        if opt:
            # We only update a gate's definition if it was actually changed.
            # This is important to preserve non-annotated singleton gates.
            op.definition = dag_to_circuit(definition_dag)

        return opt

    def _recurse(self, dag) -> Tuple[DAGCircuit, bool]:
        """
        Recursively handles gate definitions.
        Returns True if did something.
        """
        did_something = False

        for node in dag.op_nodes():
            opt = self._recursively_process_definitions(node.op)
            did_something = did_something or opt

        return dag, did_something

1	# This code is part of Qiskit.
2	#
3	# (C) Copyright IBM 2024.
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	"""Optimize annotated operations on a circuit."""
14
15	from typing import Optional, List, Tuple, Union	1✔
16
17	from qiskit.circuit.controlflow import CONTROL_FLOW_OP_NAMES	1✔
18	from qiskit.converters import circuit_to_dag, dag_to_circuit	1✔
19	from qiskit.circuit.annotated_operation import AnnotatedOperation, _canonicalize_modifiers	1✔
20	from qiskit.circuit import (	1✔
21	QuantumCircuit,
22	Instruction,
23	EquivalenceLibrary,
24	ControlledGate,
25	Operation,
26	ControlFlowOp,
27	)
28	from qiskit.transpiler.basepasses import TransformationPass	1✔
29	from qiskit.transpiler.passes.utils import control_flow	1✔
30	from qiskit.transpiler.target import Target	1✔
31	from qiskit.dagcircuit import DAGCircuit	1✔
32	from qiskit.transpiler.exceptions import TranspilerError	1✔
33
34
35	class OptimizeAnnotated(TransformationPass):	1✔
36	"""Optimization pass on circuits with annotated operations.
37
38	Implemented optimizations:
39
40	* For each annotated operation, converting the list of its modifiers to a canonical form.
41	For example, consecutively applying ``inverse()``, ``control(2)`` and ``inverse()``
42	is equivalent to applying ``control(2)``.
43
44	* Removing annotations when possible.
45	For example, ``AnnotatedOperation(SwapGate(), [InverseModifier(), InverseModifier()])``
46	is equivalent to ``SwapGate()``.
47
48	* Recursively combining annotations.
49	For example, if ``g1 = AnnotatedOperation(SwapGate(), InverseModifier())`` and
50	``g2 = AnnotatedOperation(g1, ControlModifier(2))``, then ``g2`` can be replaced with
51	``AnnotatedOperation(SwapGate(), [InverseModifier(), ControlModifier(2)])``.
52
53	* Applies conjugate reduction to annotated operations. As an example,
54	``control - [P -- Q -- P^{-1}]`` can be rewritten as ``P -- control - [Q] -- P^{-1}``,
55	that is, only the middle part needs to be controlled. This also works for inverse
56	and power modifiers.
57
58	"""
59
60	def __init__(	1✔
61	self,
62	target: Optional[Target] = None,
63	equivalence_library: Optional[EquivalenceLibrary] = None,
64	basis_gates: Optional[List[str]] = None,
65	recurse: bool = True,
66	do_conjugate_reduction: bool = True,
67	):
68	"""
69	OptimizeAnnotated initializer.
70
71	Args:
72	target: Optional, the backend target to use for this pass.
73	equivalence_library: The equivalence library used
74	(instructions in this library will not be optimized by this pass).
75	basis_gates: Optional, target basis names to unroll to, e.g. `['u3', 'cx']`
76	(instructions in this list will not be optimized by this pass).
77	Ignored if ``target`` is also specified.
78	recurse: By default, when either ``target`` or ``basis_gates`` is specified,
79	the pass recursively descends into gate definitions (and the recursion is
80	not applied when neither is specified since such objects do not need to
81	be synthesized). Setting this value to ``False`` precludes the recursion in
82	every case.
83	do_conjugate_reduction: controls whether conjugate reduction should be performed.
84	"""
85	super().__init__()	1✔
86
87	self._target = target	1✔
88	self._equiv_lib = equivalence_library	1✔
89	self._basis_gates = basis_gates	1✔
90	self._do_conjugate_reduction = do_conjugate_reduction	1✔
91
92	self._top_level_only = not recurse or (self._basis_gates is None and self._target is None)	1✔
93
94	if not self._top_level_only and self._target is None:	1✔
95	basic_insts = {"measure", "reset", "barrier", "snapshot", "delay", "store"}	1✔
96	self._device_insts = basic_insts \| set(self._basis_gates)	1✔
97
98	def run(self, dag: DAGCircuit):	1✔
99	"""Run the OptimizeAnnotated pass on `dag`.
100
101	Args:
102	dag: input dag.
103
104	Returns:
105	Output dag with higher-level operations optimized.
106
107	Raises:
108	TranspilerError: when something goes wrong.
109
110	"""
111	dag, _ = self._run_inner(dag)	1✔
112	return dag	1✔
113
114	def _run_inner(self, dag) -> Tuple[DAGCircuit, bool]:	1✔
115	"""
116	Optimizes annotated operations.
117	Returns True if did something.
118	"""
119	# Fast return
120	if self._top_level_only:	1✔
121	op_names = dag.count_ops(recurse=False)	1✔
122	if "annotated" not in op_names and not CONTROL_FLOW_OP_NAMES.intersection(op_names):	1✔
123	return dag, False	1✔
124
125	# Handle control-flow
126	for node in dag.op_nodes():	1✔
127	if isinstance(node.op, ControlFlowOp):	1✔
128	dag.substitute_node(	1✔
129	node,
130	control_flow.map_blocks(self.run, node.op),
131	)
132
133	# First, optimize every node in the DAG.
134	dag, opt1 = self._canonicalize(dag)	1✔
135
136	opt2 = False	1✔
137	if not self._top_level_only:	1✔
138	# Second, recursively descend into definitions.
139	# Note that it is important to recurse only after the optimization methods have been run,
140	# as they may remove annotated gates.
141	dag, opt2 = self._recurse(dag)	1✔
142
143	opt3 = False	1✔
144	if not self._top_level_only and self._do_conjugate_reduction:	1✔
145	dag, opt3 = self._conjugate_reduction(dag)	1✔
146
147	return dag, opt1 or opt2 or opt3	1✔
148
149	def _canonicalize(self, dag) -> Tuple[DAGCircuit, bool]:	1✔
150	"""
151	Combines recursive annotated operations and canonicalizes modifiers.
152	Returns True if did something.
153	"""
154
155	did_something = False	1✔
156	for node in dag.op_nodes(op=AnnotatedOperation):	1✔
157	modifiers = []	1✔
158	cur = node.op	1✔
159	while isinstance(cur, AnnotatedOperation):	1✔
160	modifiers.extend(cur.modifiers)	1✔
161	cur = cur.base_op	1✔
162	canonical_modifiers = _canonicalize_modifiers(modifiers)	1✔
163	if len(canonical_modifiers) > 0:	1✔
164	# this is still an annotated operation
165	node.op.base_op = cur	1✔
166	node.op.modifiers = canonical_modifiers	1✔
167	else:
168	# no need for annotated operations
169	dag.substitute_node(node, cur)	1✔
170	did_something = True	1✔
171	return dag, did_something	1✔
172
173	def _conjugate_decomposition(	1✔
174	self, dag: DAGCircuit
175	) -> Union[Tuple[DAGCircuit, DAGCircuit, DAGCircuit], None]:
176	"""
177	Decomposes a circuit ``A`` into 3 sub-circuits ``P``, ``Q``, ``R`` such that
178	``A = P -- Q -- R`` and ``R = P^{-1}``.
179
180	This is accomplished by iteratively finding inverse nodes at the front and at the back of the
181	circuit.
182	"""
183
184	front_block = [] # nodes collected from the front of the circuit (aka P)	1✔
185	back_block = [] # nodes collected from the back of the circuit (aka R)	1✔
186
187	# Stores in- and out- degree for each node. These degrees are computed at the start of this
188	# function and are updated when nodes are collected into front_block or into back_block.
189	in_degree = {}	1✔
190	out_degree = {}	1✔
191
192	# We use dicts to track for each qubit a DAG node at the front of the circuit that involves
193	# this qubit and a DAG node at the end of the circuit that involves this qubit (when exist).
194	# Note that for the DAGCircuit structure for each qubit there can be at most one such front
195	# and such back node.
196	# This allows for an efficient way to find an inverse pair of gates (one from the front and
197	# one from the back of the circuit).
198	# A qubit that was never examined does not appear in these dicts, and a qubit that was examined
199	# but currently is not involved at the front (resp. at the back) of the circuit has the value of
200	# None.
201	front_node_for_qubit = {}	1✔
202	back_node_for_qubit = {}	1✔
203
204	# Keep the set of nodes that have been moved either to front_block or to back_block
205	processed_nodes = set()	1✔
206
207	# Keep the set of qubits that are involved in nodes at the front or at the back of the circuit.
208	# When looking for inverse pairs of gates we will only iterate over these qubits.
209	active_qubits = set()	1✔
210
211	# Keep pairs of nodes for which the inverse check was performed and the nodes
212	# were found to be not inverse to each other (memoization).
213	checked_node_pairs = set()	1✔
214
215	# compute in- and out- degree for every node
216	# also update information for nodes at the start and at the end of the circuit
217	for node in dag.op_nodes():	1✔
218	preds = list(dag.op_predecessors(node))	1✔
219	in_degree[node] = len(preds)	1✔
220	if len(preds) == 0:	1✔
221	for q in node.qargs:	1✔
222	front_node_for_qubit[q] = node	1✔
223	active_qubits.add(q)	1✔
224	succs = list(dag.op_successors(node))	1✔
225	out_degree[node] = len(succs)	1✔
226	if len(succs) == 0:	1✔
227	for q in node.qargs:	1✔
228	back_node_for_qubit[q] = node	1✔
229	active_qubits.add(q)	1✔
230
231	# iterate while there is a possibility to find more inverse pairs
232	while len(active_qubits) > 0:	1✔
233	to_check = active_qubits.copy()	1✔
234	active_qubits.clear()	1✔
235
236	# For each qubit q, check whether the gate at the front of the circuit that involves q
237	# and the gate at the end of the circuit that involves q are inverse
238	for q in to_check:	1✔
239
240	if (front_node := front_node_for_qubit.get(q, None)) is None:	1✔
241	continue	1✔
242	if (back_node := back_node_for_qubit.get(q, None)) is None:	1✔
UNCOV 243	continue	×
244
245	# front_node or back_node could be already collected when considering other qubits
246	if front_node in processed_nodes or back_node in processed_nodes:	1✔
247	continue	1✔
248
249	# it is possible that the same node is both at the front and at the back,
250	# it should not be collected
251	if front_node == back_node:	1✔
252	continue	1✔
253
254	# have been checked before
255	if (front_node, back_node) in checked_node_pairs:	1✔
UNCOV 256	continue	×
257
258	# fast check based on the arguments
259	if front_node.qargs != back_node.qargs or front_node.cargs != back_node.cargs:	1✔
260	continue	1✔
261
262	# in the future we want to include a more precise check whether a pair
263	# of nodes are inverse
264	if front_node.op == back_node.op.inverse():	1✔
265	# update front_node_for_qubit and back_node_for_qubit
266	for q in front_node.qargs:	1✔
267	front_node_for_qubit[q] = None	1✔
268	for q in back_node.qargs:	1✔
269	back_node_for_qubit[q] = None	1✔
270
271	# see which other nodes become at the front and update information
272	for node in dag.op_successors(front_node):	1✔
273	if node not in processed_nodes:	1✔
274	in_degree[node] -= 1	1✔
275	if in_degree[node] == 0:	1✔
276	for q in node.qargs:	1✔
277	front_node_for_qubit[q] = node	1✔
278	active_qubits.add(q)	1✔
279
280	# see which other nodes become at the back and update information
281	for node in dag.op_predecessors(back_node):	1✔
282	if node not in processed_nodes:	1✔
283	out_degree[node] -= 1	1✔
284	if out_degree[node] == 0:	1✔
285	for q in node.qargs:	1✔
286	back_node_for_qubit[q] = node	1✔
287	active_qubits.add(q)	1✔
288
289	# collect and mark as processed
290	front_block.append(front_node)	1✔
291	back_block.append(back_node)	1✔
292	processed_nodes.add(front_node)	1✔
293	processed_nodes.add(back_node)	1✔
294
295	else:
UNCOV 296	checked_node_pairs.add((front_node, back_node))	×
297
298	# if nothing is found, return None
299	if len(front_block) == 0:	1✔
UNCOV 300	return None	×
301
302	# create the output DAGs
303	front_circuit = dag.copy_empty_like()	1✔
304	middle_circuit = dag.copy_empty_like()	1✔
305	back_circuit = dag.copy_empty_like()	1✔
306	front_circuit.global_phase = 0	1✔
307	back_circuit.global_phase = 0	1✔
308
309	for node in front_block:	1✔
310	front_circuit.apply_operation_back(node.op, node.qargs, node.cargs)	1✔
311
312	for node in back_block:	1✔
313	back_circuit.apply_operation_front(node.op, node.qargs, node.cargs)	1✔
314
315	for node in dag.op_nodes():	1✔
316	if node not in processed_nodes:	1✔
317	middle_circuit.apply_operation_back(node.op, node.qargs, node.cargs)	1✔
318
319	return front_circuit, middle_circuit, back_circuit	1✔
320
321	def _conjugate_reduce_op(	1✔
322	self, op: AnnotatedOperation, base_decomposition: Tuple[DAGCircuit, DAGCircuit, DAGCircuit]
323	) -> Operation:
324	"""
325	We are given an annotated-operation ``op = M [ B ]`` (where ``B`` is the base operation and
326	``M`` is the list of modifiers) and the "conjugate decomposition" of the definition of ``B``,
327	i.e. ``B = P * Q * R``, with ``R = P^{-1}`` (with ``P``, ``Q`` and ``R`` represented as
328	``DAGCircuit`` objects).
329
330	Let ``IQ`` denote a new custom instruction with definitions ``Q``.
331
332	We return the operation ``op_new`` which a new custom instruction with definition
333	``P * A * R``, where ``A`` is a new annotated-operation with modifiers ``M`` and
334	base gate ``IQ``.
335	"""
336	p_dag, q_dag, r_dag = base_decomposition	1✔
337
338	q_instr = Instruction(	1✔
339	name="iq", num_qubits=op.base_op.num_qubits, num_clbits=op.base_op.num_clbits, params=[]
340	)
341	q_instr.definition = dag_to_circuit(q_dag)	1✔
342
343	op_new = Instruction(	1✔
344	"optimized", num_qubits=op.num_qubits, num_clbits=op.num_clbits, params=[]
345	)
346	num_control_qubits = op.num_qubits - op.base_op.num_qubits	1✔
347
348	circ = QuantumCircuit(op.num_qubits, op.num_clbits)	1✔
349	qubits = circ.qubits	1✔
350	circ.compose(	1✔
351	dag_to_circuit(p_dag), qubits[num_control_qubits : op.num_qubits], inplace=True
352	)
353	circ.append(	1✔
354	AnnotatedOperation(base_op=q_instr, modifiers=op.modifiers), range(op.num_qubits)
355	)
356	circ.compose(	1✔
357	dag_to_circuit(r_dag), qubits[num_control_qubits : op.num_qubits], inplace=True
358	)
359	op_new.definition = circ	1✔
360	return op_new	1✔
361
362	def _conjugate_reduction(self, dag) -> Tuple[DAGCircuit, bool]:	1✔
363	"""
364	Looks for annotated operations whose base operation has a nontrivial conjugate decomposition.
365	In such cases, the modifiers of the annotated operation can be moved to the "middle" part of
366	the decomposition.
367
368	Returns the modified DAG and whether it did something.
369	"""
370	did_something = False	1✔
371	for node in dag.op_nodes(op=AnnotatedOperation):	1✔
372	base_op = node.op.base_op	1✔
373	if not self._skip_definition(base_op):	1✔
374	base_dag = circuit_to_dag(base_op.definition, copy_operations=False)	1✔
375	base_decomposition = self._conjugate_decomposition(base_dag)	1✔
376	if base_decomposition is not None:	1✔
377	new_op = self._conjugate_reduce_op(node.op, base_decomposition)	1✔
378	dag.substitute_node(node, new_op)	1✔
379	did_something = True	1✔
380	return dag, did_something	1✔
381
382	def _skip_definition(self, op: Operation) -> bool:	1✔
383	"""
384	Returns True if we should not recurse into a gate's definition.
385	"""
386	# Similar to HighLevelSynthesis transpiler pass, we do not recurse into a gate's
387	# `definition` for a gate that is supported by the target or in equivalence library.
388
389	controlled_gate_open_ctrl = isinstance(op, ControlledGate) and op._open_ctrl	1✔
390	if not controlled_gate_open_ctrl:	1✔
391	inst_supported = (	1✔
392	self._target.instruction_supported(operation_name=op.name)
393	if self._target is not None
394	else op.name in self._device_insts
395	)
396	if inst_supported or (self._equiv_lib is not None and self._equiv_lib.has_entry(op)):	1✔
397	return True	1✔
398	return False	1✔
399
400	def _recursively_process_definitions(self, op: Operation) -> bool:	1✔
401	"""
402	Recursively applies optimizations to op's definition (or to op.base_op's
403	definition if op is an annotated operation).
404	Returns True if did something.
405	"""
406
407	# If op is an annotated operation, we descend into its base_op
408	if isinstance(op, AnnotatedOperation):	1✔
409	return self._recursively_process_definitions(op.base_op)	1✔
410
411	if self._skip_definition(op):	1✔
412	return False	1✔
413
414	try:	1✔
415	# extract definition
416	definition = op.definition	1✔
417	except TypeError as err:	×
UNCOV 418	raise TranspilerError(	×
419	f"OptimizeAnnotated was unable to extract definition for {op.name}: {err}"
420	) from err
UNCOV 421	except AttributeError:	×
422	# definition is None
UNCOV 423	definition = None	×
424
425	if definition is None:	1✔
UNCOV 426	raise TranspilerError(f"OptimizeAnnotated was unable to optimize {op}.")	×
427
428	definition_dag = circuit_to_dag(definition, copy_operations=False)	1✔
429	definition_dag, opt = self._run_inner(definition_dag)	1✔
430
431	if opt:	1✔
432	# We only update a gate's definition if it was actually changed.
433	# This is important to preserve non-annotated singleton gates.
434	op.definition = dag_to_circuit(definition_dag)	1✔
435
436	return opt	1✔
437
438	def _recurse(self, dag) -> Tuple[DAGCircuit, bool]:	1✔
439	"""
440	Recursively handles gate definitions.
441	Returns True if did something.
442	"""
443	did_something = False	1✔
444
445	for node in dag.op_nodes():	1✔
446	opt = self._recursively_process_definitions(node.op)	1✔
447	did_something = did_something or opt	1✔
448
449	return dag, did_something	1✔

Qiskit / qiskit / 13540158629

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