Quellcode für qiskit.pulse.schedule
# 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.
"""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*.
"""
import abc
import copy
import itertools
import multiprocessing as mp
import sys
from typing import List, Tuple, Iterable, Union, Dict, Callable, Set, Optional
from qiskit.circuit.parameterexpression import ParameterExpression, ParameterValueType
from qiskit.pulse.channels import Channel
from qiskit.pulse.exceptions import PulseError
from qiskit.pulse.interfaces import ScheduleComponent
from qiskit.util import is_main_process
# pylint: disable=missing-return-doc
Interval = Tuple[int, int]
"""An interval type is a tuple of a start time (inclusive) and an end time (exclusive)."""
[Doku]class Schedule(ScheduleComponent):
"""A quantum program *schedule* with exact time constraints for its instructions, operating
over all input signal *channels* and supporting special syntaxes for building.
"""
# Counter for the number of instances in this class.
instances_counter = itertools.count()
# Prefix to use for auto naming.
prefix = 'sched'
[Doku] def __init__(self, *schedules: Union[ScheduleComponent, Tuple[int, ScheduleComponent]],
name: Optional[str] = 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.
"""
if name is None:
name = self.prefix + str(next(self.instances_counter))
if sys.platform != "win32" and not is_main_process():
name += '-{}'.format(mp.current_process().pid)
self._name = name
self._duration = 0
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)
@property
def name(self) -> str:
return self._name
@property
def timeslots(self) -> Dict[Channel, List[Interval]]:
"""Time keeping attribute."""
return self._timeslots
@property
def duration(self) -> int:
return self._duration
@property
def start_time(self) -> int:
return self.ch_start_time(*self.channels)
@property
def stop_time(self) -> int:
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``ScheduleComponent``s of this ``Schedule`` in the
order they were added to the schedule.
Returns:
A tuple, where each element is a two-tuple containing the initial
scheduled time of each ``ScheduleComponent`` and the component
itself.
"""
return tuple(self.__children)
@property
def instructions(self):
"""Get the time-ordered instructions from self.
ReturnType:
Tuple[Tuple[int, Instruction], ...]
"""
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))
[Doku] def ch_duration(self, *channels: List[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)
[Doku] def ch_start_time(self, *channels: List[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
[Doku] def ch_stop_time(self, *channels: List[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)
# pylint: disable=arguments-differ
[Doku] 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
"""
if name is None:
name = self.name
return Schedule((time, self), name=name)
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
# pylint: disable=arguments-differ
[Doku] 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))
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``.
"""
if name is None:
name = self.name
new_sched = Schedule(name=name)
new_sched._mutable_insert(0, self)
new_sched._mutable_insert(start_time, schedule)
return new_sched
# pylint: disable=arguments-differ
[Doku] 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)
[Doku] def flatten(self) -> 'Schedule':
"""Return a new schedule which is the flattened schedule contained all ``instructions``."""
return Schedule(*self.instructions, name=self.name)
[Doku] def filter(self, *filter_funcs: List[Callable],
channels: Optional[Iterable[Channel]] = None,
instruction_types=None,
time_ranges: Optional[Iterable[Tuple[int, int]]] = None,
intervals: Optional[Iterable[Interval]] = None) -> '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, ScheduleComponent) tuple and
return a bool.
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)]``.
"""
composed_filter = self._construct_filter(*filter_funcs,
channels=channels,
instruction_types=instruction_types,
time_ranges=time_ranges,
intervals=intervals)
return self._apply_filter(composed_filter,
new_sched_name="{name}".format(name=self.name))
[Doku] def exclude(self, *filter_funcs: List[Callable],
channels: Optional[Iterable[Channel]] = None,
instruction_types=None,
time_ranges: Optional[Iterable[Tuple[int, int]]] = None,
intervals: Optional[Iterable[Interval]] = None) -> '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 ``self.filter``, so that::
self.filter(args) | self.exclude(args) == self
Args:
filter_funcs: A list of Callables which take a (int, ScheduleComponent) tuple and
return a bool.
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)]``.
"""
composed_filter = self._construct_filter(*filter_funcs,
channels=channels,
instruction_types=instruction_types,
time_ranges=time_ranges,
intervals=intervals)
return self._apply_filter(lambda x: not composed_filter(x),
new_sched_name="{name}".format(name=self.name))
def _apply_filter(self, filter_func: Callable, new_sched_name: str) -> 'Schedule':
"""Return a Schedule containing only the instructions from this Schedule for which
``filter_func`` returns ``True``.
Args:
filter_func: Function of the form (int, ScheduleComponent) -> bool.
new_sched_name: Name of the returned ``Schedule``.
"""
subschedules = self.flatten()._children
valid_subschedules = [sched for sched in subschedules if filter_func(sched)]
return Schedule(*valid_subschedules, name=new_sched_name)
def _construct_filter(self, *filter_funcs: List[Callable],
channels: Optional[Iterable[Channel]] = None,
instruction_types: Optional[Iterable['Instruction']] = None,
time_ranges: Optional[Iterable[Tuple[int, int]]] = None,
intervals: Optional[Iterable[Interval]] = None) -> Callable:
"""Returns a boolean-valued function with input type ``(int, ScheduleComponent)`` that
returns ``True`` iff the input satisfies all of the criteria specified by the arguments;
i.e. 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``.
Args:
filter_funcs: A list of Callables which take a (int, ScheduleComponent) tuple and
return a bool
channels: For example, ``[DriveChannel(0), AcquireChannel(0)]`` or ``DriveChannel(0)``
instruction_types: For example, ``[PulseInstruction, AcquireInstruction]``
or ``DelayInstruction``
time_ranges: For example, ``[(0, 5), (6, 10)]`` or ``(0, 5)``
intervals: For example, ``[Interval(0, 5), Interval(6, 10)]`` or ``Interval(0, 5)``
"""
def if_scalar_cast_to_list(to_list):
try:
iter(to_list)
except TypeError:
to_list = [to_list]
return to_list
def only_channels(channels: Union[Set[Channel], Channel]) -> Callable:
channels = if_scalar_cast_to_list(channels)
def channel_filter(time_inst) -> bool:
"""Filter channel.
Args:
time_inst (Tuple[int, Instruction]): Time
"""
return any([chan in channels for chan in time_inst[1].channels])
return channel_filter
def only_instruction_types(types: Union[Iterable[abc.ABCMeta], abc.ABCMeta]) -> Callable:
types = if_scalar_cast_to_list(types)
def instruction_filter(time_inst) -> bool:
"""Filter instruction.
Args:
time_inst (Tuple[int, Instruction]): Time
"""
return isinstance(time_inst[1], tuple(types))
return instruction_filter
def only_intervals(ranges: Union[Iterable[Interval], Interval]) -> Callable:
ranges = if_scalar_cast_to_list(ranges)
def interval_filter(time_inst) -> bool:
"""Filter interval.
Args:
time_inst (Tuple[int, Instruction]): Time
"""
for i in ranges:
inst_start = time_inst[0]
inst_stop = inst_start + time_inst[1].duration
if i[0] <= inst_start and inst_stop <= i[1]:
return True
return False
return interval_filter
filter_func_list = list(filter_funcs)
if channels is not None:
filter_func_list.append(only_channels(channels))
if instruction_types is not None:
filter_func_list.append(only_instruction_types(instruction_types))
if time_ranges is not None:
filter_func_list.append(only_intervals(time_ranges))
if intervals is not None:
filter_func_list.append(only_intervals(intervals))
# return function returning true iff all filters are passed
return lambda x: all([filter_func(x) for filter_func in filter_func_list])
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 isinstance(time, int):
raise PulseError("Schedule start time must be an integer.")
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(schedule._timeslots[channel])
else:
self._timeslots[channel] = [(i[0] + time, i[1] + time)
for i in schedule._timeslots[channel]]
continue
for idx, interval in enumerate(schedule._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 schedule._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:
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]))
_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(
'The channel {} is not present in the schedule'.format(channel))
channel_timeslots = self._timeslots[channel]
for interval in schedule._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)
[Doku] def replace(self,
old: ScheduleComponent,
new: ScheduleComponent,
inplace: bool = False,
) -> 'Schedule':
"""Return a schedule with the ``old`` instruction replaced with a ``new``
instruction.
The replacment matching is based on an instruction equality check.
.. jupyter-kernel:: python3
:id: replace
.. jupyter-execute::
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::
.. jupyter-execute::
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.
"""
new_children = []
for time, child in self._children:
if child == old:
new_children.append((time, new))
if inplace:
self._replace_timeslots(time, old, new)
else:
new_children.append((time, child))
if inplace:
self.__children = new_children
return self
else:
try:
return Schedule(*new_children)
except PulseError as err:
raise PulseError(
'Replacement of {old} with {new} results in '
'overlapping instructions.'.format(
old=old, new=new)) from err
[Doku] def assign_parameters(self,
value_dict: Dict[ParameterExpression, ParameterValueType],
) -> '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.
Returns:
Schedule with updated parameters (a new one if not inplace, otherwise self).
"""
for _, inst in self.instructions:
inst.assign_parameters(value_dict)
for chan in copy.copy(self._timeslots):
if isinstance(chan.index, ParameterExpression):
chan_timeslots = self._timeslots.pop(chan)
# Find the channel's new assignment
new_channel = chan
for param in chan.index.parameters:
if param in value_dict:
new_index = new_channel.index.assign(param, value_dict[param])
if not new_index.parameters:
new_index = float(new_index)
if float(new_index).is_integer():
new_index = int(new_index)
new_channel = type(chan)(new_index)
# Merge with existing channel
if new_channel in self._timeslots:
sched = Schedule()
sched._timeslots = {new_channel: chan_timeslots}
self._add_timeslots(0, sched)
# Or add back under the new name
else:
self._timeslots[new_channel] = chan_timeslots
return self
[Doku] def draw(self, dt: float = 1, style=None,
filename: Optional[str] = None, interp_method: Optional[Callable] = None,
scale: Optional[float] = None,
channel_scales: Optional[Dict[Channel, float]] = None,
plot_all: bool = False, plot_range: Optional[Tuple[float]] = None,
interactive: bool = False, table: bool = False, label: bool = False,
framechange: bool = True,
channels: Optional[List[Channel]] = None,
show_framechange_channels: bool = True):
r"""Plot the schedule.
Args:
dt: Time interval of samples.
style (Optional[SchedStyle]): A style sheet to configure plot appearance.
filename: Name required to save pulse image.
interp_method: A function for interpolation.
scale: Relative visual scaling of waveform amplitudes, see Additional Information.
channel_scales: Channel independent scaling as a dictionary of ``Channel`` object.
plot_all: Plot empty channels.
plot_range: A tuple of time range to plot.
interactive: When set true show the circuit in a new window
(this depends on the matplotlib backend being used supporting this).
table: Draw event table for supported commands.
label: Label individual instructions.
framechange: Add framechange indicators.
channels: A list of Channels to plot.
show_framechange_channels: Plot channels with only framechanges.
Additional Information:
If you want to manually rescale the waveform amplitude of channels one by one,
you can set ``channel_scales`` argument instead of ``scale``.
The ``channel_scales`` should be given as a python dictionary::
channel_scales = {pulse.DriveChannels(0): 10.0,
pulse.MeasureChannels(0): 5.0}
When the channel to plot is not included in the ``channel_scales`` dictionary,
scaling factor of that channel is overwritten by the value of ``scale`` argument.
In default, waveform amplitude is normalized by the maximum amplitude of the channel.
The scaling factor is displayed under the channel name alias.
Returns:
matplotlib.Figure: A matplotlib figure object of the pulse schedule.
"""
# pylint: disable=invalid-name, cyclic-import
from qiskit import visualization
return visualization.pulse_drawer(self, dt=dt, style=style,
filename=filename, interp_method=interp_method,
scale=scale, channel_scales=channel_scales,
plot_all=plot_all, plot_range=plot_range,
interactive=interactive, table=table, label=label,
framechange=framechange, channels=channels,
show_framechange_channels=show_framechange_channels)
def __eq__(self, other: ScheduleComponent) -> bool:
"""Test if two ScheduleComponents 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
```
"""
channels = set(self.channels)
other_channels = set(other.channels)
# first check channels are the same
if channels != other_channels:
return False
# then verify same number of instructions in each
instructions = self.instructions
other_instructions = other.instructions
if len(instructions) != len(other_instructions):
return False
# finally check each instruction in `other` is in this schedule
for idx, inst in enumerate(other_instructions):
# check assumes `Schedule.instructions` is sorted consistently
if instructions[idx] != inst:
return False
return True
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 __len__(self) -> int:
"""Return number of instructions in the schedule."""
return len(self.instructions)
def __repr__(self):
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 'Schedule({}, name="{}")'.format(instructions, name)
class ParameterizedSchedule:
"""Temporary parameterized schedule class.
This should not be returned to users as it is currently only a helper class.
This class is takes an input command definition that accepts
a set of parameters. Calling ``bind`` on the class will return a ``Schedule``.
# TODO: In the near future this will be replaced with proper incorporation of parameters
into the ``Schedule`` class.
"""
def __init__(self, *schedules, parameters: Optional[Dict[str, Union[float, complex]]] = None,
name: Optional[str] = None):
full_schedules = []
parameterized = []
parameters = parameters or []
self.name = name or ''
# partition schedules into callable and schedules
for schedule in schedules:
if isinstance(schedule, ParameterizedSchedule):
parameterized.append(schedule)
parameters += schedule.parameters
elif callable(schedule):
parameterized.append(schedule)
elif isinstance(schedule, Schedule):
full_schedules.append(schedule)
else:
raise PulseError('Input type: {} not supported'.format(type(schedule)))
self._parameterized = tuple(parameterized)
self._schedules = tuple(full_schedules)
self._parameters = tuple(sorted(set(parameters)))
@property
def parameters(self) -> Tuple[str]:
"""Schedule parameters."""
return self._parameters
def bind_parameters(self,
*args: Union[int, float, complex, ParameterExpression],
**kwargs: Union[int, float, complex, ParameterExpression]) -> Schedule:
"""Generate the Schedule from params to evaluate command expressions"""
bound_schedule = Schedule(name=self.name)
schedules = list(self._schedules)
named_parameters = {}
if args:
for key, val in zip(self.parameters, args):
named_parameters[key] = val
if kwargs:
for key, val in kwargs.items():
if key in self.parameters:
if key not in named_parameters.keys():
named_parameters[key] = val
else:
raise PulseError("%s got multiple values for argument '%s'"
% (self.__class__.__name__, key))
else:
raise PulseError("%s got an unexpected keyword argument '%s'"
% (self.__class__.__name__, key))
for param_sched in self._parameterized:
# recursively call until based callable is reached
if isinstance(param_sched, type(self)):
predefined = param_sched.parameters
else:
# assuming no other parametrized instructions
predefined = self.parameters
sub_params = {k: v for k, v in named_parameters.items() if k in predefined}
schedules.append(param_sched(**sub_params))
# construct evaluated schedules
for sched in schedules:
if isinstance(sched, tuple):
bound_schedule.insert(sched[0], sched[1])
else:
bound_schedule |= sched
return bound_schedule
def __call__(self, *args: Union[int, float, complex, ParameterExpression],
**kwargs: Union[int, float, complex, ParameterExpression]) -> Schedule:
return self.bind_parameters(*args, **kwargs)
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('The interval: {} does not exist in intervals: {}'.format(
interval, 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):
"""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(
"An instruction on {} has a negative "
" starting time.".format(chan))