diff --git a/qiskit/pulse/compiler/__init__.py b/qiskit/pulse/compiler/__init__.py index e8bd600b1a05..0165f484bc35 100644 --- a/qiskit/pulse/compiler/__init__.py +++ b/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 diff --git a/qiskit/pulse/compiler/passes/__init__.py b/qiskit/pulse/compiler/passes/__init__.py index d0807e55c63d..9c6186803696 100644 --- a/qiskit/pulse/compiler/passes/__init__.py +++ b/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 diff --git a/qiskit/pulse/compiler/passes/map_mixed_frames.py b/qiskit/pulse/compiler/passes/map_mixed_frames.py new file mode 100644 index 000000000000..5e04342afeeb --- /dev/null +++ b/qiskit/pulse/compiler/passes/map_mixed_frames.py @@ -0,0 +1,60 @@ +# 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. + + .. notes:: + The pass will override results of previous ``MapMixedFrame`` runs. + """ + + 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__ diff --git a/qiskit/pulse/compiler/passes/set_sequence.py b/qiskit/pulse/compiler/passes/set_sequence.py new file mode 100644 index 000000000000..4a021da7f661 --- /dev/null +++ b/qiskit/pulse/compiler/passes/set_sequence.py @@ -0,0 +1,151 @@ +# 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. + + .. notes:: + 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. + + 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" + ) + 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. + """ + 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__ diff --git a/qiskit/pulse/exceptions.py b/qiskit/pulse/exceptions.py index 21bda97ee1b9..42d5330810eb 100644 --- a/qiskit/pulse/exceptions.py +++ b/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.""" diff --git a/qiskit/pulse/transforms/__init__.py b/qiskit/pulse/transforms/__init__.py index 4fb4bf40e9ef..b84b6ceffd05 100644 --- a/qiskit/pulse/transforms/__init__.py +++ b/qiskit/pulse/transforms/__init__.py @@ -87,6 +87,8 @@ AlignRight, AlignSequential, AlignmentKind, + ParallelAlignment, + SequentialAlignment, ) from .base_transforms import target_qobj_transform diff --git a/qiskit/pulse/transforms/alignments.py b/qiskit/pulse/transforms/alignments.py index 569219777f20..41270ca20383 100644 --- a/qiskit/pulse/transforms/alignments.py +++ b/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.""" diff --git a/test/python/pulse/compiler_passes/__init__.py b/test/python/pulse/compiler_passes/__init__.py new file mode 100644 index 000000000000..e811c8a3d61e --- /dev/null +++ b/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.""" diff --git a/test/python/pulse/compiler_passes/test_map_mixed_frames.py b/test/python/pulse/compiler_passes/test_map_mixed_frames.py new file mode 100644 index 000000000000..a437279d6004 --- /dev/null +++ b/test/python/pulse/compiler_passes/test_map_mixed_frames.py @@ -0,0 +1,112 @@ +# 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. + +"""Test MapMixedFrames""" +from test import QiskitTestCase + +from qiskit.pulse import ( + Constant, + Play, + Delay, + ShiftPhase, +) + +from qiskit.pulse.ir import ( + SequenceIR, +) + +from qiskit.pulse.model import QubitFrame, Qubit, MixedFrame +from qiskit.pulse.transforms import AlignLeft +from qiskit.pulse.compiler import MapMixedFrame + + +class TestMapMixedFrames(QiskitTestCase): + """Test MapMixedFrames analysis pass""" + + def test_basic_ir(self): + """test with basic IR""" + ir_example = SequenceIR(AlignLeft()) + mf = MixedFrame(Qubit(0), QubitFrame(1)) + ir_example.append(Play(Constant(100, 0.1), mixed_frame=mf)) + + mapping_pass = MapMixedFrame() + mapping_pass.run(ir_example) + self.assertEqual(len(mapping_pass.property_set.keys()), 1) + mapping = mapping_pass.property_set["mixed_frames_mapping"] + self.assertEqual(len(mapping), 2) + self.assertEqual(mapping[mf.pulse_target], {mf}) + self.assertEqual(mapping[mf.frame], {mf}) + + mf2 = MixedFrame(Qubit(0), QubitFrame(2)) + ir_example.append(Play(Constant(100, 0.1), mixed_frame=mf2)) + + mapping_pass.run(ir_example) + mapping = mapping_pass.property_set["mixed_frames_mapping"] + self.assertEqual(len(mapping), 3) + self.assertEqual(mapping[mf.pulse_target], {mf, mf2}) + self.assertEqual(mapping[mf.frame], {mf}) + + def test_with_several_inst_target_types(self): + """test with different inst_target types""" + ir_example = SequenceIR(AlignLeft()) + mf = MixedFrame(Qubit(0), QubitFrame(1)) + ir_example.append(Play(Constant(100, 0.1), mixed_frame=mf)) + ir_example.append(Delay(100, target=Qubit(2))) + ir_example.append(ShiftPhase(100, frame=QubitFrame(2))) + + mapping_pass = MapMixedFrame() + mapping_pass.run(ir_example) + mapping = mapping_pass.property_set["mixed_frames_mapping"] + self.assertEqual(len(mapping), 2) + self.assertEqual(mapping[Qubit(0)], {mf}) + self.assertEqual(mapping[QubitFrame(1)], {mf}) + + def test_with_sub_blocks(self): + """test with sub blocks""" + mf1 = MixedFrame(Qubit(0), QubitFrame(0)) + mf2 = MixedFrame(Qubit(0), QubitFrame(1)) + mf3 = MixedFrame(Qubit(0), QubitFrame(2)) + + sub_block_2 = SequenceIR(AlignLeft()) + sub_block_2.append(Play(Constant(100, 0.1), mixed_frame=mf1)) + + sub_block_1 = SequenceIR(AlignLeft()) + sub_block_1.append(Play(Constant(100, 0.1), mixed_frame=mf2)) + sub_block_1.append(sub_block_2) + + mapping_pass = MapMixedFrame() + mapping_pass.run(sub_block_1) + mapping = mapping_pass.property_set["mixed_frames_mapping"] + + self.assertEqual(len(mapping), 3) + self.assertEqual(mapping[Qubit(0)], {mf1, mf2}) + self.assertEqual(mapping[QubitFrame(0)], {mf1}) + self.assertEqual(mapping[QubitFrame(1)], {mf2}) + + ir_example = SequenceIR(AlignLeft()) + ir_example.append(Play(Constant(100, 0.1), mixed_frame=mf3)) + ir_example.append(sub_block_1) + + mapping_pass = MapMixedFrame() + mapping_pass.run(ir_example) + mapping = mapping_pass.property_set["mixed_frames_mapping"] + + self.assertEqual(len(mapping), 4) + self.assertEqual(mapping[Qubit(0)], {mf1, mf2, mf3}) + self.assertEqual(mapping[QubitFrame(0)], {mf1}) + self.assertEqual(mapping[QubitFrame(1)], {mf2}) + self.assertEqual(mapping[QubitFrame(2)], {mf3}) + + def test_equating(self): + """Test equating of passes""" + self.assertTrue(MapMixedFrame() == MapMixedFrame()) + self.assertFalse(MapMixedFrame() == QubitFrame(1)) diff --git a/test/python/pulse/compiler_passes/test_set_sequence.py b/test/python/pulse/compiler_passes/test_set_sequence.py new file mode 100644 index 000000000000..c6aac4d5d143 --- /dev/null +++ b/test/python/pulse/compiler_passes/test_set_sequence.py @@ -0,0 +1,380 @@ +# 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. + +"""Test SetSequence""" +import unittest +from test import QiskitTestCase +from ddt import ddt, named_data, unpack + +from qiskit.pulse import ( + Constant, + Play, + Delay, + ShiftPhase, +) + +from qiskit.pulse.ir import ( + SequenceIR, +) + +from qiskit.pulse.model import QubitFrame, Qubit, MixedFrame +from qiskit.pulse.transforms import ( + AlignLeft, + AlignRight, + AlignSequential, + AlignFunc, + AlignEquispaced, +) +from qiskit.pulse.compiler import MapMixedFrame, SetSequence +from qiskit.pulse.exceptions import PulseCompilerError +from .utils import PulseIrTranspiler + + +class TestSetSequence(QiskitTestCase): + """General tests for set sequence pass""" + + def test_equating(self): + """Test pass equating""" + self.assertTrue(SetSequence() == SetSequence()) + self.assertFalse(SetSequence() == MapMixedFrame()) + + +@ddt +class TestSetSequenceParallelAlignment(QiskitTestCase): + """Test SetSequence pass with Parallel Alignment""" + + ddt_named_data = [["align_left", AlignLeft()], ["align_right", AlignRight()]] + + def setUp(self): + super().setUp() + self._pm = PulseIrTranspiler([MapMixedFrame(), SetSequence()]) + + @named_data(*ddt_named_data) + @unpack + def test_no_mapping_pass_error(self, alignment): + """test that running without MapMixedFrame pass raises a PulseError""" + + pm = PulseIrTranspiler() + pm.append(SetSequence()) + ir_example = SequenceIR(alignment) + ir_example.append(Play(Constant(100, 0.1), mixed_frame=MixedFrame(Qubit(0), QubitFrame(1)))) + + with self.assertRaises(PulseCompilerError): + pm.run(ir_example) + + # TODO: Take care of this weird edge case + @unittest.expectedFailure + @named_data(*ddt_named_data) + @unpack + def test_instruction_not_in_mapping(self, alignment): + """test with an instruction which is not in the mapping""" + + ir_example = SequenceIR(alignment) + ir_example.append(Play(Constant(100, 0.1), mixed_frame=MixedFrame(Qubit(0), QubitFrame(1)))) + ir_example.append(Delay(100, target=Qubit(5))) + + ir_example = self._pm.run(ir_example) + edge_list = ir_example.sequence.edge_list() + self.assertEqual(len(edge_list), 4) + self.assertTrue((0, 2) in edge_list) + self.assertTrue((0, 3) in edge_list) + self.assertTrue((2, 1) in edge_list) + self.assertTrue((3, 1) in edge_list) + + @named_data(*ddt_named_data) + @unpack + def test_parallel_instructions(self, alignment): + """test with two parallel instructions""" + + ir_example = SequenceIR(alignment) + ir_example.append(Play(Constant(100, 0.1), mixed_frame=MixedFrame(Qubit(0), QubitFrame(1)))) + ir_example.append(Play(Constant(100, 0.1), mixed_frame=MixedFrame(Qubit(1), QubitFrame(1)))) + + ir_example = self._pm.run(ir_example) + edge_list = ir_example.sequence.edge_list() + self.assertEqual(len(edge_list), 4) + self.assertTrue((0, 2) in edge_list) + self.assertTrue((2, 1) in edge_list) + self.assertTrue((0, 3) in edge_list) + self.assertTrue((3, 1) in edge_list) + + @named_data(*ddt_named_data) + @unpack + def test_sequential_instructions(self, alignment): + """test with two sequential instructions""" + + ir_example = SequenceIR(alignment) + ir_example.append(Play(Constant(100, 0.1), mixed_frame=MixedFrame(Qubit(0), QubitFrame(1)))) + ir_example.append(Play(Constant(100, 0.1), mixed_frame=MixedFrame(Qubit(0), QubitFrame(1)))) + + ir_example = self._pm.run(ir_example) + edge_list = ir_example.sequence.edge_list() + self.assertEqual(len(edge_list), 3) + self.assertTrue((0, 2) in edge_list) + self.assertTrue((2, 3) in edge_list) + self.assertTrue((3, 1) in edge_list) + + @named_data(*ddt_named_data) + @unpack + def test_pulse_target_instruction_sequencing_to_dependent_instructions(self, alignment): + """test that an instruction which is defined on a PulseTarget and is sequenced correctly + to several dependent isntructions""" + + ir_example = SequenceIR(alignment) + ir_example.append(Delay(100, target=Qubit(0))) + ir_example.append(Play(Constant(100, 0.1), mixed_frame=MixedFrame(Qubit(0), QubitFrame(1)))) + ir_example.append(Play(Constant(100, 0.1), mixed_frame=MixedFrame(Qubit(0), QubitFrame(2)))) + + ir_example = self._pm.run(ir_example) + edge_list = ir_example.sequence.edge_list() + self.assertEqual(len(edge_list), 5) + self.assertTrue((0, 2) in edge_list) + self.assertTrue((2, 3) in edge_list) + self.assertTrue((2, 4) in edge_list) + self.assertTrue((3, 1) in edge_list) + self.assertTrue((4, 1) in edge_list) + + @named_data(*ddt_named_data) + @unpack + def test_frame_instruction_broadcasting_to_dependent_instructions(self, alignment): + """test that an instruction which is defined on a Frame is correctly sequenced to several + dependent instructions""" + + ir_example = SequenceIR(alignment) + ir_example.append(ShiftPhase(100, frame=QubitFrame(0))) + ir_example.append(Play(Constant(100, 0.1), mixed_frame=MixedFrame(Qubit(0), QubitFrame(0)))) + ir_example.append(Play(Constant(100, 0.1), mixed_frame=MixedFrame(Qubit(1), QubitFrame(0)))) + ir_example.append(Play(Constant(100, 0.1), mixed_frame=MixedFrame(Qubit(1), QubitFrame(1)))) + + ir_example = self._pm.run(ir_example) + edge_list = ir_example.sequence.edge_list() + self.assertEqual(len(edge_list), 7) + self.assertTrue((0, 2) in edge_list) + self.assertTrue((2, 3) in edge_list) + self.assertTrue((2, 4) in edge_list) + self.assertTrue((3, 1) in edge_list) + self.assertTrue((4, 1) in edge_list) + self.assertTrue((0, 5) in edge_list) + self.assertTrue((5, 1) in edge_list) + + @named_data(*ddt_named_data) + @unpack + def test_pulse_target_instruction_dependency(self, alignment): + """test with an instruction which is defined on a PulseTarget and depends on + several mixed frames""" + + ir_example = SequenceIR(alignment) + ir_example.append(Play(Constant(100, 0.1), mixed_frame=MixedFrame(Qubit(0), QubitFrame(1)))) + ir_example.append(Play(Constant(100, 0.1), mixed_frame=MixedFrame(Qubit(0), QubitFrame(2)))) + ir_example.append(Delay(100, target=Qubit(0))) + + ir_example = self._pm.run(ir_example) + edge_list = ir_example.sequence.edge_list() + self.assertEqual(len(edge_list), 5) + self.assertTrue((0, 2) in edge_list) + self.assertTrue((0, 3) in edge_list) + self.assertTrue((2, 4) in edge_list) + self.assertTrue((3, 4) in edge_list) + self.assertTrue((4, 1) in edge_list) + + @named_data(*ddt_named_data) + @unpack + def test_frame_instruction_dependency(self, alignment): + """test with an instruction which is defined on a Frame and depends on several mixed frames""" + + ir_example = SequenceIR(alignment) + ir_example.append(Play(Constant(100, 0.1), mixed_frame=MixedFrame(Qubit(0), QubitFrame(0)))) + ir_example.append(Play(Constant(100, 0.1), mixed_frame=MixedFrame(Qubit(1), QubitFrame(0)))) + ir_example.append(ShiftPhase(100, frame=QubitFrame(0))) + + ir_example = self._pm.run(ir_example) + edge_list = ir_example.sequence.edge_list() + self.assertEqual(len(edge_list), 5) + self.assertTrue((0, 2) in edge_list) + self.assertTrue((0, 3) in edge_list) + self.assertTrue((2, 4) in edge_list) + self.assertTrue((3, 4) in edge_list) + self.assertTrue((4, 1) in edge_list) + + @named_data(*ddt_named_data) + @unpack + def test_recursion_to_sub_blocks(self, alignment): + """test that sequencing is recursively applied to sub blocks""" + + sub_block = SequenceIR(alignment) + sub_block.append(Play(Constant(100, 0.1), mixed_frame=MixedFrame(Qubit(0), QubitFrame(1)))) + + ir_example = SequenceIR(alignment) + ir_example.append(Play(Constant(100, 0.1), mixed_frame=MixedFrame(Qubit(0), QubitFrame(0)))) + ir_example.append(sub_block) + + ir_example = self._pm.run(ir_example) + edge_list_sub_block = ir_example.elements()[1].sequence.edge_list() + self.assertEqual(len(edge_list_sub_block), 2) + self.assertTrue((0, 2) in edge_list_sub_block) + self.assertTrue((2, 1) in edge_list_sub_block) + + @named_data(*ddt_named_data) + @unpack + def test_with_parallel_sub_block(self, alignment): + """test with a sub block which doesn't depend on previous instructions""" + + sub_block = SequenceIR(alignment) + sub_block.append(Play(Constant(100, 0.1), mixed_frame=MixedFrame(Qubit(0), QubitFrame(1)))) + + ir_example = SequenceIR(alignment) + ir_example.append(Play(Constant(100, 0.1), mixed_frame=MixedFrame(Qubit(0), QubitFrame(0)))) + ir_example.append(sub_block) + + ir_example = self._pm.run(ir_example) + edge_list = ir_example.sequence.edge_list() + self.assertEqual(len(edge_list), 4) + self.assertTrue((0, 2) in edge_list) + self.assertTrue((0, 3) in edge_list) + self.assertTrue((2, 1) in edge_list) + self.assertTrue((3, 1) in edge_list) + + @named_data(*ddt_named_data) + @unpack + def test_with_simple_sequential_sub_block(self, alignment): + """test with a sub block which depends on a single previous instruction""" + + sub_block = SequenceIR(alignment) + sub_block.append(Play(Constant(100, 0.1), mixed_frame=MixedFrame(Qubit(0), QubitFrame(1)))) + + ir_example = SequenceIR(alignment) + ir_example.append(Play(Constant(100, 0.1), mixed_frame=MixedFrame(Qubit(0), QubitFrame(0)))) + ir_example.append(Play(Constant(100, 0.1), mixed_frame=MixedFrame(Qubit(0), QubitFrame(1)))) + ir_example.append(sub_block) + + ir_example = self._pm.run(ir_example) + edge_list = ir_example.sequence.edge_list() + self.assertEqual(len(edge_list), 5) + self.assertTrue((0, 2) in edge_list) + self.assertTrue((0, 3) in edge_list) + self.assertTrue((3, 4) in edge_list) + self.assertTrue((2, 1) in edge_list) + self.assertTrue((4, 1) in edge_list) + + @named_data(*ddt_named_data) + @unpack + def test_with_sequential_sub_block_with_more_dependencies(self, alignment): + """test with a sub block which depends on several previous instruction""" + + sub_block = SequenceIR(alignment) + sub_block.append(Delay(100, target=Qubit(0))) + + ir_example = SequenceIR(alignment) + ir_example.append(Play(Constant(100, 0.1), mixed_frame=MixedFrame(Qubit(0), QubitFrame(0)))) + ir_example.append(Play(Constant(100, 0.1), mixed_frame=MixedFrame(Qubit(0), QubitFrame(1)))) + ir_example.append(Play(Constant(100, 0.1), mixed_frame=MixedFrame(Qubit(1), QubitFrame(1)))) + ir_example.append(sub_block) + ir_example.append(Play(Constant(100, 0.1), mixed_frame=MixedFrame(Qubit(1), QubitFrame(1)))) + + ir_example = self._pm.run(ir_example) + edge_list = ir_example.sequence.edge_list() + self.assertEqual(len(edge_list), 8) + self.assertTrue((0, 2) in edge_list) + self.assertTrue((0, 3) in edge_list) + self.assertTrue((0, 4) in edge_list) + self.assertTrue((2, 5) in edge_list) + self.assertTrue((3, 5) in edge_list) + self.assertTrue((4, 6) in edge_list) + self.assertTrue((5, 1) in edge_list) + self.assertTrue((6, 1) in edge_list) + + +@ddt +class TestSetSequenceSequentialAlignment(QiskitTestCase): + """Test SetSequence pass with Sequential Alignment""" + + ddt_named_data = [ + ["align_sequential", AlignSequential()], + ["align_func", AlignFunc(100, lambda x: x)], + ["align_equispaced", AlignEquispaced(100)], + ] + + def setUp(self): + super().setUp() + self._pm = PulseIrTranspiler([MapMixedFrame(), SetSequence()]) + + @named_data(*ddt_named_data) + @unpack + def test_no_mapping_pass_error(self, alignment): + """test that running without MapMixedFrame pass raises a PulseError""" + + pm = PulseIrTranspiler() + pm.append(SetSequence()) + ir_example = SequenceIR(alignment) + ir_example.append(Play(Constant(100, 0.1), mixed_frame=MixedFrame(Qubit(0), QubitFrame(1)))) + + with self.assertRaises(PulseCompilerError): + pm.run(ir_example) + + @named_data(*ddt_named_data) + @unpack + def test_several_instructions(self, alignment): + """test with several instructions""" + + ir_example = SequenceIR(alignment) + ir_example.append(Play(Constant(100, 0.1), mixed_frame=MixedFrame(Qubit(0), QubitFrame(1)))) + ir_example.append(Play(Constant(100, 0.1), mixed_frame=MixedFrame(Qubit(1), QubitFrame(1)))) + ir_example.append(Play(Constant(100, 0.1), mixed_frame=MixedFrame(Qubit(2), QubitFrame(1)))) + + ir_example = self._pm.run(ir_example) + edge_list = ir_example.sequence.edge_list() + self.assertEqual(len(edge_list), 4) + self.assertTrue((0, 2) in edge_list) + self.assertTrue((2, 3) in edge_list) + self.assertTrue((3, 4) in edge_list) + self.assertTrue((4, 1) in edge_list) + + @named_data(*ddt_named_data) + @unpack + def test_recursion_to_sub_blocks(self, alignment): + """test that sequencing is recursively applied to sub blocks""" + + sub_block = SequenceIR(alignment) + sub_block.append(Play(Constant(100, 0.1), mixed_frame=MixedFrame(Qubit(0), QubitFrame(1)))) + + ir_example = SequenceIR(alignment) + ir_example.append(Play(Constant(100, 0.1), mixed_frame=MixedFrame(Qubit(0), QubitFrame(0)))) + ir_example.append(sub_block) + + ir_example = self._pm.run(ir_example) + + edge_list_sub_block = ir_example.elements()[1].sequence.edge_list() + self.assertEqual(len(edge_list_sub_block), 2) + self.assertTrue((0, 2) in edge_list_sub_block) + self.assertTrue((2, 1) in edge_list_sub_block) + + @named_data(*ddt_named_data) + @unpack + def test_sub_blocks_and_instructions(self, alignment): + """test sequencing with a mix of instructions and sub blocks""" + + sub_block = SequenceIR(alignment) + sub_block.append(Play(Constant(100, 0.1), mixed_frame=MixedFrame(Qubit(0), QubitFrame(1)))) + + ir_example = SequenceIR(alignment) + ir_example.append(Play(Constant(100, 0.1), mixed_frame=MixedFrame(Qubit(0), QubitFrame(0)))) + ir_example.append(sub_block) + ir_example.append(Play(Constant(100, 0.1), mixed_frame=MixedFrame(Qubit(0), QubitFrame(2)))) + + ir_example = self._pm.run(ir_example) + + edge_list = ir_example.sequence.edge_list() + self.assertEqual(len(edge_list), 4) + self.assertTrue((0, 2) in edge_list) + self.assertTrue((2, 3) in edge_list) + self.assertTrue((3, 4) in edge_list) + self.assertTrue((4, 1) in edge_list) diff --git a/test/python/pulse/compiler_passes/utils.py b/test/python/pulse/compiler_passes/utils.py new file mode 100644 index 000000000000..23d1f3608ead --- /dev/null +++ b/test/python/pulse/compiler_passes/utils.py @@ -0,0 +1,61 @@ +# 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. + +"""Utilities for Pulse Compiler passes tests""" + +from __future__ import annotations + +from abc import ABC +from collections.abc import Callable +from typing import Any + +from qiskit.passmanager import BasePassManager +from qiskit.pulse.ir import SequenceIR + + +PulseProgramT = Any +"""Type alias representing whatever pulse programs.""" + + +class PulseIrTranspiler(BasePassManager, ABC): + """Pass manager for Pulse IR -> Pulse IR transpilation""" + + def _passmanager_frontend( + self, + input_program: SequenceIR, + **kwargs, + ) -> SequenceIR: + + return input_program + + def _passmanager_backend( + self, + passmanager_ir: SequenceIR, + in_program: SequenceIR, + **kwargs, + ) -> SequenceIR: + + return passmanager_ir + + # pylint: disable=arguments-differ + def run( + self, + pulse_programs: SequenceIR | list[SequenceIR], + callback: Callable | None = None, + num_processes: int | None = None, + ) -> PulseProgramT | list[PulseProgramT]: + """Run all the passes on the input pulse programs.""" + return super().run( + in_programs=pulse_programs, + callback=callback, + num_processes=num_processes, + )