Código fonte de qiskit.transpiler.passes.optimization.consolidate_blocks

# 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."""

import numpy as np

from qiskit.circuit.classicalregister import ClassicalRegister
from qiskit.circuit.quantumregister import QuantumRegister
from qiskit.circuit.quantumcircuit import QuantumCircuit
from qiskit.dagcircuit.dagnode import DAGOpNode
from qiskit.quantum_info import Operator
from qiskit.quantum_info.synthesis import TwoQubitBasisDecomposer
from qiskit.extensions import UnitaryGate
from qiskit.circuit.library.standard_gates import CXGate
from qiskit.transpiler.basepasses import TransformationPass
from qiskit.circuit.controlflow import ControlFlowOp
from qiskit.transpiler.passmanager import PassManager
from qiskit.transpiler.passes.synthesis import unitary_synthesis
from qiskit.transpiler.passes.utils import _block_to_matrix
from .collect_1q_runs import Collect1qRuns
from .collect_2q_blocks import Collect2qBlocks

[documentos]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 `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 [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: self.decomposer = unitary_synthesis._decomposer_2q_from_basis_gates( basis_gates, approximation_degree=approximation_degree ) else: self.decomposer = TwoQubitBasisDecomposer(CXGate())
[documentos] 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"] or [] basis_gate_name = self.decomposer.gate.name all_block_gates = set() for block in blocks: if len(block) == 1 and self._check_not_in_basis(dag, block[0].name, block[0].qargs): all_block_gates.add(block[0]) dag.substitute_node(block[0], UnitaryGate(block[0].op.to_matrix())) else: basis_count = 0 outside_basis = False block_qargs = set() block_cargs = set() for nd in block: block_qargs |= set(nd.qargs) if isinstance(nd, DAGOpNode) and getattr(nd.op, "condition", None): block_cargs |= set(getattr(nd.op, "condition", None)[0]) all_block_gates.add(nd) block_index_map = self._block_qargs_to_indices(dag, block_qargs) for nd in block: if nd.op.name == basis_gate_name: basis_count += 1 if self._check_not_in_basis(dag, nd.op.name, nd.qargs): outside_basis = True if len(block_qargs) > 2: q = QuantumRegister(len(block_qargs)) qc = QuantumCircuit(q) if block_cargs: c = ClassicalRegister(len(block_cargs)) qc.add_register(c) for nd in block: qc.append(nd.op, [q[block_index_map[i]] for i in nd.qargs]) unitary = UnitaryGate(Operator(qc)) else: matrix = _block_to_matrix(block, block_index_map) unitary = UnitaryGate(matrix) max_2q_depth = 20 # If depth > 20, there will be 1q gates to consolidate. if ( # pylint: disable=too-many-boolean-expressions self.force_consolidate or unitary.num_qubits > 2 or self.decomposer.num_basis_gates(unitary) < basis_count or len(block) > max_2q_depth or ((self.basis_gates is not None) and outside_basis) or ((self.target is not None) and outside_basis) ): identity = np.eye(2**unitary.num_qubits) if np.allclose(identity, unitary.to_matrix()): for node in block: dag.remove_op_node(node) else: dag.replace_block_with_op( block, unitary, block_index_map, cycle_check=False ) # If 1q runs are collected before consolidate those too runs = self.property_set["run_list"] or [] identity_1q = np.eye(2) for run in runs: if any(gate in all_block_gates for gate in run): continue if len(run) == 1 and not self._check_not_in_basis(dag, run[0].name, run[0].qargs): dag.substitute_node(run[0], UnitaryGate(run[0].op.to_matrix())) else: qubit = run[0].qargs[0] operator = run[0].op.to_matrix() already_in_block = False for gate in run[1:]: if gate in all_block_gates: already_in_block = True operator = gate.op.to_matrix().dot(operator) if already_in_block: continue unitary = UnitaryGate(operator) if np.allclose(identity_1q, unitary.to_matrix()): for node in run: dag.remove_op_node(node) else: dag.replace_block_with_op(run, unitary, {qubit: 0}, cycle_check=False) 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()) if "block_list" in self.property_set: pass_manager.append(Collect2qBlocks()) pass_manager.append(self) for node in dag.op_nodes(ControlFlowOp): node.op = node.op.replace_blocks(pass_manager.run(block) for block in node.op.blocks) return dag def _check_not_in_basis(self, dag, gate_name, qargs): if self.target is not None: return not self.target.instruction_supported( gate_name, tuple(dag.find_bit(qubit).index for qubit in qargs) ) else: return self.basis_gates and gate_name not in self.basis_gates def _block_qargs_to_indices(self, dag, block_qargs): """Map each qubit in block_qargs to its wire position among the block's wires. Args: block_qargs (list): list of qubits that a block acts on global_index_map (dict): mapping from each qubit in the circuit to its wire position within that circuit Returns: dict: mapping from qarg to position in block """ block_indices = [dag.find_bit(q).index for q in block_qargs] ordered_block_indices = {bit: index for index, bit in enumerate(sorted(block_indices))} block_positions = {q: ordered_block_indices[dag.find_bit(q).index] for q in block_qargs} return block_positions