ScheduleBlock 3/4 - add ScheduleBlock

qiskit/pulse/__init__.py

@@ -102,6 +102,7 @@
    :toctree: ../stubs/
 
    Schedule
+   ScheduleBlock
    Instruction
 
 
@@ -117,7 +118,7 @@
 Schedule Transforms
 ===================
 
-These functions take :class:`Schedule` s as input and return modified
+Schedule transforms take :class:`Schedule` s as input and return modified
 :class:`Schedule` s.
 
 .. autosummary::
@@ -127,7 +128,6 @@
    transforms.add_implicit_acquires
    transforms.pad
 
-
 Exceptions
 ==========
 
@@ -469,4 +469,4 @@
     Waveform,
 )
 from qiskit.pulse.library.samplers.decorators import functional_pulse
-from qiskit.pulse.schedule import Schedule
+from qiskit.pulse.schedule import Schedule, ScheduleBlock

qiskit/pulse/builder.py

@@ -1899,9 +1899,6 @@ def measure(qubit: int):
         sched.draw()
 
 
-
-
-
     Args:
         func: The Python function to enable as a builder macro. There are no
             requirements on the signature of the function, any calls to pulse

qiskit/pulse/channels.py

@@ -21,13 +21,14 @@
 assembler.
 """
 from abc import ABCMeta
-from typing import Any, Set
+from typing import Any, Set, Union
 
 import numpy as np
 
 from qiskit.circuit import Parameter
 from qiskit.circuit.parameterexpression import ParameterExpression, ParameterValueType
 from qiskit.pulse.exceptions import PulseError
+from qiskit.pulse.utils import deprecated_functionality
 
 
 class Channel(metaclass=ABCMeta):
@@ -68,13 +69,14 @@ def __init__(self, index: int):
         """
         self._validate_index(index)
         self._index = index
-        self._hash = None
+        self._hash = hash((self.__class__.__name__, self._index))
+
         self._parameters = set()
         if isinstance(index, ParameterExpression):
             self._parameters = index.parameters
 
     @property
-    def index(self) -> int:
+    def index(self) -> Union[int, ParameterExpression]:
         """Return the index of this channel. The index is a label for a control signal line
         typically mapped trivially to a qubit index. For instance, ``DriveChannel(0)`` labels
         the signal line driving the qubit labeled with index 0.
@@ -100,14 +102,16 @@ def _validate_index(self, index: Any) -> None:
             raise PulseError('Channel index must be a nonnegative integer')
 
     @property
+    @deprecated_functionality
     def parameters(self) -> Set:
         """Parameters which determine the channel index."""
         return self._parameters
 
     def is_parameterized(self) -> bool:
         """Return True iff the channel is parameterized."""
-        return bool(self.parameters)
+        return isinstance(self.index, ParameterExpression)
 
+    @deprecated_functionality
     def assign(self, parameter: Parameter, value: ParameterValueType) -> 'Channel':
         """Return a new channel with the input Parameter assigned to value.
 
@@ -153,8 +157,6 @@ def __eq__(self, other: 'Channel') -> bool:
         return type(self) is type(other) and self._index == other._index
 
     def __hash__(self):
-        if self._hash is None:
-            self._hash = hash((type(self), self._index))
         return self._hash
 
 

qiskit/pulse/filters.py

@@ -0,0 +1,185 @@
+# This code is part of Qiskit.
+#
+# (C) Copyright IBM 2021.
+#
+# 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 collection of functions that filter instructions in a pulse program."""
+
+import abc
+from typing import Callable, List, Union, Iterable, Optional, Tuple, Any
+
+import numpy as np
+
+from qiskit.pulse import Schedule
+from qiskit.pulse.channels import Channel
+from qiskit.pulse.schedule import Interval
+
+
+def filter_instructions(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
+    filtered instructions.
+
+    Args:
+        sched: A pulse schedule 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.
+    """
+    from qiskit.pulse.transforms import flatten, inline_subroutines
+
+    target_sched = flatten(sched)
+    if recurse_subroutines:
+        target_sched = inline_subroutines(target_sched)
+
+    time_inst_tuples = np.array(target_sched.instructions)
+
+    valid_insts = np.ones(len(time_inst_tuples), dtype=bool)
+    for filt in filters:
+        valid_insts = np.logical_and(valid_insts, np.array(list(map(filt, time_inst_tuples))))
+
+    if negate and len(filters) > 0:
+        valid_insts = ~valid_insts
+
+    return Schedule(*time_inst_tuples[valid_insts], name=sched.name, metadata=sched.metadata)
+
+
+def composite_filter(channels: Optional[Union[Iterable[Channel], Channel]] = None,
+                     instruction_types: Optional[Union[Iterable[abc.ABCMeta], abc.ABCMeta]] = None,
+                     time_ranges: Optional[Iterable[Tuple[int, int]]] = None,
+                     intervals: Optional[Iterable[Interval]] = None) -> List[Callable]:
+    """A helper function to generate a list of filter functions based on
+    typical elements to be filtered.
+
+    Args:
+        channels: For example, ``[DriveChannel(0), AcquireChannel(0)]``.
+        instruction_types (Optional[Iterable[Type[qiskit.pulse.Instruction]]]): For example,
+            ``[PulseInstruction, AcquireInstruction]``.
+        time_ranges: For example, ``[(0, 5), (6, 10)]``.
+        intervals: For example, ``[(0, 5), (6, 10)]``.
+
+    Returns:
+        List of filtering functions.
+    """
+    filters = []
+
+    # An empty list is also valid input for filter generators.
+    # See unittest `test.python.pulse.test_schedule.TestScheduleFilter.test_empty_filters`.
+    if channels is not None:
+        filters.append(with_channels(channels))
+    if instruction_types is not None:
+        filters.append(with_instruction_types(instruction_types))
+    if time_ranges is not None:
+        filters.append(with_intervals(time_ranges))
+    if intervals is not None:
+        filters.append(with_intervals(intervals))
+
+    return filters
+
+
+def with_channels(channels: Union[Iterable[Channel], Channel]) -> Callable:
+    """Channel filter generator.
+
+    Args:
+        channels: List of channels to filter.
+
+    Returns:
+        A callback function to filter channels.
+    """
+    channels = _if_scalar_cast_to_list(channels)
+
+    def channel_filter(time_inst) -> bool:
+        """Filter channel.
+
+        Args:
+            time_inst (Tuple[int, Instruction]): Time
+
+        Returns:
+            If instruction matches with condition.
+        """
+        return any(chan in channels for chan in time_inst[1].channels)
+    return channel_filter
+
+
+def with_instruction_types(types: Union[Iterable[abc.ABCMeta], abc.ABCMeta]) -> Callable:
+    """Instruction type filter generator.
+
+    Args:
+        types: List of instruction types to filter.
+
+    Returns:
+        A callback function to filter instructions.
+    """
+    types = _if_scalar_cast_to_list(types)
+
+    def instruction_filter(time_inst) -> bool:
+        """Filter instruction.
+
+        Args:
+            time_inst (Tuple[int, Instruction]): Time
+
+        Returns:
+            If instruction matches with condition.
+        """
+        return isinstance(time_inst[1], tuple(types))
+
+    return instruction_filter
+
+
+def with_intervals(ranges: Union[Iterable[Interval], Interval]) -> Callable:
+    """Interval filter generator.
+
+    Args:
+        ranges: List of intervals ``[t0, t1]`` to filter.
+
+    Returns:
+        A callback function to filter intervals.
+    """
+    ranges = _if_scalar_cast_to_list(ranges)
+
+    def interval_filter(time_inst) -> bool:
+        """Filter interval.
+        Args:
+            time_inst (Tuple[int, Instruction]): Time
+
+        Returns:
+            If instruction matches with condition.
+        """
+        for t0, t1 in ranges:
+            inst_start = time_inst[0]
+            inst_stop = inst_start + time_inst[1].duration
+            if t0 <= inst_start and inst_stop <= t1:
+                return True
+        return False
+
+    return interval_filter
+
+
+def _if_scalar_cast_to_list(to_list: Any) -> List[Any]:
+    """A helper function to create python list of input arguments.
+
+    Args:
+        to_list: Arbitrary object can be converted into a python list.
+
+    Returns:
+        Python list of input object.
+    """
+    try:
+        iter(to_list)
+    except TypeError:
+        to_list = [to_list]
+    return to_list

qiskit/pulse/instructions/acquire.py

@@ -13,7 +13,7 @@
 """The Acquire instruction is used to trigger the qubit measurement unit and provide
 some metadata for the acquisition process, for example, where to store classified readout data.
 """
-from typing import Optional, Union
+from typing import Optional, Union, Tuple
 
 from qiskit.circuit import ParameterExpression
 from qiskit.pulse.channels import MemorySlot, RegisterSlot, AcquireChannel
@@ -74,8 +74,7 @@ def __init__(self,
         self._kernel = kernel
         self._discriminator = discriminator
 
-        all_channels = tuple(chan for chan in [channel, mem_slot, reg_slot] if chan is not None)
-        super().__init__((duration, channel, mem_slot, reg_slot), None, all_channels, name=name)
+        super().__init__(operands=(duration, channel, mem_slot, reg_slot), name=name)
 
     @property
     def channel(self) -> AcquireChannel:
@@ -84,6 +83,11 @@ def channel(self) -> AcquireChannel:
         """
         return self.operands[1]
 
+    @property
+    def channels(self) -> Tuple[Union[AcquireChannel, MemorySlot, RegisterSlot]]:
+        """Returns the channels that this schedule uses."""
+        return tuple(self.operands[ind] for ind in (1, 2, 3) if self.operands[ind] is not None)
+
     @property
     def duration(self) -> Union[int, ParameterExpression]:
         """Duration of this instruction."""
@@ -118,6 +122,10 @@ def reg_slot(self) -> RegisterSlot:
         """
         return self.operands[3]
 
+    def is_parameterized(self) -> bool:
+        """Return True iff the instruction is parameterized."""
+        return isinstance(self.duration, ParameterExpression) or super().is_parameterized()
+
     def __repr__(self) -> str:
         return "{}({}{}{}{}{}{})".format(
             self.__class__.__name__,