C贸digo fuente para qiskit.pulse.transforms.alignments

# This code is part of Qiskit.
#
# (C) Copyright IBM 2021.
#
# This code is licensed under the Apache License, Version 2.0. You may
# obtain a copy of this license in the LICENSE.txt file in the root directory
# of this source tree or at http://www.apache.org/licenses/LICENSE-2.0.
#
# Any modifications or derivative works of this code must retain this
# copyright notice, and modified files need to carry a notice indicating
# that they have been altered from the originals.
"""A collection of passes to reallocate the timeslots of instructions according to context."""

import abc
from typing import Callable, Dict, Any, Union, Tuple

import numpy as np

from qiskit.circuit.parameterexpression import ParameterExpression, ParameterValueType
from qiskit.pulse.exceptions import PulseError
from qiskit.pulse.schedule import Schedule, ScheduleComponent
from qiskit.pulse.utils import instruction_duration_validation
from qiskit.utils.deprecation import deprecate_func


[documentos]class AlignmentKind(abc.ABC): """An abstract class for schedule alignment.""" def __init__(self, context_params: Tuple[ParameterValueType, ...]): """Create new context.""" self._context_params = tuple(context_params) @abc.abstractmethod def align(self, schedule: Schedule) -> Schedule: """Reallocate instructions according to the policy. Only top-level sub-schedules are aligned. If sub-schedules are nested, nested schedules are not recursively aligned. Args: schedule: Schedule to align. Returns: Schedule with reallocated instructions. """ pass @deprecate_func(since="0.21") def to_dict(self) -> Dict[str, Any]: """Returns dictionary to represent this alignment.""" return {"alignment": self.__class__.__name__} @property @abc.abstractmethod def is_sequential(self) -> bool: """Return ``True`` if this is sequential alignment context. This information is used to evaluate DAG equivalency of two :class:`.ScheduleBlock`s. When the context has two pulses in different channels, a sequential context subtype intends to return following scheduling outcome. .. parsed-literal:: 鈹屸攢鈹鈹鈹鈹鈹鈹鈹鈹 D0: 鈹 pulse1 鈹溾攢鈹鈹鈹鈹鈹鈹鈹鈹鈹鈹鈹 鈹斺攢鈹鈹鈹鈹鈹鈹鈹鈹 鈹屸攢鈹鈹鈹鈹鈹鈹鈹鈹 D1: 鈹鈹鈹鈹鈹鈹鈹鈹鈹鈹鈹鈹鈹 pulse2 鈹 鈹斺攢鈹鈹鈹鈹鈹鈹鈹鈹 On the other hand, parallel context with ``is_sequential=False`` returns .. parsed-literal:: 鈹屸攢鈹鈹鈹鈹鈹鈹鈹鈹 D0: 鈹 pulse1 鈹 鈹溾攢鈹鈹鈹鈹鈹鈹鈹鈹 D1: 鈹 pulse2 鈹 鈹斺攢鈹鈹鈹鈹鈹鈹鈹鈹 All subclasses must implement this method according to scheduling strategy. """ pass def __eq__(self, other): """Check equality of two transforms.""" if type(self) is not type(other): return False if self._context_params != other._context_params: return False return True def __repr__(self): return f"{self.__class__.__name__}({', '.join(self._context_params)})"
[documentos]class AlignLeft(AlignmentKind): """Align instructions in as-soon-as-possible manner. Instructions are placed at earliest available timeslots. """ def __init__(self): """Create new left-justified context.""" super().__init__(context_params=()) @property def is_sequential(self) -> bool: return False
[documentos] def align(self, schedule: Schedule) -> Schedule: """Reallocate instructions according to the policy. Only top-level sub-schedules are aligned. If sub-schedules are nested, nested schedules are not recursively aligned. Args: schedule: Schedule to align. Returns: Schedule with reallocated instructions. """ aligned = Schedule.initialize_from(schedule) for _, child in schedule.children: self._push_left_append(aligned, child) return aligned
@staticmethod def _push_left_append(this: Schedule, other: ScheduleComponent) -> Schedule: """Return ``this`` with ``other`` inserted at the maximum time over all channels shared between ```this`` and ``other``. Args: this: Input schedule to which ``other`` will be inserted. other: Other schedule to insert. Returns: Push left appended schedule. """ this_channels = set(this.channels) other_channels = set(other.channels) shared_channels = list(this_channels & other_channels) ch_slacks = [ this.stop_time - this.ch_stop_time(channel) + other.ch_start_time(channel) for channel in shared_channels ] if ch_slacks: slack_chan = shared_channels[np.argmin(ch_slacks)] shared_insert_time = this.ch_stop_time(slack_chan) - other.ch_start_time(slack_chan) else: shared_insert_time = 0 # Handle case where channels not common to both might actually start # after ``this`` has finished. other_only_insert_time = other.ch_start_time(*(other_channels - this_channels)) # Choose whichever is greatest. insert_time = max(shared_insert_time, other_only_insert_time) return this.insert(insert_time, other, inplace=True)
[documentos]class AlignRight(AlignmentKind): """Align instructions in as-late-as-possible manner. Instructions are placed at latest available timeslots. """ def __init__(self): """Create new right-justified context.""" super().__init__(context_params=()) @property def is_sequential(self) -> bool: return False
[documentos] def align(self, schedule: Schedule) -> Schedule: """Reallocate instructions according to the policy. Only top-level sub-schedules are aligned. If sub-schedules are nested, nested schedules are not recursively aligned. Args: schedule: Schedule to align. Returns: Schedule with reallocated instructions. """ aligned = Schedule.initialize_from(schedule) for _, child in reversed(schedule.children): aligned = self._push_right_prepend(aligned, child) return aligned
@staticmethod def _push_right_prepend(this: ScheduleComponent, other: ScheduleComponent) -> Schedule: """Return ``this`` with ``other`` inserted at the latest possible time such that ``other`` ends before it overlaps with any of ``this``. If required ``this`` is shifted to start late enough so that there is room to insert ``other``. Args: this: Input schedule to which ``other`` will be inserted. other: Other schedule to insert. Returns: Push right prepended schedule. """ this_channels = set(this.channels) other_channels = set(other.channels) shared_channels = list(this_channels & other_channels) ch_slacks = [ this.ch_start_time(channel) - other.ch_stop_time(channel) for channel in shared_channels ] if ch_slacks: insert_time = min(ch_slacks) + other.start_time else: insert_time = this.stop_time - other.stop_time + other.start_time if insert_time < 0: this.shift(-insert_time, inplace=True) this.insert(0, other, inplace=True) else: this.insert(insert_time, other, inplace=True) return this
[documentos]class AlignSequential(AlignmentKind): """Align instructions sequentially. Instructions played on different channels are also arranged in a sequence. No buffer time is inserted in between instructions. """ def __init__(self): """Create new sequential context.""" super().__init__(context_params=()) @property def is_sequential(self) -> bool: return True
[documentos] def align(self, schedule: Schedule) -> Schedule: """Reallocate instructions according to the policy. Only top-level sub-schedules are aligned. If sub-schedules are nested, nested schedules are not recursively aligned. Args: schedule: Schedule to align. Returns: Schedule with reallocated instructions. """ aligned = Schedule.initialize_from(schedule) for _, child in schedule.children: aligned.insert(aligned.duration, child, inplace=True) return aligned
[documentos]class AlignEquispaced(AlignmentKind): """Align instructions with equispaced interval within a specified duration. Instructions played on different channels are also arranged in a sequence. This alignment is convenient to create dynamical decoupling sequences such as PDD. """ def __init__(self, duration: Union[int, ParameterExpression]): """Create new equispaced context. Args: duration: Duration of this context. This should be larger than the schedule duration. If the specified duration is shorter than the schedule duration, no alignment is performed and the input schedule is just returned. This duration can be parametrized. """ super().__init__(context_params=(duration,)) @property def is_sequential(self) -> bool: return True @property def duration(self): """Return context duration.""" return self._context_params[0]
[documentos] def align(self, schedule: Schedule) -> Schedule: """Reallocate instructions according to the policy. Only top-level sub-schedules are aligned. If sub-schedules are nested, nested schedules are not recursively aligned. Args: schedule: Schedule to align. Returns: Schedule with reallocated instructions. """ instruction_duration_validation(self.duration) total_duration = sum(child.duration for _, child in schedule.children) if self.duration < total_duration: return schedule total_delay = self.duration - total_duration if len(schedule.children) > 1: # Calculate the interval in between sub-schedules. # If the duration cannot be divided by the number of sub-schedules, # the modulo is appended and prepended to the input schedule. interval, mod = np.divmod(total_delay, len(schedule.children) - 1) else: interval = 0 mod = total_delay # Calculate pre schedule delay delay, mod = np.divmod(mod, 2) aligned = Schedule.initialize_from(schedule) # Insert sub-schedules with interval _t0 = int(aligned.stop_time + delay + mod) for _, child in schedule.children: aligned.insert(_t0, child, inplace=True) _t0 = int(aligned.stop_time + interval) return aligned
[documentos] @deprecate_func(since="0.21") def to_dict(self) -> Dict[str, Any]: """Returns dictionary to represent this alignment.""" return {"alignment": self.__class__.__name__, "duration": self.duration}
[documentos]class AlignFunc(AlignmentKind): """Allocate instructions at position specified by callback function. The position is specified for each instruction of index ``j`` as a fractional coordinate in [0, 1] within the specified duration. Instructions played on different channels are also arranged in a sequence. This alignment is convenient to create dynamical decoupling sequences such as UDD. For example, UDD sequence with 10 pulses can be specified with following function. .. code-block:: python def udd10_pos(j): return np.sin(np.pi*j/(2*10 + 2))**2 .. note:: This context cannot be QPY serialized because of the callable. If you use this context, your program cannot be saved in QPY format. """ def __init__(self, duration: Union[int, ParameterExpression], func: Callable): """Create new equispaced context. Args: duration: Duration of this context. This should be larger than the schedule duration. If the specified duration is shorter than the schedule duration, no alignment is performed and the input schedule is just returned. This duration can be parametrized. func: A function that takes an index of sub-schedule and returns the fractional coordinate of of that sub-schedule. The returned value should be defined within [0, 1]. The pulse index starts from 1. """ super().__init__(context_params=(duration, func)) @property def is_sequential(self) -> bool: return True @property def duration(self): """Return context duration.""" return self._context_params[0] @property def func(self): """Return context alignment function.""" return self._context_params[1]
[documentos] def align(self, schedule: Schedule) -> Schedule: """Reallocate instructions according to the policy. Only top-level sub-schedules are aligned. If sub-schedules are nested, nested schedules are not recursively aligned. Args: schedule: Schedule to align. Returns: Schedule with reallocated instructions. """ instruction_duration_validation(self.duration) if self.duration < schedule.duration: return schedule aligned = Schedule.initialize_from(schedule) for ind, (_, child) in enumerate(schedule.children): _t_center = self.duration * self.func(ind + 1) _t0 = int(_t_center - 0.5 * child.duration) if _t0 < 0 or _t0 > self.duration: raise PulseError( "Invalid schedule position t=%d is specified at index=%d" % (_t0, ind) ) aligned.insert(_t0, child, inplace=True) return aligned
[documentos] @deprecate_func(since="0.21") def to_dict(self) -> Dict[str, Any]: """Returns dictionary to represent this alignment. .. note:: ``func`` is not presented in this dictionary. Just name. """ return { "alignment": self.__class__.__name__, "duration": self.duration, "func": self.func.__name__, }