# This code is part of Qiskit.
#
# (C) Copyright IBM 2019.
#
# 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.
# pylint: disable=cyclic-import
"""
=========
Schedules
=========
.. currentmodule:: qiskit.pulse
Schedules are Pulse programs. They describe instruction sequences for the control hardware.
The Schedule is one of the most fundamental objects to this pulse-level programming module.
A ``Schedule`` is a representation of a *program* in Pulse. Each schedule tracks the time of each
instruction occuring in parallel over multiple signal *channels*.
.. autosummary::
:toctree: ../stubs/
Schedule
ScheduleBlock
"""
import abc
import copy
import functools
import itertools
import multiprocessing as mp
import re
import sys
import warnings
from typing import List, Tuple, Iterable, Union, Dict, Callable, Set, Optional, Any
import numpy as np
import rustworkx as rx
from qiskit.circuit.parameter import Parameter
from qiskit.circuit.parameterexpression import ParameterExpression, ParameterValueType
from qiskit.pulse.channels import Channel
from qiskit.pulse.exceptions import PulseError, UnassignedReferenceError
from qiskit.pulse.instructions import Instruction, Reference
from qiskit.pulse.utils import instruction_duration_validation
from qiskit.pulse.reference_manager import ReferenceManager
from qiskit.utils.multiprocessing import is_main_process
Interval = Tuple[int, int]
"""An interval type is a tuple of a start time (inclusive) and an end time (exclusive)."""
TimeSlots = Dict[Channel, List[Interval]]
"""List of timeslots occupied by instructions for each channel."""
[documentos]class Schedule:
"""A quantum program *schedule* with exact time constraints for its instructions, operating
over all input signal *channels* and supporting special syntaxes for building.
Pulse program representation for the original Qiskit Pulse model [1].
Instructions are not allowed to overlap in time
on the same channel. This overlap constraint is immediately
evaluated when a new instruction is added to the ``Schedule`` object.
It is necessary to specify the absolute start time and duration
for each instruction so as to deterministically fix its execution time.
The ``Schedule`` program supports some syntax sugar for easier programming.
- Appending an instruction to the end of a channel
.. code-block:: python
sched = Schedule()
sched += Play(Gaussian(160, 0.1, 40), DriveChannel(0))
- Appending an instruction shifted in time by a given amount
.. code-block:: python
sched = Schedule()
sched += Play(Gaussian(160, 0.1, 40), DriveChannel(0)) << 30
- Merge two schedules
.. code-block:: python
sched1 = Schedule()
sched1 += Play(Gaussian(160, 0.1, 40), DriveChannel(0))
sched2 = Schedule()
sched2 += Play(Gaussian(160, 0.1, 40), DriveChannel(1))
sched2 = sched1 | sched2
A :obj:`.PulseError` is immediately raised when the overlap constraint is violated.
In the schedule representation, we cannot parametrize the duration of instructions.
Thus we need to create a new schedule object for each duration.
To parametrize an instruction's duration, the :class:`~qiskit.pulse.ScheduleBlock`
representation may be used instead.
References:
[1]: https://arxiv.org/abs/2004.06755
"""
# Prefix to use for auto naming.
prefix = "sched"
# Counter to count instance number.
instances_counter = itertools.count()
def __init__(
self,
*schedules: Union["ScheduleComponent", Tuple[int, "ScheduleComponent"]],
name: Optional[str] = None,
metadata: Optional[dict] = None,
):
"""Create an empty schedule.
Args:
*schedules: Child Schedules of this parent Schedule. May either be passed as
the list of schedules, or a list of ``(start_time, schedule)`` pairs.
name: Name of this schedule. Defaults to an autogenerated string if not provided.
metadata: Arbitrary key value metadata to associate with the schedule. This gets
stored as free-form data in a dict in the
:attr:`~qiskit.pulse.Schedule.metadata` attribute. It will not be directly
used in the schedule.
Raises:
TypeError: if metadata is not a dict.
"""
from qiskit.pulse.parameter_manager import ParameterManager
if name is None:
name = self.prefix + str(next(self.instances_counter))
if sys.platform != "win32" and not is_main_process():
name += f"-{mp.current_process().pid}"
self._name = name
self._parameter_manager = ParameterManager()
if not isinstance(metadata, dict) and metadata is not None:
raise TypeError("Only a dictionary or None is accepted for schedule metadata")
self._metadata = metadata or {}
self._duration = 0
# These attributes are populated by ``_mutable_insert``
self._timeslots = {}
self._children = []
for sched_pair in schedules:
try:
time, sched = sched_pair
except TypeError:
# recreate as sequence starting at 0.
time, sched = 0, sched_pair
self._mutable_insert(time, sched)
[documentos] @classmethod
def initialize_from(cls, other_program: Any, name: Optional[str] = None) -> "Schedule":
"""Create new schedule object with metadata of another schedule object.
Args:
other_program: Qiskit program that provides metadata to new object.
name: Name of new schedule. Name of ``schedule`` is used by default.
Returns:
New schedule object with name and metadata.
Raises:
PulseError: When `other_program` does not provide necessary information.
"""
try:
name = name or other_program.name
if other_program.metadata:
metadata = other_program.metadata.copy()
else:
metadata = None
return cls(name=name, metadata=metadata)
except AttributeError as ex:
raise PulseError(
f"{cls.__name__} cannot be initialized from the program data "
f"{other_program.__class__.__name__}."
) from ex
@property
def name(self) -> str:
"""Name of this Schedule"""
return self._name
@property
def metadata(self) -> Dict[str, Any]:
"""The user provided metadata associated with the schedule.
User provided ``dict`` of metadata for the schedule.
The metadata contents do not affect the semantics of the program
but are used to influence the execution of the schedule. It is expected
to be passed between all transforms of the schedule and that providers
will associate any schedule metadata with the results it returns from the
execution of that schedule.
"""
return self._metadata
@metadata.setter
def metadata(self, metadata):
"""Update the schedule metadata"""
if not isinstance(metadata, dict) and metadata is not None:
raise TypeError("Only a dictionary or None is accepted for schedule metadata")
self._metadata = metadata or {}
@property
def timeslots(self) -> TimeSlots:
"""Time keeping attribute."""
return self._timeslots
@property
def duration(self) -> int:
"""Duration of this schedule."""
return self._duration
@property
def start_time(self) -> int:
"""Starting time of this schedule."""
return self.ch_start_time(*self.channels)
@property
def stop_time(self) -> int:
"""Stopping time of this schedule."""
return self.duration
@property
def channels(self) -> Tuple[Channel]:
"""Returns channels that this schedule uses."""
return tuple(self._timeslots.keys())
@property
def children(self) -> Tuple[Tuple[int, "ScheduleComponent"], ...]:
"""Return the child schedule components of this ``Schedule`` in the
order they were added to the schedule.
Notes:
Nested schedules are returned as-is. If you want to collect only instructions,
use py:meth:`~Schedule.instructions` instead.
Returns:
A tuple, where each element is a two-tuple containing the initial
scheduled time of each ``NamedValue`` and the component
itself.
"""
return tuple(self._children)
@property
def instructions(self) -> Tuple[Tuple[int, Instruction]]:
"""Get the time-ordered instructions from self."""
def key(time_inst_pair):
inst = time_inst_pair[1]
return time_inst_pair[0], inst.duration, sorted(chan.name for chan in inst.channels)
return tuple(sorted(self._instructions(), key=key))
@property
def parameters(self) -> Set:
"""Parameters which determine the schedule behavior."""
return self._parameter_manager.parameters
[documentos] def ch_duration(self, *channels: Channel) -> int:
"""Return the time of the end of the last instruction over the supplied channels.
Args:
*channels: Channels within ``self`` to include.
"""
return self.ch_stop_time(*channels)
[documentos] def ch_start_time(self, *channels: Channel) -> int:
"""Return the time of the start of the first instruction over the supplied channels.
Args:
*channels: Channels within ``self`` to include.
"""
try:
chan_intervals = (self._timeslots[chan] for chan in channels if chan in self._timeslots)
return min(intervals[0][0] for intervals in chan_intervals)
except ValueError:
# If there are no instructions over channels
return 0
[documentos] def ch_stop_time(self, *channels: Channel) -> int:
"""Return maximum start time over supplied channels.
Args:
*channels: Channels within ``self`` to include.
"""
try:
chan_intervals = (self._timeslots[chan] for chan in channels if chan in self._timeslots)
return max(intervals[-1][1] for intervals in chan_intervals)
except ValueError:
# If there are no instructions over channels
return 0
def _instructions(self, time: int = 0):
"""Iterable for flattening Schedule tree.
Args:
time: Shifted time due to parent.
Yields:
Iterable[Tuple[int, Instruction]]: Tuple containing the time each
:class:`~qiskit.pulse.Instruction`
starts at and the flattened :class:`~qiskit.pulse.Instruction` s.
"""
for insert_time, child_sched in self.children:
yield from child_sched._instructions(time + insert_time)
[documentos] def shift(self, time: int, name: Optional[str] = None, inplace: bool = False) -> "Schedule":
"""Return a schedule shifted forward by ``time``.
Args:
time: Time to shift by.
name: Name of the new schedule. Defaults to the name of self.
inplace: Perform operation inplace on this schedule. Otherwise
return a new ``Schedule``.
"""
if inplace:
return self._mutable_shift(time)
return self._immutable_shift(time, name=name)
def _immutable_shift(self, time: int, name: Optional[str] = None) -> "Schedule":
"""Return a new schedule shifted forward by `time`.
Args:
time: Time to shift by
name: Name of the new schedule if call was mutable. Defaults to name of self
"""
shift_sched = Schedule.initialize_from(self, name)
shift_sched.insert(time, self, inplace=True)
return shift_sched
def _mutable_shift(self, time: int) -> "Schedule":
"""Return this schedule shifted forward by `time`.
Args:
time: Time to shift by
Raises:
PulseError: if ``time`` is not an integer.
"""
if not isinstance(time, int):
raise PulseError("Schedule start time must be an integer.")
timeslots = {}
for chan, ch_timeslots in self._timeslots.items():
timeslots[chan] = [(ts[0] + time, ts[1] + time) for ts in ch_timeslots]
_check_nonnegative_timeslot(timeslots)
self._duration = self._duration + time
self._timeslots = timeslots
self._children = [(orig_time + time, child) for orig_time, child in self.children]
return self
[documentos] def insert(
self,
start_time: int,
schedule: "ScheduleComponent",
name: Optional[str] = None,
inplace: bool = False,
) -> "Schedule":
"""Return a new schedule with ``schedule`` inserted into ``self`` at ``start_time``.
Args:
start_time: Time to insert the schedule.
schedule: Schedule to insert.
name: Name of the new schedule. Defaults to the name of self.
inplace: Perform operation inplace on this schedule. Otherwise
return a new ``Schedule``.
"""
if inplace:
return self._mutable_insert(start_time, schedule)
return self._immutable_insert(start_time, schedule, name=name)
def _mutable_insert(self, start_time: int, schedule: "ScheduleComponent") -> "Schedule":
"""Mutably insert `schedule` into `self` at `start_time`.
Args:
start_time: Time to insert the second schedule.
schedule: Schedule to mutably insert.
"""
self._add_timeslots(start_time, schedule)
self._children.append((start_time, schedule))
self._parameter_manager.update_parameter_table(schedule)
return self
def _immutable_insert(
self,
start_time: int,
schedule: "ScheduleComponent",
name: Optional[str] = None,
) -> "Schedule":
"""Return a new schedule with ``schedule`` inserted into ``self`` at ``start_time``.
Args:
start_time: Time to insert the schedule.
schedule: Schedule to insert.
name: Name of the new ``Schedule``. Defaults to name of ``self``.
"""
new_sched = Schedule.initialize_from(self, name)
new_sched._mutable_insert(0, self)
new_sched._mutable_insert(start_time, schedule)
return new_sched
[documentos] def append(
self, schedule: "ScheduleComponent", name: Optional[str] = None, inplace: bool = False
) -> "Schedule":
r"""Return a new schedule with ``schedule`` inserted at the maximum time over
all channels shared between ``self`` and ``schedule``.
.. math::
t = \textrm{max}(\texttt{x.stop_time} |\texttt{x} \in
\texttt{self.channels} \cap \texttt{schedule.channels})
Args:
schedule: Schedule to be appended.
name: Name of the new ``Schedule``. Defaults to name of ``self``.
inplace: Perform operation inplace on this schedule. Otherwise
return a new ``Schedule``.
"""
common_channels = set(self.channels) & set(schedule.channels)
time = self.ch_stop_time(*common_channels)
return self.insert(time, schedule, name=name, inplace=inplace)
[documentos] def filter(
self,
*filter_funcs: 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,
) -> "Schedule":
"""Return a new ``Schedule`` with only the instructions from this ``Schedule`` which pass
though the provided filters; i.e. an instruction will be retained iff 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``.
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.
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.
"""
from qiskit.pulse.filters import composite_filter, filter_instructions
filters = composite_filter(channels, instruction_types, time_ranges, intervals)
filters.extend(filter_funcs)
return filter_instructions(
self, filters=filters, negate=False, recurse_subroutines=check_subroutine
)
[documentos] def exclude(
self,
*filter_funcs: 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,
) -> "Schedule":
"""Return a ``Schedule`` with only the instructions from this Schedule *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
Args:
filter_funcs: A list of Callables which take a (int, Union['Schedule', Instruction])
tuple 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.
"""
from qiskit.pulse.filters import composite_filter, filter_instructions
filters = composite_filter(channels, instruction_types, time_ranges, intervals)
filters.extend(filter_funcs)
return filter_instructions(
self, filters=filters, negate=True, recurse_subroutines=check_subroutine
)
def _add_timeslots(self, time: int, schedule: "ScheduleComponent") -> None:
"""Update all time tracking within this schedule based on the given schedule.
Args:
time: The time to insert the schedule into self.
schedule: The schedule to insert into self.
Raises:
PulseError: If timeslots overlap or an invalid start time is provided.
"""
if not np.issubdtype(type(time), np.integer):
raise PulseError("Schedule start time must be an integer.")
other_timeslots = _get_timeslots(schedule)
self._duration = max(self._duration, time + schedule.duration)
for channel in schedule.channels:
if channel not in self._timeslots:
if time == 0:
self._timeslots[channel] = copy.copy(other_timeslots[channel])
else:
self._timeslots[channel] = [
(i[0] + time, i[1] + time) for i in other_timeslots[channel]
]
continue
for idx, interval in enumerate(other_timeslots[channel]):
if interval[0] + time >= self._timeslots[channel][-1][1]:
# Can append the remaining intervals
self._timeslots[channel].extend(
[(i[0] + time, i[1] + time) for i in other_timeslots[channel][idx:]]
)
break
try:
interval = (interval[0] + time, interval[1] + time)
index = _find_insertion_index(self._timeslots[channel], interval)
self._timeslots[channel].insert(index, interval)
except PulseError as ex:
raise PulseError(
"Schedule(name='{new}') cannot be inserted into Schedule(name='{old}') at "
"time {time} because its instruction on channel {ch} scheduled from time "
"{t0} to {tf} overlaps with an existing instruction."
"".format(
new=schedule.name or "",
old=self.name or "",
time=time,
ch=channel,
t0=interval[0],
tf=interval[1],
)
) from ex
_check_nonnegative_timeslot(self._timeslots)
def _remove_timeslots(self, time: int, schedule: "ScheduleComponent"):
"""Delete the timeslots if present for the respective schedule component.
Args:
time: The time to remove the timeslots for the ``schedule`` component.
schedule: The schedule to insert into self.
Raises:
PulseError: If timeslots overlap or an invalid start time is provided.
"""
if not isinstance(time, int):
raise PulseError("Schedule start time must be an integer.")
for channel in schedule.channels:
if channel not in self._timeslots:
raise PulseError(f"The channel {channel} is not present in the schedule")
channel_timeslots = self._timeslots[channel]
other_timeslots = _get_timeslots(schedule)
for interval in other_timeslots[channel]:
if channel_timeslots:
interval = (interval[0] + time, interval[1] + time)
index = _interval_index(channel_timeslots, interval)
if channel_timeslots[index] == interval:
channel_timeslots.pop(index)
continue
raise PulseError(
"Cannot find interval ({t0}, {tf}) to remove from "
"channel {ch} in Schedule(name='{name}').".format(
ch=channel, t0=interval[0], tf=interval[1], name=schedule.name
)
)
if not channel_timeslots:
self._timeslots.pop(channel)
def _replace_timeslots(self, time: int, old: "ScheduleComponent", new: "ScheduleComponent"):
"""Replace the timeslots of ``old`` if present with the timeslots of ``new``.
Args:
time: The time to remove the timeslots for the ``schedule`` component.
old: Instruction to replace.
new: Instruction to replace with.
"""
self._remove_timeslots(time, old)
self._add_timeslots(time, new)
def _renew_timeslots(self):
"""Regenerate timeslots based on current instructions."""
self._timeslots.clear()
for t0, inst in self.instructions:
self._add_timeslots(t0, inst)
[documentos] def replace(
self,
old: "ScheduleComponent",
new: "ScheduleComponent",
inplace: bool = False,
) -> "Schedule":
"""Return a ``Schedule`` with the ``old`` instruction replaced with a ``new``
instruction.
The replacement matching is based on an instruction equality check.
.. code-block::
from qiskit import pulse
d0 = pulse.DriveChannel(0)
sched = pulse.Schedule()
old = pulse.Play(pulse.Constant(100, 1.0), d0)
new = pulse.Play(pulse.Constant(100, 0.1), d0)
sched += old
sched = sched.replace(old, new)
assert sched == pulse.Schedule(new)
Only matches at the top-level of the schedule tree. If you wish to
perform this replacement over all instructions in the schedule tree.
Flatten the schedule prior to running::
.. code-block::
sched = pulse.Schedule()
sched += pulse.Schedule(old)
sched = sched.flatten()
sched = sched.replace(old, new)
assert sched == pulse.Schedule(new)
Args:
old: Instruction to replace.
new: Instruction to replace with.
inplace: Replace instruction by mutably modifying this ``Schedule``.
Returns:
The modified schedule with ``old`` replaced by ``new``.
Raises:
PulseError: If the ``Schedule`` after replacements will has a timing overlap.
"""
from qiskit.pulse.parameter_manager import ParameterManager
new_children = []
new_parameters = ParameterManager()
for time, child in self.children:
if child == old:
new_children.append((time, new))
new_parameters.update_parameter_table(new)
else:
new_children.append((time, child))
new_parameters.update_parameter_table(child)
if inplace:
self._children = new_children
self._parameter_manager = new_parameters
self._renew_timeslots()
return self
else:
try:
new_sched = Schedule.initialize_from(self)
for time, inst in new_children:
new_sched.insert(time, inst, inplace=True)
return new_sched
except PulseError as err:
raise PulseError(
f"Replacement of {old} with {new} results in overlapping instructions."
) from err
[documentos] def is_parameterized(self) -> bool:
"""Return True iff the instruction is parameterized."""
return self._parameter_manager.is_parameterized()
[documentos] def assign_parameters(
self, value_dict: Dict[ParameterExpression, ParameterValueType], inplace: bool = True
) -> "Schedule":
"""Assign the parameters in this schedule according to the input.
Args:
value_dict: A mapping from Parameters to either numeric values or another
Parameter expression.
inplace: Set ``True`` to override this instance with new parameter.
Returns:
Schedule with updated parameters.
"""
if not inplace:
new_schedule = copy.deepcopy(self)
return new_schedule.assign_parameters(value_dict, inplace=True)
return self._parameter_manager.assign_parameters(pulse_program=self, value_dict=value_dict)
[documentos] def get_parameters(self, parameter_name: str) -> List[Parameter]:
"""Get parameter object bound to this schedule by string name.
Because different ``Parameter`` objects can have the same name,
this method returns a list of ``Parameter`` s for the provided name.
Args:
parameter_name: Name of parameter.
Returns:
Parameter objects that have corresponding name.
"""
return self._parameter_manager.get_parameters(parameter_name)
def __len__(self) -> int:
"""Return number of instructions in the schedule."""
return len(self.instructions)
def __add__(self, other: "ScheduleComponent") -> "Schedule":
"""Return a new schedule with ``other`` inserted within ``self`` at ``start_time``."""
return self.append(other)
def __or__(self, other: "ScheduleComponent") -> "Schedule":
"""Return a new schedule which is the union of `self` and `other`."""
return self.insert(0, other)
def __lshift__(self, time: int) -> "Schedule":
"""Return a new schedule which is shifted forward by ``time``."""
return self.shift(time)
def __eq__(self, other: "ScheduleComponent") -> bool:
"""Test if two Schedule are equal.
Equality is checked by verifying there is an equal instruction at every time
in ``other`` for every instruction in this ``Schedule``.
.. warning::
This does not check for logical equivalency. Ie.,
```python
>>> Delay(10, DriveChannel(0)) + Delay(10, DriveChannel(0))
== Delay(20, DriveChannel(0))
False
```
"""
# 0. type check, we consider Instruction is a subtype of schedule
if not isinstance(other, (type(self), Instruction)):
return False
# 1. channel check
if set(self.channels) != set(other.channels):
return False
# 2. size check
if len(self.instructions) != len(other.instructions):
return False
# 3. instruction check
return all(
self_inst == other_inst
for self_inst, other_inst in zip(self.instructions, other.instructions)
)
def __repr__(self) -> str:
name = format(self._name) if self._name else ""
instructions = ", ".join([repr(instr) for instr in self.instructions[:50]])
if len(self.instructions) > 25:
instructions += ", ..."
return f'{self.__class__.__name__}({instructions}, name="{name}")'
def _require_schedule_conversion(function: Callable) -> Callable:
"""A method decorator to convert schedule block to pulse schedule.
This conversation is performed for backward compatibility only if all durations are assigned.
"""
@functools.wraps(function)
def wrapper(self, *args, **kwargs):
from qiskit.pulse.transforms import block_to_schedule
return function(block_to_schedule(self), *args, **kwargs)
return wrapper
[documentos]class ScheduleBlock:
"""Time-ordered sequence of instructions with alignment context.
:class:`.ScheduleBlock` supports lazy scheduling of context instructions,
i.e. their timeslots is always generated at runtime.
This indicates we can parametrize instruction durations as well as
other parameters. In contrast to :class:`.Schedule` being somewhat static,
:class:`.ScheduleBlock` is a dynamic representation of a pulse program.
.. rubric:: Pulse Builder
The Qiskit pulse builder is a domain specific language that is developed on top of
the schedule block. Use of the builder syntax will improve the workflow of
pulse programming. See :ref:`pulse_builder` for a user guide.
.. rubric:: Alignment contexts
A schedule block is always relatively scheduled.
Instead of taking individual instructions with absolute execution time ``t0``,
the schedule block defines a context of scheduling and instructions
under the same context are scheduled in the same manner (alignment).
Several contexts are available in :ref:`pulse_alignments`.
A schedule block is instantiated with one of these alignment contexts.
The default context is :class:`AlignLeft`, for which all instructions are left-justified,
in other words, meaning they use as-soon-as-possible scheduling.
If you need an absolute-time interval in between instructions, you can explicitly
insert :class:`~qiskit.pulse.instructions.Delay` instructions.
.. rubric:: Nested blocks
A schedule block can contain other nested blocks with different alignment contexts.
This enables advanced scheduling, where a subset of instructions is
locally scheduled in a different manner.
Note that a :class:`.Schedule` instance cannot be directly added to a schedule block.
To add a :class:`.Schedule` instance, wrap it in a :class:`.Call` instruction.
This is implicitly performed when a schedule is added through the :ref:`pulse_builder`.
.. rubric:: Unsupported operations
Because the schedule block representation lacks timeslots, it cannot
perform particular :class:`.Schedule` operations such as :meth:`insert` or :meth:`shift` that
require instruction start time ``t0``.
In addition, :meth:`exclude` and :meth:`filter` methods are not supported
because these operations may identify the target instruction with ``t0``.
Except for these operations, :class:`.ScheduleBlock` provides full compatibility
with :class:`.Schedule`.
.. rubric:: Subroutine
The timeslots-free representation offers much greater flexibility for writing pulse programs.
Because :class:`.ScheduleBlock` only cares about the ordering of the child blocks
we can add an undefined pulse sequence as a subroutine of the main program.
If your program contains the same sequence multiple times, this representation may
reduce the memory footprint required by the program construction.
Such a subroutine is realized by the special compiler directive
:class:`~qiskit.pulse.instructions.Reference` that is defined by
a unique set of reference key strings to the subroutine.
The (executable) subroutine is separately stored in the main program.
Appended reference directives are resolved when the main program is executed.
Subroutines must be assigned through :meth:`assign_references` before execution.
.. rubric:: Program Scoping
When you call a subroutine from another subroutine, or append a schedule block
to another schedule block, the management of references and parameters
can be a hard task. Schedule block offers a convenient feature to help with this
by automatically scoping the parameters and subroutines.
.. code-block::
from qiskit import pulse
from qiskit.circuit.parameter import Parameter
amp1 = Parameter("amp")
with pulse.build() as sched1:
pulse.play(pulse.Constant(100, amp1), pulse.DriveChannel(0))
print(sched1.scoped_parameters())
.. parsed-literal::
(Parameter(root::amp),)
The :meth:`~ScheduleBlock.scoped_parameters` method returns all :class:`~.Parameter`
objects defined in the schedule block. The parameter name is updated to reflect
its scope information, i.e. where it is defined.
The outer scope is called "root". Since the "amp" parameter is directly used
in the current builder context, it is prefixed with "root".
Note that the :class:`Parameter` object returned by :meth:`~ScheduleBlock.scoped_parameters`
preserves the hidden `UUID`_ key, and thus the scoped name doesn't break references
to the original :class:`Parameter`.
You may want to call this program from another program.
In this example, the program is called with the reference key "grand_child".
You can call a subroutine without specifying a substantial program
(like ``sched1`` above which we will assign later).
.. code-block::
amp2 = Parameter("amp")
with pulse.build() as sched2:
with pulse.align_right():
pulse.reference("grand_child")
pulse.play(pulse.Constant(200, amp2), pulse.DriveChannel(0))
print(sched2.scoped_parameters())
.. parsed-literal::
(Parameter(root::amp),)
This only returns "root::amp" because the "grand_child" reference is unknown.
Now you assign the actual pulse program to this reference.
.. code-block::
sched2.assign_references({("grand_child", ): sched1})
print(sched2.scoped_parameters())
.. parsed-literal::
(Parameter(root::amp), Parameter(root::grand_child::amp))
Now you get two parameters "root::amp" and "root::grand_child::amp".
The second parameter name indicates it is defined within the referred program "grand_child".
The program calling the "grand_child" has a reference program description
which is accessed through :attr:`ScheduleBlock.references`.
.. code-block::
print(sched2.references)
.. parsed-literal::
ReferenceManager:
- ('grand_child',): ScheduleBlock(Play(Constant(duration=100, amp=amp,...
Finally, you may want to call this program from another program.
Here we try a different approach to define subroutine. Namely, we call
a subroutine from the root program with the actual program ``sched2``.
.. code-block::
amp3 = Parameter("amp")
with pulse.build() as main:
pulse.play(pulse.Constant(300, amp3), pulse.DriveChannel(0))
pulse.call(sched2, name="child")
print(main.scoped_parameters())
.. parsed-literal::
(Parameter(root::amp), Parameter(root::child::amp), Parameter(root::child::grand_child::amp))
This implicitly creates a reference named "child" within
the root program and assigns ``sched2`` to it.
You get three parameters "root::amp", "root::child::amp", and "root::child::grand_child::amp".
As you can see, each parameter name reflects the layer of calls from the root program.
If you know the scope of a parameter, you can directly get the parameter object
using :meth:`ScheduleBlock.search_parameters` as follows.
.. code-block::
main.search_parameters("root::child::grand_child::amp")
You can use a regular expression to specify the scope.
The following returns the parameters defined within the scope of "ground_child"
regardless of its parent scope. This is sometimes convenient if you
want to extract parameters from a deeply nested program.
.. code-block::
main.search_parameters("\\S::grand_child::amp")
Note that the root program is only aware of its direct references.
.. code-block::
print(main.references)
.. parsed-literal::
ReferenceManager:
- ('child',): ScheduleBlock(ScheduleBlock(ScheduleBlock(Play(Con...
As you can see the main program cannot directly assign a subroutine to the "grand_child" because
this subroutine is not called within the root program, i.e. it is indirectly called by "child".
However, the returned :class:`.ReferenceManager` is a dict-like object, and you can still
reach to "grand_child" via the "child" program with the following chained dict access.
.. code-block::
main.references[("child", )].references[("grand_child", )]
Note that :attr:`ScheduleBlock.parameters` and :meth:`ScheduleBlock.scoped_parameters()`
still collect all parameters also from the subroutine once it's assigned.
.. _UUID: https://docs.python.org/3/library/uuid.html#module-uuid
"""
__slots__ = (
"_parent",
"_name",
"_reference_manager",
"_parameter_manager",
"_alignment_context",
"_blocks",
"_metadata",
)
# Prefix to use for auto naming.
prefix = "block"
# Counter to count instance number.
instances_counter = itertools.count()
def __init__(
self, name: Optional[str] = None, metadata: Optional[dict] = None, alignment_context=None
):
"""Create an empty schedule block.
Args:
name: Name of this schedule. Defaults to an autogenerated string if not provided.
metadata: Arbitrary key value metadata to associate with the schedule. This gets
stored as free-form data in a dict in the
:attr:`~qiskit.pulse.ScheduleBlock.metadata` attribute. It will not be directly
used in the schedule.
alignment_context (AlignmentKind): ``AlignmentKind`` instance that manages
scheduling of instructions in this block.
Raises:
TypeError: if metadata is not a dict.
"""
from qiskit.pulse.parameter_manager import ParameterManager
from qiskit.pulse.transforms import AlignLeft
if name is None:
name = self.prefix + str(next(self.instances_counter))
if sys.platform != "win32" and not is_main_process():
name += f"-{mp.current_process().pid}"
# This points to the parent schedule object in the current scope.
# Note that schedule block can be nested without referencing, e.g. .append(child_block),
# and parent=None indicates the root program of the current scope.
# The nested schedule block objects should not have _reference_manager and
# should refer to the one of the root program.
# This also means referenced program should be assigned to the root program, not to child.
self._parent = None
self._name = name
self._parameter_manager = ParameterManager()
self._reference_manager = ReferenceManager()
self._alignment_context = alignment_context or AlignLeft()
self._blocks = []
# get parameters from context
self._parameter_manager.update_parameter_table(self._alignment_context)
if not isinstance(metadata, dict) and metadata is not None:
raise TypeError("Only a dictionary or None is accepted for schedule metadata")
self._metadata = metadata or {}
[documentos] @classmethod
def initialize_from(cls, other_program: Any, name: Optional[str] = None) -> "ScheduleBlock":
"""Create new schedule object with metadata of another schedule object.
Args:
other_program: Qiskit program that provides metadata to new object.
name: Name of new schedule. Name of ``block`` is used by default.
Returns:
New block object with name and metadata.
Raises:
PulseError: When ``other_program`` does not provide necessary information.
"""
try:
name = name or other_program.name
if other_program.metadata:
metadata = other_program.metadata.copy()
else:
metadata = None
try:
alignment_context = other_program.alignment_context
except AttributeError:
alignment_context = None
return cls(name=name, metadata=metadata, alignment_context=alignment_context)
except AttributeError as ex:
raise PulseError(
f"{cls.__name__} cannot be initialized from the program data "
f"{other_program.__class__.__name__}."
) from ex
@property
def name(self) -> str:
"""Return name of this schedule"""
return self._name
@property
def metadata(self) -> Dict[str, Any]:
"""The user provided metadata associated with the schedule.
User provided ``dict`` of metadata for the schedule.
The metadata contents do not affect the semantics of the program
but are used to influence the execution of the schedule. It is expected
to be passed between all transforms of the schedule and that providers
will associate any schedule metadata with the results it returns from the
execution of that schedule.
"""
return self._metadata
@metadata.setter
def metadata(self, metadata):
"""Update the schedule metadata"""
if not isinstance(metadata, dict) and metadata is not None:
raise TypeError("Only a dictionary or None is accepted for schedule metadata")
self._metadata = metadata or {}
@property
def alignment_context(self):
"""Return alignment instance that allocates block component to generate schedule."""
return self._alignment_context
[documentos] def is_schedulable(self) -> bool:
"""Return ``True`` if all durations are assigned."""
# check context assignment
for context_param in self._alignment_context._context_params:
if isinstance(context_param, ParameterExpression):
return False
# check duration assignment
for elm in self.blocks:
if isinstance(elm, ScheduleBlock):
if not elm.is_schedulable():
return False
else:
try:
if not isinstance(elm.duration, int):
return False
except UnassignedReferenceError:
return False
return True
@property
@_require_schedule_conversion
def duration(self) -> int:
"""Duration of this schedule block."""
return self.duration
@property
def channels(self) -> Tuple[Channel]:
"""Returns channels that this schedule block uses."""
chans = set()
for elm in self.blocks:
if isinstance(elm, Reference):
raise UnassignedReferenceError(
f"This schedule contains unassigned reference {elm.ref_keys} "
"and channels are ambiguous. Please assign the subroutine first."
)
chans = chans | set(elm.channels)
return tuple(chans)
@property
@_require_schedule_conversion
def instructions(self) -> Tuple[Tuple[int, Instruction]]:
"""Get the time-ordered instructions from self."""
return self.instructions
@property
def blocks(self) -> Tuple["BlockComponent", ...]:
"""Get the block elements added to self.
.. note::
The sequence of elements is returned in order of addition. Because the first element is
schedule first, e.g. FIFO, the returned sequence is roughly time-ordered.
However, in the parallel alignment context, especially in
the as-late-as-possible scheduling, or :class:`.AlignRight` context,
the actual timing of when the instructions are issued is unknown until
the :class:`.ScheduleBlock` is scheduled and converted into a :class:`.Schedule`.
"""
blocks = []
for elm in self._blocks:
if isinstance(elm, Reference):
elm = self.references.get(elm.ref_keys, None) or elm
blocks.append(elm)
return tuple(blocks)
@property
def parameters(self) -> Set[Parameter]:
"""Return unassigned parameters with raw names."""
# Need new object not to mutate parameter_manager.parameters
out_params = set()
out_params |= self._parameter_manager.parameters
for subroutine in self.references.values():
if subroutine is None:
continue
out_params |= subroutine.parameters
return out_params
[documentos] def scoped_parameters(self) -> Tuple[Parameter]:
"""Return unassigned parameters with scoped names.
.. note::
If a parameter is defined within a nested scope,
it is prefixed with all parent-scope names with the delimiter string,
which is "::". If a reference key of the scope consists of
multiple key strings, it will be represented by a single string joined with ",".
For example, "root::xgate,q0::amp" for the parameter "amp" defined in the
reference specified by the key strings ("xgate", "q0").
"""
return tuple(
sorted(
_collect_scoped_parameters(self, current_scope="root").values(),
key=lambda p: p.name,
)
)
@property
def references(self) -> ReferenceManager:
"""Return a reference manager of the current scope."""
if self._parent is not None:
return self._parent.references
return self._reference_manager
[documentos] @_require_schedule_conversion
def ch_duration(self, *channels: Channel) -> int:
"""Return the time of the end of the last instruction over the supplied channels.
Args:
*channels: Channels within ``self`` to include.
"""
return self.ch_duration(*channels)
[documentos] def append(
self, block: "BlockComponent", name: Optional[str] = None, inplace: bool = True
) -> "ScheduleBlock":
"""Return a new schedule block with ``block`` appended to the context block.
The execution time is automatically assigned when the block is converted into schedule.
Args:
block: ScheduleBlock to be appended.
name: Name of the new ``Schedule``. Defaults to name of ``self``.
inplace: Perform operation inplace on this schedule. Otherwise,
return a new ``Schedule``.
Returns:
Schedule block with appended schedule.
Raises:
PulseError: When invalid schedule type is specified.
"""
if not isinstance(block, (ScheduleBlock, Instruction)):
raise PulseError(
f"Appended `schedule` {block.__class__.__name__} is invalid type. "
"Only `Instruction` and `ScheduleBlock` can be accepted."
)
if not inplace:
schedule = copy.deepcopy(self)
schedule._name = name or self.name
schedule.append(block, inplace=True)
return schedule
if isinstance(block, Reference) and block.ref_keys not in self.references:
self.references[block.ref_keys] = None
elif isinstance(block, ScheduleBlock):
block = copy.deepcopy(block)
# Expose subroutines to the current main scope.
# Note that this 'block' is not called.
# The block is just directly appended to the current scope.
if block.is_referenced():
if block._parent is not None:
# This is an edge case:
# If this is not a parent, block.references points to the parent's reference
# where subroutine not referred within the 'block' may exist.
# Move only references existing in the 'block'.
# See 'test.python.pulse.test_reference.TestReference.test_appending_child_block'
for ref in _get_references(block._blocks):
self.references[ref.ref_keys] = block.references[ref.ref_keys]
else:
# Avoid using dict.update and explicitly call __set_item__ for validation.
# Reference manager of appended block is cleared because of data reduction.
for ref_keys, ref in block._reference_manager.items():
self.references[ref_keys] = ref
block._reference_manager.clear()
# Now switch the parent because block is appended to self.
block._parent = self
self._blocks.append(block)
self._parameter_manager.update_parameter_table(block)
return self
[documentos] def filter(
self,
*filter_funcs: List[Callable],
channels: Optional[Iterable[Channel]] = None,
instruction_types: Union[Iterable[abc.ABCMeta], abc.ABCMeta] = None,
check_subroutine: bool = True,
):
"""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``, 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.
Args:
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]``.
check_subroutine: Set `True` to individually filter instructions inside a subroutine
defined by the :py:class:`~qiskit.pulse.instructions.Call` instruction.
Returns:
``ScheduleBlock`` consisting of instructions that matches with filtering condition.
"""
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
)
[documentos] def exclude(
self,
*filter_funcs: List[Callable],
channels: Optional[Iterable[Channel]] = None,
instruction_types: Union[Iterable[abc.ABCMeta], abc.ABCMeta] = None,
check_subroutine: bool = True,
):
"""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 in terms of instructions included.
.. 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 ``Instruction`` and return a bool.
channels: For example, ``[DriveChannel(0), AcquireChannel(0)]``.
instruction_types: For example, ``[PulseInstruction, AcquireInstruction]``.
check_subroutine: Set `True` to individually filter instructions inside of a subroutine
defined by the :py:class:`~qiskit.pulse.instructions.Call` instruction.
Returns:
``ScheduleBlock`` consisting of instructions that do not match with
at least one of filtering conditions.
"""
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
)
[documentos] def replace(
self,
old: "BlockComponent",
new: "BlockComponent",
inplace: bool = True,
) -> "ScheduleBlock":
"""Return a ``ScheduleBlock`` with the ``old`` component replaced with a ``new``
component.
Args:
old: Schedule block component to replace.
new: Schedule block component to replace with.
inplace: Replace instruction by mutably modifying this ``ScheduleBlock``.
Returns:
The modified schedule block with ``old`` replaced by ``new``.
"""
if not inplace:
schedule = copy.deepcopy(self)
return schedule.replace(old, new, inplace=True)
if old not in self._blocks:
# Avoid unnecessary update of reference and parameter manager
return self
# Temporarily copies references
all_references = ReferenceManager()
if isinstance(new, ScheduleBlock):
new = copy.deepcopy(new)
all_references.update(new.references)
new._reference_manager.clear()
new._parent = self
for ref_key, subroutine in self.references.items():
if ref_key in all_references:
warnings.warn(
f"Reference {ref_key} conflicts with substituted program {new.name}. "
"Existing reference has been replaced with new reference.",
UserWarning,
)
continue
all_references[ref_key] = subroutine
# Regenerate parameter table by regenerating elements.
# Note that removal of parameters in old is not sufficient,
# because corresponding parameters might be also used in another block element.
self._parameter_manager.clear()
self._parameter_manager.update_parameter_table(self._alignment_context)
new_elms = []
for elm in self._blocks:
if elm == old:
elm = new
self._parameter_manager.update_parameter_table(elm)
new_elms.append(elm)
self._blocks = new_elms
# Regenerate reference table
# Note that reference is attached to the outer schedule if nested.
# Thus, this investigates all references within the scope.
self.references.clear()
root = self
while root._parent is not None:
root = root._parent
for ref in _get_references(root._blocks):
self.references[ref.ref_keys] = all_references[ref.ref_keys]
return self
[documentos] def is_parameterized(self) -> bool:
"""Return True iff the instruction is parameterized."""
return any(self.parameters)
[documentos] def is_referenced(self) -> bool:
"""Return True iff the current schedule block contains reference to subroutine."""
return len(self.references) > 0
[documentos] def assign_parameters(
self,
value_dict: Dict[ParameterExpression, ParameterValueType],
inplace: bool = True,
) -> "ScheduleBlock":
"""Assign the parameters in this schedule according to the input.
Args:
value_dict: A mapping from Parameters to either numeric values or another
Parameter expression.
inplace: Set ``True`` to override this instance with new parameter.
Returns:
Schedule with updated parameters.
Raises:
PulseError: When the block is nested into another block.
"""
if not inplace:
new_schedule = copy.deepcopy(self)
return new_schedule.assign_parameters(value_dict, inplace=True)
# Update parameters in the current scope
self._parameter_manager.assign_parameters(pulse_program=self, value_dict=value_dict)
for subroutine in self._reference_manager.values():
# Also assigning parameters to the references associated with self.
# Note that references are always stored in the root program.
# So calling assign_parameters from nested block doesn't update references.
if subroutine is None:
continue
subroutine.assign_parameters(value_dict=value_dict, inplace=True)
return self
[documentos] def assign_references(
self,
subroutine_dict: Dict[Union[str, Tuple[str, ...]], "ScheduleBlock"],
inplace: bool = True,
) -> "ScheduleBlock":
"""Assign schedules to references.
It is only capable of assigning a schedule block to immediate references
which are directly referred within the current scope.
Let's see following example:
.. code-block:: python
from qiskit import pulse
with pulse.build() as subroutine:
pulse.delay(10, pulse.DriveChannel(0))
with pulse.build() as sub_prog:
pulse.reference("A")
with pulse.build() as main_prog:
pulse.reference("B")
In above example, the ``main_prog`` can refer to the subroutine "root::B" and the
reference of "B" to program "A", i.e., "B::A", is not defined in the root namespace.
This prevents breaking the reference "root::B::A" by the assignment of "root::B".
For example, if a user could indirectly assign "root::B::A" from the root program,
one can later assign another program to "root::B" that doesn't contain "A" within it.
In this situation, a reference "root::B::A" would still live in
the reference manager of the root.
However, the subroutine "root::B::A" would no longer be used in the actual pulse program.
To assign subroutine "A" to ``nested_prog`` as a nested subprogram of ``main_prog``,
you must first assign "A" of the ``sub_prog``,
and then assign the ``sub_prog`` to the ``main_prog``.
.. code-block:: python
sub_prog.assign_references({("A", ): nested_prog}, inplace=True)
main_prog.assign_references({("B", ): sub_prog}, inplace=True)
Alternatively, you can also write
.. code-block:: python
main_prog.assign_references({("B", ): sub_prog}, inplace=True)
main_prog.references[("B", )].assign_references({"A": nested_prog}, inplace=True)
Here :attr:`.references` returns a dict-like object, and you can
mutably update the nested reference of the particular subroutine.
.. note::
Assigned programs are deep-copied to prevent an unexpected update.
Args:
subroutine_dict: A mapping from reference key to schedule block of the subroutine.
inplace: Set ``True`` to override this instance with new subroutine.
Returns:
Schedule block with assigned subroutine.
Raises:
PulseError: When reference key is not defined in the current scope.
"""
if not inplace:
new_schedule = copy.deepcopy(self)
return new_schedule.assign_references(subroutine_dict, inplace=True)
for key, subroutine in subroutine_dict.items():
if key not in self.references:
unassigned_keys = ", ".join(map(repr, self.references.unassigned()))
raise PulseError(
f"Reference instruction with {key} doesn't exist "
f"in the current scope: {unassigned_keys}"
)
self.references[key] = copy.deepcopy(subroutine)
return self
[documentos] def get_parameters(self, parameter_name: str) -> List[Parameter]:
"""Get parameter object bound to this schedule by string name.
Note that we can define different parameter objects with the same name,
because these different objects are identified by their unique uuid.
For example,
.. code-block:: python
from qiskit import pulse, circuit
amp1 = circuit.Parameter("amp")
amp2 = circuit.Parameter("amp")
with pulse.build() as sub_prog:
pulse.play(pulse.Constant(100, amp1), pulse.DriveChannel(0))
with pulse.build() as main_prog:
pulse.call(sub_prog, name="sub")
pulse.play(pulse.Constant(100, amp2), pulse.DriveChannel(0))
main_prog.get_parameters("amp")
This returns a list of two parameters ``amp1`` and ``amp2``.
Args:
parameter_name: Name of parameter.
Returns:
Parameter objects that have corresponding name.
"""
matched = [p for p in self.parameters if p.name == parameter_name]
return matched
[documentos] def search_parameters(self, parameter_regex: str) -> List[Parameter]:
"""Search parameter with regular expression.
This method looks for the scope-aware parameters.
For example,
.. code-block:: python
from qiskit import pulse, circuit
amp1 = circuit.Parameter("amp")
amp2 = circuit.Parameter("amp")
with pulse.build() as sub_prog:
pulse.play(pulse.Constant(100, amp1), pulse.DriveChannel(0))
with pulse.build() as main_prog:
pulse.call(sub_prog, name="sub")
pulse.play(pulse.Constant(100, amp2), pulse.DriveChannel(0))
main_prog.search_parameters("root::sub::amp")
This finds ``amp1`` with scoped name "root::sub::amp".
Args:
parameter_regex: Regular expression for scoped parameter name.
Returns:
Parameter objects that have corresponding name.
"""
pattern = re.compile(parameter_regex)
return sorted(
_collect_scoped_parameters(self, current_scope="root", filter_regex=pattern).values(),
key=lambda p: p.name,
)
def __len__(self) -> int:
"""Return number of instructions in the schedule."""
return len(self.blocks)
def __eq__(self, other: "ScheduleBlock") -> bool:
"""Test if two ScheduleBlocks are equal.
Equality is checked by verifying there is an equal instruction at every time
in ``other`` for every instruction in this ``ScheduleBlock``. This check is
performed by converting the instruction representation into directed acyclic graph,
in which execution order of every instruction is evaluated correctly across all channels.
Also ``self`` and ``other`` should have the same alignment context.
.. warning::
This does not check for logical equivalency. Ie.,
```python
>>> Delay(10, DriveChannel(0)) + Delay(10, DriveChannel(0))
== Delay(20, DriveChannel(0))
False
```
"""
# 0. type check
if not isinstance(other, type(self)):
return False
# 1. transformation check
if self.alignment_context != other.alignment_context:
return False
# 2. size check
if len(self) != len(other):
return False
# 3. instruction check with alignment
from qiskit.pulse.transforms.dag import block_to_dag as dag
if not rx.is_isomorphic_node_match(dag(self), dag(other), lambda x, y: x == y):
return False
return True
def __repr__(self) -> str:
name = format(self._name) if self._name else ""
blocks = ", ".join([repr(instr) for instr in self.blocks[:50]])
if len(self.blocks) > 25:
blocks += ", ..."
return '{}({}, name="{}", transform={})'.format(
self.__class__.__name__, blocks, name, repr(self.alignment_context)
)
def __add__(self, other: "BlockComponent") -> "ScheduleBlock":
"""Return a new schedule with ``other`` inserted within ``self`` at ``start_time``."""
return self.append(other)
def _common_method(*classes):
"""A function decorator to attach the function to specified classes as a method.
.. note:: For developer: A method attached through this decorator may hurt readability
of the codebase, because the method may not be detected by a code editor.
Thus, this decorator should be used to a limited extent, i.e. huge helper method.
By using this decorator wisely, we can reduce code maintenance overhead without
losing readability of the codebase.
"""
def decorator(method):
@functools.wraps(method)
def wrapper(*args, **kwargs):
return method(*args, **kwargs)
for cls in classes:
setattr(cls, method.__name__, wrapper)
return method
return decorator
@_common_method(Schedule, ScheduleBlock)
def draw(
self,
style: Optional[Dict[str, Any]] = None,
backend=None, # importing backend causes cyclic import
time_range: Optional[Tuple[int, int]] = None,
time_unit: str = "dt",
disable_channels: Optional[List[Channel]] = None,
show_snapshot: bool = True,
show_framechange: bool = True,
show_waveform_info: bool = True,
show_barrier: bool = True,
plotter: str = "mpl2d",
axis: Optional[Any] = None,
):
"""Plot the schedule.
Args:
style: Stylesheet options. This can be dictionary or preset stylesheet classes. See
:py:class:`~qiskit.visualization.pulse_v2.stylesheets.IQXStandard`,
:py:class:`~qiskit.visualization.pulse_v2.stylesheets.IQXSimple`, and
:py:class:`~qiskit.visualization.pulse_v2.stylesheets.IQXDebugging` for details of
preset stylesheets.
backend (Optional[BaseBackend]): Backend object to play the input pulse program.
If provided, the plotter may use to make the visualization hardware aware.
time_range: Set horizontal axis limit. Tuple `(tmin, tmax)`.
time_unit: The unit of specified time range either `dt` or `ns`.
The unit of `ns` is available only when `backend` object is provided.
disable_channels: A control property to show specific pulse channel.
Pulse channel instances provided as a list are not shown in the output image.
show_snapshot: Show snapshot instructions.
show_framechange: Show frame change instructions. The frame change represents
instructions that modulate phase or frequency of pulse channels.
show_waveform_info: Show additional information about waveforms such as their name.
show_barrier: Show barrier lines.
plotter: Name of plotter API to generate an output image.
One of following APIs should be specified::
mpl2d: Matplotlib API for 2D image generation.
Matplotlib API to generate 2D image. Charts are placed along y axis with
vertical offset. This API takes matplotlib.axes.Axes as ``axis`` input.
``axis`` and ``style`` kwargs may depend on the plotter.
axis: Arbitrary object passed to the plotter. If this object is provided,
the plotters use a given ``axis`` instead of internally initializing
a figure object. This object format depends on the plotter.
See plotter argument for details.
Returns:
Visualization output data.
The returned data type depends on the ``plotter``.
If matplotlib family is specified, this will be a ``matplotlib.pyplot.Figure`` data.
"""
# pylint: disable=cyclic-import
from qiskit.visualization import pulse_drawer
return pulse_drawer(
program=self,
style=style,
backend=backend,
time_range=time_range,
time_unit=time_unit,
disable_channels=disable_channels,
show_snapshot=show_snapshot,
show_framechange=show_framechange,
show_waveform_info=show_waveform_info,
show_barrier=show_barrier,
plotter=plotter,
axis=axis,
)
def _interval_index(intervals: List[Interval], interval: Interval) -> int:
"""Find the index of an interval.
Args:
intervals: A sorted list of non-overlapping Intervals.
interval: The interval for which the index into intervals will be found.
Returns:
The index of the interval.
Raises:
PulseError: If the interval does not exist.
"""
index = _locate_interval_index(intervals, interval)
found_interval = intervals[index]
if found_interval != interval:
raise PulseError(f"The interval: {interval} does not exist in intervals: {intervals}")
return index
def _locate_interval_index(intervals: List[Interval], interval: Interval, index: int = 0) -> int:
"""Using binary search on start times, find an interval.
Args:
intervals: A sorted list of non-overlapping Intervals.
interval: The interval for which the index into intervals will be found.
index: A running tally of the index, for recursion. The user should not pass a value.
Returns:
The index into intervals that new_interval would be inserted to maintain
a sorted list of intervals.
"""
if not intervals or len(intervals) == 1:
return index
mid_idx = len(intervals) // 2
mid = intervals[mid_idx]
if interval[1] <= mid[0] and (interval != mid):
return _locate_interval_index(intervals[:mid_idx], interval, index=index)
else:
return _locate_interval_index(intervals[mid_idx:], interval, index=index + mid_idx)
def _find_insertion_index(intervals: List[Interval], new_interval: Interval) -> int:
"""Using binary search on start times, return the index into `intervals` where the new interval
belongs, or raise an error if the new interval overlaps with any existing ones.
Args:
intervals: A sorted list of non-overlapping Intervals.
new_interval: The interval for which the index into intervals will be found.
Returns:
The index into intervals that new_interval should be inserted to maintain a sorted list
of intervals.
Raises:
PulseError: If new_interval overlaps with the given intervals.
"""
index = _locate_interval_index(intervals, new_interval)
if index < len(intervals):
if _overlaps(intervals[index], new_interval):
raise PulseError("New interval overlaps with existing.")
return index if new_interval[1] <= intervals[index][0] else index + 1
return index
def _overlaps(first: Interval, second: Interval) -> bool:
"""Return True iff first and second overlap.
Note: first.stop may equal second.start, since Interval stop times are exclusive.
"""
if first[0] == second[0] == second[1]:
# They fail to overlap if one of the intervals has duration 0
return False
if first[0] > second[0]:
first, second = second, first
return second[0] < first[1]
def _check_nonnegative_timeslot(timeslots: TimeSlots):
"""Test that a channel has no negative timeslots.
Raises:
PulseError: If a channel timeslot is negative.
"""
for chan, chan_timeslots in timeslots.items():
if chan_timeslots:
if chan_timeslots[0][0] < 0:
raise PulseError(f"An instruction on {chan} has a negative starting time.")
def _get_timeslots(schedule: "ScheduleComponent") -> TimeSlots:
"""Generate timeslots from given schedule component.
Args:
schedule: Input schedule component.
Raises:
PulseError: When invalid schedule type is specified.
"""
if isinstance(schedule, Instruction):
duration = schedule.duration
instruction_duration_validation(duration)
timeslots = {channel: [(0, duration)] for channel in schedule.channels}
elif isinstance(schedule, Schedule):
timeslots = schedule.timeslots
else:
raise PulseError(f"Invalid schedule type {type(schedule)} is specified.")
return timeslots
def _get_references(block_elms: List["BlockComponent"]) -> Set[Reference]:
"""Recursively get reference instructions in the current scope.
Args:
block_elms: List of schedule block elements to investigate.
Returns:
A set of unique reference instructions.
"""
references = set()
for elm in block_elms:
if isinstance(elm, ScheduleBlock):
references |= _get_references(elm._blocks)
elif isinstance(elm, Reference):
references.add(elm)
return references
def _collect_scoped_parameters(
schedule: ScheduleBlock,
current_scope: str,
filter_regex: Optional[re.Pattern] = None,
) -> Dict[Tuple[str, int], Parameter]:
"""A helper function to collect parameters from all references in scope-aware fashion.
Parameter object is renamed with attached scope information but its UUID is remained.
This means object is treated identically on the assignment logic.
This function returns a dictionary of all parameters existing in the target program
including its reference, which is keyed on the unique identifier consisting of
scoped parameter name and parameter object UUID.
This logic prevents parameter clash in the different scope.
For example, when two parameter objects with the same UUID exist in different references,
both of them appear in the output dictionary, even though they are technically the same object.
This feature is particularly convenient to search parameter object with associated scope.
Args:
schedule: Schedule to get parameters.
current_scope: Name of scope where schedule exist.
filter_regex: Optional. Compiled regex to sort parameter by name.
Returns:
A dictionary of scoped parameter objects.
"""
parameters_out = {}
for param in schedule._parameter_manager.parameters:
new_name = f"{current_scope}{Reference.scope_delimiter}{param.name}"
if filter_regex and not re.search(filter_regex, new_name):
continue
scoped_param = Parameter.__new__(Parameter, new_name, uuid=getattr(param, "_uuid"))
scoped_param.__init__(new_name)
unique_key = new_name, hash(param)
parameters_out[unique_key] = scoped_param
for sub_namespace, subroutine in schedule.references.items():
if subroutine is None:
continue
composite_key = Reference.key_delimiter.join(sub_namespace)
full_path = f"{current_scope}{Reference.scope_delimiter}{composite_key}"
sub_parameters = _collect_scoped_parameters(
subroutine, current_scope=full_path, filter_regex=filter_regex
)
parameters_out.update(sub_parameters)
return parameters_out
# These type aliases are defined at the bottom of the file, because as of 2022-01-18 they are
# imported into other parts of Terra. Previously, the aliases were at the top of the file and used
# forwards references within themselves. This was fine within the same file, but causes scoping
# issues when the aliases are imported into different scopes, in which the `ForwardRef` instances
# would no longer resolve. Instead, we only use forward references in the annotations of _this_
# file to reference the aliases, which are guaranteed to resolve in scope, so the aliases can all be
# concrete.
ScheduleComponent = Union[Schedule, Instruction]
"""An element that composes a pulse schedule."""
BlockComponent = Union[ScheduleBlock, Instruction]
"""An element that composes a pulse schedule block."""