Pulse compiler skeleton

qiskit/pulse/compiler/__init__.py

@@ -0,0 +1,15 @@
+# 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.
+
+"""Pass-based Qiskit pulse program compiler."""
+
+from .passmanager import PulsePassManager

qiskit/pulse/compiler/basepasses.py

@@ -0,0 +1,93 @@
+# 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.
+
+"""A base pass for Qiskit PulseIR compilation."""
+
+import warnings
+from abc import ABC, abstractmethod
+
+from qiskit.passmanager.base_tasks import GenericPass
+from qiskit.transpiler.target import Target
+from qiskit.pulse.ir import IrBlock
+
+
+class TransformationPass(GenericPass, ABC):
+    """A base transform pass for Qiskit PulseIR.
+
+    A transform pass modifies the input Qiskit PulseIR and returns an updated PulseIR.
+    The returned object can be new instance, or the pass can mutate and return the same object.
+    """
+
+    def __init__(
+        self,
+        target: Target,
+    ):
+        """Create new transform pass.
+
+        Args:
+            target: System configuration information presented in the form of Qiskit model.
+        """
+        super().__init__()
+        self.target = target
+
+    @abstractmethod
+    def run(
+        self,
+        passmanager_ir: IrBlock,
+    ) -> IrBlock:
+        pass
+
+    def __eq__(self, other):
+        warnings.warn(
+            f"{self.__class__} does not explicitly define a protocol to evaluate equality. "
+            "Two pass objects instantiated individually with the same configuration may be "
+            "considered as different passes.",
+            RuntimeWarning,
+        )
+        super().__eq__(other)
+
+
+class AnalysisPass(GenericPass, ABC):
+    """A base analysis pass for Qiskit PulseIR.
+
+    An analysis pass performs investigation on the input Qiskit PulseIR.
+    The information obtained may be stored in the property set.
+    This pass returns nothing.
+    """
+
+    def __init__(
+        self,
+        target: Target,
+    ):
+        """Create new transform pass.
+
+        Args:
+            target: System configuration information presented in the form of Qiskit model.
+        """
+        super().__init__()
+        self.target = target
+
+    @abstractmethod
+    def run(
+        self,
+        passmanager_ir: IrBlock,
+    ) -> None:
+        pass
+
+    def __eq__(self, other):
+        warnings.warn(
+            f"{self.__class__} does not explicitly define a protocol to evaluate equality. "
+            "Two pass objects instantiated individually with the same configuration may be "
+            "considered as different passes.",
+            RuntimeWarning,
+        )
+        super().__eq__(other)

qiskit/pulse/compiler/converters.py

@@ -0,0 +1,62 @@
+# 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.
+
+"""A base pass for Qiskit PulseIR compilation."""
+
+
+from qiskit.pulse.schedule import ScheduleBlock
+from qiskit.pulse.ir import IrBlock, IrInstruction
+
+
+def schedule_to_ir(schedule: ScheduleBlock) -> IrBlock:
+    """Convert ScheduleBlock into PulseIR.
+
+    Args:
+        schedule: Schedule to convert.
+
+    Returns:
+        PulseIR used internally in the pulse compiler.
+    """
+    out = IrBlock(alignment=schedule.alignment_context)
+
+    def _wrap_recursive(_elm):
+        if isinstance(_elm, ScheduleBlock):
+            return schedule_to_ir(_elm)
+        return IrInstruction(instruction=_elm)
+
+    for element in schedule.blocks:
+        wrapped_element = _wrap_recursive(element)
+        out.add_element(wrapped_element)
+    return out
+
+
+def ir_to_schedule(pulse_ir: IrBlock) -> ScheduleBlock:
+    """Convert PulseIR to ScheduleBlock.
+
+    Args:
+        pulse_ir: PulseIR to convert.
+
+    Returns:
+        ScheduleBlock that end-user may interact with.
+    """
+    out = ScheduleBlock(alignment_context=pulse_ir.alignment)
+
+    def _unwrap_recursive(_elm):
+        if isinstance(_elm, IrBlock):
+            return ir_to_schedule(_elm)
+        return _elm.instruction
+
+    for element in pulse_ir.elements:
+        unwrapped_element = _unwrap_recursive(element)
+        out.append(unwrapped_element, inplace=True)
+
+    return out

qiskit/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.
+
+"""Built-in pulse compile passes."""

qiskit/pulse/compiler/passmanager.py

@@ -0,0 +1,108 @@
+# 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.
+
+"""Pass manager for pulse schedules."""
+
+from __future__ import annotations
+
+from collections.abc import Callable
+from typing import Any
+
+from qiskit.passmanager import BasePassManager
+from qiskit.pulse.ir import IrBlock
+from qiskit.pulse.compiler.converters import schedule_to_ir, ir_to_schedule
+from qiskit.pulse.schedule import ScheduleBlock
+
+
+class PulsePassManager(BasePassManager):
+    """A pass manager for compiling Qiskit ScheduleBlock programs."""
+
+    def _passmanager_frontend(
+        self,
+        input_program: ScheduleBlock,
+        **kwargs,
+    ) -> IrBlock:
+        return schedule_to_ir(input_program)
+
+    def _passmanager_backend(
+        self,
+        passmanager_ir: IrBlock,
+        in_program: ScheduleBlock,
+        **kwargs,
+    ) -> Any:
+        output = kwargs.get("output", "schedule_block")
+
+        if output == "pulse_ir":
+            return passmanager_ir
+        if output == "schedule_block":
+            return ir_to_schedule(passmanager_ir)
+
+        raise ValueError(f"Specified target format '{output}' is not supported.")
+
+    # pylint: disable=arguments-differ
+    def run(
+        self,
+        schedules: ScheduleBlock | list[ScheduleBlock],
+        output: str = "schedule_block",
+        callback: Callable | None = None,
+        num_processes: int | None = None,
+    ) -> Any:
+        """Run all the passes on the input pulse schedules.
+
+        Args:
+            schedules: Input pulse programs to transform via all the registered passes.
+                When a list of schedules are passed, the transform is performed in parallel
+                for each input schedule with multiprocessing.
+            output: Format of the output program::
+
+                    schedule_block: Return in :class:`.ScheduleBlock` format.
+                    pulse_ir: Return in :class:`.PulseIR` format.
+
+            callback: A callback function that will be called after each pass execution. The
+                function will be called with 5 keyword arguments::
+
+                    task (GenericPass): the pass being run
+                    passmanager_ir (Any): depending on pass manager subclass
+                    property_set (PropertySet): the property set
+                    running_time (float): the time to execute the pass
+                    count (int): the index for the pass execution
+
+                The exact arguments pass expose the internals of the pass
+                manager and are subject to change as the pass manager internals
+                change. If you intend to reuse a callback function over
+                multiple releases be sure to check that the arguments being
+                passed are the same.
+
+                To use the callback feature you define a function that will
+                take in kwargs dict and access the variables. For example::
+
+                    def callback_func(**kwargs):
+                        task = kwargs['task']
+                        passmanager_ir = kwargs['passmanager_ir']
+                        property_set = kwargs['property_set']
+                        running_time = kwargs['running_time']
+                        count = kwargs['count']
+                        ...
+            num_processes: The maximum number of parallel processes to launch if parallel
+                execution is enabled. This argument overrides ``num_processes`` in the user
+                configuration file, and the ``QISKIT_NUM_PROCS`` environment variable. If set
+                to ``None`` the system default or local user configuration will be used.
+
+        Returns:
+            The transformed program(s) in specified program format.
+        """
+        return super().run(
+            in_programs=schedules,
+            callback=callback,
+            num_processes=num_processes,
+            output=output,
+        )

test/python/pulse/_dummy_programs.py

@@ -0,0 +1,34 @@
+# 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.
+
+"""A pulse schedule library for unit tests."""
+
+from qiskit.pulse import builder
+from qiskit.pulse.library import symbolic_pulses
+from qiskit.pulse import channels
+from qiskit.pulse.schedule import ScheduleBlock
+
+
+def play_gaussian() -> ScheduleBlock:
+    """Create simple schedule of playing a single Gaussian pulse."""
+
+    with builder.build() as out:
+        builder.play(
+            pulse=symbolic_pulses.Gaussian(
+                duration=100,
+                amp=0.1,
+                sigma=25,
+                angle=0.0,
+            ),
+            channel=channels.DriveChannel(0),
+        )
+    return out

test/python/pulse/test_compile.py

@@ -0,0 +1,31 @@
+# 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.
+
+"""A test cases for pulse compilation."""
+
+from qiskit.pulse.compiler import PulsePassManager
+
+from test import QiskitTestCase  # pylint: disable=wrong-import-order
+from . import _dummy_programs as schedule_lib
+
+
+class TestCompiler(QiskitTestCase):
+    """Test case for pulse compiler."""
+
+    def test_roundtrip(self):
+        """Test just returns the input program."""
+        # Just convert an input to PulseIR and convert it back to ScheduleBlock.
+        pm = PulsePassManager()
+
+        in_prog = schedule_lib.play_gaussian()
+        out_prog = pm.run(schedules=in_prog, output="schedule_block")
+        self.assertEqual(in_prog, out_prog)