Pulse builder interface (#4174)

* Fix setting contexts.

* Make builder class private.

* Add group transform.

* Initial demo test is running (not necessarily doing the right thing.

* Typed context gate call parameters.

* Fix imports for builder.

* Fix bugs in tests.

* Minor changes.

* Add pad context. Rename active -> current.

* Add barrier test, expected fail.

* Added group context test.

* Fix bug in acquire instruction.

* Add instruction tests.

* Add builder utilities tests.

* Fix bug in scheduler when no gates present in circuit.

* Test transpiler and scheduler context settings / utilities.

* Made children, a public method.

* add default alignment context for builder.

* Get tests working for call_schedule and call_circuit. Rework right align transform.

* Rename test file.

* Test measure macro.

* Add delay_qubit macro test.

* Update context builder test docstring.

* Review suggestions

* Add gate tests. Running into a bug with the schedule implementation.

* Fix get_qubit_channels

* Fix bug in acquire round 2.

* Deprecation warning to use control(qubits: List) instead of control(channel: int)

* Fix mutable insert flattening schedule tree.

* Fix delay test using qubit index.

* Fix using deprecated cnotgate.

* typos

* Rename alignment contexts.

* update the control based on the new backend schema

* typos

* Unify alignment context naming.

* Unify naming of transforms and fix broken align_right.

* More logic fixes

* lint

* lint

* add older code

* More logic fixes

* lint

* Update docstrings and typehints

* Unused imports

* Release notes

* Review suggestions

* Bug fix for get_qubit_channels(0)

Updated tests too

* Review Suggestions

* lint

* Change error message of _get_channel_prefix_index

* Added raise error for backends without 'channel' information

* Update docstrings for raise errors

* Fix error messages and docstrings

* make self.channels private

* Remove collections.Counter

* Revert "Remove collections.Counter"

This reverts commit d602738387.

* Add complex test.

* Rename _schedule_lazy_circuit_before to _compile_lazy_circuit_before

* Add delay_qubits instruction. Fix bug in schedule equality test.

* Add frequency and phase offset contexts.

* Renamed builder_context to builder.

* Add inline directive.

* linting.

* Fix linting errors.

* Add rescheduler linting.

* Add transforms documentation.

* Lint test builder.

* Add macro linting.

* Schedule linting.

* lint pulse builder.

* Remove cyclical imports with compiler import in pulse builder.

* Add channel builder helper functions. Rename current to active.

* Add relative barrier directive.

* Refactor visualization namespacing.

* Add barrier directive to builder.

* Add num_qubits.

* Add measure_all builder command.

* Add acquire on qubit.

* Add error if not all channels supplied to an Instruction is a channel.

* Remove notebooks directory from hackathon.

* linting

* Fix issue with deprecated Delay.

* Add backport packages for contextvars.

* Reorder schedule and backend arguments. Make backend no longer required.

* Small builder doc update.

* Add simpler calling syntax for the builder.

* Fix python versioninign.

* Add __all__ for builder.

* Add exceptions for no active builder or no backend set.

* Update automodule

Adds documentation of all non-private member function and properties

* Remove unwanted docstrings

Also fixes some typos

* Fix pylint error - BackendNotSet not added in docstring

* Fix weird bug

* Fix 'commands' not imported

* Fix typo

* Rename pulse.reschedule -> pulse.transforms

* Update surrounding code to support rename (releasenotes, add rescheduler back for import path deprecation, API documentation, update imports in tests)

* Remove unwanted newlines

* Reapply feedback from code review

* Add test cases for issues with timeslots.'

* Fix small bug in timeslot insertion finder.

* Fixed another error in the code path.

* Remove accidental comment.

* Fix another small timeslot bug.

(cherry picked from commit 28bbbdec13ebd137ca4e5fd1c61aff0c0304988e)

* Fix small issue.

* Moved circuit_scheduler back to scheduler.

* Properly deprecate scheduler utils.

* linting

* Fix missed pieces when moving back scheduler.

* Fix circular imports by using from instead of import

* linting.

* use Constant instead of ConstantPulse.

* Rename pulse.reschedule -> pulse.transforms

* Update surrounding code to support rename (releasenotes, add rescheduler back for import path deprecation, API documentation, update imports in tests)

* Remove unwanted newlines

* Reapply feedback from code review

* Fix docs unexpected indent

* Linting.

* Change deprecation warning to emit warning at proper stack location.

* Make changes to remove deprecation warnings.

* Remove leftover files from moving of scheduler.

* Revert ordering of channel and qubit arguments compared to parameters.

* Add pulse builder documentation.

* fix up builder tests.

* Linting.

* Docs update.

* Fix deprecation warning.

* Fix reno warnings

* Some builder.py docstring updates

* Documentation updates.

* Fix issue with wrapped function not being properly documented.

* Update builder docs to use jupyter-executer

* Builder interface documentation.

* Fix issue with pulse barrier padding.

* Fix builder barrier doc.

* Structure builder imports.

* Update qiskit/pulse/instructions/directives.py

Co-authored-by: Lauren Capelluto <laurencapelluto@gmail.com>

* Update qiskit/pulse/instructions/acquire.py

Co-authored-by: Lauren Capelluto <laurencapelluto@gmail.com>

* Update qiskit/pulse/instructions/directives.py

Co-authored-by: Lauren Capelluto <laurencapelluto@gmail.com>

* Fix call circuit.

* Enable a samples_to_time and time_to_samples method.

* Add test to builder for calling circuit with cregs.

* Update qiskit/pulse/builder.py

Co-authored-by: Lauren Capelluto <laurencapelluto@gmail.com>

* Update qiskit/pulse/builder.py

Co-authored-by: Lauren Capelluto <laurencapelluto@gmail.com>

* Update qiskit/pulse/builder.py

Co-authored-by: Lauren Capelluto <laurencapelluto@gmail.com>

* Revert acquire import changes.

* Ignore trivial barriers.

* Add note on future deprecation of calling gates within the pulse builder interface.

* Remove commented out execution.

* Update qiskit/pulse/transforms.py

Co-authored-by: Lauren Capelluto <laurencapelluto@gmail.com>

* Remove mention of quantum circuit in builder documentation intro.

* Update qiskit/pulse/builder.py

Co-authored-by: Lauren Capelluto <laurencapelluto@gmail.com>

* Fix jupyter execute import style.

* Fix unclear arguments in delay_qubits doc.

* Make pulse builder warning more prominent and revise message.

* Fixed import asthetic.

* Move BackendNotSet to exceptions.

* Fix more asthetic setting of default settings in builder.

* Remove left over assert in builder.

* Remove dead comment.

* Reorder _call_circuit.

* Fix small linting error

* Fix commented out example code.

* Remove comment headers.

* Move NoActiveBuilder exception to exceptions.

* Rename time to sample and vice versa to seconds to samples

* Update macros module docstring.

* Fix measure docstring spacing.

* Update test_builder copyright.

* Fix test docstrings.

* Add notes for expected test failures.

* Rename types sections to channels.

* Fix test spacing.

* Revert test modification.

* Fix test_transform docstring.

* Add better coverage for TestRemoveTrivialBarriers

* Readd transform tests.

* Change example usage to examples for uniformity.

* linting invalid name.

* Replace mutate flag with inplace to mirror the circuit model.

* Allow passing name to pulse.build.

* Fix bad builder example assert.

* Allow frequency and phase offset contexts to broacast across channels.

* Revert delay instruction error.

* Remove unnecessary assignments for inplace schedule operations.

* Reorder sequential transform reference ordering.

* Removed unnecessary removal of barriers in transform test.

* Verify that flattened works correctly by testing aligned on grouped schedule.

* Make align right test depending on nesting. Refactor implementation of align_right.

* Use append instead of append where it makes sense. Use inplace where I can.

* Remove the pulse group instruction for now.

* Added better inlining test.

* Implement shift frequency and frequency_offset.

* Clean up builder tests #1

* Deprecate warns only once.

* Remove append_instruction from pulse namespace.

* Fix bad test name in test_instructions.

* Added beta flag.

* Update flatten transform documentation.

* Operator spaces.

* Fix unnecessary else in schedule.

* Fix operator spacing in test-transforms.

* time to seconds

* Fix misnamed test.

* Revert children to _children.

* Update pulse module docstring

* linting.

* Revert apidocs change in pulse.

* Remove call_schedule, call_circuit, call_gate from top-level import.

* Rename block to context_schedule.

* Better error message.

* Fix module level import.

* revert reno files.

* Added space in test builder.

* Added set_phase instruction.

* Revert visualization changes.

* Remove unused import

* Standardize imports for builder PR.

* Docs updates.

Co-authored-by: SooluThomas <soolu.elto@gmail.com>
Co-authored-by: Lauren Capelluto <laurencapelluto@gmail.com>
Co-authored-by: mergify[bot] <37929162+mergify[bot]@users.noreply.github.com>
This commit is contained in:
Thomas Alexander 2020-07-08 09:24:50 -04:00 committed by GitHub
parent d40911dbc9
commit 5bdee8cf84
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
31 changed files with 4157 additions and 429 deletions

View File

@ -52,7 +52,7 @@ RunConfig
RunConfig
"""
from .assemble_circuits import assemble_circuits
from .assemble_schedules import assemble_schedules
from .disassemble import disassemble
from .run_config import RunConfig
from qiskit.assembler.assemble_circuits import assemble_circuits
from qiskit.assembler.assemble_schedules import assemble_schedules
from qiskit.assembler.disassemble import disassemble
from qiskit.assembler.run_config import RunConfig

View File

@ -17,25 +17,18 @@ from collections import defaultdict
from typing import Any, Dict, List, Tuple
import hashlib
from qiskit import qobj, pulse
from qiskit.assembler.run_config import RunConfig
from qiskit.exceptions import QiskitError
from qiskit.pulse import Schedule, Acquire, Delay, Play, transforms
from qiskit.pulse.pulse_lib import ParametricPulse, SamplePulse
from qiskit.pulse.commands import (Command, PulseInstruction, AcquireInstruction,
DelayInstruction, ParametricInstruction)
from qiskit.qobj import (PulseQobj, QobjHeader, QobjExperimentHeader,
PulseQobjInstruction, PulseQobjExperimentConfig,
PulseQobjExperiment, PulseQobjConfig, PulseLibraryItem)
from qiskit.qobj.converters import InstructionToQobjConverter, LoConfigConverter
from qiskit.pulse import instructions, transforms, pulse_lib, commands
from qiskit.qobj import utils as qobj_utils, converters
from qiskit.qobj.converters.pulse_instruction import ParametricPulseShapes
from qiskit.qobj.utils import MeasLevel, MeasReturnType
from .run_config import RunConfig
def assemble_schedules(schedules: List[Schedule],
def assemble_schedules(schedules: List[pulse.Schedule],
qobj_id: int,
qobj_header: QobjHeader,
run_config: RunConfig) -> PulseQobj:
qobj_header: qobj.QobjHeader,
run_config: RunConfig) -> qobj.PulseQobj:
"""Assembles a list of schedules into a qobj that can be run on the backend.
Args:
@ -55,24 +48,24 @@ def assemble_schedules(schedules: List[Schedule],
if not hasattr(run_config, 'meas_lo_freq'):
raise QiskitError('meas_lo_freq must be supplied.')
lo_converter = LoConfigConverter(PulseQobjExperimentConfig,
**run_config.to_dict())
lo_converter = converters.LoConfigConverter(qobj.PulseQobjExperimentConfig,
**run_config.to_dict())
experiments, experiment_config = _assemble_experiments(schedules,
lo_converter,
run_config)
qobj_config = _assemble_config(lo_converter, experiment_config, run_config)
return PulseQobj(experiments=experiments,
qobj_id=qobj_id,
header=qobj_header,
config=qobj_config)
return qobj.PulseQobj(experiments=experiments,
qobj_id=qobj_id,
header=qobj_header,
config=qobj_config)
def _assemble_experiments(
schedules: List[Schedule],
lo_converter: LoConfigConverter,
schedules: List[pulse.Schedule],
lo_converter: converters.LoConfigConverter,
run_config: RunConfig
) -> Tuple[List[PulseQobjExperiment], Dict[str, Any]]:
) -> Tuple[List[qobj.PulseQobjExperiment], Dict[str, Any]]:
"""Assembles a list of schedules into PulseQobjExperiments, and returns related metadata that
will be assembled into the Qobj configuration.
@ -95,8 +88,11 @@ def _assemble_experiments(
'schedule, or a list of frequencies should be provided for a single '
'frequency sweep schedule.')
instruction_converter = getattr(run_config, 'instruction_converter', InstructionToQobjConverter)
instruction_converter = instruction_converter(PulseQobjInstruction, **run_config.to_dict())
instruction_converter = getattr(run_config,
'instruction_converter',
converters.InstructionToQobjConverter)
instruction_converter = instruction_converter(qobj.PulseQobjInstruction,
**run_config.to_dict())
compressed_schedules = transforms.compress_pulses(schedules)
user_pulselib = {}
@ -109,11 +105,11 @@ def _assemble_experiments(
user_pulselib)
# TODO: add other experimental header items (see circuit assembler)
qobj_experiment_header = QobjExperimentHeader(
qobj_experiment_header = qobj.QobjExperimentHeader(
memory_slots=max_memory_slot + 1, # Memory slots are 0 indexed
name=schedule.name or 'Experiment-%d' % idx)
experiment = PulseQobjExperiment(
experiment = qobj.PulseQobjExperiment(
header=qobj_experiment_header,
instructions=qobj_instructions)
if freq_configs:
@ -129,14 +125,14 @@ def _assemble_experiments(
experiment = experiments[0]
experiments = []
for freq_config in freq_configs:
experiments.append(PulseQobjExperiment(
experiments.append(qobj.PulseQobjExperiment(
header=experiment.header,
instructions=experiment.instructions,
config=freq_config))
# Top level Qobj configuration
experiment_config = {
'pulse_library': [PulseLibraryItem(name=name, samples=samples)
'pulse_library': [qobj.PulseLibraryItem(name=name, samples=samples)
for name, samples in user_pulselib.items()],
'memory_slots': max([exp.header.memory_slots for exp in experiments])
}
@ -145,11 +141,11 @@ def _assemble_experiments(
def _assemble_instructions(
schedule: Schedule,
instruction_converter: InstructionToQobjConverter,
schedule: pulse.Schedule,
instruction_converter: converters.InstructionToQobjConverter,
run_config: RunConfig,
user_pulselib: Dict[str, Command]
) -> Tuple[List[PulseQobjInstruction], int]:
user_pulselib: Dict[str, commands.Command]
) -> Tuple[List[qobj.PulseQobjInstruction], int]:
"""Assembles the instructions in a schedule into a list of PulseQobjInstructions and returns
related metadata that will be assembled into the Qobj configuration. Lookup table for
pulses defined in all experiments are registered in ``user_pulselib``. This object should be
@ -173,28 +169,35 @@ def _assemble_instructions(
acquire_instruction_map = defaultdict(list)
for time, instruction in schedule.instructions:
if isinstance(instruction, ParametricInstruction): # deprecated
instruction = Play(instruction.command, instruction.channels[0], name=instruction.name)
if isinstance(instruction, commands.ParametricInstruction): # deprecated
instruction = instructions.Play(instruction.command,
instruction.channels[0],
name=instruction.name)
if isinstance(instruction, Play) and isinstance(instruction.pulse, ParametricPulse):
if (isinstance(instruction, instructions.Play) and
isinstance(instruction.pulse, pulse_lib.ParametricPulse)):
pulse_shape = ParametricPulseShapes(type(instruction.pulse)).name
if pulse_shape not in run_config.parametric_pulses:
instruction = Play(instruction.pulse.get_sample_pulse(),
instruction.channel,
name=instruction.name)
instruction = instructions.Play(instruction.pulse.get_sample_pulse(),
instruction.channel,
name=instruction.name)
if isinstance(instruction, PulseInstruction): # deprecated
instruction = Play(SamplePulse(name=name, samples=instruction.command.samples),
instruction.channels[0], name=name)
if isinstance(instruction, commands.PulseInstruction): # deprecated
name = instruction.command.name
instruction = instructions.Play(
pulse_lib.SamplePulse(name=name, samples=instruction.command.samples),
instruction.channels[0], name=name)
if isinstance(instruction, Play) and isinstance(instruction.pulse, SamplePulse):
if (isinstance(instruction, instructions.Play) and
isinstance(instruction.pulse, pulse_lib.SamplePulse)):
name = hashlib.sha256(instruction.pulse.samples).hexdigest()
instruction = Play(SamplePulse(name=name, samples=instruction.pulse.samples),
channel=instruction.channel,
name=name)
instruction = instructions.Play(
pulse_lib.SamplePulse(name=name, samples=instruction.pulse.samples),
channel=instruction.channel,
name=name)
user_pulselib[name] = instruction.pulse.samples
if isinstance(instruction, (AcquireInstruction, Acquire)):
if isinstance(instruction, (commands.AcquireInstruction, instructions.Acquire)):
if instruction.mem_slot:
max_memory_slot = max(max_memory_slot, instruction.mem_slot.index)
# Acquires have a single AcquireChannel per inst, but we have to bundle them
@ -202,7 +205,9 @@ def _assemble_instructions(
acquire_instruction_map[(time, instruction.command)].append(instruction)
continue
if isinstance(instruction, (DelayInstruction, Delay)):
if isinstance(instruction, (commands.DelayInstruction,
instructions.Delay,
instructions.Directive)):
# delay instructions are ignored as timing is explicit within qobj
continue
@ -211,17 +216,18 @@ def _assemble_instructions(
if acquire_instruction_map:
if hasattr(run_config, 'meas_map'):
_validate_meas_map(acquire_instruction_map, run_config.meas_map)
for (time, _), instructions in acquire_instruction_map.items():
qubits, mem_slots, reg_slots = _bundle_channel_indices(instructions)
for (time, _), instrs in acquire_instruction_map.items():
qubits, mem_slots, reg_slots = _bundle_channel_indices(instrs)
qobj_instructions.append(
instruction_converter.convert_single_acquires(
time, instructions[0],
time, instrs[0],
qubits=qubits, memory_slot=mem_slots, register_slot=reg_slots))
return qobj_instructions, max_memory_slot
def _validate_meas_map(instruction_map: Dict[Tuple[int, Acquire], List[AcquireInstruction]],
def _validate_meas_map(instruction_map: Dict[Tuple[int, instructions.Acquire],
List[commands.AcquireInstruction]],
meas_map: List[List[int]]) -> None:
"""Validate all qubits tied in ``meas_map`` are to be acquired.
@ -236,9 +242,9 @@ def _validate_meas_map(instruction_map: Dict[Tuple[int, Acquire], List[AcquireIn
meas_map_sets = [set(m) for m in meas_map]
# Check each acquisition time individually
for _, instructions in instruction_map.items():
for _, instrs in instruction_map.items():
measured_qubits = set()
for inst in instructions:
for inst in instrs:
measured_qubits.add(inst.channel.index)
for meas_set in meas_map_sets:
@ -249,13 +255,13 @@ def _validate_meas_map(instruction_map: Dict[Tuple[int, Acquire], List[AcquireIn
def _bundle_channel_indices(
instructions: List[AcquireInstruction]
instrs: List[commands.AcquireInstruction]
) -> Tuple[List[int], List[int], List[int]]:
"""From the list of AcquireInstructions, bundle the indices of the acquire channels,
memory slots, and register slots into a 3-tuple of lists.
Args:
instructions: A list of AcquireInstructions to be bundled.
instrs: A list of AcquireInstructions to be bundled.
Returns:
The qubit indices, the memory slot indices, and register slot indices from instructions.
@ -263,7 +269,7 @@ def _bundle_channel_indices(
qubits = []
mem_slots = []
reg_slots = []
for inst in instructions:
for inst in instrs:
qubits.append(inst.channel.index)
if inst.mem_slot:
mem_slots.append(inst.mem_slot.index)
@ -272,9 +278,9 @@ def _bundle_channel_indices(
return qubits, mem_slots, reg_slots
def _assemble_config(lo_converter: LoConfigConverter,
def _assemble_config(lo_converter: converters.LoConfigConverter,
experiment_config: Dict[str, Any],
run_config: RunConfig) -> PulseQobjConfig:
run_config: RunConfig) -> qobj.PulseQobjConfig:
"""Assembles the QobjConfiguration from experimental config and runtime config.
Args:
@ -295,11 +301,11 @@ def _assemble_config(lo_converter: LoConfigConverter,
# convert enums to serialized values
meas_return = qobj_config.get('meas_return', 'avg')
if isinstance(meas_return, MeasReturnType):
if isinstance(meas_return, qobj_utils.MeasReturnType):
qobj_config['meas_return'] = meas_return.value
meas_level = qobj_config.get('meas_level', 2)
if isinstance(meas_level, MeasLevel):
if isinstance(meas_level, qobj_utils.MeasLevel):
qobj_config['meas_level'] = meas_level.value
# convert lo frequencies to Hz
@ -318,4 +324,4 @@ def _assemble_config(lo_converter: LoConfigConverter,
if m_los:
qobj_config['meas_lo_freq'] = [freq / 1e9 for freq in m_los]
return PulseQobjConfig(**qobj_config)
return qobj.PulseQobjConfig(**qobj_config)

View File

@ -25,7 +25,8 @@ from qiskit.circuit.quantumcircuit import QuantumCircuit
from qiskit.exceptions import QiskitError
from qiskit.pulse import InstructionScheduleMap, Schedule
from qiskit.providers import BaseBackend
from qiskit.scheduler import schedule_circuit, ScheduleConfig
from qiskit.scheduler import ScheduleConfig
from qiskit.scheduler.schedule_circuit import schedule_circuit
LOG = logging.getLogger(__name__)

View File

@ -13,25 +13,39 @@
# that they have been altered from the originals.
"""
.. currentmodule:: qiskit.pulse
===========================
Pulse (:mod:`qiskit.pulse`)
===========================
Qiskit-Pulse is a pulse-level quantum programming kit. This lower level of programming offers the
user more control than programming with :py:class:`~qiskit.circuit.QuantumCircuit` s.
Qiskit-Pulse is a pulse-level quantum programming kit. This lower level of
programming offers the user more control than programming with
:py:class:`~qiskit.circuit.QuantumCircuit` s.
Extracting the greatest performance from quantum hardware requires real-time pulse-level
instructions. Pulse answers that need: it enables the quantum physicist *user* to specify the
exact time dynamics of an experiment. It is especially powerful for error mitigation techniques.
Extracting the greatest performance from quantum hardware requires real-time
pulse-level instructions. Pulse answers that need: it enables the quantum
physicist *user* to specify the exact time dynamics of an experiment.
It is especially powerful for error mitigation techniques.
The input is given as arbitrary, time-ordered signals (see: :ref:`pulse-insts`) scheduled in
parallel over multiple virtual hardware or simulator resources (see: :ref:`pulse-channels`). The
system also allows the user to recover the time dynamics of the measured output.
The input is given as arbitrary, time-ordered signals (see: :ref:`pulse-insts`)
scheduled in parallel over multiple virtual hardware or simulator resources
(see: :ref:`pulse-channels`). The system also allows the user to recover the
time dynamics of the measured output.
This is sufficient to allow the quantum physicist to explore and correct for
noise in a quantum system.
.. _pulse-builder:
(BETA) Pulse Builder (:mod:`~qiskit.pulse.builder`)
===================================================
.. autosummary::
:toctree: ../stubs/
~builder
This is sufficient to allow the quantum physicist to explore and correct for noise in a quantum
system.
.. _pulse-insts:
@ -133,16 +147,95 @@ Exceptions
"""
from .channels import (DriveChannel, MeasureChannel, AcquireChannel,
ControlChannel, RegisterSlot, MemorySlot)
from .commands import AcquireInstruction, FrameChange, PersistentValue
from .configuration import LoConfig, LoRange, Kernel, Discriminator
from .exceptions import PulseError
from .instruction_schedule_map import InstructionScheduleMap
from .instructions import (Acquire, Instruction, Delay, Play, ShiftPhase, Snapshot,
SetFrequency, ShiftFrequency, SetPhase)
from .interfaces import ScheduleComponent
from .pulse_lib import (SamplePulse, Gaussian, GaussianSquare, Drag,
Constant, ConstantPulse, ParametricPulse)
from .pulse_lib.samplers.decorators import functional_pulse
from .schedule import Schedule
# Builder imports.
from qiskit.pulse.builder import (
# Construction methods.
active_backend,
active_transpiler_settings,
active_circuit_scheduler_settings,
build,
num_qubits,
qubit_channels,
samples_to_seconds,
seconds_to_samples,
# Instructions.
acquire,
barrier,
call,
delay,
play,
set_frequency,
set_phase,
shift_frequency,
shift_phase,
snapshot,
# Channels.
acquire_channel,
control_channels,
drive_channel,
measure_channel,
# Contexts.
align_left,
align_right,
align_sequential,
circuit_scheduler_settings,
frequency_offset,
inline,
pad,
phase_offset,
transpiler_settings,
# Macros.
measure,
measure_all,
delay_qubits,
# Circuit instructions.
cx,
u1,
u2,
u3,
x,
)
from qiskit.pulse.channels import (
AcquireChannel,
ControlChannel,
DriveChannel,
MeasureChannel,
MemorySlot,
RegisterSlot,
)
from qiskit.pulse.commands import (
AcquireInstruction,
FrameChange,
PersistentValue,
)
from qiskit.pulse.configuration import (
Discriminator,
Kernel,
LoConfig,
LoRange,
)
from qiskit.pulse.exceptions import PulseError
from qiskit.pulse.instruction_schedule_map import InstructionScheduleMap
from qiskit.pulse.instructions import (
Acquire,
Delay,
Instruction,
Play,
SetFrequency,
SetPhase,
ShiftFrequency,
ShiftPhase,
Snapshot,
)
from qiskit.pulse.interfaces import ScheduleComponent
from qiskit.pulse.pulse_lib import (
Constant,
ConstantPulse,
Drag,
Gaussian,
GaussianSquare,
ParametricPulse,
SamplePulse,
)
from qiskit.pulse.pulse_lib.samplers.decorators import functional_pulse
from qiskit.pulse.schedule import Schedule

1968
qiskit/pulse/builder.py Normal file

File diff suppressed because it is too large Load Diff

View File

@ -27,3 +27,11 @@ class PulseError(QiskitError):
def __str__(self):
"""Return the message."""
return repr(self.message)
class BackendNotSet(PulseError):
"""Raised if the builder context does not have a backend."""
class NoActiveBuilder(PulseError):
"""Raised if no builder context is active."""

View File

@ -54,6 +54,7 @@ Abstract Classes
"""
from .acquire import Acquire
from .delay import Delay
from .directives import Directive, RelativeBarrier
from .instruction import Instruction
from .frequency import SetFrequency, ShiftFrequency
from .phase import ShiftPhase, SetPhase

View File

@ -13,7 +13,7 @@
# that they have been altered from the originals.
"""The Acquire instruction is used to trigger the qubit measurement unit and provide
some metadata for the acquisition process; for example, where to store classified readout data.
some metadata for the acquisition process, for example, where to store classified readout data.
"""
import warnings

View File

@ -50,12 +50,14 @@ class Delay(Instruction):
channel: The channel that will have the delay.
name: Name of the delay for display purposes.
"""
self._channel = channel
if channel is None:
warnings.warn("Usage of Delay without specifying a channel is deprecated. For "
"example, Delay(5)(DriveChannel(0)) should be replaced by "
"Delay(5, DriveChannel(0)).", DeprecationWarning)
self._channel = channel
super().__init__((duration, channel), duration, (channel,), name=name)
super().__init__((duration, channel), duration, tuple(), name=name)
else:
super().__init__((duration, channel), duration, (channel,), name=name)
@property
def channel(self) -> Channel:

View File

@ -0,0 +1,51 @@
# -*- coding: utf-8 -*-
# This code is part of Qiskit.
#
# (C) Copyright IBM 2020.
#
# 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.
"""Directives are hints to the pulse compiler for how to process its input programs."""
from abc import ABC
from typing import Optional
from qiskit.pulse import channels as chans
from qiskit.pulse.instructions import instruction
class Directive(instruction.Instruction, ABC):
"""A compiler directive.
This is a hint to the pulse compiler and is not loaded into hardware.
"""
class RelativeBarrier(Directive):
"""Pulse ``RelativeBarrier`` directive."""
def __init__(self,
*channels: chans.Channel,
name: Optional[str] = None):
"""Create a relative barrier directive.
The barrier directive blocks instructions within the same schedule
as the barrier on channels contained within this barrier from moving
through the barrier in time.
Args:
channels: The channel that the barrier applies to.
name: Name of the directive for display purposes.
"""
super().__init__(tuple(channels), 0, tuple(channels), name=name)
def __eq__(self, other):
"""Verify two barriers are equivalent."""
return (isinstance(other, type(self)) and
set(self.channels) == set(other.channels))

View File

@ -27,14 +27,13 @@ import warnings
from abc import ABC
from typing import Callable, Dict, Iterable, List, Optional, Tuple, Union
from typing import Callable, Dict, Iterable, List, Optional, Tuple
import numpy as np
from ..channels import Channel
from ..exceptions import PulseError
from ..interfaces import ScheduleComponent
from ..schedule import Schedule
from .. import commands # pylint: disable=unused-import
# pylint: disable=missing-return-doc
@ -46,20 +45,22 @@ class Instruction(ScheduleComponent, ABC):
def __init__(self,
operands: Tuple,
duration: Union['commands.Command', int],
duration,
channels: Tuple[Channel],
name: Optional[str] = None):
"""Instruction initializer.
Args:
operands: The argument list.
duration: Length of time taken by the instruction in terms of dt.
Deprecated: the first argument used to be the Command.
duration (Union['commands.Command', int]): Length of time taken by the instruction in
terms of dt. **Deprecated: the first argument used to be the Command.**
channels: Tuple of pulse channels that this instruction operates on.
name: Optional display name for this instruction.
Raises:
PulseError: If duration is negative.
PulseError: If the input ``channels`` are not all of
type :class:`Channel`.
"""
self._command = None
if isinstance(duration, (float, np.float)):
@ -77,6 +78,12 @@ class Instruction(ScheduleComponent, ABC):
raise PulseError("{} duration of {} is invalid: must be nonnegative."
"".format(self.__class__.__name__, duration))
self._duration = duration
for channel in channels:
if not isinstance(channel, Channel):
raise PulseError(
"Expected a channel, got {} instead.".format(channel))
self._channels = channels
self._timeslots = {channel: [(0, self.duration)] for channel in channels}
self._operands = operands
@ -89,9 +96,12 @@ class Instruction(ScheduleComponent, ABC):
return self._name
@property
def command(self) -> 'commands.Command':
def command(self):
"""The associated command. Commands are deprecated, so this method will be deprecated
shortly.
Returns:
Command: The deprecated command if available.
"""
return self._command

110
qiskit/pulse/macros.py Normal file
View File

@ -0,0 +1,110 @@
# -*- coding: utf-8 -*-
# This code is part of Qiskit.
#
# (C) Copyright IBM 2020.
#
# 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.
"""Module for common pulse programming macros."""
from typing import Dict, List, Optional, Union
from qiskit.pulse import channels, commands, exceptions, instructions, utils
from qiskit.pulse.instruction_schedule_map import InstructionScheduleMap
from qiskit.pulse.schedule import Schedule
def measure(qubits: List[int],
backend=None,
inst_map: Optional[InstructionScheduleMap] = None,
meas_map: Optional[Union[List[List[int]], Dict[int, List[int]]]] = None,
qubit_mem_slots: Optional[Dict[int, int]] = None,
measure_name: str = 'measure') -> Schedule:
"""Return a schedule which measures the requested qubits according to the given
instruction mapping and measure map, or by using the defaults provided by the backend.
By default, the measurement results for each qubit are trivially mapped to the qubit
index. This behavior is overridden by qubit_mem_slots. For instance, to measure
qubit 0 into MemorySlot(1), qubit_mem_slots can be provided as {0: 1}.
Args:
qubits: List of qubits to be measured.
backend (BaseBackend): A backend instance, which contains hardware-specific data
required for scheduling.
inst_map: Mapping of circuit operations to pulse schedules. If None, defaults to the
``instruction_schedule_map`` of ``backend``.
meas_map: List of sets of qubits that must be measured together. If None, defaults to
the ``meas_map`` of ``backend``.
qubit_mem_slots: Mapping of measured qubit index to classical bit index.
measure_name: Name of the measurement schedule.
Returns:
A measurement schedule corresponding to the inputs provided.
Raises:
PulseError: If both ``inst_map`` or ``meas_map``, and ``backend`` is None.
"""
schedule = Schedule(name="Default measurement schedule for qubits {}".format(qubits))
try:
inst_map = inst_map or backend.defaults().instruction_schedule_map
meas_map = meas_map or backend.configuration().meas_map
except AttributeError:
raise exceptions.PulseError(
'inst_map or meas_map, and backend cannot be None simultaneously')
if isinstance(meas_map, List):
meas_map = utils.format_meas_map(meas_map)
measure_groups = set()
for qubit in qubits:
measure_groups.add(tuple(meas_map[qubit]))
for measure_group_qubits in measure_groups:
if qubit_mem_slots is not None:
unused_mem_slots = set(measure_group_qubits) - set(qubit_mem_slots.values())
try:
default_sched = inst_map.get(measure_name, measure_group_qubits)
except exceptions.PulseError:
raise exceptions.PulseError(
"We could not find a default measurement schedule called '{}'. "
"Please provide another name using the 'measure_name' keyword "
"argument. For assistance, the instructions which are defined are: "
"{}".format(measure_name, inst_map.instructions))
for time, inst in default_sched.instructions:
if qubit_mem_slots and isinstance(
inst, (instructions.Acquire, commands.AcquireInstruction)):
if inst.channel.index in qubit_mem_slots:
mem_slot = channels.MemorySlot(qubit_mem_slots[inst.channel.index])
else:
mem_slot = channels.MemorySlot(unused_mem_slots.pop())
schedule = schedule.insert(time, instructions.Acquire(
inst.duration, inst.channel, mem_slot=mem_slot))
elif qubit_mem_slots is None and isinstance(
inst, (instructions.Acquire, commands.AcquireInstruction)):
schedule = schedule.insert(time, inst)
# Measurement pulses should only be added if its qubit was measured by the user
elif inst.channels[0].index in qubits:
schedule = schedule.insert(time, inst)
return schedule
def measure_all(backend) -> Schedule:
"""
Return a Schedule which measures all qubits of the given backend.
Args:
backend (BaseBackend): A backend instance, which contains hardware-specific data
required for scheduling.
Returns:
A schedule corresponding to the inputs provided.
"""
return measure(qubits=list(range(backend.configuration().n_qubits)),
backend=backend)

View File

@ -16,8 +16,12 @@
import warnings
# pylint: disable=unused-import
from .transforms import align_measures, add_implicit_acquires, pad, compress_pulses
from qiskit.pulse.transforms import (
align_measures,
add_implicit_acquires,
pad,
compress_pulses,
)
warnings.warn("The reschedule module has been renamed to transforms. This import path "

View File

@ -18,7 +18,7 @@ instruction occuring in parallel over multiple signal *channels*.
"""
import abc
from copy import copy
import copy
import itertools
import multiprocessing as mp
import sys
@ -26,9 +26,9 @@ from typing import List, Tuple, Iterable, Union, Dict, Callable, Set, Optional
import warnings
from qiskit.util import is_main_process
from .channels import Channel
from .interfaces import ScheduleComponent
from .exceptions import PulseError
from qiskit.pulse.channels import Channel
from qiskit.pulse.interfaces import ScheduleComponent
from qiskit.pulse.exceptions import PulseError
# pylint: disable=missing-return-doc
@ -64,18 +64,15 @@ class Schedule(ScheduleComponent):
self._duration = 0
self._timeslots = {}
_children = []
self.__children = []
for sched_pair in schedules:
if isinstance(sched_pair, list):
sched_pair = tuple(sched_pair)
if not isinstance(sched_pair, tuple):
try:
time, sched = sched_pair
except TypeError:
# recreate as sequence starting at 0.
sched_pair = (0, sched_pair)
insert_time, sched = sched_pair
# This will also update duration
self._add_timeslots(insert_time, sched)
_children.append(sched_pair)
self.__children = tuple(_children)
time, sched = 0, sched_pair
self._mutable_insert(time, sched)
@property
def name(self) -> str:
@ -105,7 +102,15 @@ class Schedule(ScheduleComponent):
@property
def _children(self) -> Tuple[Tuple[int, ScheduleComponent], ...]:
return self.__children
"""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):
@ -118,7 +123,7 @@ class Schedule(ScheduleComponent):
def key(time_inst_pair):
inst = time_inst_pair[1]
return (time_inst_pair[0], inst.duration,
min(chan.index for chan in inst.channels))
sorted(chan.name for chan in inst.channels))
return tuple(sorted(self._instructions(), key=key))
@ -170,27 +175,107 @@ class Schedule(ScheduleComponent):
for insert_time, child_sched in self._children:
yield from child_sched._instructions(time + insert_time)
def union(self, *schedules: Union[ScheduleComponent, Tuple[int, ScheduleComponent]],
name: Optional[str] = None) -> 'Schedule':
"""Return a new schedule which is the union of both ``self`` and ``schedules``.
def union(self,
*schedules: Union[ScheduleComponent, Tuple[int, ScheduleComponent]],
name: Optional[str] = None,
inplace: bool = False
) -> 'Schedule':
"""Return a schedule which is the union of both ``self`` and ``schedules``.
Args:
*schedules: Schedules to be take the union with this ``Schedule``.
schedules: Schedules to be take the union with this ``Schedule``.
name: Name of the new schedule. Defaults to the name of self.
inplace: Perform operation inplace on this schedule. Otherwise return
a new ``Schedule``.
"""
warnings.warn("The union method is deprecated. Use insert with start_time=0.",
DeprecationWarning)
return self.insert(0, *schedules, name=name, inplace=inplace)
# pylint: disable=arguments-differ
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
new_sched = Schedule(name=name)
new_sched._insert(0, self)
for sched_pair in schedules:
if not isinstance(sched_pair, tuple):
sched_pair = (0, sched_pair)
new_sched._insert(sched_pair[0], sched_pair[1])
return new_sched
return Schedule((time, self), name=name)
def _insert(self, start_time: int, schedule: ScheduleComponent) -> 'Schedule':
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
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:
@ -198,43 +283,32 @@ class Schedule(ScheduleComponent):
schedule: Schedule to mutably insert.
"""
self._add_timeslots(start_time, schedule)
if isinstance(schedule, Schedule):
shifted_children = schedule._children
if start_time != 0:
shifted_children = tuple((t + start_time, child) for t, child in shifted_children)
self.__children += shifted_children
else: # isinstance(schedule, Instruction)
self.__children += ((start_time, schedule),)
self.__children.append((start_time, schedule))
return self
def 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. Defaults to the name of self.
"""
if name is None:
name = self.name
return Schedule((time, self), name=name)
def insert(self, start_time: int, schedule: ScheduleComponent,
name: Optional[str] = None) -> 'Schedule':
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 the name of self.
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._insert(0, self)
new_sched._insert(start_time, schedule)
new_sched._mutable_insert(0, self)
new_sched._mutable_insert(start_time, schedule)
return new_sched
# pylint: disable=arguments-differ
def append(self, schedule: ScheduleComponent,
name: Optional[str] = None) -> 'Schedule':
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``.
@ -246,10 +320,12 @@ class Schedule(ScheduleComponent):
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)
return self.insert(time, schedule, name=name, inplace=inplace)
def flatten(self) -> 'Schedule':
"""Return a new schedule which is the flattened schedule contained all ``instructions``."""
@ -404,15 +480,16 @@ class Schedule(ScheduleComponent):
Raises:
PulseError: If timeslots overlap or an invalid start time is provided.
"""
if (not isinstance(time, int)) or (time + schedule.start_time < 0):
raise PulseError("Schedule start time must be a non-negative integer.")
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(schedule._timeslots[channel])
self._timeslots[channel] = copy.copy(schedule._timeslots[channel])
else:
self._timeslots[channel] = [(i[0] + time, i[1] + time)
for i in schedule._timeslots[channel]]
@ -438,6 +515,8 @@ class Schedule(ScheduleComponent):
"".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 draw(self, dt: float = 1, style=None,
filename: Optional[str] = None, interp_method: Optional[Callable] = None,
scale: Optional[float] = None,
@ -686,3 +765,17 @@ def _overlaps(first: Interval, second: Interval) -> bool:
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))

View File

@ -12,8 +12,8 @@
# copyright notice, and modified files need to carry a notice indicating
# that they have been altered from the originals.
"""Basic transformation functions which take a Schedule (and possibly some arguments) and return
a new Schedule.
"""Basic rescheduling functions which take schedules or instructions
(and possibly some arguments) and return new schedules.
"""
import warnings
@ -21,17 +21,18 @@ from typing import List, Optional, Iterable
import numpy as np
from qiskit.pulse import (Acquire, AcquireInstruction, Delay, Play,
InstructionScheduleMap, ScheduleComponent, Schedule)
from .channels import Channel, AcquireChannel, MeasureChannel, MemorySlot
from .exceptions import PulseError
from qiskit.pulse import channels as chans, commands, exceptions, instructions, interfaces
from qiskit.pulse.instructions import directives
from qiskit.pulse.instruction_schedule_map import InstructionScheduleMap
from qiskit.pulse.schedule import Schedule
def align_measures(schedules: Iterable[ScheduleComponent],
def align_measures(schedules: Iterable[interfaces.ScheduleComponent],
inst_map: Optional[InstructionScheduleMap] = None,
cal_gate: str = 'u3',
max_calibration_duration: Optional[int] = None,
align_time: Optional[int] = None) -> Schedule:
align_time: Optional[int] = None
) -> List[Schedule]:
"""Return new schedules where measurements occur at the same physical time. Minimum measurement
wait time (to allow for calibration pulses) is enforced.
@ -47,7 +48,7 @@ def align_measures(schedules: Iterable[ScheduleComponent],
align_time: If provided, this will be used as final align time.
Returns:
Schedule
The input list of schedules transformed to have their measurements aligned.
Raises:
PulseError: if an acquire or pulse is encountered on a channel that has already been part
@ -64,7 +65,8 @@ def align_measures(schedules: Iterable[ScheduleComponent],
for schedule in schedules:
last_acquire = 0
acquire_times = [time for time, inst in schedule.instructions
if isinstance(inst, (Acquire, AcquireInstruction))]
if isinstance(inst, (instructions.Acquire,
commands.AcquireInstruction))]
if acquire_times:
last_acquire = max(acquire_times)
align_time = max(align_time, last_acquire)
@ -79,9 +81,10 @@ def align_measures(schedules: Iterable[ScheduleComponent],
return max_calibration_duration
if align_time is None and max_calibration_duration is None and inst_map is None:
raise PulseError("Must provide a inst_map, an alignment time, or a calibration duration.")
raise exceptions.PulseError("Must provide a inst_map, an alignment time, "
"or a calibration duration.")
if align_time is not None and align_time < 0:
raise PulseError("Align time cannot be negative.")
raise exceptions.PulseError("Align time cannot be negative.")
if align_time is None:
align_time = calculate_align_time()
@ -94,21 +97,23 @@ def align_measures(schedules: Iterable[ScheduleComponent],
for time, inst in schedule.instructions:
for chan in inst.channels:
if isinstance(chan, MeasureChannel):
if isinstance(chan, chans.MeasureChannel):
if chan.index in measured_channels:
raise PulseError("Multiple measurements are not supported by this "
"rescheduling pass.")
raise exceptions.PulseError("Multiple measurements are "
"not supported by this "
"rescheduling pass.")
elif chan.index in acquired_channels:
raise PulseError("Pulse encountered on channel {0} after acquire on "
"same channel.".format(chan.index))
raise exceptions.PulseError("Pulse encountered on channel "
"{0} after acquire on "
"same channel.".format(chan.index))
if isinstance(inst, (Acquire, AcquireInstruction)):
if isinstance(inst, (instructions.Acquire, commands.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.add(inst.channel.index)
elif isinstance(inst.channels[0], MeasureChannel):
elif isinstance(inst.channels[0], chans.MeasureChannel):
new_schedule |= inst << align_time
measured_channels.update({a.index for a in inst.channels})
else:
@ -119,7 +124,9 @@ def align_measures(schedules: Iterable[ScheduleComponent],
return new_schedules
def add_implicit_acquires(schedule: ScheduleComponent, meas_map: List[List[int]]) -> Schedule:
def add_implicit_acquires(schedule: interfaces.ScheduleComponent,
meas_map: List[List[int]]
) -> Schedule:
"""Return a new schedule with implicit acquires from the measurement mapping replaced by
explicit ones.
@ -137,7 +144,7 @@ def add_implicit_acquires(schedule: ScheduleComponent, meas_map: List[List[int]]
acquire_map = dict()
for time, inst in schedule.instructions:
if isinstance(inst, (Acquire, AcquireInstruction)):
if isinstance(inst, (instructions.Acquire, commands.AcquireInstruction)):
if inst.mem_slot and inst.mem_slot.index != inst.channel.index:
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.")
@ -150,10 +157,11 @@ def add_implicit_acquires(schedule: ScheduleComponent, meas_map: List[List[int]]
# Replace the old acquire instruction by a new one explicitly acquiring all qubits in
# the measurement group.
for i in all_qubits:
explicit_inst = Acquire(inst.duration, AcquireChannel(i),
mem_slot=MemorySlot(i),
kernel=inst.kernel,
discriminator=inst.discriminator) << time
explicit_inst = instructions.Acquire(inst.duration,
chans.AcquireChannel(i),
mem_slot=chans.MemorySlot(i),
kernel=inst.kernel,
discriminator=inst.discriminator) << time
if time not in acquire_map:
new_schedule |= explicit_inst
acquire_map = {time: {i}}
@ -167,16 +175,20 @@ def add_implicit_acquires(schedule: ScheduleComponent, meas_map: List[List[int]]
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 ``until``
if it is provided, otherwise until ``schedule.duration``.
channels: Optional[Iterable[chans.Channel]] = None,
until: Optional[int] = None,
inplace: bool = False
) -> Schedule:
r"""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.
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.
inplace: Pad this schedule by mutating rather than returning a new schedule.
Returns:
The padded schedule.
@ -186,7 +198,7 @@ def pad(schedule: Schedule,
for channel in channels:
if channel not in schedule.channels:
schedule |= Delay(until, channel)
schedule |= instructions.Delay(until, channel)
continue
curr_time = 0
@ -196,10 +208,16 @@ def pad(schedule: Schedule,
break
if interval[0] != curr_time:
end_time = min(interval[0], until)
schedule = schedule.insert(curr_time, Delay(end_time - curr_time, channel))
schedule = schedule.insert(
curr_time,
instructions.Delay(end_time - curr_time, channel),
inplace=inplace)
curr_time = interval[1]
if curr_time < until:
schedule = schedule.insert(curr_time, Delay(until - curr_time, channel))
schedule = schedule.insert(
curr_time,
instructions.Delay(until - curr_time, channel),
inplace=inplace)
return schedule
@ -221,11 +239,12 @@ def compress_pulses(schedules: List[Schedule]) -> List[Schedule]:
new_schedule = Schedule(name=schedule.name)
for time, inst in schedule.instructions:
if isinstance(inst, Play):
if isinstance(inst, instructions.Play):
if inst.pulse in existing_pulses:
idx = existing_pulses.index(inst.pulse)
identical_pulse = existing_pulses[idx]
new_schedule |= Play(identical_pulse, inst.channel, inst.name) << time
new_schedule |= instructions.Play(
identical_pulse, inst.channel, inst.name) << time
else:
existing_pulses.append(inst.pulse)
new_schedule |= inst << time
@ -235,3 +254,142 @@ def compress_pulses(schedules: List[Schedule]) -> List[Schedule]:
new_schedules.append(new_schedule)
return new_schedules
def _push_left_append(this: Schedule,
other: interfaces.ScheduleComponent,
) -> Schedule:
r"""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)
def align_left(schedule: Schedule) -> Schedule:
"""Align a list of pulse instructions on the left.
Args:
schedule: Input schedule of which top-level ``child`` nodes will be
reschedulued.
Returns:
New schedule with input `schedule`` child schedules and instructions
left aligned.
"""
aligned = Schedule()
for _, child in schedule._children:
_push_left_append(aligned, child)
return aligned
def _push_right_prepend(this: interfaces.ScheduleComponent,
other: interfaces.ScheduleComponent,
) -> Schedule:
r"""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
def align_right(schedule: Schedule) -> Schedule:
"""Align a list of pulse instructions on the right.
Args:
schedule: Input schedule of which top-level ``child`` nodes will be
reschedulued.
Returns:
New schedule with input `schedule`` child schedules and instructions
right aligned.
"""
aligned = Schedule()
for _, child in reversed(schedule._children):
aligned = _push_right_prepend(aligned, child)
return aligned
def align_sequential(schedule: Schedule) -> Schedule:
"""Schedule all top-level nodes in parallel.
Args:
schedule: Input schedule of which top-level ``child`` nodes will be
reschedulued.
Returns:
New schedule with input `schedule`` child schedules and instructions
applied sequentially across channels
"""
aligned = Schedule()
for _, child in schedule._children:
aligned.insert(aligned.duration, child, inplace=True)
return aligned
def flatten(schedule: Schedule) -> Schedule:
"""Flatten any called nodes into a Schedule tree with no nested children."""
return schedule.flatten()
def remove_directives(schedule: Schedule) -> Schedule:
"""Remove directives."""
return schedule.exclude(instruction_types=[directives.Directive])
def remove_trivial_barriers(schedule: Schedule) -> Schedule:
"""Remove trivial barriers with 0 or 1 channels."""
def filter_func(inst):
return (isinstance(inst[1], directives.RelativeBarrier) and
len(inst[1].channels) < 2)
return schedule.exclude(filter_func)

35
qiskit/pulse/utils.py Normal file
View File

@ -0,0 +1,35 @@
# -*- coding: utf-8 -*-
# This code is part of Qiskit.
#
# (C) Copyright IBM 2020.
#
# 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.
"""Module for common pulse programming utilies."""
from typing import List, Dict
def format_meas_map(meas_map: List[List[int]]) -> Dict[int, List[int]]:
"""
Return a mapping from qubit label to measurement group given the nested list meas_map returned
by a backend configuration. (Qubits can not always be measured independently.) Sorts the
measurement group for consistency.
Args:
meas_map: Groups of qubits that get measured together, for example: [[0, 1], [2, 3, 4]]
Returns:
Measure map in map format
"""
qubit_mapping = {}
for sublist in meas_map:
sublist.sort()
for q in sublist:
qubit_mapping[q] = sublist
return qubit_mapping

View File

@ -19,7 +19,7 @@ Circuit Scheduler (:mod:`qiskit.scheduler`)
.. currentmodule:: qiskit.scheduler
A scheduler compiles a circuit program to a pulse program.
A circuit scheduler compiles a circuit program to a pulse program.
.. autosummary::
:toctree: ../stubs/
@ -27,16 +27,8 @@ A scheduler compiles a circuit program to a pulse program.
schedule_circuit
ScheduleConfig
Scheduling utility functions
.. autosummary::
:toctree: ../stubs/
qiskit.scheduler.utils
.. automodule:: qiskit.scheduler.methods
"""
from qiskit.scheduler import schedule_circuit
from qiskit.scheduler.config import ScheduleConfig
from qiskit.scheduler.schedule_circuit import schedule_circuit
from qiskit.scheduler.utils import measure, measure_all

View File

@ -17,8 +17,7 @@
from typing import List
from qiskit.pulse.instruction_schedule_map import InstructionScheduleMap
from qiskit.scheduler.utils import format_meas_map
from qiskit.pulse.utils import format_meas_map
class ScheduleConfig():

View File

@ -23,4 +23,4 @@ Pulse scheduling methods.
basic
"""
from .basic import *
from qiskit.scheduler.methods.basic import *

View File

@ -23,10 +23,10 @@ from qiskit.circuit.measure import Measure
from qiskit.circuit.quantumcircuit import QuantumCircuit
from qiskit.exceptions import QiskitError
from qiskit.circuit.barrier import Barrier
from qiskit.pulse.exceptions import PulseError
from qiskit.pulse.schedule import Schedule
from qiskit.pulse.channels import AcquireChannel
from qiskit.scheduler.utils import measure
from qiskit.pulse.exceptions import PulseError
from qiskit.pulse.macros import measure
from qiskit.pulse.schedule import Schedule
from qiskit.scheduler.config import ScheduleConfig

View File

@ -17,8 +17,8 @@ from typing import Optional
from qiskit.circuit.quantumcircuit import QuantumCircuit
from qiskit.exceptions import QiskitError
from qiskit.pulse.schedule import Schedule
from qiskit.pulse.schedule import Schedule
from qiskit.scheduler.config import ScheduleConfig
from qiskit.scheduler.methods.basic import as_soon_as_possible, as_late_as_possible

View File

@ -14,112 +14,15 @@
"""Scheduling utility functions."""
from typing import Dict, List, Optional, Union
from qiskit.pulse import Schedule, MemorySlot, Acquire, PulseError, InstructionScheduleMap
from qiskit.pulse.commands import AcquireInstruction
# pylint: disable=invalid-name
from qiskit.util import deprecate_function
from qiskit.pulse import macros, utils
format_meas_map = deprecate_function(
'"format_meas_map" has been moved to "qiskit.pulse.utils"')(utils.format_meas_map)
def format_meas_map(meas_map: List[List[int]]) -> Dict[int, List[int]]:
"""
Return a mapping from qubit label to measurement group given the nested list meas_map returned
by a backend configuration. (Qubits can not always be measured independently.) Sorts the
measurement group for consistency.
measure = deprecate_function(
'"measure" has been moved to "qiskit.pulse.macros"')(macros.measure)
Args:
meas_map: Groups of qubits that get measured together, for example: [[0, 1], [2, 3, 4]]
Returns:
Measure map in map format
"""
qubit_mapping = {}
for sublist in meas_map:
sublist.sort()
for q in sublist:
qubit_mapping[q] = sublist
return qubit_mapping
def measure(qubits: List[int],
backend=None,
inst_map: Optional[InstructionScheduleMap] = None,
meas_map: Optional[Union[List[List[int]], Dict[int, List[int]]]] = None,
qubit_mem_slots: Optional[Dict[int, int]] = None,
measure_name: str = 'measure') -> Schedule:
"""
Return a schedule which measures the requested qubits according to the given
instruction mapping and measure map, or by using the defaults provided by the backend.
By default, the measurement results for each qubit are trivially mapped to the qubit
index. This behavior is overridden by qubit_mem_slots. For instance, to measure
qubit 0 into MemorySlot(1), qubit_mem_slots can be provided as {0: 1}.
Args:
qubits: List of qubits to be measured.
backend (BaseBackend): A backend instance, which contains hardware-specific data
required for scheduling.
inst_map: Mapping of circuit operations to pulse schedules. If None, defaults to the
``instruction_schedule_map`` of ``backend``.
meas_map: List of sets of qubits that must be measured together. If None, defaults to
the ``meas_map`` of ``backend``.
qubit_mem_slots: Mapping of measured qubit index to classical bit index.
measure_name: Name of the measurement schedule.
Returns:
A measurement schedule corresponding to the inputs provided.
Raises:
PulseError: If both ``inst_map`` or ``meas_map``, and ``backend`` is None.
"""
schedule = Schedule(name="Default measurement schedule for qubits {}".format(qubits))
try:
inst_map = inst_map or backend.defaults().instruction_schedule_map
meas_map = meas_map or backend.configuration().meas_map
except AttributeError:
raise PulseError('inst_map or meas_map, and backend cannot be None simultaneously')
if isinstance(meas_map, List):
meas_map = format_meas_map(meas_map)
measure_groups = set()
for qubit in qubits:
measure_groups.add(tuple(meas_map[qubit]))
for measure_group_qubits in measure_groups:
if qubit_mem_slots is not None:
unused_mem_slots = set(measure_group_qubits) - set(qubit_mem_slots.values())
try:
default_sched = inst_map.get(measure_name, measure_group_qubits)
except PulseError:
raise PulseError("We could not find a default measurement schedule called '{}'. "
"Please provide another name using the 'measure_name' keyword "
"argument. For assistance, the instructions which are defined are: "
"{}".format(measure_name, inst_map.instructions))
for time, inst in default_sched.instructions:
if qubit_mem_slots and isinstance(inst, (Acquire, AcquireInstruction)):
if inst.channel.index in qubit_mem_slots:
mem_slot = MemorySlot(qubit_mem_slots[inst.channel.index])
else:
mem_slot = MemorySlot(unused_mem_slots.pop())
schedule = schedule.insert(time, Acquire(inst.duration,
inst.channel,
mem_slot=mem_slot))
elif qubit_mem_slots is None and isinstance(inst, (Acquire, AcquireInstruction)):
schedule = schedule.insert(time, inst)
# Measurement pulses should only be added if its qubit was measured by the user
elif inst.channels[0].index in qubits:
schedule = schedule.insert(time, inst)
return schedule
def measure_all(backend) -> Schedule:
"""
Return a Schedule which measures all qubits of the given backend.
Args:
backend (BaseBackend): A backend instance, which contains hardware-specific data
required for scheduling.
Returns:
A schedule corresponding to the inputs provided.
"""
return measure(qubits=list(range(backend.configuration().n_qubits)),
backend=backend)
measure_all = deprecate_function(
'"measure_all" has been moved to "qiskit.pulse.macros"')(macros.measure_all)

View File

@ -112,6 +112,28 @@ def deprecate_arguments(kwarg_map):
return decorator
def deprecate_function(msg):
"""Emit a warning prior to calling decorated function.
Args:
msg (str): Warning message to emit.
Returns:
Callable: The decorated, deprecated callable.
"""
def decorator(func):
@functools.wraps(func)
def wrapper(*args, **kwargs):
# warn only once
if not wrapper._warned:
warnings.warn(msg, DeprecationWarning, stacklevel=2)
wrapper._warned = True
return func(*args, **kwargs)
wrapper._warned = False
return wrapper
return decorator
def _rename_kwargs(func_name, kwargs, kwarg_map):
for old_arg, new_arg in kwarg_map.items():
if old_arg in kwargs:

View File

@ -1,3 +1,4 @@
contextvars>=2.4;python_version<'3.7'
jsonschema>=2.6
networkx>=2.2;python_version>'3.5'
networkx>=2.2,<2.4;python_version=='3.5'

View File

@ -25,6 +25,7 @@ except ImportError:
from Cython.Build import cythonize
REQUIREMENTS = [
"contextvars>=2.4;python_version<'3.7'",
"jsonschema>=2.6",
"networkx>=2.2;python_version>'3.5'",
# Networkx 2.4 is the final version with python 3.5 support.

View File

@ -30,7 +30,7 @@ from qiskit.pulse.channels import MemorySlot, AcquireChannel, DriveChannel, Meas
from qiskit.pulse.pulse_lib import gaussian
from qiskit.qobj import QasmQobj, validate_qobj_against_schema
from qiskit.qobj.utils import MeasLevel, MeasReturnType
from qiskit.scheduler import measure
from qiskit.pulse.macros import measure
from qiskit.test import QiskitTestCase
from qiskit.test.mock import FakeOpenPulse2Q, FakeOpenPulse3Q, FakeYorktown, FakeAlmaden
from qiskit.validation.jsonschema import SchemaValidationError

View File

@ -0,0 +1,929 @@
# -*- coding: utf-8 -*-
# This code is part of Qiskit.
#
# (C) Copyright IBM 2020.
#
# 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.
"""Test pulse builder context utilities."""
from math import pi
import numpy as np
from qiskit import circuit, compiler, pulse
from qiskit.pulse import builder, exceptions, macros, transforms
from qiskit.pulse.instructions import directives
from qiskit.test import QiskitTestCase
from qiskit.test.mock import FakeOpenPulse2Q
from qiskit.pulse import pulse_lib, instructions
# pylint: disable=invalid-name
class TestBuilder(QiskitTestCase):
"""Test the pulse builder context."""
def setUp(self):
self.backend = FakeOpenPulse2Q()
self.configuration = self.backend.configuration()
self.defaults = self.backend.defaults()
self.inst_map = self.defaults.instruction_schedule_map
class TestBuilderBase(TestBuilder):
"""Test builder base."""
def test_schedule_supplied(self):
"""Test that schedule is used if it is supplied to the builder."""
d0 = pulse.DriveChannel(0)
with pulse.build(name='reference') as reference:
with pulse.align_sequential():
pulse.delay(10, d0)
with pulse.build(schedule=reference) as schedule:
pass
self.assertEqual(schedule, reference)
self.assertEqual(schedule.name, 'reference')
def test_default_alignment_left(self):
"""Test default left alignment setting."""
d0 = pulse.DriveChannel(0)
d1 = pulse.DriveChannel(0)
with pulse.build(default_alignment='left') as schedule:
pulse.delay(10, d0)
pulse.delay(20, d1)
with pulse.build(self.backend) as reference:
with pulse.align_left():
pulse.delay(10, d0)
pulse.delay(20, d1)
self.assertEqual(schedule, reference)
def test_default_alignment_right(self):
"""Test default right alignment setting."""
d0 = pulse.DriveChannel(0)
d1 = pulse.DriveChannel(0)
with pulse.build(default_alignment='right') as schedule:
pulse.delay(10, d0)
pulse.delay(20, d1)
with pulse.build() as reference:
with pulse.align_right():
pulse.delay(10, d0)
pulse.delay(20, d1)
self.assertEqual(schedule, reference)
def test_default_alignment_sequential(self):
"""Test default sequential alignment setting."""
d0 = pulse.DriveChannel(0)
d1 = pulse.DriveChannel(0)
with pulse.build(default_alignment='sequential') as schedule:
pulse.delay(10, d0)
pulse.delay(20, d1)
reference = pulse.Schedule()
with pulse.build(reference) as reference:
with pulse.align_sequential():
pulse.delay(10, d0)
pulse.delay(20, d1)
self.assertEqual(schedule, reference)
class TestContexts(TestBuilder):
"""Test builder contexts."""
def test_align_sequential(self):
"""Test the sequential alignment context."""
d0 = pulse.DriveChannel(0)
d1 = pulse.DriveChannel(1)
with pulse.build() as schedule:
with pulse.align_sequential():
pulse.delay(3, d0)
pulse.delay(5, d1)
pulse.delay(7, d0)
reference = pulse.Schedule()
# d0
reference.insert(0, instructions.Delay(3, d0), inplace=True)
reference.insert(8, instructions.Delay(7, d0), inplace=True)
# d1
reference.insert(3, instructions.Delay(5, d1), inplace=True)
self.assertEqual(schedule, reference)
def test_align_left(self):
"""Test the left alignment context."""
d0 = pulse.DriveChannel(0)
d1 = pulse.DriveChannel(1)
d2 = pulse.DriveChannel(2)
with pulse.build() as schedule:
with pulse.align_left():
pulse.delay(11, d2)
pulse.delay(3, d0)
with pulse.align_left():
pulse.delay(5, d1)
pulse.delay(7, d0)
reference = pulse.Schedule()
# d0
reference.insert(0, instructions.Delay(3, d0), inplace=True)
reference.insert(3, instructions.Delay(7, d0), inplace=True)
# d1
reference.insert(3, instructions.Delay(5, d1), inplace=True)
# d2
reference.insert(0, instructions.Delay(11, d2), inplace=True)
self.assertEqual(schedule, reference)
def test_align_right(self):
"""Test the right alignment context."""
d0 = pulse.DriveChannel(0)
d1 = pulse.DriveChannel(1)
d2 = pulse.DriveChannel(2)
with pulse.build() as schedule:
with pulse.align_right():
with pulse.align_right():
pulse.delay(11, d2)
pulse.delay(3, d0)
pulse.delay(13, d0)
pulse.delay(5, d1)
reference = pulse.Schedule()
# d0
reference.insert(8, instructions.Delay(3, d0), inplace=True)
reference.insert(11, instructions.Delay(13, d0), inplace=True)
# d1
reference.insert(19, instructions.Delay(5, d1), inplace=True)
# d2
reference.insert(0, instructions.Delay(11, d2), inplace=True)
self.assertEqual(schedule, reference)
def test_inline(self):
"""Test the inlining context."""
d0 = pulse.DriveChannel(0)
d1 = pulse.DriveChannel(1)
with pulse.build() as schedule:
pulse.delay(3, d0)
with pulse.inline():
# this alignment will be ignored due to inlining.
with pulse.align_right():
pulse.delay(5, d1)
pulse.delay(7, d0)
reference = pulse.Schedule()
# d0
reference += instructions.Delay(3, d0)
reference += instructions.Delay(7, d0)
# d1
reference += instructions.Delay(5, d1)
self.assertEqual(schedule, reference)
def test_transpiler_settings(self):
"""Test the transpiler settings context.
Tests that two cx gates are optimized away with higher optimization level.
"""
twice_cx_qc = circuit.QuantumCircuit(2)
twice_cx_qc.cx(0, 1)
twice_cx_qc.cx(0, 1)
with pulse.build(self.backend) as schedule:
with pulse.transpiler_settings(optimization_level=0):
builder.call_circuit(twice_cx_qc)
self.assertNotEqual(len(schedule.instructions), 0)
with pulse.build(self.backend) as schedule:
with pulse.transpiler_settings(optimization_level=3):
builder.call_circuit(twice_cx_qc)
self.assertEqual(len(schedule.instructions), 0)
def test_scheduler_settings(self):
"""Test the circuit scheduler settings context."""
inst_map = pulse.InstructionScheduleMap()
d0 = pulse.DriveChannel(0)
test_x_sched = pulse.Schedule()
test_x_sched += instructions.Delay(10, d0)
inst_map.add('x', (0,), test_x_sched)
x_qc = circuit.QuantumCircuit(2)
x_qc.x(0)
with pulse.build(backend=self.backend) as schedule:
with pulse.transpiler_settings(basis_gates=['x']):
with pulse.circuit_scheduler_settings(inst_map=inst_map):
builder.call_circuit(x_qc)
self.assertEqual(schedule, test_x_sched)
def test_phase_offset(self):
"""Test the phase offset context."""
d0 = pulse.DriveChannel(0)
with pulse.build() as schedule:
with pulse.phase_offset(3.14, d0):
pulse.delay(10, d0)
reference = pulse.Schedule()
reference += instructions.ShiftPhase(3.14, d0)
reference += instructions.Delay(10, d0)
reference += instructions.ShiftPhase(-3.14, d0)
self.assertEqual(schedule, reference)
def test_frequency_offset(self):
"""Test the frequency offset context."""
d0 = pulse.DriveChannel(0)
with pulse.build() as schedule:
with pulse.frequency_offset(1e9, d0):
pulse.delay(10, d0)
reference = pulse.Schedule()
reference += instructions.ShiftFrequency(1e9, d0) # pylint: disable=no-member
reference += instructions.Delay(10, d0)
reference += instructions.ShiftFrequency(-1e9, d0) # pylint: disable=no-member
self.assertEqual(schedule, reference)
def test_phase_compensated_frequency_offset(self):
"""Test that the phase offset context properly compensates for phase
accumulation."""
d0 = pulse.DriveChannel(0)
with pulse.build(self.backend) as schedule:
with pulse.frequency_offset(1e9, d0, compensate_phase=True):
pulse.delay(10, d0)
reference = pulse.Schedule()
reference += instructions.ShiftFrequency(1e9, d0) # pylint: disable=no-member
reference += instructions.Delay(10, d0)
reference += instructions.ShiftPhase(
-(1e9*10*self.configuration.dt % (2*np.pi)), d0)
reference += instructions.ShiftFrequency(-1e9, d0) # pylint: disable=no-member
self.assertEqual(schedule, reference)
class TestChannels(TestBuilder):
"""Test builder channels."""
def test_drive_channel(self):
"""Text context builder drive channel."""
with pulse.build(self.backend):
self.assertEqual(pulse.drive_channel(0), pulse.DriveChannel(0))
def test_measure_channel(self):
"""Text context builder measure channel."""
with pulse.build(self.backend):
self.assertEqual(pulse.measure_channel(0), pulse.MeasureChannel(0))
def test_acquire_channel(self):
"""Text context builder acquire channel."""
with pulse.build(self.backend):
self.assertEqual(pulse.acquire_channel(0), pulse.AcquireChannel(0))
def test_control_channel(self):
"""Text context builder control channel."""
with pulse.build(self.backend):
self.assertEqual(pulse.control_channels(0, 1)[0],
pulse.ControlChannel(0))
class TestInstructions(TestBuilder):
"""Test builder instructions."""
def test_delay(self):
"""Test delay instruction."""
d0 = pulse.DriveChannel(0)
with pulse.build() as schedule:
pulse.delay(10, d0)
reference = pulse.Schedule()
reference += instructions.Delay(10, d0)
self.assertEqual(schedule, reference)
def test_play_parametric_pulse(self):
"""Test play instruction with parametric pulse."""
d0 = pulse.DriveChannel(0)
test_pulse = pulse_lib.Constant(10, 1.0)
with pulse.build() as schedule:
pulse.play(test_pulse, d0)
reference = pulse.Schedule()
reference += instructions.Play(test_pulse, d0)
self.assertEqual(schedule, reference)
def test_play_sample_pulse(self):
"""Test play instruction with sample pulse."""
d0 = pulse.DriveChannel(0)
test_pulse = pulse_lib.SamplePulse([0.0, 0.0])
with pulse.build() as schedule:
pulse.play(test_pulse, d0)
reference = pulse.Schedule()
reference += instructions.Play(test_pulse, d0)
self.assertEqual(schedule, reference)
def test_play_array_pulse(self):
"""Test play instruction on an array directly."""
d0 = pulse.DriveChannel(0)
test_array = np.array([0., 0.], dtype=np.complex_)
with pulse.build() as schedule:
pulse.play(test_array, d0)
reference = pulse.Schedule()
test_pulse = pulse.SamplePulse(test_array)
reference += instructions.Play(test_pulse, d0)
self.assertEqual(schedule, reference)
def test_acquire_memory_slot(self):
"""Test acquire instruction into memory slot."""
acquire0 = pulse.AcquireChannel(0)
mem0 = pulse.MemorySlot(0)
with pulse.build() as schedule:
pulse.acquire(10, acquire0, mem0)
reference = pulse.Schedule()
reference += pulse.Acquire(10, acquire0, mem_slot=mem0)
self.assertEqual(schedule, reference)
def test_acquire_register_slot(self):
"""Test acquire instruction into register slot."""
acquire0 = pulse.AcquireChannel(0)
reg0 = pulse.RegisterSlot(0)
with pulse.build() as schedule:
pulse.acquire(10, acquire0, reg0)
reference = pulse.Schedule()
reference += pulse.Acquire(10, acquire0, reg_slot=reg0)
self.assertEqual(schedule, reference)
def test_acquire_qubit(self):
"""Test acquire instruction on qubit."""
acquire0 = pulse.AcquireChannel(0)
mem0 = pulse.MemorySlot(0)
with pulse.build() as schedule:
pulse.acquire(10, 0, mem0)
reference = pulse.Schedule()
reference += pulse.Acquire(10, acquire0, mem_slot=mem0)
self.assertEqual(schedule, reference)
def test_set_frequency(self):
"""Test set frequency instruction."""
d0 = pulse.DriveChannel(0)
with pulse.build() as schedule:
pulse.set_frequency(1e9, d0)
reference = pulse.Schedule()
reference += instructions.SetFrequency(1e9, d0)
self.assertEqual(schedule, reference)
def test_shift_frequency(self): # pylint: disable=no-member
"""Test shift frequency instruction."""
d0 = pulse.DriveChannel(0)
with pulse.build() as schedule:
pulse.shift_frequency(0.1e9, d0)
reference = pulse.Schedule()
reference += instructions.ShiftFrequency(0.1e9, d0) # pylint: disable=no-member
self.assertEqual(schedule, reference)
def test_set_phase(self): # pylint: disable=no-member
"""Test set phase instruction."""
d0 = pulse.DriveChannel(0)
with pulse.build() as schedule:
pulse.set_phase(3.14, d0)
reference = pulse.Schedule()
reference += instructions.SetPhase(3.14, d0) # pylint: disable=no-member
self.assertEqual(schedule, reference)
def test_shift_phase(self):
"""Test shift phase instruction."""
d0 = pulse.DriveChannel(0)
with pulse.build() as schedule:
pulse.shift_phase(3.14, d0)
reference = pulse.Schedule()
reference += instructions.ShiftPhase(3.14, d0)
self.assertEqual(schedule, reference)
def test_snapshot(self):
"""Test snapshot instruction."""
with pulse.build() as schedule:
pulse.snapshot('test', 'state')
reference = pulse.Schedule()
reference += instructions.Snapshot('test', 'state')
self.assertEqual(schedule, reference)
def test_call_schedule(self):
"""Test calling schedule instruction."""
schedule = pulse.Schedule()
d0 = pulse.DriveChannel(0)
d1 = pulse.DriveChannel(1)
reference = pulse.Schedule()
reference = reference.insert(10, instructions.Delay(10, d0))
reference += instructions.Delay(20, d1)
with pulse.build() as schedule:
with pulse.align_right():
builder.call_schedule(reference)
self.assertEqual(schedule, reference)
with pulse.build() as schedule:
with pulse.align_right():
pulse.call(reference)
self.assertEqual(schedule, reference)
def test_call_circuit(self):
"""Test calling circuit instruction."""
inst_map = self.inst_map
reference = inst_map.get('u1', (0,), 0.0)
u1_qc = circuit.QuantumCircuit(2)
u1_qc.u1(0.0, 0)
transpiler_settings = {'optimization_level': 0}
with pulse.build(self.backend,
default_transpiler_settings=transpiler_settings
) as schedule:
with pulse.align_right():
builder.call_circuit(u1_qc)
self.assertEqual(schedule, reference)
class TestDirectives(TestBuilder):
"""Test builder directives."""
def test_barrier_with_align_right(self):
"""Test barrier directive with right alignment context."""
d0 = pulse.DriveChannel(0)
d1 = pulse.DriveChannel(1)
d2 = pulse.DriveChannel(2)
with pulse.build() as schedule:
with pulse.align_right():
pulse.delay(3, d0)
pulse.barrier(d0, d1, d2)
pulse.delay(11, d2)
with pulse.align_right():
pulse.delay(5, d1)
pulse.delay(7, d0)
schedule = transforms.remove_directives(schedule)
reference = pulse.Schedule()
# d0
reference.insert(0, instructions.Delay(3, d0), inplace=True)
reference.insert(7, instructions.Delay(7, d0), inplace=True)
# d1
reference.insert(9, instructions.Delay(5, d1), inplace=True)
# d2
reference.insert(3, instructions.Delay(11, d2), inplace=True)
self.assertEqual(schedule, reference)
def test_barrier_with_align_left(self):
"""Test barrier directive with left alignment context."""
d0 = pulse.DriveChannel(0)
d1 = pulse.DriveChannel(1)
d2 = pulse.DriveChannel(2)
with pulse.build() as schedule:
with pulse.align_left():
pulse.delay(3, d0)
pulse.barrier(d0, d1, d2)
pulse.delay(11, d2)
with pulse.align_left():
pulse.delay(5, d1)
pulse.delay(7, d0)
schedule = transforms.remove_directives(schedule)
reference = pulse.Schedule()
# d0
reference.insert(0, instructions.Delay(3, d0), inplace=True)
reference.insert(3, instructions.Delay(7, d0), inplace=True)
# d1
reference.insert(3, instructions.Delay(5, d1), inplace=True)
# d2
reference.insert(3, instructions.Delay(11, d2), inplace=True)
self.assertEqual(schedule, reference)
def test_barrier_on_qubits(self):
"""Test barrier directive on qubits."""
with pulse.build(self.backend) as schedule:
pulse.barrier(0, 1)
reference = pulse.Schedule()
reference += directives.RelativeBarrier(
pulse.DriveChannel(0),
pulse.DriveChannel(1),
pulse.MeasureChannel(0),
pulse.MeasureChannel(1),
pulse.ControlChannel(0),
pulse.ControlChannel(1),
pulse.AcquireChannel(0),
pulse.AcquireChannel(1)
)
self.assertEqual(schedule, reference)
def test_trivial_barrier(self):
"""Test that trivial barrier is not added."""
with pulse.build() as schedule:
pulse.barrier(pulse.DriveChannel(0))
self.assertEqual(schedule, pulse.Schedule())
class TestUtilities(TestBuilder):
"""Test builder utilities."""
def test_active_backend(self):
"""Test getting active builder backend."""
with pulse.build(self.backend):
self.assertEqual(pulse.active_backend(), self.backend)
def test_append_schedule(self):
"""Test appending a schedule to the active builder."""
d0 = pulse.DriveChannel(0)
reference = pulse.Schedule()
reference += instructions.Delay(10, d0)
with pulse.build() as schedule:
builder.append_schedule(reference)
self.assertEqual(schedule, reference)
def test_append_instruction(self):
"""Test appending an instruction to the active builder."""
d0 = pulse.DriveChannel(0)
instruction = instructions.Delay(10, d0)
with pulse.build() as schedule:
builder.append_instruction(instruction)
self.assertEqual(schedule, instruction)
def test_qubit_channels(self):
"""Test getting the qubit channels of the active builder's backend."""
with pulse.build(self.backend):
qubit_channels = pulse.qubit_channels(0)
self.assertEqual(qubit_channels,
{pulse.DriveChannel(0),
pulse.MeasureChannel(0),
pulse.AcquireChannel(0),
pulse.ControlChannel(0),
pulse.ControlChannel(1)})
def test_active_transpiler_settings(self):
"""Test setting settings of active builder's transpiler."""
with pulse.build(self.backend):
self.assertFalse(pulse.active_transpiler_settings())
with pulse.transpiler_settings(test_setting=1):
self.assertEqual(
pulse.active_transpiler_settings()['test_setting'], 1)
def test_active_circuit_scheduler_settings(self):
"""Test setting settings of active builder's circuit scheduler."""
with pulse.build(self.backend):
self.assertFalse(pulse.active_circuit_scheduler_settings())
with pulse.circuit_scheduler_settings(test_setting=1):
self.assertEqual(
pulse.active_circuit_scheduler_settings()['test_setting'], 1)
def test_num_qubits(self):
"""Test builder utility to get number of qubits."""
with pulse.build(self.backend):
self.assertEqual(pulse.num_qubits(), 2)
def test_samples_to_seconds(self):
"""Test samples to time"""
config = self.backend.configuration()
config.dt = 0.1
with pulse.build(self.backend):
time = pulse.samples_to_seconds(100)
self.assertTrue(isinstance(time, float))
self.assertEqual(pulse.samples_to_seconds(100), 10)
def test_seconds_to_samples(self):
"""Test samples to time"""
config = self.backend.configuration()
config.dt = 0.1
with pulse.build(self.backend):
samples = pulse.seconds_to_samples(10)
self.assertTrue(isinstance(samples, int))
self.assertEqual(pulse.seconds_to_samples(10), 100)
class TestMacros(TestBuilder):
"""Test builder macros."""
def test_measure(self):
"""Test utility function - measure."""
with pulse.build(self.backend) as schedule:
reg = pulse.measure(0)
self.assertEqual(reg, pulse.MemorySlot(0))
reference = macros.measure(
qubits=[0],
inst_map=self.inst_map,
meas_map=self.configuration.meas_map)
self.assertEqual(schedule, reference)
def test_measure_all(self):
"""Test utility function - measure."""
with pulse.build(self.backend) as schedule:
regs = pulse.measure_all()
self.assertEqual(regs, [pulse.MemorySlot(0), pulse.MemorySlot(1)])
reference = macros.measure_all(self.backend)
self.assertEqual(schedule, reference)
def test_delay_qubit(self):
"""Test delaying on a qubit macro."""
with pulse.build(self.backend) as schedule:
pulse.delay_qubits(10, 0)
d0 = pulse.DriveChannel(0)
m0 = pulse.MeasureChannel(0)
a0 = pulse.AcquireChannel(0)
u0 = pulse.ControlChannel(0)
u1 = pulse.ControlChannel(1)
reference = pulse.Schedule()
reference += instructions.Delay(10, d0)
reference += instructions.Delay(10, m0)
reference += instructions.Delay(10, a0)
reference += instructions.Delay(10, u0)
reference += instructions.Delay(10, u1)
self.assertEqual(schedule, reference)
def test_delay_qubits(self):
"""Test delaying on multiple qubits to make sure we don't insert delays twice."""
with pulse.build(self.backend) as schedule:
pulse.delay_qubits(10, 0, 1)
d0 = pulse.DriveChannel(0)
d1 = pulse.DriveChannel(1)
m0 = pulse.MeasureChannel(0)
m1 = pulse.MeasureChannel(1)
a0 = pulse.AcquireChannel(0)
a1 = pulse.AcquireChannel(1)
u0 = pulse.ControlChannel(0)
u1 = pulse.ControlChannel(1)
reference = pulse.Schedule()
reference += instructions.Delay(10, d0)
reference += instructions.Delay(10, d1)
reference += instructions.Delay(10, m0)
reference += instructions.Delay(10, m1)
reference += instructions.Delay(10, a0)
reference += instructions.Delay(10, a1)
reference += instructions.Delay(10, u0)
reference += instructions.Delay(10, u1)
self.assertEqual(schedule, reference)
class TestGates(TestBuilder):
"""Test builder gates."""
def test_cx(self):
"""Test cx gate."""
with pulse.build(self.backend) as schedule:
pulse.cx(0, 1)
reference_qc = circuit.QuantumCircuit(2)
reference_qc.cx(0, 1)
reference = compiler.schedule(reference_qc, self.backend)
self.assertEqual(schedule, reference)
def test_u1(self):
"""Test u1 gate."""
with pulse.build(self.backend) as schedule:
pulse.u1(np.pi, 0)
reference_qc = circuit.QuantumCircuit(1)
reference_qc.u1(np.pi, 0)
reference = compiler.schedule(reference_qc, self.backend)
self.assertEqual(schedule, reference)
def test_u2(self):
"""Test u2 gate."""
with pulse.build(self.backend) as schedule:
pulse.u2(np.pi, 0, 0)
reference_qc = circuit.QuantumCircuit(1)
reference_qc.u2(np.pi, 0, 0)
reference = compiler.schedule(reference_qc, self.backend)
self.assertEqual(schedule, reference)
def test_u3(self):
"""Test u3 gate."""
with pulse.build(self.backend) as schedule:
pulse.u3(np.pi, 0, np.pi/2, 0)
reference_qc = circuit.QuantumCircuit(1)
reference_qc.u3(np.pi, 0, np.pi/2, 0)
reference = compiler.schedule(reference_qc, self.backend)
self.assertEqual(schedule, reference)
def test_x(self):
"""Test x gate."""
with pulse.build(self.backend) as schedule:
pulse.x(0)
reference_qc = circuit.QuantumCircuit(1)
reference_qc.x(0)
reference_qc = compiler.transpile(reference_qc, self.backend)
reference = compiler.schedule(reference_qc, self.backend)
self.assertEqual(schedule, reference)
def test_lazy_evaluation_with_transpiler(self):
"""Test that the two cx gates are optimizied away by the transpiler."""
with pulse.build(self.backend) as schedule:
pulse.cx(0, 1)
pulse.cx(0, 1)
reference_qc = circuit.QuantumCircuit(2)
reference = compiler.schedule(reference_qc, self.backend)
self.assertEqual(schedule, reference)
def test_measure(self):
"""Test pulse measurement macro against circuit measurement and
ensure agreement."""
with pulse.build(self.backend) as schedule:
with pulse.align_sequential():
pulse.x(0)
pulse.measure(0)
reference_qc = circuit.QuantumCircuit(1, 1)
reference_qc.x(0)
reference_qc.measure(0, 0)
reference_qc = compiler.transpile(reference_qc, self.backend)
reference = compiler.schedule(reference_qc, self.backend)
self.assertEqual(schedule, reference)
def test_backend_require(self):
"""Test that a backend is required to use a gate."""
with self.assertRaises(exceptions.BackendNotSet):
with pulse.build():
pulse.x(0)
def test_call_circuit_with_cregs(self):
"""Test calling of circuit wiht classical registers."""
qc = circuit.QuantumCircuit(2, 2)
qc.h(0)
qc.cx(0, 1)
qc.measure([0, 1], [0, 1])
with pulse.build(self.backend) as schedule:
pulse.call(qc)
reference_qc = compiler.transpile(qc, self.backend)
reference = compiler.schedule(reference_qc, self.backend)
self.assertEqual(schedule, reference)
class TestBuilderComposition(TestBuilder):
"""Test more sophisticated composite builder examples."""
def test_complex_build(self):
"""Test a general program build with nested contexts,
circuits and macros."""
d0 = pulse.DriveChannel(0)
d1 = pulse.DriveChannel(1)
d2 = pulse.DriveChannel(2)
delay_dur = 19
short_dur = 31
long_dur = 101
with pulse.build(self.backend) as schedule:
with pulse.align_sequential():
pulse.delay(delay_dur, d0)
pulse.u2(0, pi/2, 1)
with pulse.align_right():
pulse.play(pulse_lib.Constant(short_dur, 0.1), d1)
pulse.play(pulse_lib.Constant(long_dur, 0.1), d2)
pulse.u2(0, pi/2, 1)
with pulse.align_left():
pulse.u2(0, pi/2, 0)
pulse.u2(0, pi/2, 1)
pulse.u2(0, pi/2, 0)
pulse.measure(0)
# prepare and schedule circuits that will be used.
single_u2_qc = circuit.QuantumCircuit(2)
single_u2_qc.u2(0, pi/2, 1)
single_u2_qc = compiler.transpile(single_u2_qc, self.backend)
single_u2_sched = compiler.schedule(single_u2_qc, self.backend)
# sequential context
sequential_reference = pulse.Schedule()
sequential_reference += instructions.Delay(delay_dur, d0)
sequential_reference.insert(delay_dur, single_u2_sched, inplace=True)
# align right
align_right_reference = pulse.Schedule()
align_right_reference += pulse.Play(
pulse_lib.Constant(long_dur, 0.1), d2)
align_right_reference.insert(long_dur-single_u2_sched.duration,
single_u2_sched,
inplace=True)
align_right_reference.insert(
long_dur-single_u2_sched.duration-short_dur,
pulse.Play(pulse_lib.Constant(short_dur, 0.1), d1),
inplace=True)
# align left
triple_u2_qc = circuit.QuantumCircuit(2)
triple_u2_qc.u2(0, pi/2, 0)
triple_u2_qc.u2(0, pi/2, 1)
triple_u2_qc.u2(0, pi/2, 0)
triple_u2_qc = compiler.transpile(triple_u2_qc, self.backend)
align_left_reference = compiler.schedule(
triple_u2_qc, self.backend, method='alap')
# measurement
measure_reference = macros.measure(qubits=[0],
inst_map=self.inst_map,
meas_map=self.configuration.meas_map)
reference = pulse.Schedule()
reference += sequential_reference
# Insert so that the long pulse on d2 occurs as early as possible
# without an overval on d1.
insert_time = (reference.ch_stop_time(d1) -
align_right_reference.ch_start_time(d1))
reference.insert(insert_time,
align_right_reference,
inplace=True)
reference.insert(reference.ch_stop_time(d0, d1),
align_left_reference,
inplace=True)
reference += measure_reference
self.assertEqual(schedule, reference)

View File

@ -15,9 +15,8 @@
"""Unit tests for pulse instructions."""
import numpy as np
from qiskit.pulse import DriveChannel, AcquireChannel, MemorySlot, pulse_lib
from qiskit.pulse import Delay, Play, ShiftPhase, Snapshot, SetFrequency, Acquire
from qiskit.pulse.configuration import Kernel, Discriminator
from qiskit.pulse import channels, configuration, instructions, pulse_lib
from qiskit.test import QiskitTestCase
@ -30,17 +29,22 @@ class TestAcquire(QiskitTestCase):
'start_window': 0,
'stop_window': 10
}
kernel = Kernel(name='boxcar', **kernel_opts)
kernel = configuration.Kernel(name='boxcar', **kernel_opts)
discriminator_opts = {
'neighborhoods': [{'qubits': 1, 'channels': 1}],
'cal': 'coloring',
'resample': False
}
discriminator = Discriminator(name='linear_discriminator', **discriminator_opts)
discriminator = configuration.Discriminator(name='linear_discriminator',
**discriminator_opts)
acq = Acquire(10, AcquireChannel(0), MemorySlot(0),
kernel=kernel, discriminator=discriminator, name='acquire')
acq = instructions.Acquire(10,
channels.AcquireChannel(0),
channels.MemorySlot(0),
kernel=kernel,
discriminator=discriminator,
name='acquire')
self.assertEqual(acq.duration, 10)
self.assertEqual(acq.discriminator.name, 'linear_discriminator')
@ -49,26 +53,37 @@ class TestAcquire(QiskitTestCase):
self.assertEqual(acq.kernel.params, kernel_opts)
self.assertIsInstance(acq.id, int)
self.assertEqual(acq.name, 'acquire')
self.assertEqual(acq.operands, (10, AcquireChannel(0), MemorySlot(0), None))
self.assertEqual(acq.operands, (10,
channels.AcquireChannel(0),
channels.MemorySlot(0), None))
def test_isntructions_hash(self):
def test_instructions_hash(self):
"""Test hashing for acquire instruction."""
kernel_opts = {
'start_window': 0,
'stop_window': 10
}
kernel = Kernel(name='boxcar', **kernel_opts)
kernel = configuration.Kernel(name='boxcar', **kernel_opts)
discriminator_opts = {
'neighborhoods': [{'qubits': 1, 'channels': 1}],
'cal': 'coloring',
'resample': False
}
discriminator = Discriminator(name='linear_discriminator', **discriminator_opts)
acq_1 = Acquire(10, AcquireChannel(0), MemorySlot(0),
kernel=kernel, discriminator=discriminator, name='acquire')
acq_2 = Acquire(10, AcquireChannel(0), MemorySlot(0),
kernel=kernel, discriminator=discriminator, name='acquire')
discriminator = configuration.Discriminator(
name='linear_discriminator', **discriminator_opts)
acq_1 = instructions.Acquire(10,
channels.AcquireChannel(0),
channels.MemorySlot(0),
kernel=kernel,
discriminator=discriminator,
name='acquire')
acq_2 = instructions.Acquire(10,
channels.AcquireChannel(0),
channels.MemorySlot(0),
kernel=kernel,
discriminator=discriminator,
name='acquire')
hash_1 = hash(acq_1)
hash_2 = hash(acq_2)
@ -81,19 +96,21 @@ class TestDelay(QiskitTestCase):
def test_delay(self):
"""Test delay."""
delay = Delay(10, DriveChannel(0), name='test_name')
delay = instructions.Delay(10, channels.DriveChannel(0), name='test_name')
self.assertIsInstance(delay.id, int)
self.assertEqual(delay.name, 'test_name')
self.assertEqual(delay.duration, 10)
self.assertIsInstance(delay.duration, int)
self.assertEqual(delay.operands, (10, DriveChannel(0)))
self.assertEqual(delay, Delay(10, DriveChannel(0)))
self.assertNotEqual(delay, Delay(11, DriveChannel(1)))
self.assertEqual(delay.operands, (10, channels.DriveChannel(0)))
self.assertEqual(delay, instructions.Delay(10, channels.DriveChannel(0)))
self.assertNotEqual(delay, instructions.Delay(11, channels.DriveChannel(1)))
self.assertEqual(repr(delay), "Delay(10, DriveChannel(0), name='test_name')")
# Test numpy int for duration
delay = Delay(np.int32(10), DriveChannel(0), name='test_name2')
delay = instructions.Delay(np.int32(10),
channels.DriveChannel(0),
name='test_name2')
self.assertEqual(delay.duration, 10)
self.assertIsInstance(delay.duration, np.integer)
@ -103,14 +120,20 @@ class TestSetFrequency(QiskitTestCase):
def test_freq(self):
"""Test set frequency basic functionality."""
set_freq = SetFrequency(4.5e9, DriveChannel(1), name='test')
set_freq = instructions.SetFrequency(4.5e9, channels.DriveChannel(1), name='test')
self.assertIsInstance(set_freq.id, int)
self.assertEqual(set_freq.duration, 0)
self.assertEqual(set_freq.frequency, 4.5e9)
self.assertEqual(set_freq.operands, (4.5e9, DriveChannel(1)))
self.assertEqual(set_freq, SetFrequency(4.5e9, DriveChannel(1), name='test'))
self.assertNotEqual(set_freq, SetFrequency(4.5e8, DriveChannel(1), name='test'))
self.assertEqual(set_freq.operands, (4.5e9, channels.DriveChannel(1)))
self.assertEqual(set_freq,
instructions.SetFrequency(4.5e9,
channels.DriveChannel(1),
name='test'))
self.assertNotEqual(set_freq,
instructions.SetFrequency(4.5e8,
channels.DriveChannel(1),
name='test'))
self.assertEqual(repr(set_freq),
"SetFrequency(4500000000.0, DriveChannel(1), name='test')")
@ -120,15 +143,21 @@ class TestShiftPhase(QiskitTestCase):
def test_default(self):
"""Test basic ShiftPhase."""
shift_phase = ShiftPhase(1.57, DriveChannel(0))
shift_phase = instructions.ShiftPhase(1.57, channels.DriveChannel(0))
self.assertIsInstance(shift_phase.id, int)
self.assertEqual(shift_phase.name, None)
self.assertEqual(shift_phase.duration, 0)
self.assertEqual(shift_phase.phase, 1.57)
self.assertEqual(shift_phase.operands, (1.57, DriveChannel(0)))
self.assertEqual(shift_phase, ShiftPhase(1.57, DriveChannel(0), name='test'))
self.assertNotEqual(shift_phase, ShiftPhase(1.57j, DriveChannel(0), name='test'))
self.assertEqual(shift_phase.operands, (1.57, channels.DriveChannel(0)))
self.assertEqual(shift_phase,
instructions.ShiftPhase(1.57,
channels.DriveChannel(0),
name='test'))
self.assertNotEqual(shift_phase,
instructions.ShiftPhase(1.57j,
channels.DriveChannel(0),
name='test'))
self.assertEqual(repr(shift_phase), "ShiftPhase(1.57, DriveChannel(0))")
@ -137,13 +166,13 @@ class TestSnapshot(QiskitTestCase):
def test_default(self):
"""Test default snapshot."""
snapshot = Snapshot(label='test_name', snapshot_type='state')
snapshot = instructions.Snapshot(label='test_name', snapshot_type='state')
self.assertIsInstance(snapshot.id, int)
self.assertEqual(snapshot.name, 'test_name')
self.assertEqual(snapshot.type, 'state')
self.assertEqual(snapshot.duration, 0)
self.assertNotEqual(snapshot, Delay(10, DriveChannel(0)))
self.assertNotEqual(snapshot, instructions.Delay(10, channels.DriveChannel(0)))
self.assertEqual(repr(snapshot), "Snapshot(test_name, state, name='test_name')")
@ -154,7 +183,7 @@ class TestPlay(QiskitTestCase):
"""Test basic play instruction."""
duration = 4
pulse = pulse_lib.SamplePulse([1.0] * duration, name='test')
play = Play(pulse, DriveChannel(1))
play = instructions.Play(pulse, channels.DriveChannel(1))
self.assertIsInstance(play.id, int)
self.assertEqual(play.name, pulse.name)
@ -162,3 +191,25 @@ class TestPlay(QiskitTestCase):
self.assertEqual(repr(play),
"Play(SamplePulse(array([1.+0.j, 1.+0.j, 1.+0.j, 1.+0.j]), name='test'),"
" DriveChannel(1), name='test')")
class TestDirectives(QiskitTestCase):
"""Test pulse directives."""
def test_relative_barrier(self):
"""Test the relative barrier directive."""
# pylint: disable=invalid-name
a0 = channels.AcquireChannel(0)
d0 = channels.DriveChannel(0)
m0 = channels.MeasureChannel(0)
u0 = channels.ControlChannel(0)
mem0 = channels.MemorySlot(0)
reg0 = channels.RegisterSlot(0)
chans = (a0, d0, m0, u0, mem0, reg0)
name = 'barrier'
barrier = instructions.RelativeBarrier(*chans, name=name)
self.assertEqual(barrier.name, name)
self.assertEqual(barrier.duration, 0)
self.assertEqual(barrier.channels, chans)
self.assertEqual(barrier.operands, chans)

View File

@ -12,27 +12,35 @@
# copyright notice, and modified files need to carry a notice indicating
# that they have been altered from the originals.
"""Test cases for Scheduler Utility functions."""
"""Test cases for Pulse Macro functions."""
from qiskit.pulse import (Schedule, AcquireChannel, Acquire, InstructionScheduleMap,
MeasureChannel, MemorySlot, GaussianSquare, Play)
from qiskit.scheduler import measure, measure_all
from qiskit.pulse import (
Schedule,
AcquireChannel,
Acquire,
InstructionScheduleMap,
MeasureChannel,
MemorySlot,
GaussianSquare,
Play,
)
from qiskit.pulse import macros
from qiskit.pulse.exceptions import PulseError
from qiskit.test.mock import FakeOpenPulse2Q
from qiskit.test import QiskitTestCase
class TestUtils(QiskitTestCase):
"""Utils tests."""
class TestMacros(QiskitTestCase):
"""Pulse Macro tests."""
def setUp(self):
self.backend = FakeOpenPulse2Q()
self.inst_map = self.backend.defaults().instruction_schedule_map
def test_measure(self):
"""Test utility function - measure."""
sched = measure(qubits=[0],
backend=self.backend)
"""Test macro - measure."""
sched = macros.measure(qubits=[0],
backend=self.backend)
expected = Schedule(
self.inst_map.get('measure', [0, 1]).filter(channels=[MeasureChannel(0)]),
Acquire(10, AcquireChannel(0), MemorySlot(0)),
@ -41,9 +49,9 @@ class TestUtils(QiskitTestCase):
def test_measure_sched_with_qubit_mem_slots(self):
"""Test measure with custom qubit_mem_slots."""
sched = measure(qubits=[0],
backend=self.backend,
qubit_mem_slots={0: 1})
sched = macros.measure(qubits=[0],
backend=self.backend,
qubit_mem_slots={0: 1})
expected = Schedule(
self.inst_map.get('measure', [0, 1]).filter(channels=[MeasureChannel(0)]),
Acquire(10, AcquireChannel(0), MemorySlot(1)),
@ -52,12 +60,12 @@ class TestUtils(QiskitTestCase):
def test_measure_sched_with_meas_map(self):
"""Test measure with custom meas_map as list and dict."""
sched_with_meas_map_list = measure(qubits=[0],
backend=self.backend,
meas_map=[[0, 1]])
sched_with_meas_map_dict = measure(qubits=[0],
backend=self.backend,
meas_map={0: [0, 1], 1: [0, 1]})
sched_with_meas_map_list = macros.measure(qubits=[0],
backend=self.backend,
meas_map=[[0, 1]])
sched_with_meas_map_dict = macros.measure(qubits=[0],
backend=self.backend,
meas_map={0: [0, 1], 1: [0, 1]})
expected = Schedule(
self.inst_map.get('measure', [0, 1]).filter(channels=[MeasureChannel(0)]),
Acquire(10, AcquireChannel(0), MemorySlot(0)),
@ -71,29 +79,29 @@ class TestUtils(QiskitTestCase):
q0_sched += Acquire(1200, AcquireChannel(0), MemorySlot(0))
inst_map = InstructionScheduleMap()
inst_map.add('my_sched', 0, q0_sched)
sched = measure(qubits=[0],
measure_name='my_sched',
inst_map=inst_map,
meas_map=[[0]])
sched = macros.measure(qubits=[0],
measure_name='my_sched',
inst_map=inst_map,
meas_map=[[0]])
self.assertEqual(sched.instructions, q0_sched.instructions)
with self.assertRaises(PulseError):
measure(qubits=[0],
measure_name="name",
inst_map=inst_map,
meas_map=[[0]])
macros.measure(qubits=[0],
measure_name="name",
inst_map=inst_map,
meas_map=[[0]])
def test_fail_measure(self):
"""Test failing measure."""
with self.assertRaises(PulseError):
measure(qubits=[0],
meas_map=self.backend.configuration().meas_map)
macros.measure(qubits=[0],
meas_map=self.backend.configuration().meas_map)
with self.assertRaises(PulseError):
measure(qubits=[0],
inst_map=self.inst_map)
macros.measure(qubits=[0],
inst_map=self.inst_map)
def test_measure_all(self):
"""Test measure_all function."""
sched = measure_all(self.backend)
sched = macros.measure_all(self.backend)
expected = Schedule(self.inst_map.get('measure', [0, 1]))
self.assertEqual(sched.instructions, expected.instructions)

View File

@ -18,11 +18,32 @@ from unittest.mock import patch
import numpy as np
from qiskit.pulse import (Play, SamplePulse, ShiftPhase, Instruction, SetFrequency, Acquire,
pulse_lib, Snapshot, Delay, Gaussian, Drag, GaussianSquare, Constant,
functional_pulse, ShiftFrequency, SetPhase)
from qiskit.pulse.channels import (MemorySlot, RegisterSlot, DriveChannel, AcquireChannel,
SnapshotChannel, MeasureChannel)
from qiskit.pulse import (
Play,
SamplePulse,
ShiftPhase,
Instruction,
SetFrequency,
Acquire,
pulse_lib,
Snapshot,
Delay,
Gaussian,
Drag,
GaussianSquare,
Constant,
functional_pulse,
ShiftFrequency,
SetPhase,
)
from qiskit.pulse.channels import (
MemorySlot,
RegisterSlot,
DriveChannel,
AcquireChannel,
SnapshotChannel,
MeasureChannel,
)
from qiskit.pulse.commands import PersistentValue, PulseInstruction
from qiskit.pulse.exceptions import PulseError
from qiskit.pulse.schedule import Schedule, ParameterizedSchedule, _overlaps, _insertion_index
@ -263,19 +284,19 @@ class TestScheduleBuilding(BaseTestSchedule):
self.assertEqual([100, 130, 140], start_times)
def test_keep_original_schedule_after_attached_to_another_schedule(self):
"""Test if a schedule keeps its _children after attached to another schedule."""
_children = (Acquire(10, self.config.acquire(0), MemorySlot(0)).shift(20) +
Acquire(10, self.config.acquire(0), MemorySlot(0)))
self.assertEqual(2, len(list(_children.instructions)))
"""Test if a schedule keeps its children after attached to another schedule."""
children = (Acquire(10, self.config.acquire(0), MemorySlot(0)).shift(20) +
Acquire(10, self.config.acquire(0), MemorySlot(0)))
self.assertEqual(2, len(list(children.instructions)))
sched = Acquire(10, self.config.acquire(0), MemorySlot(0)).append(_children)
sched = Acquire(10, self.config.acquire(0), MemorySlot(0)).append(children)
self.assertEqual(3, len(list(sched.instructions)))
# add 2 instructions to _children (2 instructions -> 4 instructions)
_children = _children.append(Acquire(10, self.config.acquire(0), MemorySlot(0)))
_children = _children.insert(100, Acquire(10, self.config.acquire(0),
MemorySlot(0)))
self.assertEqual(4, len(list(_children.instructions)))
# add 2 instructions to children (2 instructions -> 4 instructions)
children = children.append(Acquire(10, self.config.acquire(0), MemorySlot(0)))
children = children.insert(
100, Acquire(10, self.config.acquire(0), MemorySlot(0)))
self.assertEqual(4, len(list(children.instructions)))
# sched must keep 3 instructions (must not update to 5 instructions)
self.assertEqual(3, len(list(sched.instructions)))

View File

@ -19,14 +19,25 @@ from typing import List, Set
import numpy as np
from qiskit import pulse
from qiskit.pulse import (Play, Delay, Acquire, Schedule, SamplePulse, Drag,
Gaussian, GaussianSquare, Constant)
from qiskit.pulse import (
Play,
Delay,
Acquire,
Schedule,
SamplePulse,
Drag,
Gaussian,
GaussianSquare,
Constant,
)
from qiskit.pulse import transforms, instructions
from qiskit.pulse.channels import MeasureChannel, MemorySlot, DriveChannel, AcquireChannel
from qiskit.pulse.exceptions import PulseError
from qiskit.pulse.instructions import directives
from qiskit.test import QiskitTestCase
from qiskit.test.mock import FakeOpenPulse2Q
from qiskit.pulse.transforms import add_implicit_acquires, align_measures, pad, compress_pulses
# pylint: disable=invalid-name
class TestAutoMerge(QiskitTestCase):
@ -47,12 +58,13 @@ class TestAutoMerge(QiskitTestCase):
sched = sched.insert(10, Acquire(5, self.config.acquire(1), MemorySlot(1)))
sched = sched.insert(10, Play(self.short_pulse, self.config.measure(0)))
sched = sched.insert(10, Play(self.short_pulse, self.config.measure(1)))
sched = align_measures([sched], self.inst_map)[0]
sched = transforms.align_measures([sched], self.inst_map)[0]
self.assertEqual(sched.name, 'fake_experiment')
for time, inst in sched.instructions:
if isinstance(inst, Acquire):
self.assertEqual(time, 10)
sched = align_measures([sched], self.inst_map, align_time=20)[0]
sched = transforms.align_measures(
[sched], self.inst_map, align_time=20)[0]
for time, inst in sched.instructions:
if isinstance(inst, Acquire):
self.assertEqual(time, 20)
@ -65,11 +77,12 @@ class TestAutoMerge(QiskitTestCase):
sched = pulse.Schedule(name='fake_experiment')
sched = sched.insert(0, Play(self.short_pulse, self.config.drive(0)))
sched = sched.insert(1, Acquire(5, self.config.acquire(0), MemorySlot(0)))
sched = align_measures([sched], self.inst_map)[0]
sched = transforms.align_measures([sched], self.inst_map)[0]
for time, inst in sched.instructions:
if isinstance(inst, Acquire):
self.assertEqual(time, 4)
sched = align_measures([sched], self.inst_map, max_calibration_duration=10)[0]
sched = transforms.align_measures(
[sched], self.inst_map, max_calibration_duration=10)[0]
for time, inst in sched.instructions:
if isinstance(inst, Acquire):
self.assertEqual(time, 10)
@ -81,22 +94,22 @@ class TestAutoMerge(QiskitTestCase):
sched = sched.insert(4, Acquire(5, self.config.acquire(0), MemorySlot(0)))
sched = sched.insert(10, Acquire(5, self.config.acquire(0), MemorySlot(0)))
with self.assertRaises(PulseError):
align_measures([sched], self.inst_map)
transforms.align_measures([sched], self.inst_map)
# Test for measure channel
sched = pulse.Schedule(name='fake_experiment')
sched = sched.insert(10, Play(self.short_pulse, self.config.measure(0)))
sched = sched.insert(30, Play(self.short_pulse, self.config.measure(0)))
with self.assertRaises(PulseError):
align_measures([sched], self.inst_map)
transforms.align_measures([sched], self.inst_map)
# Test both using inst_map
sched = pulse.Schedule()
sched += self.inst_map.get('measure', (0, 1))
align_measures([sched], align_time=50)
transforms.align_measures([sched], align_time=50)
sched += self.inst_map.get('measure', (0, 1))
with self.assertRaises(PulseError):
align_measures([sched], align_time=50)
transforms.align_measures([sched], align_time=50)
def test_error_post_acquire_pulse(self):
"""Test that an error is raised if a pulse occurs on a channel after an acquire."""
@ -105,10 +118,10 @@ class TestAutoMerge(QiskitTestCase):
sched = sched.insert(4, Acquire(5, self.config.acquire(0), MemorySlot(0)))
# No error with separate channel
sched = sched.insert(10, Play(self.short_pulse, self.config.drive(1)))
align_measures([sched], self.inst_map)
transforms.align_measures([sched], self.inst_map)
sched = sched.insert(10, Play(self.short_pulse, self.config.drive(0)))
with self.assertRaises(PulseError):
align_measures([sched], self.inst_map)
transforms.align_measures([sched], self.inst_map)
def test_align_across_schedules(self):
"""Test that acquires are aligned together across multiple schedules."""
@ -118,7 +131,7 @@ class TestAutoMerge(QiskitTestCase):
sched2 = pulse.Schedule(name='fake_experiment')
sched2 = sched2.insert(3, Play(self.short_pulse, self.config.drive(0)))
sched2 = sched2.insert(25, Acquire(5, self.config.acquire(0), MemorySlot(0)))
schedules = align_measures([sched1, sched2], self.inst_map)
schedules = transforms.align_measures([sched1, sched2], self.inst_map)
for time, inst in schedules[0].instructions:
if isinstance(inst, Acquire):
self.assertEqual(time, 25)
@ -143,7 +156,7 @@ class TestAddImplicitAcquires(QiskitTestCase):
def test_add_implicit(self):
"""Test that implicit acquires are made explicit according to the meas map."""
sched = add_implicit_acquires(self.sched, [[0, 1]])
sched = transforms.add_implicit_acquires(self.sched, [[0, 1]])
acquired_qubits = set()
for _, inst in sched.instructions:
if isinstance(inst, Acquire):
@ -152,7 +165,7 @@ class TestAddImplicitAcquires(QiskitTestCase):
def test_add_across_meas_map_sublists(self):
"""Test that implicit acquires in separate meas map sublists are all added."""
sched = add_implicit_acquires(self.sched, [[0, 2], [1, 3]])
sched = transforms.add_implicit_acquires(self.sched, [[0, 2], [1, 3]])
acquired_qubits = set()
for _, inst in sched.instructions:
if isinstance(inst, Acquire):
@ -161,7 +174,8 @@ class TestAddImplicitAcquires(QiskitTestCase):
def test_dont_add_all(self):
"""Test that acquires aren't added if no qubits in the sublist aren't being acquired."""
sched = add_implicit_acquires(self.sched, [[4, 5], [0, 2], [1, 3]])
sched = transforms.add_implicit_acquires(
self.sched, [[4, 5], [0, 2], [1, 3]])
acquired_qubits = set()
for _, inst in sched.instructions:
if isinstance(inst, Acquire):
@ -174,7 +188,7 @@ class TestAddImplicitAcquires(QiskitTestCase):
acq_q0 = pulse.Acquire(1200, AcquireChannel(0), MemorySlot(0))
sched += acq_q0
sched += acq_q0 << sched.duration
sched = add_implicit_acquires(sched, meas_map=[[0]])
sched = transforms.add_implicit_acquires(sched, meas_map=[[0]])
self.assertEqual(sched.instructions, ((0, acq_q0), (2400, acq_q0)))
@ -183,7 +197,7 @@ class TestPad(QiskitTestCase):
def test_padding_empty_schedule(self):
"""Test padding of empty schedule."""
self.assertEqual(pulse.Schedule(), pad(pulse.Schedule()))
self.assertEqual(pulse.Schedule(), transforms.pad(pulse.Schedule()))
def test_padding_schedule(self):
"""Test padding schedule."""
@ -198,7 +212,7 @@ class TestPad(QiskitTestCase):
Delay(delay, DriveChannel(1)) |
Delay(2 * delay, DriveChannel(1)).shift(20))
self.assertEqual(pad(sched), ref_sched)
self.assertEqual(transforms.pad(sched), ref_sched)
def test_padding_schedule_inverse_order(self):
"""Test padding schedule is insensitive to order in which commands were added.
@ -217,7 +231,7 @@ class TestPad(QiskitTestCase):
Delay(delay, DriveChannel(1)) |
Delay(2 * delay, DriveChannel(1)).shift(20))
self.assertEqual(pad(sched), ref_sched)
self.assertEqual(transforms.pad(sched), ref_sched)
def test_padding_until_less(self):
"""Test padding until time that is less than schedule duration."""
@ -230,7 +244,7 @@ class TestPad(QiskitTestCase):
Delay(delay, DriveChannel(0)) |
Delay(5, DriveChannel(1)).shift(10))
self.assertEqual(pad(sched, until=15), ref_sched)
self.assertEqual(transforms.pad(sched, until=15), ref_sched)
def test_padding_until_greater(self):
"""Test padding until time that is greater than schedule duration."""
@ -244,7 +258,7 @@ class TestPad(QiskitTestCase):
Delay(30, DriveChannel(0)).shift(20) |
Delay(40, DriveChannel(1)).shift(10))
self.assertEqual(pad(sched, until=50), ref_sched)
self.assertEqual(transforms.pad(sched, until=50), ref_sched)
def test_padding_supplied_channels(self):
"""Test padding of only specified channels."""
@ -258,7 +272,7 @@ class TestPad(QiskitTestCase):
channels = [DriveChannel(0), DriveChannel(2)]
self.assertEqual(pad(sched, channels=channels), ref_sched)
self.assertEqual(transforms.pad(sched, channels=channels), ref_sched)
def test_padding_less_than_sched_duration(self):
"""Test that the until arg is respected even for less than the input schedule duration."""
@ -266,7 +280,7 @@ class TestPad(QiskitTestCase):
sched = (Delay(delay, DriveChannel(0)) +
Delay(delay, DriveChannel(0)).shift(20))
ref_sched = (sched | pulse.Delay(5, DriveChannel(0)).shift(10))
self.assertEqual(pad(sched, until=15), ref_sched)
self.assertEqual(transforms.pad(sched, until=15), ref_sched)
def get_pulse_ids(schedules: List[Schedule]) -> Set[int]:
@ -288,7 +302,7 @@ class TestCompressTransform(QiskitTestCase):
schedule += Play(SamplePulse([0.0, 0.1]), drive_channel)
schedule += Play(SamplePulse([0.0, 0.1]), drive_channel)
compressed_schedule = compress_pulses([schedule])
compressed_schedule = transforms.compress_pulses([schedule])
original_pulse_ids = get_pulse_ids([schedule])
compressed_pulse_ids = get_pulse_ids(compressed_schedule)
@ -304,7 +318,7 @@ class TestCompressTransform(QiskitTestCase):
schedule += Play(SamplePulse([0.0, 1.001], epsilon=1e-3), drive_channel)
schedule += Play(SamplePulse([0.0, 1.0000000001]), drive_channel)
compressed_schedule = compress_pulses([schedule])
compressed_schedule = transforms.compress_pulses([schedule])
original_pulse_ids = get_pulse_ids([schedule])
compressed_pulse_ids = get_pulse_ids(compressed_schedule)
@ -320,7 +334,7 @@ class TestCompressTransform(QiskitTestCase):
schedule += Play(SamplePulse([0.0, 0.9]), drive_channel)
schedule += Play(SamplePulse([0.0, 0.3]), drive_channel)
compressed_schedule = compress_pulses([schedule])
compressed_schedule = transforms.compress_pulses([schedule])
original_pulse_ids = get_pulse_ids([schedule])
compressed_pulse_ids = get_pulse_ids(compressed_schedule)
self.assertEqual(len(original_pulse_ids), len(compressed_pulse_ids))
@ -340,7 +354,7 @@ class TestCompressTransform(QiskitTestCase):
schedule += Play(Drag(duration=25, amp=0.2 + 0.3j, sigma=7.8, beta=4), drive_channel)
schedule += Play(Drag(duration=25, amp=0.2 + 0.3j, sigma=7.8, beta=4), drive_channel)
compressed_schedule = compress_pulses([schedule])
compressed_schedule = transforms.compress_pulses([schedule])
original_pulse_ids = get_pulse_ids([schedule])
compressed_pulse_ids = get_pulse_ids(compressed_schedule)
self.assertEqual(len(original_pulse_ids), 8)
@ -361,7 +375,7 @@ class TestCompressTransform(QiskitTestCase):
schedule += Play(Drag(duration=25, amp=0.2 + 0.3j, sigma=7.8, beta=4), drive_channel)
schedule += Play(Drag(duration=25, amp=0.2 + 0.31j, sigma=7.8, beta=4), drive_channel)
compressed_schedule = compress_pulses([schedule])
compressed_schedule = transforms.compress_pulses([schedule])
original_pulse_ids = get_pulse_ids([schedule])
compressed_pulse_ids = get_pulse_ids(compressed_schedule)
self.assertEqual(len(original_pulse_ids), len(compressed_pulse_ids))
@ -372,7 +386,7 @@ class TestCompressTransform(QiskitTestCase):
schedule += Play(SamplePulse([0.0, 0.1]), DriveChannel(0))
schedule += Play(SamplePulse([0.0, 0.1]), DriveChannel(1))
compressed_schedule = compress_pulses([schedule])
compressed_schedule = transforms.compress_pulses([schedule])
original_pulse_ids = get_pulse_ids([schedule])
compressed_pulse_ids = get_pulse_ids(compressed_schedule)
self.assertEqual(len(original_pulse_ids), 2)
@ -384,7 +398,7 @@ class TestCompressTransform(QiskitTestCase):
schedule += Play(SamplePulse([0.0, 0.1001], epsilon=1e-3), DriveChannel(0))
schedule += Play(SamplePulse([0.0, 0.1], epsilon=1e-3), DriveChannel(1))
compressed_schedule = compress_pulses([schedule])
compressed_schedule = transforms.compress_pulses([schedule])
original_pulse_ids = get_pulse_ids([schedule])
compressed_pulse_ids = get_pulse_ids(compressed_schedule)
self.assertEqual(len(original_pulse_ids), 2)
@ -401,12 +415,259 @@ class TestCompressTransform(QiskitTestCase):
schedule += Play(SamplePulse([0.0, 0.2]), drive_channel)
schedules.append(schedule)
compressed_schedule = compress_pulses(schedules)
compressed_schedule = transforms.compress_pulses(schedules)
original_pulse_ids = get_pulse_ids(schedules)
compressed_pulse_ids = get_pulse_ids(compressed_schedule)
self.assertEqual(len(original_pulse_ids), 6)
self.assertEqual(len(compressed_pulse_ids), 2)
class TestAlignSequential(QiskitTestCase):
"""Test sequential alignment transform."""
def test_align_sequential(self):
"""Test sequential alignment without a barrier."""
d0 = pulse.DriveChannel(0)
d1 = pulse.DriveChannel(1)
schedule = pulse.Schedule()
schedule.insert(1, instructions.Delay(3, d0), inplace=True)
schedule.insert(4, instructions.Delay(5, d1), inplace=True)
schedule.insert(12, instructions.Delay(7, d0), inplace=True)
schedule = transforms.align_sequential(schedule)
reference = pulse.Schedule()
# d0
reference.insert(0, instructions.Delay(3, d0), inplace=True)
reference.insert(8, instructions.Delay(7, d0), inplace=True)
# d1
reference.insert(3, instructions.Delay(5, d1), inplace=True)
self.assertEqual(schedule, reference)
def test_align_sequential_with_barrier(self):
"""Test sequential alignment with a barrier."""
d0 = pulse.DriveChannel(0)
d1 = pulse.DriveChannel(1)
schedule = pulse.Schedule()
schedule.insert(1, instructions.Delay(3, d0), inplace=True)
schedule.append(directives.RelativeBarrier(d0, d1), inplace=True)
schedule.insert(4, instructions.Delay(5, d1), inplace=True)
schedule.insert(12, instructions.Delay(7, d0), inplace=True)
schedule = transforms.align_sequential(schedule)
reference = pulse.Schedule()
reference.insert(0, instructions.Delay(3, d0), inplace=True)
reference.insert(3, directives.RelativeBarrier(d0, d1), inplace=True)
reference.insert(3, instructions.Delay(5, d1), inplace=True)
reference.insert(8, instructions.Delay(7, d0), inplace=True)
self.assertEqual(schedule, reference)
class TestAlignLeft(QiskitTestCase):
"""Test left alignment transform."""
def test_align_left(self):
"""Test left alignment without a barrier."""
d0 = pulse.DriveChannel(0)
d1 = pulse.DriveChannel(1)
d2 = pulse.DriveChannel(2)
schedule = pulse.Schedule()
schedule.insert(1, instructions.Delay(3, d0), inplace=True)
schedule.insert(17, instructions.Delay(11, d2), inplace=True)
sched_grouped = pulse.Schedule()
sched_grouped += instructions.Delay(5, d1)
sched_grouped += instructions.Delay(7, d0)
schedule.append(sched_grouped, inplace=True)
schedule = transforms.align_left(schedule)
reference = pulse.Schedule()
# d0
reference.insert(0, instructions.Delay(3, d0), inplace=True)
reference.insert(3, instructions.Delay(7, d0), inplace=True)
# d1
reference.insert(3, instructions.Delay(5, d1), inplace=True)
# d2
reference.insert(0, instructions.Delay(11, d2), inplace=True)
self.assertEqual(schedule, reference)
def test_align_left_with_barrier(self):
"""Test left alignment with a barrier."""
d0 = pulse.DriveChannel(0)
d1 = pulse.DriveChannel(1)
d2 = pulse.DriveChannel(2)
schedule = pulse.Schedule()
schedule.insert(1, instructions.Delay(3, d0), inplace=True)
schedule.append(directives.RelativeBarrier(d0, d1, d2), inplace=True)
schedule.insert(17, instructions.Delay(11, d2), inplace=True)
sched_grouped = pulse.Schedule()
sched_grouped += instructions.Delay(5, d1)
sched_grouped += instructions.Delay(7, d0)
schedule.append(sched_grouped, inplace=True)
schedule = transforms.remove_directives(
transforms.align_left(schedule))
reference = pulse.Schedule()
# d0
reference.insert(0, instructions.Delay(3, d0), inplace=True)
reference.insert(3, instructions.Delay(7, d0), inplace=True)
# d1
reference = reference.insert(3, instructions.Delay(5, d1))
# d2
reference = reference.insert(3, instructions.Delay(11, d2))
self.assertEqual(schedule, reference)
class TestAlignRight(QiskitTestCase):
"""Test right alignment transform."""
def test_align_right(self):
"""Test right alignment without a barrier."""
d0 = pulse.DriveChannel(0)
d1 = pulse.DriveChannel(1)
d2 = pulse.DriveChannel(2)
schedule = pulse.Schedule()
schedule.insert(1, instructions.Delay(3, d0), inplace=True)
schedule.insert(17, instructions.Delay(11, d2), inplace=True)
sched_grouped = pulse.Schedule()
sched_grouped.insert(2, instructions.Delay(5, d1), inplace=True)
sched_grouped += instructions.Delay(7, d0)
schedule.append(sched_grouped, inplace=True)
schedule = transforms.align_right(schedule)
reference = pulse.Schedule()
# d0
reference.insert(1, instructions.Delay(3, d0), inplace=True)
reference.insert(4, instructions.Delay(7, d0), inplace=True)
# d1
reference.insert(6, instructions.Delay(5, d1), inplace=True)
# d2
reference.insert(0, instructions.Delay(11, d2), inplace=True)
self.assertEqual(schedule, reference)
def test_align_right_with_barrier(self):
"""Test right alignment with a barrier."""
d0 = pulse.DriveChannel(0)
d1 = pulse.DriveChannel(1)
d2 = pulse.DriveChannel(2)
schedule = pulse.Schedule()
schedule.insert(1, instructions.Delay(3, d0), inplace=True)
schedule.append(directives.RelativeBarrier(d0, d1, d2), inplace=True)
schedule.insert(17, instructions.Delay(11, d2), inplace=True)
sched_grouped = pulse.Schedule()
sched_grouped.insert(2, instructions.Delay(5, d1), inplace=True)
sched_grouped += instructions.Delay(7, d0)
schedule.append(sched_grouped, inplace=True)
schedule = transforms.remove_directives(transforms.align_right(schedule))
reference = pulse.Schedule()
# d0
reference.insert(0, instructions.Delay(3, d0), inplace=True)
reference.insert(7, instructions.Delay(7, d0), inplace=True)
# d1
reference.insert(9, instructions.Delay(5, d1), inplace=True)
# d2
reference.insert(3, instructions.Delay(11, d2), inplace=True)
self.assertEqual(schedule, reference)
class TestFlatten(QiskitTestCase):
"""Test flattening transform."""
def test_flatten(self):
"""Test the flatten transform."""
d0 = pulse.DriveChannel(0)
d1 = pulse.DriveChannel(1)
schedule = pulse.Schedule()
schedule += instructions.Delay(3, d0)
grouped = pulse.Schedule()
grouped += instructions.Delay(5, d1)
grouped += instructions.Delay(7, d0)
# include a grouped schedule
grouped = schedule + grouped
# flatten the schedule inline internal groups
flattened = transforms.flatten(grouped)
# align all the instructions to the left after flattening
flattened = transforms.align_left(flattened)
grouped = transforms.align_left(grouped)
reference = pulse.Schedule()
# d0
reference.insert(0, instructions.Delay(3, d0), inplace=True)
reference.insert(3, instructions.Delay(7, d0), inplace=True)
# d1
reference.insert(0, instructions.Delay(5, d1), inplace=True)
self.assertEqual(flattened, reference)
self.assertNotEqual(grouped, reference)
class _TestDirective(directives.Directive):
"""Pulse ``RelativeBarrier`` directive."""
def __init__(self, *channels):
"""Test directive"""
super().__init__(tuple(channels), 0, tuple(channels))
class TestRemoveDirectives(QiskitTestCase):
"""Test removing of directives."""
def test_remove_directives(self):
"""Test that all directives are removed."""
d0 = pulse.DriveChannel(0)
d1 = pulse.DriveChannel(1)
schedule = pulse.Schedule()
schedule += _TestDirective(d0, d1)
schedule += instructions.Delay(3, d0)
schedule += _TestDirective(d0, d1)
schedule = transforms.remove_directives(schedule)
reference = pulse.Schedule()
# d0
reference += instructions.Delay(3, d0)
self.assertEqual(schedule, reference)
class TestRemoveTrivialBarriers(QiskitTestCase):
"""Test scheduling transforms."""
def test_remove_trivial_barriers(self):
"""Test that trivial barriers are properly removed."""
schedule = pulse.Schedule()
schedule += directives.RelativeBarrier()
schedule += directives.RelativeBarrier(pulse.DriveChannel(0))
schedule += directives.RelativeBarrier(pulse.DriveChannel(0),
pulse.DriveChannel(1))
schedule = transforms.remove_trivial_barriers(schedule)
reference = pulse.Schedule()
reference += directives.RelativeBarrier(pulse.DriveChannel(0),
pulse.DriveChannel(1))
self.assertEqual(schedule, reference)
if __name__ == '__main__':
unittest.main()