Generalize CXCancellation optimization pass to generic gate-inverse pairs

qiskit/transpiler/passes/__init__.py

@@ -173,6 +173,7 @@
 from .optimization import CrosstalkAdaptiveSchedule
 from .optimization import HoareOptimizer
 from .optimization import TemplateOptimization
+from .optimization import Cancellation
 
 # circuit analysis
 from .analysis import ResourceEstimation

qiskit/transpiler/passes/optimization/__init__.py

@@ -25,3 +25,4 @@
 from .crosstalk_adaptive_schedule import CrosstalkAdaptiveSchedule
 from .hoare_opt import HoareOptimizer
 from .template_optimization import TemplateOptimization
+from .inverse_cancellation import Cancellation

qiskit/transpiler/passes/optimization/inverse_cancellation.py

@@ -0,0 +1,136 @@
+# 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.
+
+"""
+A generic InverseCancellation pass for any set of gate-inverse pairs.
+"""
+from typing import List, Tuple, Union
+
+from qiskit.circuit import Gate
+from qiskit.dagcircuit import DAGCircuit
+from qiskit.transpiler.basepasses import TransformationPass
+from qiskit.transpiler.exceptions import TranspilerError
+
+
+class InverseCancellation(TransformationPass):
+    """Cancel specific Gates which are inverses of each other when they occur back-to-
+    back."""
+
+    def __init__(self, gates_to_cancel: List[Union[Gate, Tuple[Gate, Gate]]]):
+        """Initialize InverseCancellation pass.
+
+        Args:
+            gates_to_cancel: list of gates to cancel
+
+        Raises:
+            TranspilerError:
+                Initalization raises an error when the input is not a self-inverse gate
+                or a two-tuple of inverse gates.
+        """
+
+        for gates in gates_to_cancel:
+            if isinstance(gates, Gate):
+                if gates != gates.inverse():
+                    raise TranspilerError("Gate {} is not self-inverse".format(gates.name))
+            elif isinstance(gates, tuple):
+                if len(gates) != 2:
+                    raise TranspilerError(
+                        "Too many or too few inputs: {}. Only two are allowed.".format(gates)
+                    )
+                elif gates[0] != gates[1].inverse():
+                    raise TranspilerError(
+                        "Gate {} and {} are not inverse.".format(gates[0].name, gates[1].name)
+                    )
+            else:
+                raise TranspilerError(
+                    "InverseCancellation pass does not take input type {}. Input must be"
+                    " a Gate.".format(type(gates))
+                )
+
+        self.gates_to_cancel = gates_to_cancel
+        super().__init__()
+
+        self_inverse_gates = []
+        inverse_gate_pairs = []
+
+        for gates in self.gates_to_cancel:
+            if isinstance(gates, Gate):
+                self_inverse_gates.append(gates)
+            else:
+                inverse_gate_pairs.append(gates)
+
+    def run(self, dag: DAGCircuit):
+        """Run the InverseCancellation pass on `dag`.
+
+        Args:
+            dag: the directed acyclic graph to run on.
+
+        Returns:
+            DAGCircuit: Transformed DAG.
+        """
+
+        dag = self._run_on_self_inverse(dag, self_inverse_gates)
+        return self._run_on_inverse_pairs(dag, inverse_gate_pairs)
+
+    def _run_on_self_inverse(self, dag: DAGCircuit, self_inverse_gates: List[Gate]):
+        """
+        Run self-inverse gates on `dag`.
+
+        Args:
+            dag: the directed acyclic graph to run on.
+            self_inverse_gates: list of gates who cancel themeselves in pairs
+
+        Returns:
+            DAGCircuit: Transformed DAG.
+        """
+        gate_cancel_runs = [dag.collect_runs([gate.name]) for gate in self_inverse_gates]
+        for gate_cancel_run in gate_cancel_runs:
+            partitions = []
+            chunk = []
+            for i in range(len(gate_cancel_run) - 1):
+                chunk.append(gate_cancel_run[i])
+                if gate_cancel_run[i].qargs != gate_cancel_run[i + 1].qargs:
+                    partitions.append(chunk)
+                    chunk = []
+            chunk.append(gate_cancel_run[-1])
+            partitions.append(chunk)
+            # Remove an even number of gates from each chunk
+            for chunk in partitions:
+                if len(chunk) % 2 == 0:
+                    dag.remove_op_node(chunk[0])
+                for node in chunk[1:]:
+                    dag.remove_op_node(node)
+        return dag
+
+    def _run_on_inverse_pairs(self, dag: DAGCircuit, inverse_gate_pairs: List[Tuple[Gate, Gate]]):
+        """
+        Run inverse gate pairs on `dag`.
+
+        Args:
+            dag: the directed acyclic graph to run on.
+            inverse_gate_pairs: list of gates with inverse angles that cancel each other.
+
+        Returns:
+            DAGCircuit: Transformed DAG.
+        """
+        for pair in inverse_gate_pairs:
+            gate_cancel_runs = dag.collect_runs([pair[0].name])
+            for dag_nodes in gate_cancel_runs:
+                for i in range(len(dag_nodes) - 1):
+                    if dag_nodes[i].op == pair[0] and dag_nodes[i + 1].op == pair[1]:
+                        dag.remove_op_node(dag_nodes[i])
+                        dag.remove_op_node(dag_nodes[i + 1])
+                    elif dag_nodes[i].op == pair[1] and dag_nodes[i + 1].op == pair[0]:
+                        dag.remove_op_node(dag_nodes[i])
+                        dag.remove_op_node(dag_nodes[i + 1])
+
+        return dag

releasenotes/notes/cx-cancellation-pass-generalization-538fb7cfe49b3fd5.yaml

@@ -0,0 +1,17 @@
+---
+features:
+  - |
+    Introduced a new feature ``InverseCancellation`` that generalizes the ``CXInverseCancellation`` 
+    pass to cancel any self-inverse gates or gate-inverse pairs. It can be used by 
+    initializing ``InverseCancellation`` and passing a gate to cancel, for example::
+
+      from qiskit.transpiler.passes import InverseCancellation
+      from qiskit import QuantumCircuit
+      from qiskit.circuit.library import HGate
+      from qiskit.transpiler import PassManager
+      qc = QuantumCircuit(2, 2)
+      qc.h(0)
+      qc.h(0)
+      pass_ = InverseCancellation([HGate()])
+      pm = PassManager(pass_)
+      new_circ = pm.run(qc)

test/python/transpiler/test_inverse_cancellation.py

@@ -0,0 +1,153 @@
+# This code is part of Qiskit.
+#
+# (C) Copyright IBM 2020.
+#
+# 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.
+
+"""
+Testing InverseCancellation
+"""
+
+import numpy as np
+
+from qiskit import QuantumCircuit
+from qiskit.transpiler.exceptions import TranspilerError
+from qiskit.transpiler.passes import InverseCancellation
+from qiskit.transpiler import PassManager
+from qiskit.test import QiskitTestCase
+from qiskit.circuit.library import RXGate, HGate, CXGate, PhaseGate, XGate
+
+
+class TestInverseCancellation(QiskitTestCase):
+    def test_InverseCancellation_h(self):
+        qc = QuantumCircuit(2, 2)
+        qc.h(0)
+        qc.h(0)
+        pass_ = InverseCancellation([HGate()])
+        pm = PassManager(pass_)
+        new_circ = pm.run(qc)
+        gates_after = new_circ.count_ops()
+        self.assertNotIn("h", gates_after)
+
+    def test_InverseCancellation_h3(self):
+        qc = QuantumCircuit(2, 2)
+        qc.h(0)
+        qc.h(0)
+        qc.h(0)
+        pass_ = InverseCancellation([HGate()])
+        pm = PassManager(pass_)
+        new_circ = pm.run(qc)
+        gates_after = new_circ.count_ops()
+        self.assertIn("h", gates_after)
+        self.assertEqual(gates_after["h"], 1)
+
+    def test_InverseCancellation_cx(self):
+        qc = QuantumCircuit(2, 2)
+        qc.cx(0, 1)
+        qc.cx(0, 1)
+        pass_ = InverseCancellation([CXGate()])
+        pm = PassManager(pass_)
+        new_circ = pm.run(qc)
+        gates_after = new_circ.count_ops()
+        self.assertNotIn("cx", gates_after)
+
+    def test_InverseCancellation_rx1(self):
+        qc = QuantumCircuit(2, 2)
+        qc.rx(np.pi / 4, 0)
+        qc.rx(-np.pi / 4, 0)
+        pass_ = InverseCancellation([(RXGate(np.pi / 4), RXGate(-np.pi / 4))])
+        pm = PassManager(pass_)
+        new_circ = pm.run(qc)
+        gates_after = new_circ.count_ops()
+        self.assertNotIn("rx", gates_after)
+
+    def test_InverseCancellation_rx2(self):
+        qc = QuantumCircuit(2, 2)
+        qc.rx(np.pi / 4, 0)
+        qc.rx(np.pi / 4, 0)
+        pass_ = InverseCancellation([(RXGate(np.pi / 4), RXGate(-np.pi / 4))])
+        pm = PassManager(pass_)
+        new_circ = pm.run(qc)
+        gates_after = new_circ.count_ops()
+        self.assertIn("rx", gates_after)
+        self.assertEqual(gates_after["rx"], 2)
+
+    def test_InverseCancellation_rx3(self):
+        qc = QuantumCircuit(2, 2)
+        qc.rx(np.pi / 2, 0)
+        qc.rx(np.pi / 4, 0)
+        with self.assertRaises(TranspilerError):
+            pass_ = InverseCancellation([RXGate(0.5)])
+
+    def test_InverseCancellation_hcx(self):
+        qc = QuantumCircuit(2, 2)
+        qc.h(0)
+        qc.h(0)
+        qc.h(0)
+        qc.cx(0, 1)
+        qc.cx(0, 1)
+        qc.h(0)
+        pass_ = InverseCancellation([HGate(), CXGate()])
+        pm = PassManager(pass_)
+        new_circ = pm.run(qc)
+        gates_after = new_circ.count_ops()
+        self.assertNotIn("cx", gates_after)
+        self.assertEqual(gates_after["h"], 2)
+
+    def test_InverseCancellation_p(self):
+        qc = QuantumCircuit(2, 2)
+        qc.p(np.pi / 4, 0)
+        qc.p(-np.pi / 4, 0)
+        pass_ = InverseCancellation([(PhaseGate(np.pi / 4), PhaseGate(-np.pi / 4))])
+        pm = PassManager(pass_)
+        new_circ = pm.run(qc)
+        gates_after = new_circ.count_ops()
+        self.assertNotIn("p", gates_after)
+
+    def test_InverseCancellation_h4(self):
+        qc = QuantumCircuit(2, 2)
+        qc.h(0)
+        qc.h(1)
+        qc.h(0)
+        qc.h(1)
+        pass_ = InverseCancellation([HGate()])
+        pm = PassManager(pass_)
+        new_circ = pm.run(qc)
+        gates_after = new_circ.count_ops()
+        self.assertNotIn("h", gates_after)
+
+    def test_InverseCancellation_rx4(self):
+        qc = QuantumCircuit(2, 2)
+        qc.rx(np.pi / 4, 0)
+        qc.rx(np.pi / 4, 0)
+        with self.assertRaises(TranspilerError):
+            pass_ = InverseCancellation([(RXGate(np.pi / 4))])
+
+    def test_InverseCancellation_herror(self):
+        qc = QuantumCircuit(2, 2)
+        qc.h(0)
+        qc.h(0)
+        with self.assertRaises(TranspilerError):
+            pass_ = InverseCancellation(["h"])
+
+    def test_InverseCancellation_hcx(self):
+        qc = QuantumCircuit(2, 2)
+        qc.h(0)
+        qc.h(0)
+        qc.h(0)
+        qc.x(0)
+        qc.x(0)
+        qc.h(0)
+        pass_ = InverseCancellation([HGate(), XGate()])
+        pm = PassManager(pass_)
+        new_circ = pm.run(qc)
+        gates_after = new_circ.count_ops()
+        self.assertNotIn("x", gates_after)
+        self.assertEqual(gates_after["h"], 2)
+