scnc
/
qiskit
mirror of https://github.com/Qiskit/qiskit.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Move pulse utils to reschedule (#3154) 

* Move pulse utils, which are Schedule -> Schedule functions, to a reschedule file

* Raise helpful warning

* Style

* Ignore style warning for unused imports, because the functions aren't being called, they are only raising warnings for the user
This commit is contained in:
Lauren Capelluto 2019-09-27 12:07:24 -04:00 committed by Thomas Alexander
parent 5c5a1d04b0
commit 0b197d1fe4
3 changed files with 184 additions and 139 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

167
qiskit/pulse/reschedule.py Normal file
View File

@ -0,0 +1,167 @@
# -*- coding: utf-8 -*-
# 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.
"""
Basic rescheduling functions which take a Schedule (and possibly some arguments) and return
a new Schedule.
"""
import warnings
from typing import List, Optional, Iterable
import numpy as np
from .channels import Channel, AcquireChannel, MemorySlot
from .cmd_def import CmdDef
from .commands import Acquire, AcquireInstruction, Delay
from .exceptions import PulseError
from .interfaces import ScheduleComponent
from .schedule import Schedule
def align_measures(schedules: Iterable[ScheduleComponent], cmd_def: CmdDef,
cal_gate: str = 'u3', max_calibration_duration: Optional[int] = None,
align_time: Optional[int] = None) -> Schedule:
"""Return new schedules where measurements occur at the same physical time. Minimum measurement
wait time (to allow for calibration pulses) is enforced.
This is only defined for schedules that are acquire-less or acquire-final per channel: a
schedule with pulses or acquires occuring on a channel which has already had an acquire will
throw an error.
Args:
schedules: Collection of schedules to be aligned together
cmd_def: Command definition list
cal_gate: The name of the gate to inspect for the calibration time
max_calibration_duration: If provided, cmd_def and cal_gate will be ignored
align_time: If provided, this will be used as final align time.
Returns:
Schedule
Raises:
PulseError: if an acquire or pulse is encountered on a channel that has already been part
of an acquire, or
if align_time is negative
"""
if align_time is not None and align_time < 0:
raise PulseError("Align time cannot be negative.")
if align_time is None:
# Need time to allow for calibration pulses to be played for result classification
if max_calibration_duration is None:
max_calibration_duration = 0
for qubits in cmd_def.cmd_qubits(cal_gate):
cmd = cmd_def.get(cal_gate, qubits, np.pi, 0, np.pi)
max_calibration_duration = max(cmd.duration, max_calibration_duration)
# Schedule the acquires to be either at the end of the needed calibration time, or when the
# last acquire is scheduled, whichever comes later
align_time = max_calibration_duration
for schedule in schedules:
last_acquire = max([time for time, inst in schedule.instructions
if isinstance(inst, AcquireInstruction)])
align_time = max(align_time, last_acquire)
# Shift acquires according to the new scheduled time
new_schedules = []
for schedule in schedules:
new_schedule = Schedule()
acquired_channels = set()
for time, inst in schedule.instructions:
for chan in inst.channels:
if chan.index in acquired_channels:
raise PulseError("Pulse encountered on channel {0} after acquire on "
"same channel.".format(chan.index))
if isinstance(inst, AcquireInstruction):
if time > align_time:
warnings.warn("You provided an align_time which is scheduling an acquire "
"sooner than it was scheduled for in the original Schedule.")
new_schedule |= inst << align_time
acquired_channels.update({a.index for a in inst.acquires})
else:
new_schedule |= inst << time
new_schedules.append(new_schedule)
return new_schedules
def add_implicit_acquires(schedule: ScheduleComponent, meas_map: List[List[int]]) -> Schedule:
"""Return a new schedule with implicit acquires from the measurement mapping replaced by
explicit ones.
Warning:
Since new acquires are being added, Memory Slots will be set to match the qubit index. This
may overwrite your specification.
Args:
schedule: Schedule to be aligned
meas_map: List of lists of qubits that are measured together
Returns:
Schedule
"""
new_schedule = Schedule(name=schedule.name)
for time, inst in schedule.instructions:
if isinstance(inst, AcquireInstruction):
if any([acq.index != mem.index for acq, mem in zip(inst.acquires, inst.mem_slots)]):
warnings.warn("One of your acquires was mapped to a memory slot which didn't match"
" the qubit index. I'm relabeling them to match.")
cmd = Acquire(inst.duration, inst.command.discriminator, inst.command.kernel)
# Get the label of all qubits that are measured with the qubit(s) in this instruction
existing_qubits = {chan.index for chan in inst.acquires}
all_qubits = []
for sublist in meas_map:
if existing_qubits.intersection(set(sublist)):
all_qubits.extend(sublist)
# Replace the old acquire instruction by a new one explicitly acquiring all qubits in
# the measurement group.
new_schedule |= AcquireInstruction(
cmd,
[AcquireChannel(i) for i in all_qubits],
[MemorySlot(i) for i in all_qubits]) << time
else:
new_schedule |= inst << time
return new_schedule
def pad(schedule: Schedule, channels: Optional[Iterable[Channel]] = None,
until: Optional[int] = None) -> Schedule:
"""Pad the input Schedule with `Delay`s on all unoccupied timeslots until
`schedule.duration` or `until` if not `None`.
Args:
schedule: Schedule to pad.
channels: Channels to pad. Defaults to all channels in
`schedule` if not provided. If the supplied channel is not a member
of `schedule` it will be added.
until: Time to pad until. Defaults to `schedule.duration` if not provided.
Returns:
Schedule: The padded schedule
"""
until = until or schedule.duration
channels = channels or schedule.channels
occupied_channels = schedule.channels
unoccupied_channels = set(channels) - set(occupied_channels)
empty_timeslot_collection = schedule.timeslots.complement(until)
for channel in channels:
for timeslot in empty_timeslot_collection.ch_timeslots(channel):
schedule |= Delay(timeslot.duration)(timeslot.channel).shift(timeslot.start)
for channel in unoccupied_channels:
schedule |= Delay(until)(channel)
return schedule

154
qiskit/pulse/utils.py
View File

@ -17,150 +17,28 @@ Pulse utilities.
"""
import warnings
from typing import List, Optional, Iterable
import numpy as np
from .channels import Channel, AcquireChannel, MemorySlot
from .cmd_def import CmdDef
from .commands import Acquire, AcquireInstruction, Delay
from .exceptions import PulseError
from .interfaces import ScheduleComponent
from .schedule import Schedule
# pylint: disable=unused-argument
def align_measures(schedules: Iterable[ScheduleComponent], cmd_def: CmdDef,
cal_gate: str = 'u3', max_calibration_duration: Optional[int] = None,
align_time: Optional[int] = None) -> Schedule:
"""Return new schedules where measurements occur at the same physical time. Minimum measurement
wait time (to allow for calibration pulses) is enforced.
This is only defined for schedules that are acquire-less or acquire-final per channel: a
schedule with pulses or acquires occuring on a channel which has already had an acquire will
throw an error.
Args:
schedules: Collection of schedules to be aligned together
cmd_def: Command definition list
cal_gate: The name of the gate to inspect for the calibration time
max_calibration_duration: If provided, cmd_def and cal_gate will be ignored
align_time: If provided, this will be used as final align time.
Returns:
Schedule
Raises:
PulseError: if an acquire or pulse is encountered on a channel that has already been part
of an acquire, or
if align_time is negative
def align_measures(schedules, cmd_def, cal_gate, max_calibration_duration=None, align_time=None):
"""
if align_time is not None and align_time < 0:
raise PulseError("Align time cannot be negative.")
if align_time is None:
# Need time to allow for calibration pulses to be played for result classification
if max_calibration_duration is None:
max_calibration_duration = 0
for qubits in cmd_def.cmd_qubits(cal_gate):
cmd = cmd_def.get(cal_gate, qubits, np.pi, 0, np.pi)
max_calibration_duration = max(cmd.duration, max_calibration_duration)
# Schedule the acquires to be either at the end of the needed calibration time, or when the
# last acquire is scheduled, whichever comes later
align_time = max_calibration_duration
for schedule in schedules:
last_acquire = max([time for time, inst in schedule.instructions
if isinstance(inst, AcquireInstruction)])
align_time = max(align_time, last_acquire)
# Shift acquires according to the new scheduled time
new_schedules = []
for schedule in schedules:
new_schedule = Schedule()
acquired_channels = set()
for time, inst in schedule.instructions:
for chan in inst.channels:
if chan.index in acquired_channels:
raise PulseError("Pulse encountered on channel {0} after acquire on "
"same channel.".format(chan.index))
if isinstance(inst, AcquireInstruction):
if time > align_time:
warnings.warn("You provided an align_time which is scheduling an acquire "
"sooner than it was scheduled for in the original Schedule.")
new_schedule |= inst << align_time
acquired_channels.update({a.index for a in inst.acquires})
else:
new_schedule |= inst << time
new_schedules.append(new_schedule)
return new_schedules
def add_implicit_acquires(schedule: ScheduleComponent, meas_map: List[List[int]]) -> Schedule:
"""Return a new schedule with implicit acquires from the measurement mapping replaced by
explicit ones.
Warning:
Since new acquires are being added, Memory Slots will be set to match the qubit index. This
may overwrite your specification.
Args:
schedule: Schedule to be aligned
meas_map: List of lists of qubits that are measured together
Returns:
Schedule
This function has been moved!
"""
new_schedule = Schedule(name=schedule.name)
for time, inst in schedule.instructions:
if isinstance(inst, AcquireInstruction):
if any([acq.index != mem.index for acq, mem in zip(inst.acquires, inst.mem_slots)]):
warnings.warn("One of your acquires was mapped to a memory slot which didn't match"
" the qubit index. I'm relabeling them to match.")
cmd = Acquire(inst.duration, inst.command.discriminator, inst.command.kernel)
# Get the label of all qubits that are measured with the qubit(s) in this instruction
existing_qubits = {chan.index for chan in inst.acquires}
all_qubits = []
for sublist in meas_map:
if existing_qubits.intersection(set(sublist)):
all_qubits.extend(sublist)
# Replace the old acquire instruction by a new one explicitly acquiring all qubits in
# the measurement group.
new_schedule |= AcquireInstruction(
cmd,
[AcquireChannel(i) for i in all_qubits],
[MemorySlot(i) for i in all_qubits]) << time
else:
new_schedule |= inst << time
return new_schedule
warnings.warn("The function `align_measures` has been moved to qiskit.pulse.reschedule. "
"It cannot be invoked from `utils` anymore (this call returns None).")
def pad(schedule: Schedule, channels: Optional[Iterable[Channel]] = None,
until: Optional[int] = None) -> Schedule:
"""Pad the input Schedule with `Delay`s on all unoccupied timeslots until
`schedule.duration` or `until` if not `None`.
Args:
schedule: Schedule to pad.
channels: Channels to pad. Defaults to all channels in
`schedule` if not provided. If the supplied channel is not a member
of `schedule` it will be added.
until: Time to pad until. Defaults to `schedule.duration` if not provided.
Returns:
Schedule: The padded schedule
def add_implicit_acquires(schedule, meas_map):
"""
until = until or schedule.duration
This function has been moved!
"""
warnings.warn("The function `add_implicit_acquires` has been moved to qiskit.pulse.reschedule."
" It cannot be invoked from `utils` anymore (this call returns None).")
channels = channels or schedule.channels
occupied_channels = schedule.channels
unoccupied_channels = set(channels) - set(occupied_channels)
empty_timeslot_collection = schedule.timeslots.complement(until)
for channel in channels:
for timeslot in empty_timeslot_collection.ch_timeslots(channel):
schedule |= Delay(timeslot.duration)(timeslot.channel).shift(timeslot.start)
for channel in unoccupied_channels:
schedule |= Delay(until)(channel)
return schedule
def pad(schedule, channels=None, until=None):
"""
This function has been moved!
"""
warnings.warn("The function `pad` has been moved to qiskit.pulse.reschedule. It cannot be "
"invoked from `utils` anymore (this call returns None).")

2
test/python/pulse/test_utils.py → test/python/pulse/test_reschedule.py
View File

@ -24,7 +24,7 @@ from qiskit.pulse.exceptions import PulseError
from qiskit.test import QiskitTestCase
from qiskit.test.mock import FakeOpenPulse2Q
from qiskit.pulse.utils import add_implicit_acquires, align_measures, pad
from qiskit.pulse.reschedule import add_implicit_acquires, align_measures, pad
class TestAutoMerge(QiskitTestCase):