Pulse Compiler SetSequence pass

qiskit/pulse/compiler/__init__.py

@@ -13,3 +13,4 @@
 """Pass-based Qiskit pulse program compiler."""
 
 from .passmanager import BlockTranspiler, BlockToIrCompiler
+from .passes import MapMixedFrame, SetSequence

qiskit/pulse/compiler/passes/__init__.py

@@ -11,3 +11,6 @@
 # that they have been altered from the originals.
 
 """Built-in pulse compile passes."""
+
+from .map_mixed_frames import MapMixedFrame
+from .set_sequence import SetSequence

qiskit/pulse/compiler/passes/map_mixed_frames.py

@@ -0,0 +1,57 @@
+# 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.
+
+"""MixedFrames mapping analysis pass"""
+
+from __future__ import annotations
+from collections import defaultdict
+
+from qiskit.pulse.compiler.basepasses import AnalysisPass
+from qiskit.pulse.ir import SequenceIR
+from qiskit.pulse.model import MixedFrame
+
+
+class MapMixedFrame(AnalysisPass):
+    r"""Map the dependencies of all class:`.MixedFrame`\s
+    on class:`~qiskit.pulse.PulseTaraget` and :class:`~qiskit.pulse.Frame`.
+
+    The pass recursively scans the :class:`.SequenceIR`, identifies all :class:`.MixedFrame`\s and
+    tracks the dependencies of them on class:`~qiskit.pulse.PulseTaraget` and
+    :class:`~qiskit.pulse.Frame`. The analysis result
+    is added as a dictionary to the property set under key "mixed_frames_mapping". The
+    added dictionary is keyed on every class:`~qiskit.pulse.PulseTaraget` and
+    :class:`~qiskit.pulse.Frame` in :class:`.SequenceIR`
+    with the value being a set of all class:`.MixedFrame`\s associated with the key.
+    """
+
+    def __init__(self):
+        """Create new ``MapMixedFrames`` pass"""
+        super().__init__(target=None)
+
+    def run(
+        self,
+        passmanager_ir: SequenceIR,
+    ) -> None:
+        """Run ``MapMixedFrame`` pass"""
+        mixed_frames_mapping = defaultdict(set)
+
+        for inst_target in passmanager_ir.inst_targets:
+            if isinstance(inst_target, MixedFrame):
+                mixed_frames_mapping[inst_target.frame].add(inst_target)
+                mixed_frames_mapping[inst_target.pulse_target].add(inst_target)
+        self.property_set["mixed_frames_mapping"] = mixed_frames_mapping
+
+    def __hash__(self):
+        return hash((self.__class__.__name__,))
+
+    def __eq__(self, other):
+        return self.__class__.__name__ == other.__class__.__name__

qiskit/pulse/compiler/passes/set_sequence.py

@@ -0,0 +1,152 @@
+# 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.
+
+"""Sequencing pass for Qiskit PulseIR compilation."""
+
+from __future__ import annotations
+from functools import singledispatchmethod
+from rustworkx import PyDAG
+
+from qiskit.pulse.compiler.basepasses import TransformationPass
+from qiskit.pulse.ir import SequenceIR
+from qiskit.pulse.model import MixedFrame
+from qiskit.pulse.transforms import AlignmentKind, SequentialAlignment, ParallelAlignment
+from qiskit.pulse.exceptions import PulseCompilerError
+
+
+class SetSequence(TransformationPass):
+    """Sets the sequence of a :class:`.SequenceIR` object.
+
+    The pass traverses the :class:`.SequenceIR` and recursively sets the sequence, by adding edges to
+    the ``sequence`` property. Sequencing is done according to the alignment strategy.
+
+    For parallel alignment types, the pass depends on the results of the analysis pass
+    :class:`~qiskit.pulse.compiler.passes.MapMixedFrame`.
+    """
+
+    def __init__(self):
+        """Create new SetSequence pass"""
+        super().__init__(target=None)
+
+    def run(
+        self,
+        passmanager_ir: SequenceIR,
+    ) -> SequenceIR:
+        """Run sequencing pass.
+
+        Arguments:
+            passmanager_ir: The IR object to be sequenced.
+        """
+
+        self._sequence_instructions(passmanager_ir.alignment, passmanager_ir.sequence)
+        return passmanager_ir
+
+    @singledispatchmethod
+    def _sequence_instructions(self, alignment: AlignmentKind, sequence: PyDAG):
+        """Finalize the sequence by adding edges to the DAG
+
+        ``sequence`` is mutated to include all the edges
+        connecting the elements of the sequence.
+
+        Nested :class:`.SequenceIR` objects are sequenced recursively.
+        """
+        raise NotImplementedError
+
+    # pylint: disable=unused-argument
+    @_sequence_instructions.register(ParallelAlignment)
+    def _sequence_instructions_parallel(self, alignment: ParallelAlignment, sequence: PyDAG):
+        """Finalize the sequence by adding edges to the DAG, following parallel alignment.
+
+        ``sequence`` is mutated to include all the edges
+        connecting the elements of the sequence in parallel.
+
+        Nested :class:`.SequenceIR` objects are sequenced recursively.
+
+        Args:
+            alignment: The IR alignment.
+            sequence: The graph object to be sequenced.
+
+        Raises:
+            PulseCompilerError: if ``property_set`` does not include a mixed_frames_mapping dictionary.
+        """
+        if self.property_set["mixed_frames_mapping"] is None:
+            raise PulseCompilerError(
+                "Parallel sequencing requires mixed frames mapping."
+                " Run MapMixedFrame before sequencing"
+            )
+
+        mixed_frame_mapping = self.property_set["mixed_frames_mapping"]
+
+        idle_after = {}
+        for ind in sequence.node_indices():
+            if ind in (0, 1):
+                # In, Out node
+                continue
+            node = sequence.get_node_data(ind)
+            node_mixed_frames = set()
+
+            if isinstance(node, SequenceIR):
+                self._sequence_instructions(node.alignment, node.sequence)
+                inst_targets = node.inst_targets
+            else:
+                inst_targets = [node.inst_target]
+
+            for inst_target in inst_targets:
+                if isinstance(inst_target, MixedFrame):
+                    node_mixed_frames.add(inst_target)
+                else:
+                    node_mixed_frames |= mixed_frame_mapping[inst_target]
+
+            pred_nodes = [
+                idle_after[mixed_frame]
+                for mixed_frame in node_mixed_frames
+                if mixed_frame in idle_after
+            ]
+            if len(pred_nodes) == 0:
+                pred_nodes = [0]
+            for pred_node in pred_nodes:
+                sequence.add_edge(pred_node, ind, None)
+            for mixed_frame in node_mixed_frames:
+                idle_after[mixed_frame] = ind
+        sequence.add_edges_from_no_data([(ni, 1) for ni in idle_after.values()])
+
+    # pylint: disable=unused-argument
+    @_sequence_instructions.register(SequentialAlignment)
+    def _sequence_instructions_sequential(self, alignment: SequentialAlignment, sequence: PyDAG):
+        """Finalize the sequence by adding edges to the DAG, following sequential alignment.
+
+        ``sequence`` is mutated to include all the edges
+        connecting the elements of the sequence sequentially.
+
+        Nested :class:`.SequenceIR` objects are sequenced recursively.
+
+        Args:
+            alignment: The IR alignment.
+            sequence: The graph object to be sequenced.
+        """
+        nodes = sequence.node_indices()
+        prev = 0
+        # TODO : What's the correct order to use here? Addition index? Actual index?
+        #  Should at least be documented.
+        # The first two nodes are the in\out nodes.
+        for ind in nodes[2:]:
+            sequence.add_edge(prev, ind, None)
+            prev = ind
+            if isinstance(elem := sequence.get_node_data(ind), SequenceIR):
+                self._sequence_instructions(elem.alignment, elem.sequence)
+        sequence.add_edge(prev, 1, None)
+
+    def __hash__(self):
+        return hash((self.__class__.__name__,))
+
+    def __eq__(self, other):
+        return self.__class__.__name__ == other.__class__.__name__

qiskit/pulse/exceptions.py

@@ -27,6 +27,10 @@ def __str__(self):
         return repr(self.message)
 
 
+class PulseCompilerError(PulseError):
+    """Errors raised by the pulse module compiler"""
+
+
 class BackendNotSet(PulseError):
     """Raised if the builder context does not have a backend."""
 

qiskit/pulse/transforms/__init__.py

@@ -87,6 +87,8 @@
     AlignRight,
     AlignSequential,
     AlignmentKind,
+    ParallelAlignment,
+    SequentialAlignment,
 )
 
 from .base_transforms import target_qobj_transform

qiskit/pulse/transforms/alignments.py

@@ -87,7 +87,23 @@ def __repr__(self):
         return f"{self.__class__.__name__}({', '.join(self._context_params)})"
 
 
-class AlignLeft(AlignmentKind):
+class SequentialAlignment(AlignmentKind, abc.ABC):
+    """Abstract base class for ``AlignmentKind`` which aligns instructions sequentially"""
+
+    @property
+    def is_sequential(self) -> bool:
+        return True
+
+
+class ParallelAlignment(AlignmentKind, abc.ABC):
+    """Abstract base class for ``AlignmentKind`` which aligns instructions in parallel"""
+
+    @property
+    def is_sequential(self) -> bool:
+        return False
+
+
+class AlignLeft(ParallelAlignment):
     """Align instructions in as-soon-as-possible manner.
 
     Instructions are placed at earliest available timeslots.
@@ -97,10 +113,6 @@ def __init__(self):
         """Create new left-justified context."""
         super().__init__(context_params=())
 
-    @property
-    def is_sequential(self) -> bool:
-        return False
-
     def align(self, schedule: Schedule) -> Schedule:
         """Reallocate instructions according to the policy.
 
@@ -154,7 +166,7 @@ def _push_left_append(this: Schedule, other: ScheduleComponent) -> Schedule:
         return this.insert(insert_time, other, inplace=True)
 
 
-class AlignRight(AlignmentKind):
+class AlignRight(ParallelAlignment):
     """Align instructions in as-late-as-possible manner.
 
     Instructions are placed at latest available timeslots.
@@ -164,10 +176,6 @@ def __init__(self):
         """Create new right-justified context."""
         super().__init__(context_params=())
 
-    @property
-    def is_sequential(self) -> bool:
-        return False
-
     def align(self, schedule: Schedule) -> Schedule:
         """Reallocate instructions according to the policy.
 
@@ -222,7 +230,7 @@ def _push_right_prepend(this: Schedule, other: ScheduleComponent) -> Schedule:
         return this
 
 
-class AlignSequential(AlignmentKind):
+class AlignSequential(SequentialAlignment):
     """Align instructions sequentially.
 
     Instructions played on different channels are also arranged in a sequence.
@@ -233,10 +241,6 @@ def __init__(self):
         """Create new sequential context."""
         super().__init__(context_params=())
 
-    @property
-    def is_sequential(self) -> bool:
-        return True
-
     def align(self, schedule: Schedule) -> Schedule:
         """Reallocate instructions according to the policy.
 
@@ -256,7 +260,7 @@ def align(self, schedule: Schedule) -> Schedule:
         return aligned
 
 
-class AlignEquispaced(AlignmentKind):
+class AlignEquispaced(SequentialAlignment):
     """Align instructions with equispaced interval within a specified duration.
 
     Instructions played on different channels are also arranged in a sequence.
@@ -274,10 +278,6 @@ def __init__(self, duration: int | ParameterExpression):
         """
         super().__init__(context_params=(duration,))
 
-    @property
-    def is_sequential(self) -> bool:
-        return True
-
     @property
     def duration(self):
         """Return context duration."""
@@ -325,7 +325,7 @@ def align(self, schedule: Schedule) -> Schedule:
         return aligned
 
 
-class AlignFunc(AlignmentKind):
+class AlignFunc(SequentialAlignment):
     """Allocate instructions at position specified by callback function.
 
     The position is specified for each instruction of index ``j`` as a
@@ -362,10 +362,6 @@ def __init__(self, duration: int | ParameterExpression, func: Callable):
         """
         super().__init__(context_params=(duration, func))
 
-    @property
-    def is_sequential(self) -> bool:
-        return True
-
     @property
     def duration(self):
         """Return context duration."""

test/python/pulse/compiler_passes/__init__.py

@@ -0,0 +1,13 @@
+# 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.
+
+"""Qiskit pulse compiler tests."""