Add StagedPassManager class for pass manager with defined stages

qiskit/transpiler/__init__.py

@@ -595,6 +595,74 @@
 
       <br>
 
+.. dropdown:: Working with preset :class:`~.PassManager`
+   :animate: fade-in-slide-down
+
+   By default Qiskit includes functions to build preset :class:`~.PassManager` objects.
+   These preset passmangers are what get used by the :func:`~.transpile` function
+   for each optimization level. If you'd like to work directly with a
+   preset pass manager you can use the :func:`~.generate_preset_pass_manager`
+   function to easily generate one. For example:
+
+   .. code-block:: python
+
+       from qiskit.transpiler.preset_passmanager import generate_preset_pass_manager
+       from qiskit.providers.fake_provider import FakeLagosV2
+
+       backend = FakeLagosV2()
+       pass_manager = generate_preset_pass_manager(3, backend)
+
+   which will generate a :class:`~.StagedPassManager` object for optimization level 3
+   targetting the :class:`~.FakeLagosV2` backend (equivalent to what is used internally
+   by :func:`~.transpile` with ``backend=FakeLagosV2()`` and ``optimization_level=3``).
+   You can use this just like working with any other :class:`~.PassManager`. However,
+   because it is a :class:`~.StagedPassManager` it also makes it easy to compose and/or
+   replace stages of the pipeline. For example, if you wanted to run a custom scheduling
+   stage using dynamical decoupling (via the :class:`~.PadDynamicalDecoupling` pass) and
+   also add initial logical optimization prior to routing you would do something like
+   (building off the previous example):
+
+   .. code-block:: python
+
+       from qiskit.circuit.library import XGate, HGate, RXGate, PhaseGate, TGate, TdgGate
+       from qiskit.transpiler import PassManager
+       from qiskit.transpiler.passes import ALAPScheduleAnalysis, PadDynamicalDecoupling
+       from qiskit.transpiler.passes import CXCancellation, InverseCancellation
+
+       backend_durations = backend.target.durations()
+       dd_sequence = [XGate(), XGate()]
+       scheduling_pm = PassManager([
+           ALAPScheduleAnalysis(backend_durations),
+           PadDynamicalDecoupling(backend_durations, dd_sequence),
+       ])
+       inverse_gate_list = [
+           HGate(),
+           (RXGate(np.pi / 4), RXGate(-np.pi / 4)),
+           (PhaseGate(np.pi / 4), PhaseGate(-np.pi / 4)),
+           (TGate(), TdgGate()),
+
+       ])
+       logical_opt = PassManager([
+           CXCancellation(),
+           InverseCancellation([HGate(), (RXGate(np.pi / 4), RXGate(-np.pi / 4))
+       ])
+
+
+       # Add pre-layout stage to run extra logical optimization
+       pass_manager.pre_layout = logical_opt
+       # Set scheduling stage to custom pass manager
+       pass_manager.scheduling = scheduling_pm
+
+
+   Then when :meth:`~.StagedPassManager.run` is called on ``pass_manager`` the
+   ``logical_opt`` :class:`~.PassManager` will be called prior to the ``layout`` stage
+   and for the ``scheduling`` stage our custom :class:`~.PassManager`
+   ``scheduling_pm`` will be used.
+
+   .. raw:: html
+
+      <br>
+
 
 Transpiler API
 ==============
@@ -614,6 +682,7 @@
 .. autosummary::
    :toctree: ../stubs/
 
+   StagedPassManager
    PassManager
    PassManagerConfig
    PropertySet
@@ -669,6 +738,7 @@
 from .runningpassmanager import FlowController, ConditionalController, DoWhileController
 from .passmanager import PassManager
 from .passmanager_config import PassManagerConfig
+from .passmanager import StagedPassManager
 from .propertyset import PropertySet
 from .exceptions import TranspilerError, TranspilerAccessError
 from .fencedobjs import FencedDAGCircuit, FencedPropertySet

qiskit/transpiler/passes/optimization/consolidate_blocks.py

@@ -117,7 +117,7 @@ def run(self, dag):
         # If 1q runs are collected before consolidate those too
         runs = self.property_set["run_list"] or []
         for run in runs:
-            if run[0] in all_block_gates:
+            if any(gate in all_block_gates for gate in run):
                 continue
             if len(run) == 1 and not self._check_not_in_basis(
                 run[0].name, run[0].qargs, global_index_map
@@ -135,6 +135,11 @@ def run(self, dag):
                     continue
                 unitary = UnitaryGate(operator)
                 dag.replace_block_with_op(run, unitary, {qubit: 0}, cycle_check=False)
+        # 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 _check_not_in_basis(self, gate_name, qargs, global_index_map):

qiskit/transpiler/passmanager.py

@@ -12,6 +12,8 @@
 
 """Manager for a set of Passes and their scheduling during transpilation."""
 
+import io
+import re
 from typing import Union, List, Callable, Dict, Any
 
 import dill
@@ -319,3 +321,185 @@ def passes(self) -> List[Dict[str, BasePass]]:
                 item["flow_controllers"] = {}
             ret.append(item)
         return ret
+
+
+class StagedPassManager(PassManager):
+    """A Pass manager pipeline built up of individual stages
+
+    This class enables building a compilation pipeline out of fixed stages.
+    Each ``StagedPassManager`` defines a list of stages which are executed in
+    a fixed order, and each stage is defined as a standalone :class:`~.PassManager`
+    instance. There are also ``pre_`` and ``post_`` stages for each defined stage.
+    This enables easily composing and replacing different stages and also adding
+    hook points to enable programmtic modifications to a pipeline. When using a staged
+    pass manager you are not able to modify the individual passes and are only able
+    to modify stages.
+
+    By default instances of StagedPassManager define a typical full compilation
+    pipeline from an abstract virtual circuit to one that is optimized and
+    capable of running on the specified backend. The default pre-defined stages are:
+
+    #. ``init`` - any initial passes that are run before we start embedding the circuit to the backend
+    #. ``layout`` - This stage runs layout and maps the virtual qubits in the
+       circuit to the physical qubits on a backend
+    #. ``routing`` - This stage runs after a layout has been run and will insert any
+       necessary gates to move the qubit states around until it can be run on
+       backend's compuling map.
+    #. ``translation`` - Perform the basis gate translation, in other words translate the gates
+       in the circuit to the target backend's basis set
+    #. ``optimization`` - The main optimization loop, this will typically run in a loop trying to
+       optimize the circuit until a condtion (such as fixed depth) is reached.
+    #. ``scheduling`` - Any hardware aware scheduling passes
+
+    .. note::
+
+        For backwards compatibility the relative positioning of these default
+        stages will remain stable moving forward. However, new stages may be
+        added to the default stage list in between current stages. For example,
+        in a future release a new phase, something like ``logical_optimization``, could be added
+        immediately after the existing ``init`` stage in the default stage list.
+        This would preserve compatibility for pre-existing ``StagedPassManager``
+        users as the relative positions of the stage are preserved so the behavior
+        will not change between releases.
+
+    These stages will be executed in order and any stage set to ``None`` will be skipped. If
+    a :class:`~qiskit.transpiler.PassManager` input is being used for more than 1 stage here
+    (for example in the case of a :class:`~.Pass` that covers both Layout and Routing) you will want
+    to set that to the earliest stage in sequence that it covers.
+    """
+
+    invalid_stage_regex = re.compile(
+        r"\s|\+|\-|\*|\/|\\|\%|\<|\>|\@|\!|\~|\^|\&|\:|\[|\]|\{|\}|\(|\)"
+    )
+
+    def __init__(self, stages=None, **kwargs):
+        """Initialize a new StagedPassManager object
+
+        Args:
+            stages (List[str]): An optional list of stages to use for this
+                instance. If this is not specified the default stages list
+                ``['init', 'layout', 'routing', 'translation', 'optimization', 'scheduling']`` is
+                used
+            kwargs: The initial :class:`~.PassManager` values for any stages
+                defined in ``stages``. If a argument is not defined the
+                stages will default to ``None`` indicating an empty/undefined
+                stage.
+
+        Raises:
+            AttributeError: If a stage in the input keyword arguments is not defined.
+            ValueError: If an invalid stage name is specified.
+        """
+        if stages is None:
+            self.stages = [
+                "init",
+                "layout",
+                "routing",
+                "translation",
+                "optimization",
+                "scheduling",
+            ]
+        else:
+            invalid_stages = [
+                stage for stage in stages if self.invalid_stage_regex.search(stage) is not None
+            ]
+            if invalid_stages:
+                with io.StringIO() as msg:
+                    msg.write(f"The following stage names are not valid: {invalid_stages[0]}")
+                    for invalid_stage in invalid_stages[1:]:
+                        msg.write(f", {invalid_stage}")
+                    raise ValueError(msg.getvalue())
+
+            self.stages = stages
+        super().__init__()
+        for stage in self.stages:
+            pre_stage = "pre_" + stage
+            post_stage = "post_" + stage
+            setattr(self, pre_stage, None)
+            setattr(self, stage, None)
+            setattr(self, post_stage, None)
+        for stage, pm in kwargs.items():
+            if (
+                stage not in self.stages
+                and not (stage.startswith("pre_") and stage[4:] in self.stages)
+                and not (stage.startswith("post_") and stage[5:] in self.stages)
+            ):
+                raise AttributeError(f"{stage} is not a valid stage.")
+            setattr(self, stage, pm)
+        self._update_passmanager()
+
+    def _update_passmanager(self):
+        self._pass_sets = []
+        for stage in self.stages:
+            # Add pre-stage PM
+            pre_stage_pm = getattr(self, "pre_" + stage, None)
+            if pre_stage_pm is not None:
+                self._pass_sets.extend(pre_stage_pm._pass_sets)
+            # Add stage PM
+            stage_pm = getattr(self, stage, None)
+            if stage_pm is not None:
+                self._pass_sets.extend(stage_pm._pass_sets)
+            # Add post-stage PM
+            post_stage_pm = getattr(self, "post_" + stage, None)
+            if post_stage_pm is not None:
+                self._pass_sets.extend(post_stage_pm._pass_sets)
+
+    def __setattr__(self, attr, value):
+        super().__setattr__(attr, value)
+        if (
+            attr in self.stages
+            or (attr.startswith("pre_") and attr[4:] not in self.stages)
+            or (attr.startswith("post_") and attr[5:] not in self.stages)
+        ):
+            self._update_passmanager()
+
+    def append(
+        self,
+        passes: Union[BasePass, List[BasePass]],
+        max_iteration: int = None,
+        **flow_controller_conditions: Any,
+    ) -> None:
+        raise NotImplementedError
+
+    def replace(
+        self,
+        index: int,
+        passes: Union[BasePass, List[BasePass]],
+        max_iteration: int = None,
+        **flow_controller_conditions: Any,
+    ) -> None:
+        raise NotImplementedError
+
+    # Raise NotImplemntedError on individual pass manipulation
+    def remove(self, index: int) -> None:
+        raise NotImplementedError
+
+    def __getitem(self, index):
+        self._update_passmanager()
+        return super().__getitem__(index)
+
+    def __len__(self):
+        self._update_passmanager()
+        return super().__len__()
+
+    def __setitem__(self, index, item):
+        raise NotImplementedError
+
+    def __add__(self, other):
+        raise NotImplementedError
+
+    def _create_running_passmanager(self) -> RunningPassManager:
+        self._update_passmanager()
+        return super()._create_running_passmanager()
+
+    def passes(self) -> List[Dict[str, BasePass]]:
+        self._update_passmanager()
+        return super().passes()
+
+    def run(
+        self,
+        circuits: Union[QuantumCircuit, List[QuantumCircuit]],
+        output_name: str = None,
+        callback: Callable = None,
+    ) -> Union[QuantumCircuit, List[QuantumCircuit]]:
+        self._update_passmanager()
+        return super().run(circuits, output_name, callback)

qiskit/transpiler/preset_passmanagers/__init__.py

@@ -136,7 +136,7 @@ def generate_preset_pass_manager(
             to use this option.
 
     Returns:
-        PassManager: The preset pass manager for the given options
+        StagedPassManager: The preset pass manager for the given options
 
     Raises:
         ValueError: if an invalid value for ``optimization_level`` is passed in.