scnc
/
qiskit
mirror of https://github.com/Qiskit/qiskit.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

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>
This commit is contained in:
Junya Nakamura 2023-06-07 20:03:04 +09:00 committed by GitHub
parent 218c7735d4
commit 2d8300c834
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
4 changed files with 401 additions and 49 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

119
qiskit/pulse/filters.py
View File

@ -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

80
qiskit/pulse/schedule.py
View File

@ -1326,44 +1326,38 @@ class ScheduleBlock:
*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``.
.. 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.
If no arguments are provided, ``self`` is returned.
.. note:: This method is currently not supported. Support will be soon added
please create an issue if you believe this must be prioritized.
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(
@ -1371,41 +1365,37 @@ class ScheduleBlock:
*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(

27
releasenotes/notes/filter-schedule-block-29d392ca351f1fb1.yaml Normal file
View File

@ -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.

224
test/python/pulse/test_block.py
View File

@ -13,7 +13,9 @@
# pylint: disable=invalid-name
"""Test cases for the pulse schedule block."""
import re
import unittest
from typing import List, Any
from qiskit import pulse, circuit
from qiskit.pulse import transforms
from qiskit.pulse.exceptions import PulseError
@ -749,3 +751,225 @@ class TestParametrizedBlockOperation(BaseTestBlock):
ref_sched = ref_sched.insert(90, pulse.Delay(10, self.d0))
self.assertScheduleEqual(block, ref_sched)
class TestBlockFilter(BaseTestBlock):
"""Test ScheduleBlock filtering methods."""
def test_filter_channels(self):
"""Test filtering over channels."""
with pulse.build() as blk:
pulse.play(self.test_waveform0, self.d0)
pulse.delay(10, self.d0)
pulse.play(self.test_waveform1, self.d1)
filtered_blk = self._filter_and_test_consistency(blk, channels=[self.d0])
self.assertEqual(len(filtered_blk.channels), 1)
self.assertTrue(self.d0 in filtered_blk.channels)
with pulse.build() as ref_blk:
pulse.play(self.test_waveform0, self.d0)
pulse.delay(10, self.d0)
self.assertEqual(filtered_blk, ref_blk)
filtered_blk = self._filter_and_test_consistency(blk, channels=[self.d1])
self.assertEqual(len(filtered_blk.channels), 1)
self.assertTrue(self.d1 in filtered_blk.channels)
with pulse.build() as ref_blk:
pulse.play(self.test_waveform1, self.d1)
self.assertEqual(filtered_blk, ref_blk)
filtered_blk = self._filter_and_test_consistency(blk, channels=[self.d0, self.d1])
self.assertEqual(len(filtered_blk.channels), 2)
for ch in [self.d0, self.d1]:
self.assertTrue(ch in filtered_blk.channels)
self.assertEqual(filtered_blk, blk)
def test_filter_channels_nested_block(self):
"""Test filtering over channels in a nested block."""
with pulse.build() as blk:
with pulse.align_sequential():
pulse.play(self.test_waveform0, self.d0)
pulse.delay(5, self.d0)
with pulse.build(self.backend) as cx_blk:
pulse.cx(0, 1)
pulse.call(cx_blk)
for ch in [self.d0, self.d1, pulse.ControlChannel(0)]:
filtered_blk = self._filter_and_test_consistency(blk, channels=[ch])
self.assertEqual(len(filtered_blk.channels), 1)
self.assertTrue(ch in filtered_blk.channels)
def test_filter_inst_types(self):
"""Test filtering on instruction types."""
with pulse.build() as blk:
pulse.acquire(5, pulse.AcquireChannel(0), pulse.MemorySlot(0))
with pulse.build() as blk_internal:
pulse.play(self.test_waveform1, self.d1)
pulse.call(blk_internal)
pulse.reference(name="dummy_reference")
pulse.delay(10, self.d0)
pulse.play(self.test_waveform0, self.d0)
pulse.barrier(self.d0, self.d1, pulse.AcquireChannel(0), pulse.MemorySlot(0))
pulse.set_frequency(10, self.d0)
pulse.shift_frequency(5, self.d1)
pulse.set_phase(3.14 / 4.0, self.d0)
pulse.shift_phase(-3.14 / 2.0, self.d1)
pulse.snapshot(label="dummy_snapshot")
# test filtering Acquire
filtered_blk = self._filter_and_test_consistency(blk, instruction_types=[pulse.Acquire])
self.assertEqual(len(filtered_blk.blocks), 1)
self.assertIsInstance(filtered_blk.blocks[0], pulse.Acquire)
self.assertEqual(len(filtered_blk.channels), 2)
# test filtering Reference
filtered_blk = self._filter_and_test_consistency(
blk, instruction_types=[pulse.instructions.Reference]
)
self.assertEqual(len(filtered_blk.blocks), 1)
self.assertIsInstance(filtered_blk.blocks[0], pulse.instructions.Reference)
# test filtering Delay
filtered_blk = self._filter_and_test_consistency(blk, instruction_types=[pulse.Delay])
self.assertEqual(len(filtered_blk.blocks), 1)
self.assertIsInstance(filtered_blk.blocks[0], pulse.Delay)
self.assertEqual(len(filtered_blk.channels), 1)
# test filtering Play
filtered_blk = self._filter_and_test_consistency(blk, instruction_types=[pulse.Play])
self.assertEqual(len(filtered_blk.blocks), 2)
self.assertIsInstance(filtered_blk.blocks[0].blocks[0], pulse.Play)
self.assertIsInstance(filtered_blk.blocks[1], pulse.Play)
self.assertEqual(len(filtered_blk.channels), 2)
# test filtering RelativeBarrier
filtered_blk = self._filter_and_test_consistency(
blk, instruction_types=[pulse.instructions.RelativeBarrier]
)
self.assertEqual(len(filtered_blk.blocks), 1)
self.assertIsInstance(filtered_blk.blocks[0], pulse.instructions.RelativeBarrier)
self.assertEqual(len(filtered_blk.channels), 4)
# test filtering SetFrequency
filtered_blk = self._filter_and_test_consistency(
blk, instruction_types=[pulse.SetFrequency]
)
self.assertEqual(len(filtered_blk.blocks), 1)
self.assertIsInstance(filtered_blk.blocks[0], pulse.SetFrequency)
self.assertEqual(len(filtered_blk.channels), 1)
# test filtering ShiftFrequency
filtered_blk = self._filter_and_test_consistency(
blk, instruction_types=[pulse.ShiftFrequency]
)
self.assertEqual(len(filtered_blk.blocks), 1)
self.assertIsInstance(filtered_blk.blocks[0], pulse.ShiftFrequency)
self.assertEqual(len(filtered_blk.channels), 1)
# test filtering SetPhase
filtered_blk = self._filter_and_test_consistency(blk, instruction_types=[pulse.SetPhase])
self.assertEqual(len(filtered_blk.blocks), 1)
self.assertIsInstance(filtered_blk.blocks[0], pulse.SetPhase)
self.assertEqual(len(filtered_blk.channels), 1)
# test filtering ShiftPhase
filtered_blk = self._filter_and_test_consistency(blk, instruction_types=[pulse.ShiftPhase])
self.assertEqual(len(filtered_blk.blocks), 1)
self.assertIsInstance(filtered_blk.blocks[0], pulse.ShiftPhase)
self.assertEqual(len(filtered_blk.channels), 1)
# test filtering SnapShot
filtered_blk = self._filter_and_test_consistency(blk, instruction_types=[pulse.Snapshot])
self.assertEqual(len(filtered_blk.blocks), 1)
self.assertIsInstance(filtered_blk.blocks[0], pulse.Snapshot)
self.assertEqual(len(filtered_blk.channels), 1)
def test_filter_functionals(self):
"""Test functional filtering."""
with pulse.build() as blk:
pulse.play(self.test_waveform0, self.d0, "play0")
pulse.delay(10, self.d0, "delay0")
with pulse.build() as blk_internal:
pulse.play(self.test_waveform1, self.d1, "play1")
pulse.call(blk_internal)
pulse.play(self.test_waveform1, self.d1)
def filter_with_inst_name(inst: pulse.Instruction) -> bool:
try:
if isinstance(inst.name, str):
match_obj = re.search(pattern="play", string=inst.name)
if match_obj is not None:
return True
except AttributeError:
pass
return False
filtered_blk = self._filter_and_test_consistency(blk, filter_with_inst_name)
self.assertEqual(len(filtered_blk.blocks), 2)
self.assertIsInstance(filtered_blk.blocks[0], pulse.Play)
self.assertIsInstance(filtered_blk.blocks[1].blocks[0], pulse.Play)
self.assertEqual(len(filtered_blk.channels), 2)
def test_filter_multiple(self):
"""Test filter composition."""
with pulse.build() as blk:
pulse.play(pulse.Constant(100, 0.1, name="play0"), self.d0)
pulse.delay(10, self.d0, "delay0")
with pulse.build(name="internal_blk") as blk_internal:
pulse.play(pulse.Constant(50, 0.1, name="play1"), self.d0)
pulse.call(blk_internal)
pulse.barrier(self.d0, self.d1)
pulse.play(pulse.Constant(100, 0.1, name="play2"), self.d1)
def filter_with_pulse_name(inst: pulse.Instruction) -> bool:
try:
if isinstance(inst.pulse.name, str):
match_obj = re.search(pattern="play", string=inst.pulse.name)
if match_obj is not None:
return True
except AttributeError:
pass
return False
filtered_blk = self._filter_and_test_consistency(
blk, filter_with_pulse_name, channels=[self.d1], instruction_types=[pulse.Play]
)
self.assertEqual(len(filtered_blk.blocks), 1)
self.assertIsInstance(filtered_blk.blocks[0], pulse.Play)
self.assertEqual(len(filtered_blk.channels), 1)
def _filter_and_test_consistency(
self, sched_blk: pulse.ScheduleBlock, *args: Any, **kwargs: Any
) -> pulse.ScheduleBlock:
"""
Returns sched_blk.filter(*args, **kwargs),
including a test that sched_blk.filter | sched_blk.exclude == sched_blk
in terms of instructions.
"""
filtered = sched_blk.filter(*args, **kwargs)
excluded = sched_blk.exclude(*args, **kwargs)
def list_instructions(blk: pulse.ScheduleBlock) -> List[pulse.Instruction]:
insts = list()
for element in blk.blocks:
if isinstance(element, pulse.ScheduleBlock):
inner_insts = list_instructions(element)
if len(inner_insts) != 0:
insts.extend(inner_insts)
elif isinstance(element, pulse.Instruction):
insts.append(element)
return insts
sum_insts = list_instructions(filtered) + list_instructions(excluded)
ref_insts = list_instructions(sched_blk)
self.assertEqual(len(sum_insts), len(ref_insts))
self.assertTrue(all(inst in ref_insts for inst in sum_insts))
return filtered