Source code for 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)."""


[docs]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' 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))
[docs] 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)
[docs] 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
[docs] 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
[docs] 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
[docs] 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
[docs] 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)
[docs] def flatten(self) -> 'Schedule': """Return a new schedule which is the flattened schedule contained all ``instructions``.""" return Schedule(*self.instructions, name=self.name)
[docs] 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))
[docs] 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)
[docs] 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
[docs] 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
[docs] 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)
[docs] 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))