Support instruction filtering method in ScheduleBlock class (#9971)

* activate filter method in ScheduleBlock class

* use functools.singledispatch

* add tests for filter method in ScheduleBlock class

* filter out empty schedule_blocks

* update doc

* rm logical_and

* add catch-TypeError functions decorated with singledispatch

* use pulse builder in test

* make another test function for nested block

* add a release note

* Rm optional args

Co-authored-by: Naoki Kanazawa <nkanazawa1989@gmail.com>

* add warning

Co-authored-by: Naoki Kanazawa <nkanazawa1989@gmail.com>

* update the release note

Co-authored-by: Naoki Kanazawa <nkanazawa1989@gmail.com>

* activate exclude method in ScheduleBlock class

---------

Co-authored-by: Naoki Kanazawa <nkanazawa1989@gmail.com>

qiskit/pulse/filters.py

@@ -13,16 +13,31 @@
 """A collection of functions that filter instructions in a pulse program."""
 
 import abc
+from functools import singledispatch
 from typing import Callable, List, Union, Iterable, Optional, Tuple, Any
 
 import numpy as np
 
-from qiskit.pulse import Schedule
+from qiskit.pulse import Schedule, ScheduleBlock, Instruction
 from qiskit.pulse.channels import Channel
 from qiskit.pulse.schedule import Interval
+from qiskit.pulse.exceptions import PulseError
 
 
+@singledispatch
 def filter_instructions(
+    sched, filters: List[Callable], negate: bool = False, recurse_subroutines: bool = True
+):
+    """A catch-TypeError function which will only get called if none of the other decorated
+    functions, namely handle_schedule() and handle_scheduleblock(), handle the type passed.
+    """
+    raise TypeError(
+        f"Type '{type(sched)}' is not valid data format as an input to filter_instructions."
+    )
+
+
+@filter_instructions.register
+def handle_schedule(
     sched: Schedule, filters: List[Callable], negate: bool = False, recurse_subroutines: bool = True
 ) -> Schedule:
     """A filtering function that takes a schedule and returns a schedule consisting of
@@ -61,6 +76,58 @@ def filter_instructions(
     return filter_schedule
 
 
+@filter_instructions.register
+def handle_scheduleblock(
+    sched_blk: ScheduleBlock,
+    filters: List[Callable],
+    negate: bool = False,
+    recurse_subroutines: bool = True,
+) -> ScheduleBlock:
+    """A filtering function that takes a schedule_block and returns a schedule_block consisting of
+    filtered instructions.
+
+    Args:
+        sched_blk: A pulse schedule_block to be filtered.
+        filters: List of callback functions that take an instruction and return boolean.
+        negate: Set `True` to accept an instruction if a filter function returns `False`.
+            Otherwise the instruction is accepted when the filter function returns `False`.
+        recurse_subroutines: Set `True` to individually filter instructions inside of a subroutine
+            defined by the :py:class:`~qiskit.pulse.instructions.Call` instruction.
+
+    Returns:
+        Filtered pulse schedule_block.
+    """
+    from qiskit.pulse.transforms import inline_subroutines
+
+    target_sched_blk = sched_blk
+    if recurse_subroutines:
+        target_sched_blk = inline_subroutines(target_sched_blk)
+
+    def apply_filters_to_insts_in_scheblk(blk: ScheduleBlock) -> ScheduleBlock:
+        blk_new = ScheduleBlock.initialize_from(blk)
+        for element in blk.blocks:
+            if isinstance(element, ScheduleBlock):
+                inner_blk = apply_filters_to_insts_in_scheblk(element)
+                if len(inner_blk) > 0:
+                    blk_new.append(inner_blk)
+
+            elif isinstance(element, Instruction):
+                valid_inst = all(filt(element) for filt in filters)
+                if negate:
+                    valid_inst ^= True
+                if valid_inst:
+                    blk_new.append(element)
+
+            else:
+                raise PulseError(
+                    f"An unexpected element '{element}' is included in ScheduleBlock.blocks."
+                )
+        return blk_new
+
+    filter_sched_blk = apply_filters_to_insts_in_scheblk(target_sched_blk)
+    return filter_sched_blk
+
+
 def composite_filter(
     channels: Optional[Union[Iterable[Channel], Channel]] = None,
     instruction_types: Optional[Union[Iterable[abc.ABCMeta], abc.ABCMeta]] = None,
@@ -107,17 +174,39 @@ def with_channels(channels: Union[Iterable[Channel], Channel]) -> Callable:
     """
     channels = _if_scalar_cast_to_list(channels)
 
-    def channel_filter(time_inst) -> bool:
+    @singledispatch
+    def channel_filter(time_inst):
+        """A catch-TypeError function which will only get called if none of the other decorated
+        functions, namely handle_numpyndarray() and handle_instruction(), handle the type passed.
+        """
+        raise TypeError(
+            f"Type '{type(time_inst)}' is not valid data format as an input to channel_filter."
+        )
+
+    @channel_filter.register
+    def handle_numpyndarray(time_inst: np.ndarray) -> bool:
         """Filter channel.
 
         Args:
-            time_inst (Tuple[int, Instruction]): Time
+            time_inst (numpy.ndarray([int, Instruction])): Time
 
         Returns:
             If instruction matches with condition.
         """
         return any(chan in channels for chan in time_inst[1].channels)
 
+    @channel_filter.register
+    def handle_instruction(inst: Instruction) -> bool:
+        """Filter channel.
+
+        Args:
+            inst: Instruction
+
+        Returns:
+            If instruction matches with condition.
+        """
+        return any(chan in channels for chan in inst.channels)
+
     return channel_filter
 
 
@@ -132,17 +221,39 @@ def with_instruction_types(types: Union[Iterable[abc.ABCMeta], abc.ABCMeta]) ->
     """
     types = _if_scalar_cast_to_list(types)
 
+    @singledispatch
     def instruction_filter(time_inst) -> bool:
+        """A catch-TypeError function which will only get called if none of the other decorated
+        functions, namely handle_numpyndarray() and handle_instruction(), handle the type passed.
+        """
+        raise TypeError(
+            f"Type '{type(time_inst)}' is not valid data format as an input to instruction_filter."
+        )
+
+    @instruction_filter.register
+    def handle_numpyndarray(time_inst: np.ndarray) -> bool:
         """Filter instruction.
 
         Args:
-            time_inst (Tuple[int, Instruction]): Time
+            time_inst (numpy.ndarray([int, Instruction])): Time
 
         Returns:
             If instruction matches with condition.
         """
         return isinstance(time_inst[1], tuple(types))
 
+    @instruction_filter.register
+    def handle_instruction(inst: Instruction) -> bool:
+        """Filter instruction.
+
+        Args:
+            inst: Instruction
+
+        Returns:
+            If instruction matches with condition.
+        """
+        return isinstance(inst, tuple(types))
+
     return instruction_filter
 
 

qiskit/pulse/schedule.py

@@ -1326,86 +1326,76 @@ def filter(
         *filter_funcs: List[Callable],
         channels: Optional[Iterable[Channel]] = None,
         instruction_types: Union[Iterable[abc.ABCMeta], abc.ABCMeta] = None,
-        time_ranges: Optional[Iterable[Tuple[int, int]]] = None,
-        intervals: Optional[Iterable[Interval]] = None,
         check_subroutine: bool = True,
     ):
-        """Return a new ``Schedule`` with only the instructions from this ``ScheduleBlock``
-        which pass though the provided filters; i.e. an instruction will be retained iff
+        """Return a new ``ScheduleBlock`` with only the instructions from this ``ScheduleBlock``
+        which pass though the provided filters; i.e. an instruction will be retained if
         every function in ``filter_funcs`` returns ``True``, the instruction occurs on
-        a channel type contained in ``channels``, the instruction type is contained
-        in ``instruction_types``, and the period over which the instruction operates
-        is *fully* contained in one specified in ``time_ranges`` or ``intervals``.
+        a channel type contained in ``channels``, and the instruction type is contained
+        in ``instruction_types``.
 
-        If no arguments are provided, ``self`` is returned.
+        .. warning::
+            Because ``ScheduleBlock`` is not aware of the execution time of
+            the context instructions, filtering out some instructions may
+            change the execution time of the remaining instructions.
 
-        .. note:: This method is currently not supported. Support will be soon added
-            please create an issue if you believe this must be prioritized.
+        If no arguments are provided, ``self`` is returned.
 
         Args:
-            filter_funcs: A list of Callables which take a (int, Union['Schedule', Instruction])
-                tuple and return a bool.
+            filter_funcs: A list of Callables which take a ``Instruction`` and return a bool.
             channels: For example, ``[DriveChannel(0), AcquireChannel(0)]``.
             instruction_types: For example, ``[PulseInstruction, AcquireInstruction]``.
-            time_ranges: For example, ``[(0, 5), (6, 10)]``.
-            intervals: For example, ``[(0, 5), (6, 10)]``.
             check_subroutine: Set `True` to individually filter instructions inside a subroutine
                 defined by the :py:class:`~qiskit.pulse.instructions.Call` instruction.
 
         Returns:
-            ``Schedule`` consisting of instructions that matches with filtering condition.
-
-        Raises:
-            PulseError: When this method is called. This method will be supported soon.
+            ``ScheduleBlock`` consisting of instructions that matches with filtering condition.
         """
-        raise PulseError(
-            "Method ``ScheduleBlock.filter`` is not supported as this program "
-            "representation does not have the notion of an explicit instruction "
-            "time. Apply ``qiskit.pulse.transforms.block_to_schedule`` function to "
-            "this program to obtain the ``Schedule`` representation supporting "
-            "this method."
+        from qiskit.pulse.filters import composite_filter, filter_instructions
+
+        filters = composite_filter(channels, instruction_types)
+        filters.extend(filter_funcs)
+
+        return filter_instructions(
+            self, filters=filters, negate=False, recurse_subroutines=check_subroutine
         )
 
     def exclude(
         self,
         *filter_funcs: List[Callable],
         channels: Optional[Iterable[Channel]] = None,
         instruction_types: Union[Iterable[abc.ABCMeta], abc.ABCMeta] = None,
-        time_ranges: Optional[Iterable[Tuple[int, int]]] = None,
-        intervals: Optional[Iterable[Interval]] = None,
         check_subroutine: bool = True,
     ):
-        """Return a ``Schedule`` with only the instructions from this Schedule *failing*
-        at least one of the provided filters.
+        """Return a new ``ScheduleBlock`` with only the instructions from this ``ScheduleBlock``
+        *failing* at least one of the provided filters.
         This method is the complement of py:meth:`~self.filter`, so that::
 
-            self.filter(args) | self.exclude(args) == self
+            self.filter(args) + self.exclude(args) == self in terms of instructions included.
 
-        .. note:: This method is currently not supported. Support will be soon added
-            please create an issue if you believe this must be prioritized.
+        .. warning::
+            Because ``ScheduleBlock`` is not aware of the execution time of
+            the context instructions, excluding some instructions may
+            change the execution time of the remaining instructions.
 
         Args:
-            filter_funcs: A list of Callables which take a (int, Union['Schedule', Instruction])
-                tuple and return a bool.
+            filter_funcs: A list of Callables which take a ``Instruction`` and return a bool.
             channels: For example, ``[DriveChannel(0), AcquireChannel(0)]``.
             instruction_types: For example, ``[PulseInstruction, AcquireInstruction]``.
-            time_ranges: For example, ``[(0, 5), (6, 10)]``.
-            intervals: For example, ``[(0, 5), (6, 10)]``.
             check_subroutine: Set `True` to individually filter instructions inside of a subroutine
                 defined by the :py:class:`~qiskit.pulse.instructions.Call` instruction.
 
         Returns:
-            ``Schedule`` consisting of instructions that are not match with filtering condition.
-
-        Raises:
-            PulseError: When this method is called. This method will be supported soon.
+            ``ScheduleBlock`` consisting of instructions that do not match with
+            at least one of filtering conditions.
         """
-        raise PulseError(
-            "Method ``ScheduleBlock.exclude`` is not supported as this program "
-            "representation does not have the notion of instruction "
-            "time. Apply ``qiskit.pulse.transforms.block_to_schedule`` function to "
-            "this program to obtain the ``Schedule`` representation supporting "
-            "this method."
+        from qiskit.pulse.filters import composite_filter, filter_instructions
+
+        filters = composite_filter(channels, instruction_types)
+        filters.extend(filter_funcs)
+
+        return filter_instructions(
+            self, filters=filters, negate=True, recurse_subroutines=check_subroutine
         )
 
     def replace(

releasenotes/notes/filter-schedule-block-29d392ca351f1fb1.yaml

@@ -0,0 +1,27 @@
+---
+features:
+  - |
+    The method :meth:`~qiskit.pulse.schedule.ScheduleBlock.filter` is activated
+    in the :class:`~qiskit.pulse.schedule.ScheduleBlock` class.
+    This method enables users to retain only :class:`~qiskit.pulse.Instruction`
+    objects which pass through all the provided filters.
+    As builtin filter conditions, pulse :class:`~qiskit.pulse.channels.Channel`
+    subclass instance and :class:`~qiskit.pulse.instructions.Instruction`
+    subclass type can be specified.
+    User-defined callbacks taking :class:`~qiskit.pulse.instructions.Instruction` instance
+    can be added to the filters, too.
+  - |
+    The method :meth:`~qiskit.pulse.schedule.ScheduleBlock.exclude` is activated
+    in the :class:`~qiskit.pulse.schedule.ScheduleBlock` class.
+    This method enables users to retain only :class:`~qiskit.pulse.Instruction`
+    objects which do not pass at least one of all the provided filters.
+    As builtin filter conditions, pulse :class:`~qiskit.pulse.channels.Channel`
+    subclass instance and :class:`~qiskit.pulse.instructions.Instruction`
+    subclass type can be specified.
+    User-defined callbacks taking :class:`~qiskit.pulse.instructions.Instruction` instance
+    can be added to the filters, too.
+    This method is the complement of :meth:`~qiskit.pulse.schedule.ScheduleBlock.filter`, so
+    the following condition is always satisfied:
+    ``block.filter(*filters) + block.exclude(*filters) == block`` in terms of
+    instructions included, where ``block`` is a :class:`~qiskit.pulse.schedule.ScheduleBlock`
+    instance.