Delay instruction in QuantumCircuit and circuit scheduling (#4555)

* add circuit.delay

* add scheduling passes

* add sequence() that translates scheduled circuit to pulse schedule

* add unit in drawing delays

* add to_matrix to circuit.Delay

* fix a critical bug not able to transpile normal circuits

* Fix a bug embeded at the latest merge

fix a bug embeded at the latest merge

fix a bug embeded at the latest merge

* Drop timestep

* Rename file

* Rename mirror to reverse_ops

* Fix num_qubits of Delay to be always one

Fix num_qubits of Delay to be always one

Fix num_qubits of Delay to be always one

* lint

* Add arguments checks for transpile() with scheduling_method and add some tests

* Accept Parameter for duration of delay

* lint

* Raise error when executing circuits with delays except for schedule_circuit on pulse backends

* Remove duration from Instruction

* Move apply_prefix to util

* lint

lint

lint

* Put dt conversion logics into one place (delayindt) and make rounding logics consistent with #4480

Put dt conversion logics into one place (delayindt) and make rounding logics consistent with #4480

Remove unnecessary input check in DelayInDt

* lint

* Add and update tests

* Move setting of scheduling passes to preset managers

* Fix scheduling passes

* Fix Schedule.__eq__

* Add RemoveOpsOnIdleQubits pass

* Remove MeasureReschedule pass

* Update InstructionDurations and remove DurationMapper

* Add QuantumCircuit.instruction_durations

* Add instruction_durations argument to schedule()

* Remove use of circuit.Instruction.duration

* Fix a bug in instruction durations

* Catch up with the latest codes

* Update qiskit/circuit/quantumcircuit.py

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

* Update qiskit/circuit/quantumcircuit.py

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

* Update qiskit/execute.py

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

* Update qiskit/execute.py

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

* Update test/python/circuit/test_scheduled_circuit.py

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

* Fix bugs suggested by linter

* Move type and value checks in delay() to Delay

* docstring

* Revert "Update qiskit/execute.py"(7593df8)

Fix a critical bug embedded at 7593df8

Revert "Fix a critical bug embedded at 7593df8"

This reverts commit 1de91d4951054889302dcc703e8a396483d9e88f.

Revert "Update qiskit/execute.py"

This reverts commit 7593df8f58.

* Update qiskit/compiler/transpile.py

Update qiskit/compiler/transpile.py

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

lint

* Remove unnecessary asserts

* Add InstructionDurations to InstructionDurationsType

* Fix a bug on circuit.instruction_durations

* Update sequence() to be independent of schedule()

* Remove QuantumCircuit.instruction_durations

* Add release note for delay in circuit

* Attempt to fix type hint for docs

* Update releasenotes/notes/delay-in-circuit-33f0d81783ac12ea.yaml

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

* Update releasenotes/notes/delay-in-circuit-33f0d81783ac12ea.yaml

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

* Update releasenotes/notes/delay-in-circuit-33f0d81783ac12ea.yaml

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

* Update qiskit/transpiler/instruction_durations.py

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

* Small fixes

* Move up lowering.py and sequence.py

Move up lowering.py and sequence.py

Fix sequence

* Remove sequence from scheduling methods

* Expose sequence at top level

* Remove coding: utf-8 per PEP 3120

* lint

* WIP: make scheduling dt-agnostic

* small fixes

* update test

* Remove support of Parameter for duration of delay

* Change default unit to None

* Add checking consistency of duration unit

* Improve interface for unit conversion

* Clean up _parse_transpile_args

* lint

* Update qiskit/compiler/sequence.py

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

* Rename convert_durations_in_dt to convert_durations_to_dt

* Fix arguments check in sequence

* Update docstring

* Fix to draw unit of delays only if unit is not empty

* Improve when and what errors to be raised

* Move convert_durations_to_dt() to circuit/duration.py

* Add missing super() calls

* Keep unit of delays until transpiling

* Improve InstructionDurations to convert unit of delays on demand

Improve InstructionDurations to convert unit of delays on demand

Simplyfy and add tests

Fix a degraded bug

* Change default unit of delays from None to dt

* Add a mock TimeUnitAnalysis pass

* Change scheduling passes to use property_set['time_unit']

* Update qiskit/circuit/quantumcircuit.py

Co-authored-by: Ali Javadi-Abhari <ajavadia@users.noreply.github.com>

* Add validation to force integer duration if unit is 'dt'

* Move imports of scheduling passes at the top

* Complete TimeUnitAnalysis pass

* Update docstring

* Add a new argument dt=None to transpile

* Add time_unit property to Instruction

* Fix to check time unit of duration in assemble

* Update doctring

* Add dt==dtm check

* Add QuantumCircuit.unit

* Update qiskit/compiler/transpile.py

Co-authored-by: Ali Javadi-Abhari <ajavadia@users.noreply.github.com>

* Update qiskit/compiler/transpile.py

Co-authored-by: Ali Javadi-Abhari <ajavadia@users.noreply.github.com>

* Update qiskit/compiler/transpile.py

Co-authored-by: Ali Javadi-Abhari <ajavadia@users.noreply.github.com>

* Update qiskit/compiler/transpile.py

Co-authored-by: Ali Javadi-Abhari <ajavadia@users.noreply.github.com>

* Enable schedule to accept delays with SI units

* Remove RemoveOpsOnIdleQubits pass and add a tweak to sequence

* Lint and fix a test

* Move any_delay_in() to execute.py and make it private

* lint

Co-authored-by: Lauren Capelluto <laurencapelluto@gmail.com>
Co-authored-by: Ali Javadi-Abhari <ajavadia@users.noreply.github.com>
This commit is contained in:
itoko 2020-09-23 21:55:47 +09:00 committed by GitHub
parent cef3f1d6f6
commit 6adb90feb3
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
42 changed files with 1695 additions and 50 deletions

View File

@ -116,7 +116,7 @@ evaluation=10.0 - ((float(5 * error + warning + refactor + convention) / stateme
# pi = the PI constant
# op = operation iterator
# b = basis iterator
good-names=a,b,i,j,k,d,n,m,ex,v,w,x,y,z,Run,_,logger,q,c,r,qr,cr,qc,nd,pi,op,b,ar,br,p,cp,
good-names=a,b,i,j,k,d,n,m,ex,v,w,x,y,z,Run,_,logger,q,c,r,qr,cr,qc,nd,pi,op,b,ar,br,p,cp,dt,
__unittest,iSwapGate
# Bad variable names which should always be refused, separated by a comma

View File

@ -77,7 +77,7 @@ except ImportError:
# Moved to after IBMQ and Aer imports due to import issues
# with other modules that check for IBMQ (tools)
from qiskit.execute import execute # noqa
from qiskit.compiler import transpile, assemble, schedule # noqa
from qiskit.compiler import transpile, assemble, schedule, sequence # noqa
from .version import __version__ # noqa
from .version import _get_qiskit_versions # noqa

View File

@ -17,6 +17,7 @@ from typing import Dict, List, Optional, Tuple
from qiskit.assembler.run_config import RunConfig
from qiskit.assembler.assemble_schedules import _assemble_instructions as _assemble_schedule
from qiskit.circuit import QuantumCircuit
from qiskit.exceptions import QiskitError
from qiskit.qobj import (QasmQobj, QobjExperimentHeader,
QasmQobjInstruction, QasmQobjExperimentConfig, QasmQobjExperiment,
QasmQobjConfig, QasmExperimentCalibrations, GateCalibration,
@ -39,7 +40,14 @@ def _assemble_circuit(
Returns:
One experiment for the QasmQobj, and pulse library for pulse gates (which could be None)
Raises:
QiskitError: when the circuit has unit other than 'dt'.
"""
if circuit.unit != 'dt':
raise QiskitError("Unable to assemble circuit with unit '{}', which must be 'dt'."
.format(circuit.unit))
# header data
num_qubits = 0
memory_slots = 0

View File

@ -189,6 +189,7 @@ Gates and Instructions
Gate
ControlledGate
Delay
Measure
Reset
Instruction
@ -222,6 +223,7 @@ from .controlledgate import ControlledGate
from .instruction import Instruction
from .instructionset import InstructionSet
from .barrier import Barrier
from .delay import Delay
from .measure import Measure
from .reset import Reset
from .parameter import Parameter

65
qiskit/circuit/delay.py Normal file
View File

@ -0,0 +1,65 @@
# 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.
"""
Delay instruction (for circuit module).
"""
import numpy as np
from qiskit.circuit.exceptions import CircuitError
from qiskit.circuit.instruction import Instruction
class Delay(Instruction):
"""Do nothing and just delay/wait/idle for a specified duration."""
def __init__(self, duration, unit='dt'):
"""Create new delay instruction."""
if not isinstance(duration, (float, int)):
raise CircuitError('Unsupported duration type.')
if unit == 'dt' and not isinstance(duration, int):
raise CircuitError("Integer duration is required for 'dt' unit.")
if unit not in {'s', 'ms', 'us', 'ns', 'ps', 'dt'}:
raise CircuitError('Unknown unit %s is specified.' % unit)
super().__init__("delay", 1, 0, params=[duration], unit=unit)
def inverse(self):
"""Special case. Return self."""
return self
def broadcast_arguments(self, qargs, cargs):
yield [qarg for sublist in qargs for qarg in sublist], []
def c_if(self, classical, val):
raise CircuitError('Conditional Delay is not yet implemented.')
@property
def duration(self):
"""Get the duration of this delay."""
return self.params[0]
@duration.setter
def duration(self, duration):
"""Set the duration of this delay."""
self.params = [duration]
def to_matrix(self) -> np.ndarray:
"""Return the identity matrix."""
return np.array([[1, 0],
[0, 1]], dtype=complex)
def __repr__(self):
"""Return the official string representing the delay."""
return "%s(duration=%s[unit=%s])" % \
(self.__class__.__name__, self.params[0], self.unit)

View File

@ -0,0 +1,85 @@
# 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.
"""
Utilities for handling duration of a circuit instruction.
"""
import warnings
from qiskit.circuit import QuantumCircuit
from qiskit.circuit.exceptions import CircuitError
from qiskit.util import apply_prefix
def duration_in_dt(duration_in_sec: float, dt_in_sec: float) -> int:
"""
Return duration in dt.
Args:
duration_in_sec: duration [s] to be converted.
dt_in_sec: duration of dt in seconds used for conversion.
Returns:
Duration in dt.
"""
res = round(duration_in_sec / dt_in_sec)
rounding_error = abs(duration_in_sec - res * dt_in_sec)
if rounding_error > 1e-15:
warnings.warn("Duration is rounded to %d [dt] = %e [s] from %e [s]"
% (res, res * dt_in_sec, duration_in_sec),
UserWarning)
return res
def convert_durations_to_dt(qc: QuantumCircuit, dt_in_sec: float, inplace=True):
"""Convert all the durations in SI (seconds) into those in dt.
Returns a new circuit if `inplace=False`.
Parameters:
qc (QuantumCircuit): Duration of dt in seconds used for conversion.
dt_in_sec (float): Duration of dt in seconds used for conversion.
inplace (bool): All durations are converted inplace or return new circuit.
Returns:
QuantumCircuit: Converted circuit if `inplace = False`, otherwise None.
Raises:
CircuitError: if fail to convert durations.
"""
if inplace:
circ = qc
else:
circ = qc.copy()
for inst, _, _ in circ.data:
if inst.unit == 'dt' or inst.duration is None:
continue
if not inst.unit.endswith('s'):
raise CircuitError("Invalid time unit: '{0}'".format(inst.unit))
duration = inst.duration
if inst.unit != 's':
duration = apply_prefix(duration, inst.unit)
inst.duration = duration_in_dt(duration, dt_in_sec)
inst.unit = 'dt'
if circ.duration is not None:
circ.duration = duration_in_dt(circ.duration, dt_in_sec)
circ.unit = 'dt'
if not inplace:
return circ
else:
return None

View File

@ -49,7 +49,7 @@ _CUTOFF_PRECISION = 1E-10
class Instruction:
"""Generic quantum instruction."""
def __init__(self, name, num_qubits, num_clbits, params):
def __init__(self, name, num_qubits, num_clbits, params, duration=None, unit='dt'):
"""Create a new instruction.
Args:
@ -58,6 +58,8 @@ class Instruction:
num_clbits (int): instruction's clbit width
params (list[int|float|complex|str|ndarray|list|ParameterExpression]):
list of parameters
duration (int or float): instruction's duration. it must be integer if ``unit`` is 'dt'
unit (str): time unit of duration
Raises:
CircuitError: when the register is not in the correct format.
@ -81,6 +83,9 @@ class Instruction:
self._definition = None
self.params = params
self._duration = duration
self._unit = unit
def __eq__(self, other):
"""Two instructions are the same if they have the same name,
same dimensions, and same params.
@ -181,6 +186,26 @@ class Instruction:
from qiskit.circuit.equivalence_library import SessionEquivalenceLibrary as sel
sel.add_equivalence(self, decomposition)
@property
def duration(self):
"""Get the duration."""
return self._duration
@duration.setter
def duration(self, duration):
"""Set the duration."""
self._duration = duration
@property
def unit(self):
"""Get the time unit of duration."""
return self._unit
@unit.setter
def unit(self, unit):
"""Set the time unit of duration."""
self._unit = unit
def assemble(self):
"""Assemble a QasmQobjInstruction"""
instruction = QasmQobjInstruction(name=self.name)

View File

@ -174,6 +174,9 @@ class QuantumCircuit:
self._global_phase = 0
self.global_phase = global_phase
self.duration = None
self.unit = 'dt'
@property
def data(self):
"""Return the circuit data (instructions and context).
@ -311,6 +314,9 @@ class QuantumCircuit:
for inst, qargs, cargs in reversed(self.data):
reverse_circ._append(inst.reverse_ops(), qargs, cargs)
reverse_circ.duration = self.duration
reverse_circ.unit = self.unit
return reverse_circ
def reverse_bits(self):
@ -852,6 +858,10 @@ class QuantumCircuit:
self._update_parameter_table(instruction)
# mark as normal circuit if a new instruction is added
self.duration = None
self.unit = 'dt'
return instruction
def _update_parameter_table(self, instruction):
@ -973,6 +983,7 @@ class QuantumCircuit:
Returns:
QuantumCircuit: a circuit one level decomposed
"""
# pylint: disable=cyclic-import
from qiskit.transpiler.passes.basis.decompose import Decompose
from qiskit.converters.circuit_to_dag import circuit_to_dag
from qiskit.converters.dag_to_circuit import dag_to_circuit
@ -1947,6 +1958,46 @@ class QuantumCircuit:
return self.append(Barrier(len(qubits)), qubits, [])
def delay(self, duration, qarg=None, unit='dt'):
"""Apply :class:`~qiskit.circuit.Delay`. If qarg is None, applies to all qubits.
When applying to multiple qubits, delays with the same duration will be created.
Args:
duration (int or float): duration of the delay.
qarg (Object): qubit argument to apply this delay.
unit (str): unit of the duration. Supported units: 's', 'ms', 'us', 'ns', 'ps', 'dt'.
Default is ``dt``, i.e. integer time unit depending on the target backend.
Returns:
qiskit.Instruction: the attached delay instruction.
Raises:
CircuitError: if arguments have bad format.
"""
from .delay import Delay
qubits = []
if qarg is None: # -> apply delays to all qubits
for q in self.qubits:
qubits.append(q)
else:
if isinstance(qarg, QuantumRegister):
qubits.extend([qarg[j] for j in range(qarg.size)])
elif isinstance(qarg, list):
qubits.extend(qarg)
elif isinstance(qarg, (range, tuple)):
qubits.extend(list(qarg))
elif isinstance(qarg, slice):
qubits.extend(self.qubits[qarg])
else:
qubits.append(qarg)
instructions = InstructionSet()
for q in qubits:
inst = (Delay(duration, unit), [q], [])
self.append(*inst)
instructions.add(*inst)
return instructions
def h(self, qubit): # pylint: disable=invalid-name
"""Apply :class:`~qiskit.circuit.library.HGate`."""
from .library.standard_gates.h import HGate

View File

@ -26,9 +26,11 @@ Circuit and Pulse Compilation Functions
assemble
schedule
transpile
sequence
"""
from .assemble import assemble
from .transpile import transpile
from .schedule import schedule
from .sequence import sequence

View File

@ -38,6 +38,7 @@ def schedule(circuits: Union[QuantumCircuit, List[QuantumCircuit]],
backend: Optional[BaseBackend] = None,
inst_map: Optional[InstructionScheduleMap] = None,
meas_map: Optional[List[List[int]]] = None,
dt: Optional[float] = None,
method: Optional[Union[str, List[str]]] = None) -> Union[Schedule, List[Schedule]]:
"""
Schedule a circuit to a pulse ``Schedule``, using the backend, according to any specified
@ -50,6 +51,9 @@ def schedule(circuits: Union[QuantumCircuit, List[QuantumCircuit]],
``backend``\'s ``instruction_schedule_map``
meas_map: List of sets of qubits that must be measured together. If ``None``, defaults to
the ``backend``\'s ``meas_map``
dt: The output sample rate of backend control electronics. For scheduled circuits
which contain time information, dt is required. If not provided, it will be
obtained from the backend configuration
method: Optionally specify a particular scheduling method
Returns:
@ -72,8 +76,11 @@ def schedule(circuits: Union[QuantumCircuit, List[QuantumCircuit]],
if backend is None:
raise QiskitError("Must supply either a backend or a meas_map for scheduling passes.")
meas_map = backend.configuration().meas_map
if dt is None:
if backend is not None:
dt = backend.configuration().dt
schedule_config = ScheduleConfig(inst_map=inst_map, meas_map=meas_map)
schedule_config = ScheduleConfig(inst_map=inst_map, meas_map=meas_map, dt=dt)
circuits = circuits if isinstance(circuits, list) else [circuits]
schedules = [schedule_circuit(circuit, schedule_config, method) for circuit in circuits]
end_time = time()

View File

@ -0,0 +1,67 @@
# 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.
"""
Mapping a scheduled ``QuantumCircuit`` to a pulse ``Schedule``.
"""
from typing import List, Optional, Union
from qiskit.circuit.quantumcircuit import QuantumCircuit
from qiskit.exceptions import QiskitError
from qiskit.providers import BaseBackend
from qiskit.pulse import InstructionScheduleMap, Schedule
from qiskit.scheduler import ScheduleConfig
from qiskit.scheduler.sequence import sequence as _sequence
def sequence(scheduled_circuits: Union[QuantumCircuit, List[QuantumCircuit]],
backend: Optional[BaseBackend] = None,
inst_map: Optional[InstructionScheduleMap] = None,
meas_map: Optional[List[List[int]]] = None,
dt: Optional[float] = None) -> Union[Schedule, List[Schedule]]:
"""
Schedule a scheduled circuit to a pulse ``Schedule``, using the backend.
Args:
scheduled_circuits: Scheduled circuit(s) to be translated
backend: 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
``backend``\'s ``instruction_schedule_map``
meas_map: List of sets of qubits that must be measured together. If ``None``, defaults to
the ``backend``\'s ``meas_map``
dt: The output sample rate of backend control electronics. For scheduled circuits
which contain time information, dt is required. If not provided, it will be
obtained from the backend configuration
Returns:
A pulse ``Schedule`` that implements the input circuit
Raises:
QiskitError: If ``inst_map`` and ``meas_map`` are not passed and ``backend`` is not passed
"""
if inst_map is None:
if backend is None:
raise QiskitError("Must supply either a backend or inst_map for sequencing.")
inst_map = backend.defaults().instruction_schedule_map
if meas_map is None:
if backend is None:
raise QiskitError("Must supply either a backend or a meas_map for sequencing.")
meas_map = backend.configuration().meas_map
if dt is None:
if backend is None:
raise QiskitError("Must supply either a backend or a dt for sequencing.")
dt = backend.configuration().dt
schedule_config = ScheduleConfig(inst_map=inst_map, meas_map=meas_map, dt=dt)
circuits = scheduled_circuits if isinstance(scheduled_circuits, list) else [scheduled_circuits]
schedules = [_sequence(circuit, schedule_config) for circuit in circuits]
return schedules[0] if len(schedules) == 1 else schedules

View File

@ -12,24 +12,26 @@
"""Circuit transpile function"""
import logging
from time import time
import warnings
from time import time
from typing import List, Union, Dict, Callable, Any, Optional, Tuple
from qiskit import user_config
from qiskit.circuit.quantumcircuit import QuantumCircuit
from qiskit.circuit.quantumregister import Qubit
from qiskit.converters import isinstanceint, isinstancelist, dag_to_circuit, circuit_to_dag
from qiskit.dagcircuit import DAGCircuit
from qiskit.providers import BaseBackend
from qiskit.providers.models import BackendProperties
from qiskit.providers.models.backendproperties import Gate
from qiskit.pulse import Schedule
from qiskit.tools.parallel import parallel_map
from qiskit.transpiler import Layout, CouplingMap, PropertySet, PassManager
from qiskit.transpiler.basepasses import BasePass
from qiskit.dagcircuit import DAGCircuit
from qiskit.tools.parallel import parallel_map
from qiskit.transpiler.passmanager_config import PassManagerConfig
from qiskit.pulse import Schedule
from qiskit.circuit.quantumregister import Qubit
from qiskit import user_config
from qiskit.transpiler.exceptions import TranspilerError
from qiskit.transpiler.instruction_durations import InstructionDurationsType
from qiskit.transpiler.passes import ApplyLayout
from qiskit.converters import isinstanceint, isinstancelist, dag_to_circuit, circuit_to_dag
from qiskit.transpiler.passmanager_config import PassManagerConfig
from qiskit.transpiler.preset_passmanagers import (level_0_pass_manager,
level_1_pass_manager,
level_2_pass_manager,
@ -47,6 +49,9 @@ def transpile(circuits: Union[QuantumCircuit, List[QuantumCircuit]],
layout_method: Optional[str] = None,
routing_method: Optional[str] = None,
translation_method: Optional[str] = None,
scheduling_method: Optional[str] = None,
instruction_durations: Optional[InstructionDurationsType] = None,
dt: Optional[float] = None,
seed_transpiler: Optional[int] = None,
optimization_level: Optional[int] = None,
pass_manager: Optional[PassManager] = None,
@ -120,6 +125,24 @@ def transpile(circuits: Union[QuantumCircuit, List[QuantumCircuit]],
may not run.
routing_method: Name of routing pass ('basic', 'lookahead', 'stochastic', 'sabre')
translation_method: Name of translation pass ('unroller', 'translator', 'synthesis')
scheduling_method: Name of scheduling pass.
* ``'as_soon_as_possible'``: Schedule instructions greedily, as early as possible
on a qubit resource. alias: ``'asap'``)
* ``'as_late_as_possible'``: Schedule instructions late, i.e. keeping qubits
in the ground state when possible. (alias: ``'alap'``)
If ``None``, no scheduling will be done.
instruction_durations: Durations of instructions.
The gate lengths defined in ``backend.properties`` are used as default and
they are updated (overwritten) if this ``instruction_durations`` is specified.
The format of ``instruction_durations`` must be as follows.
The `instruction_durations` must be given as a list of tuples
[(instruction_name, qubits, duration, unit), ...].
| [('cx', [0, 1], 12.3, 'ns'), ('u3', [0], 4.56, 'ns')]
| [('cx', [0, 1], 1000), ('u3', [0], 300)]
If unit is omitted, the default is 'dt', which is a sample time depending on backend.
If the time unit is 'dt', the duration must be an integer.
dt: Backend sample time (resolution) in seconds.
If ``None`` (default), ``backend.configuration().dt`` is used.
seed_transpiler: Sets random seed for the stochastic parts of the transpiler
optimization_level: How much optimization to perform on the circuits.
Higher levels generate more optimized circuits,
@ -205,10 +228,16 @@ def transpile(circuits: Union[QuantumCircuit, List[QuantumCircuit]],
config = user_config.get_config()
optimization_level = config.get('transpile_optimization_level', 1)
if scheduling_method is not None and backend is None and not instruction_durations:
warnings.warn("When scheduling circuits without backend,"
" 'instruction_durations' should be usually provided.",
UserWarning)
# Get transpile_args to configure the circuit transpilation job(s)
transpile_args = _parse_transpile_args(circuits, backend, basis_gates, coupling_map,
backend_properties, initial_layout,
layout_method, routing_method, translation_method,
scheduling_method, instruction_durations, dt,
seed_transpiler, optimization_level,
callback, output_name)
@ -298,6 +327,13 @@ def _transpile_circuit(circuit_config_tuple: Tuple[QuantumCircuit, Dict]) -> Qua
else:
raise TranspilerError("optimization_level can range from 0 to 3.")
if pass_manager_config.scheduling_method is not None:
if pass_manager_config.basis_gates:
if 'delay' not in pass_manager_config.basis_gates:
pass_manager_config.basis_gates.append('delay')
else:
pass_manager_config.basis_gates = ['delay']
result = pass_manager.run(circuit, callback=transpile_config['callback'],
output_name=transpile_config['output_name'])
@ -365,6 +401,7 @@ def _remap_layout_faulty_backend(layout, faulty_qubits_map):
def _parse_transpile_args(circuits, backend,
basis_gates, coupling_map, backend_properties,
initial_layout, layout_method, routing_method, translation_method,
scheduling_method, instruction_durations, dt,
seed_transpiler, optimization_level,
callback, output_name) -> List[Dict]:
"""Resolve the various types of args allowed to the transpile() function through
@ -394,15 +431,18 @@ def _parse_transpile_args(circuits, backend,
layout_method = _parse_layout_method(layout_method, num_circuits)
routing_method = _parse_routing_method(routing_method, num_circuits)
translation_method = _parse_translation_method(translation_method, num_circuits)
durations = _parse_instruction_durations(backend, instruction_durations, dt,
scheduling_method, num_circuits)
scheduling_method = _parse_scheduling_method(scheduling_method, num_circuits)
seed_transpiler = _parse_seed_transpiler(seed_transpiler, num_circuits)
optimization_level = _parse_optimization_level(optimization_level, num_circuits)
output_name = _parse_output_name(output_name, circuits)
callback = _parse_callback(callback, num_circuits)
list_transpile_args = []
for args in zip(basis_gates, coupling_map, backend_properties,
initial_layout, layout_method, routing_method, translation_method,
seed_transpiler, optimization_level,
for args in zip(basis_gates, coupling_map, backend_properties, initial_layout,
layout_method, routing_method, translation_method, scheduling_method,
durations, seed_transpiler, optimization_level,
output_name, callback, backend_num_qubits, faulty_qubits_map):
transpile_args = {'pass_manager_config': PassManagerConfig(basis_gates=args[0],
coupling_map=args[1],
@ -411,12 +451,14 @@ def _parse_transpile_args(circuits, backend,
layout_method=args[4],
routing_method=args[5],
translation_method=args[6],
seed_transpiler=args[7]),
'optimization_level': args[8],
'output_name': args[9],
'callback': args[10],
'backend_num_qubits': args[11],
'faulty_qubits_map': args[12]}
scheduling_method=args[7],
instruction_durations=args[8],
seed_transpiler=args[9]),
'optimization_level': args[10],
'output_name': args[11],
'callback': args[12],
'backend_num_qubits': args[13],
'faulty_qubits_map': args[14]}
list_transpile_args.append(transpile_args)
return list_transpile_args
@ -588,6 +630,26 @@ def _parse_translation_method(translation_method, num_circuits):
return translation_method
def _parse_scheduling_method(scheduling_method, num_circuits):
if not isinstance(scheduling_method, list):
scheduling_method = [scheduling_method] * num_circuits
return scheduling_method
def _parse_instruction_durations(backend, inst_durations, dt, scheduling_method, num_circuits):
durations = None
if scheduling_method is not None:
from qiskit.transpiler.instruction_durations import InstructionDurations
if backend:
durations = InstructionDurations.from_backend(backend).update(inst_durations, dt)
else:
durations = InstructionDurations(inst_durations, dt)
if not isinstance(durations, list):
durations = [durations] * num_circuits
return durations
def _parse_seed_transpiler(seed_transpiler, num_circuits):
if not isinstance(seed_transpiler, list):
seed_transpiler = [seed_transpiler] * num_circuits

View File

@ -55,4 +55,7 @@ def circuit_to_dag(circuit):
for instruction, qargs, cargs in circuit.data:
dagcircuit.apply_operation_back(instruction.copy(), qargs, cargs)
dagcircuit.duration = circuit.duration
dagcircuit.unit = circuit.unit
return dagcircuit

View File

@ -11,7 +11,7 @@
# that they have been altered from the originals.
"""Helper function for converting a dag to a circuit."""
from qiskit.circuit import QuantumCircuit
from qiskit.circuit.quantumcircuit import QuantumCircuit
def dag_to_circuit(dag):
@ -56,4 +56,6 @@ def dag_to_circuit(dag):
inst.condition = node.condition
circuit._append(inst, node.qargs, node.cargs)
circuit.duration = dag.duration
circuit.unit = dag.unit
return circuit

View File

@ -93,6 +93,9 @@ class DAGCircuit:
self._global_phase = 0
self._calibrations = defaultdict(dict)
self.duration = None
self.unit = 'dt'
def to_networkx(self):
"""Returns a copy of the DAGCircuit in networkx format."""
G = nx.MultiDiGraph()
@ -663,6 +666,20 @@ class DAGCircuit:
else:
return None
def reverse_ops(self):
"""Reverse the operations in the ``self`` circuit.
Returns:
DAGCircuit: the reversed dag.
"""
# TODO: speed up
# pylint: disable=cyclic-import
from qiskit.converters import dag_to_circuit, circuit_to_dag
qc = dag_to_circuit(self)
reversed_qc = qc.reverse_ops()
reversed_dag = circuit_to_dag(reversed_qc)
return reversed_dag
def idle_wires(self, ignore=None):
"""Return idle wires.

View File

@ -21,6 +21,7 @@ Executing Experiments (:mod:`qiskit.execute`)
"""
import logging
from time import time
from qiskit.circuit import QuantumCircuit
from qiskit.compiler import transpile, assemble, schedule
from qiskit.qobj.utils import MeasLevel, MeasReturnType
from qiskit.pulse import Schedule
@ -246,6 +247,14 @@ def execute(experiments, backend,
initial_layout=initial_layout)
experiments = pass_manager.run(experiments)
else:
if 'delay' not in backend.configuration().basis_gates and _any_delay_in(experiments):
if schedule_circuit and backend.configuration().open_pulse:
pass # the delay will be handled in the Pulse schedule
else:
raise QiskitError("Backend {} does not support delay instruction. "
"Use 'schedule_circuit=True' for pulse-enabled backends."
.format(backend.name()))
# transpiling the circuits using given transpile options
experiments = transpile(experiments,
basis_gates=basis_gates,
@ -298,3 +307,22 @@ def _check_conflicting_argument(**kargs):
if conflicting_args:
raise QiskitError("The parameters pass_manager conflicts with the following "
"parameter(s): {}.".format(', '.join(conflicting_args)))
def _any_delay_in(circuits):
"""Check if the circuits have any delay instruction.
Args:
circuits (QuantumCircuit or list[QuantumCircuit]): Circuits to be checked
Returns:
bool: True if there is any delay in either of the circuit, otherwise False.
"""
if isinstance(circuits, QuantumCircuit):
circuits = [circuits]
has_delay = False
for qc in circuits:
if 'delay' in qc.count_ops():
has_delay = True
break
return has_delay

View File

@ -18,6 +18,7 @@ from typing import Any, Iterable, Tuple, Union
import dateutil.parser
from qiskit.providers.exceptions import BackendPropertyError
from qiskit.util import apply_prefix
class Nduv:
@ -475,24 +476,7 @@ class BackendProperties:
Raises:
BackendPropertyError: If the units aren't recognized.
"""
downfactors = {
'p': 1e12,
'n': 1e9,
'u': 1e6,
'µ': 1e6,
'm': 1e3
}
upfactors = {
'k': 1e3,
'M': 1e6,
'G': 1e9
}
if not unit:
return value
if unit[0] in downfactors:
return value / downfactors[unit[0]]
elif unit[0] in upfactors:
return value * upfactors[unit[0]]
else:
raise BackendPropertyError(
"Could not understand units: {u}".format(u=unit))
try:
return apply_prefix(value, unit)
except Exception:
raise BackendPropertyError("Could not understand units: {u}".format(u=unit))

View File

@ -23,13 +23,16 @@ class ScheduleConfig():
def __init__(self,
inst_map: InstructionScheduleMap,
meas_map: List[List[int]]):
meas_map: List[List[int]],
dt: float):
"""
Container for information needed to schedule a QuantumCircuit into a pulse Schedule.
Args:
inst_map: The schedule definition of all gates supported on a backend.
meas_map: A list of groups of qubits which have to be measured together.
dt: Sample duration.
"""
self.inst_map = inst_map
self.meas_map = format_meas_map(meas_map)
self.dt = dt

View File

@ -17,16 +17,18 @@ from collections import namedtuple
from typing import Dict, List
from qiskit.circuit.barrier import Barrier
from qiskit.circuit.delay import Delay
from qiskit.circuit.duration import convert_durations_to_dt
from qiskit.circuit.measure import Measure
from qiskit.circuit.quantumcircuit import QuantumCircuit
from qiskit.exceptions import QiskitError
from qiskit.pulse import Schedule
from qiskit.pulse.channels import AcquireChannel, MemorySlot
from qiskit.pulse import instructions as pulse_inst
from qiskit.pulse.channels import AcquireChannel, MemorySlot, DriveChannel
from qiskit.pulse.exceptions import PulseError
from qiskit.pulse.macros import measure
from qiskit.scheduler.config import ScheduleConfig
CircuitPulseDef = namedtuple('CircuitPulseDef', [
'schedule', # The schedule which implements the quantum circuit command
'qubits']) # The labels of the qubits involved in the command according to the circuit
@ -35,7 +37,6 @@ CircuitPulseDef = namedtuple('CircuitPulseDef', [
def lower_gates(circuit: QuantumCircuit, schedule_config: ScheduleConfig) -> List[CircuitPulseDef]:
"""
Return a list of Schedules and the qubits they operate on, for each element encountered in the
``
input circuit.
Without concern for the final schedule, extract and return a list of Schedules and the qubits
@ -58,6 +59,9 @@ def lower_gates(circuit: QuantumCircuit, schedule_config: ScheduleConfig) -> Lis
inst_map = schedule_config.inst_map
qubit_mem_slots = {} # Map measured qubit index to classical bit index
# convert the unit of durations from SI to dt before lowering
circuit = convert_durations_to_dt(circuit, dt_in_sec=schedule_config.dt, inplace=False)
def get_measure_schedule(qubit_mem_slots: Dict[int, int]) -> CircuitPulseDef:
"""Create a schedule to measure the qubits queued for measuring."""
sched = Schedule()
@ -105,6 +109,12 @@ def lower_gates(circuit: QuantumCircuit, schedule_config: ScheduleConfig) -> Lis
if isinstance(inst, Barrier):
circ_pulse_defs.append(CircuitPulseDef(schedule=inst, qubits=inst_qubits))
elif isinstance(inst, Delay):
sched = Schedule(name=inst.name)
for qubit in inst_qubits:
for channel in [DriveChannel]:
sched += pulse_inst.Delay(duration=inst.duration, channel=channel(qubit))
circ_pulse_defs.append(CircuitPulseDef(schedule=sched, qubits=inst_qubits))
elif isinstance(inst, Measure):
if (len(inst_qubits) != 1 and len(clbits) != 1):
raise QiskitError("Qubit '{}' or classical bit '{}' errored because the "

View File

@ -18,7 +18,7 @@ from qiskit.exceptions import QiskitError
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
from qiskit.scheduler.methods import as_soon_as_possible, as_late_as_possible
def schedule_circuit(circuit: QuantumCircuit,

View File

@ -0,0 +1,92 @@
# 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.
"""
Mapping a scheduled QuantumCircuit to a pulse Schedule.
"""
from collections import defaultdict
from qiskit.circuit.barrier import Barrier
from qiskit.circuit.measure import Measure
from qiskit.circuit.quantumcircuit import QuantumCircuit
from qiskit.exceptions import QiskitError
from qiskit.pulse.schedule import Schedule
from qiskit.pulse.transforms import pad
from qiskit.scheduler.config import ScheduleConfig
from qiskit.scheduler.lowering import lower_gates
def sequence(scheduled_circuit: QuantumCircuit, schedule_config: ScheduleConfig) -> Schedule:
"""
Return the pulse Schedule which implements the input scheduled circuit.
Assume all measurements are done at once at the last of the circuit.
Schedules according to the command definition given by the schedule_config.
Args:
scheduled_circuit: The scheduled quantum circuit to translate.
schedule_config: Backend specific parameters used for building the Schedule.
Returns:
A schedule corresponding to the input ``circuit``.
Raises:
QiskitError: If invalid scheduled circuit is supplied.
"""
circ_pulse_defs = lower_gates(scheduled_circuit, schedule_config)
# find the measurement start time (assume measurement once)
def _meas_start_time():
_qubit_time_available = defaultdict(int)
for inst, qubits, _ in scheduled_circuit.data:
if isinstance(inst, Measure):
return _qubit_time_available[qubits[0]]
for q in qubits:
_qubit_time_available[q] += inst.duration
return None
meas_time = _meas_start_time()
# restore start times
qubit_time_available = {}
start_times = []
out_circ_pulse_defs = []
for circ_pulse_def in circ_pulse_defs:
active_qubits = [q for q in circ_pulse_def.qubits if q in qubit_time_available]
start_time = max([qubit_time_available[q] for q in active_qubits], default=0)
for q in active_qubits:
if qubit_time_available[q] != start_time:
# print(q, ":", qubit_time_available[q], "!=", start_time)
raise QiskitError("Invalid scheduled circuit.")
stop_time = start_time
if not isinstance(circ_pulse_def.schedule, Barrier):
stop_time += circ_pulse_def.schedule.duration
delay_overlaps_meas = False
for q in circ_pulse_def.qubits:
qubit_time_available[q] = stop_time
if meas_time is not None \
and circ_pulse_def.schedule.name == "delay" \
and stop_time > meas_time:
qubit_time_available[q] = meas_time
delay_overlaps_meas = True
# skip to delays overlapping measures and barriers
if not delay_overlaps_meas and not isinstance(circ_pulse_def.schedule, Barrier):
start_times.append(start_time)
out_circ_pulse_defs.append(circ_pulse_def)
timed_schedules = [(time, cpd.schedule) for time, cpd in zip(start_times, out_circ_pulse_defs)]
sched = Schedule(*timed_schedules, name=scheduled_circuit.name)
return pad(sched)

View File

@ -0,0 +1,218 @@
# 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.
"""Durations of instructions, one of transpiler configurations."""
from typing import Optional, List, Tuple, Union, Iterable, Set
from qiskit.circuit import Barrier, Delay
from qiskit.circuit import Instruction, Qubit
from qiskit.providers import BaseBackend
from qiskit.transpiler.exceptions import TranspilerError
from qiskit.util import apply_prefix
from qiskit.circuit.duration import duration_in_dt
class InstructionDurations:
"""Helper class to provide durations of instructions for scheduling.
It stores durations (gate lengths) and dt to be used at the scheduling stage of transpiling.
It can be constructed from ``backend`` or ``instruction_durations``,
which is an argument of :func:`transpile`.
"""
def __init__(self,
instruction_durations: Optional['InstructionDurationsType'] = None,
dt: float = None):
self.duration_by_name = {}
self.duration_by_name_qubits = {}
self.dt = dt # pylint: disable=invalid-name
if instruction_durations:
self.update(instruction_durations)
@classmethod
def from_backend(cls, backend: BaseBackend):
"""Construct an :class:`InstructionDurations` object from the backend.
Args:
backend: backend from which durations (gate lengths) and dt are extracted.
Returns:
InstructionDurations: The InstructionDurations constructed from backend.
Raises:
TranspilerError: If dt and dtm is different in the backend.
"""
# All durations in seconds in gate_length
instruction_durations = []
for gate, insts in backend.properties()._gates.items():
for qubits, props in insts.items():
if 'gate_length' in props:
gate_length = props['gate_length'][0] # Throw away datetime at index 1
instruction_durations.append((gate, qubits, gate_length, 's'))
try:
dt = backend.configuration().dt # pylint: disable=invalid-name
except AttributeError:
dt = None
# TODO: backend.properties() should tell us durations of measurements
# TODO: Remove the following lines after that
try:
dtm = backend.configuration().dtm
if dtm != dt:
raise TranspilerError("dtm != dt case is not supported.")
inst_map = backend.defaults().instruction_schedule_map
all_qubits = tuple(range(backend.configuration().num_qubits))
meas_duration = inst_map.get('measure', all_qubits).duration
for q in all_qubits:
instruction_durations.append(('measure', [q], meas_duration, 'dt'))
except AttributeError:
pass
return InstructionDurations(instruction_durations, dt=dt)
def update(self,
inst_durations: Optional['InstructionDurationsType'],
dt: float = None):
"""Update self with inst_durations (inst_durations overwrite self).
Args:
inst_durations: Instruction durations to be merged into self (overwriting self).
dt: Sampling duration in seconds of the target backend.
Returns:
InstructionDurations: The updated InstructionDurations.
Raises:
TranspilerError: If the format of instruction_durations is invalid.
"""
if dt:
self.dt = dt
if inst_durations is None:
return self
if isinstance(inst_durations, InstructionDurations):
self.duration_by_name.update(inst_durations.duration_by_name)
self.duration_by_name_qubits.update(inst_durations.duration_by_name_qubits)
else:
for i, items in enumerate(inst_durations):
if len(items) == 3:
inst_durations[i] = (*items, 'dt') # set default unit
elif len(items) != 4:
raise TranspilerError("Each entry of inst_durations dictionary must be "
"(inst_name, qubits, duration) or "
"(inst_name, qubits, duration, unit)")
for name, qubits, duration, unit in inst_durations:
if isinstance(qubits, int):
qubits = [qubits]
if qubits is None:
self.duration_by_name[name] = duration, unit
else:
self.duration_by_name_qubits[(name, tuple(qubits))] = duration, unit
return self
def get(self,
inst: Union[str, Instruction],
qubits: Union[int, List[int], Qubit, List[Qubit]],
unit: str = 'dt') -> Union[float, int]:
"""Get the duration of the instruction with the name and the qubits.
Args:
inst: An instruction or its name to be queried.
qubits: Qubits or its indices that the instruction acts on.
unit: The unit of duration to be returned. It must be 's' or 'dt'.
Returns:
float|int: The duration of the instruction on the qubits.
Raises:
TranspilerError: No duration is defined for the instruction.
"""
if isinstance(inst, Barrier):
return 0
elif isinstance(inst, Delay):
return self._convert_unit(inst.duration, inst.unit, unit)
if isinstance(inst, Instruction):
inst_name = inst.name
else:
inst_name = inst
if isinstance(qubits, (int, Qubit)):
qubits = [qubits]
if isinstance(qubits[0], Qubit):
qubits = [q.index for q in qubits]
try:
return self._get(inst_name, qubits, unit)
except TranspilerError:
raise TranspilerError("Duration of {} on qubits {} is not found."
.format(inst_name, qubits))
def _get(self, name: str, qubits: List[int], to_unit: str) -> Union[float, int]:
"""Get the duration of the instruction with the name and the qubits."""
if name == 'barrier':
return 0
key = (name, tuple(qubits))
if key in self.duration_by_name_qubits:
duration, unit = self.duration_by_name_qubits[key]
elif name in self.duration_by_name:
duration, unit = self.duration_by_name[name]
else:
raise TranspilerError("No value is found for key={}".format(key))
return self._convert_unit(duration, unit, to_unit)
def _convert_unit(self, duration: float, from_unit: str, to_unit: str) -> Union[float, int]:
if from_unit.endswith('s') and from_unit != 's':
duration = apply_prefix(duration, from_unit)
from_unit = 's'
# assert both from_unit and to_unit in {'s', 'dt'}
if from_unit == to_unit:
return duration
if self.dt is None:
raise TranspilerError("dt is necessary to convert durations from '{}' to '{}'"
.format(from_unit, to_unit))
if from_unit == 's' and to_unit == 'dt':
return duration_in_dt(duration, self.dt)
elif from_unit == 'dt' and to_unit == 's':
return duration * self.dt
else:
raise TranspilerError("Conversion from '{}' to '{}' is not supported"
.format(from_unit, to_unit))
def units_used(self) -> Set[str]:
"""Get the set of all units used in this instruction durations.
Returns:
Set of units used in this instruction durations.
"""
units_used = set()
for _, unit in self.duration_by_name_qubits.values():
units_used.add(unit)
for _, unit in self.duration_by_name.values():
units_used.add(unit)
return units_used
InstructionDurationsType = Union[List[Tuple[str, Optional[Iterable[int]], Union[float, int], str]],
List[Tuple[str, Optional[Iterable[int]], Union[float, int]]],
InstructionDurations]
"""List of tuples representing (instruction name, qubits indices, duration)."""

View File

@ -74,6 +74,15 @@ Optimizations
CrosstalkAdaptiveSchedule
TemplateOptimization
Scheduling
=============
.. autosummary::
:toctree: ../stubs/
ALAPSchedule
ASAPSchedule
Circuit Analysis
================
@ -165,6 +174,11 @@ from .analysis import DAGLongestPath
# synthesis
from .synthesis import UnitarySynthesis
# circuit scheduling
from .scheduling import ALAPSchedule
from .scheduling import ASAPSchedule
from .scheduling import TimeUnitAnalysis
# additional utility passes
from .utils import CheckMap
from .utils import CheckCXDirection

View File

@ -0,0 +1,17 @@
# 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 containing circuit scheduling passes."""
from .alap import ALAPSchedule
from .asap import ASAPSchedule
from .time_unit_analysis import TimeUnitAnalysis

View File

@ -0,0 +1,56 @@
# 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.
"""ALAP Scheduling."""
from qiskit.transpiler.basepasses import TransformationPass
from qiskit.transpiler.exceptions import TranspilerError
from qiskit.transpiler.passes.scheduling.asap import ASAPSchedule
class ALAPSchedule(TransformationPass):
"""ALAP Scheduling."""
def __init__(self, durations):
"""ALAPSchedule initializer.
Args:
durations (InstructionDurations): Durations of instructions to be used in scheduling
"""
super().__init__()
self._asap = ASAPSchedule(durations)
def run(self, dag, time_unit=None): # pylint: disable=arguments-differ
"""Run the ALAPSchedule pass on `dag`.
Args:
dag (DAGCircuit): DAG to schedule.
time_unit (str): Time unit to be used in scheduling: 'dt' or 's'.
Returns:
DAGCircuit: A scheduled DAG.
Raises:
TranspilerError: if the circuit is not mapped on physical qubits.
"""
if len(dag.qregs) != 1 or dag.qregs.get('q', None) is None:
raise TranspilerError('ALAP schedule runs on physical circuits only')
if not time_unit:
time_unit = self.property_set['time_unit']
new_dag = dag.reverse_ops()
new_dag = self._asap.run(new_dag, time_unit)
new_dag = new_dag.reverse_ops()
new_dag.name = dag.name
return new_dag

View File

@ -0,0 +1,90 @@
# 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.
"""ASAP Scheduling."""
from collections import defaultdict
from typing import List
from qiskit.circuit.delay import Delay
from qiskit.dagcircuit import DAGCircuit
from qiskit.transpiler.basepasses import TransformationPass
from qiskit.transpiler.exceptions import TranspilerError
class ASAPSchedule(TransformationPass):
"""ASAP Scheduling."""
def __init__(self, durations):
"""ASAPSchedule initializer.
Args:
durations (InstructionDurations): Durations of instructions to be used in scheduling
"""
super().__init__()
self.durations = durations
def run(self, dag, time_unit=None): # pylint: disable=arguments-differ
"""Run the ASAPSchedule pass on `dag`.
Args:
dag (DAGCircuit): DAG to schedule.
time_unit (str): Time unit to be used in scheduling: 'dt' or 's'.
Returns:
DAGCircuit: A scheduled DAG.
Raises:
TranspilerError: if the circuit is not mapped on physical qubits.
"""
if len(dag.qregs) != 1 or dag.qregs.get('q', None) is None:
raise TranspilerError('ASAP schedule runs on physical circuits only')
if not time_unit:
time_unit = self.property_set['time_unit']
new_dag = DAGCircuit()
for qreg in dag.qregs.values():
new_dag.add_qreg(qreg)
for creg in dag.cregs.values():
new_dag.add_creg(creg)
qubit_time_available = defaultdict(int)
def pad_with_delays(qubits: List[int], until, unit) -> None:
"""Pad idle time-slots in ``qubits`` with delays in ``unit`` until ``until``."""
for q in qubits:
if qubit_time_available[q] < until:
idle_duration = until - qubit_time_available[q]
new_dag.apply_operation_back(Delay(idle_duration, unit), [q])
for node in dag.topological_op_nodes():
start_time = max(qubit_time_available[q] for q in node.qargs)
pad_with_delays(node.qargs, until=start_time, unit=time_unit)
new_node = new_dag.apply_operation_back(node.op, node.qargs, node.cargs, node.condition)
duration = self.durations.get(node.op, node.qargs, unit=time_unit)
# set duration for each instruction (tricky but necessary)
new_node.op.duration = duration
new_node.op.unit = time_unit
stop_time = start_time + duration
# update time table
for q in node.qargs:
qubit_time_available[q] = stop_time
working_qubits = qubit_time_available.keys()
circuit_duration = max(qubit_time_available[q] for q in working_qubits)
pad_with_delays(new_dag.qubits, until=circuit_duration, unit=time_unit)
new_dag.name = dag.name
new_dag.duration = circuit_duration
new_dag.unit = time_unit
return new_dag

View File

@ -0,0 +1,98 @@
# 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.
"""Choose a time unit to be used in the scheduling and its following passes."""
from typing import Set
from qiskit.circuit import Delay
from qiskit.dagcircuit import DAGCircuit
from qiskit.transpiler.basepasses import AnalysisPass
from qiskit.transpiler.exceptions import TranspilerError
from qiskit.transpiler.instruction_durations import InstructionDurations
class TimeUnitAnalysis(AnalysisPass):
"""Choose a time unit to be used in the following passes
(e.g. scheduling pass and dynamical decoupling pass).
If dt (dt in seconds) is known to transpiler, the unit 'dt' is chosen. Otherwise,
the unit to be selected depends on what units are used in delays and instruction durations:
* 's': if they are all in SI units.
* 'dt': if they are all in the unit 'dt'.
* raise error: if they are a mix of SI units and 'dt'.
"""
def __init__(self, inst_durations: InstructionDurations):
"""TimeUnitAnalysis initializer.
Args:
inst_durations (InstructionDurations): A dictionary of durations of instructions.
"""
super().__init__()
self.inst_durations = inst_durations
def run(self, dag: DAGCircuit):
"""Run the TimeUnitAnalysis pass on `dag`.
Args:
dag (DAGCircuit): DAG to be checked.
Raises:
TranspilerError: if the units are not unifiable
"""
if self.inst_durations.dt is not None:
self.property_set['time_unit'] = 'dt'
else:
# Check what units are used in delays and other instructions: dt or SI or mixed
units_delay = self._units_used_in_delays(dag)
if self._unified(units_delay) == "mixed":
raise TranspilerError("Fail to unify time units in delays. SI units "
"and dt unit must not be mixed when dt is not supplied.")
units_other = self.inst_durations.units_used()
if self._unified(units_other) == "mixed":
raise TranspilerError("Fail to unify time units in instruction_durations. SI units "
"and dt unit must not be mixed when dt is not supplied.")
unified_unit = self._unified(units_delay | units_other)
if unified_unit == "SI":
self.property_set['time_unit'] = 's'
elif unified_unit == "dt":
self.property_set['time_unit'] = 'dt'
else:
raise TranspilerError("Fail to unify time units. SI units "
"and dt unit must not be mixed when dt is not supplied.")
@staticmethod
def _units_used_in_delays(dag: DAGCircuit) -> Set[str]:
units_used = set()
for node in dag.op_nodes(op=Delay):
units_used.add(node.op.unit)
return units_used
@staticmethod
def _unified(unit_set: Set[str]) -> str:
if not unit_set:
return "dt"
if len(unit_set) == 1 and 'dt' in unit_set:
return "dt"
all_si = True
for unit in unit_set:
if not unit.endswith('s'):
all_si = False
break
if all_si:
return "SI"
return "mixed"

View File

@ -24,6 +24,8 @@ class PassManagerConfig:
layout_method=None,
routing_method=None,
translation_method=None,
scheduling_method=None,
instruction_durations=None,
backend_properties=None,
seed_transpiler=None):
"""Initialize a PassManagerConfig object
@ -40,6 +42,9 @@ class PassManagerConfig:
architecture.
translation_method (str): the pass to use for translating gates to
basis_gates.
scheduling_method (str): the pass to use for scheduling instructions.
instruction_durations (InstructionDurations): Dictionary of duration
(in dt) for each instruction.
backend_properties (BackendProperties): Properties returned by a
backend, including information on gate errors, readout errors,
qubit coherence times, etc.
@ -52,5 +57,7 @@ class PassManagerConfig:
self.layout_method = layout_method
self.routing_method = routing_method
self.translation_method = translation_method
self.scheduling_method = scheduling_method
self.instruction_durations = instruction_durations
self.backend_properties = backend_properties
self.seed_transpiler = seed_transpiler

View File

@ -41,6 +41,9 @@ from qiskit.transpiler.passes import CheckCXDirection
from qiskit.transpiler.passes import Collect2qBlocks
from qiskit.transpiler.passes import ConsolidateBlocks
from qiskit.transpiler.passes import UnitarySynthesis
from qiskit.transpiler.passes import TimeUnitAnalysis
from qiskit.transpiler.passes import ALAPSchedule
from qiskit.transpiler.passes import ASAPSchedule
from qiskit.transpiler import TranspilerError
@ -74,6 +77,8 @@ def level_0_pass_manager(pass_manager_config: PassManagerConfig) -> PassManager:
layout_method = pass_manager_config.layout_method or 'trivial'
routing_method = pass_manager_config.routing_method or 'stochastic'
translation_method = pass_manager_config.translation_method or 'translator'
scheduling_method = pass_manager_config.scheduling_method
instruction_durations = pass_manager_config.instruction_durations
seed_transpiler = pass_manager_config.seed_transpiler
backend_properties = pass_manager_config.backend_properties
@ -143,6 +148,16 @@ def level_0_pass_manager(pass_manager_config: PassManagerConfig) -> PassManager:
_direction = [CXDirection(coupling_map)]
# 7. Schedule the circuit only when scheduling_method is supplied
if scheduling_method:
_scheduling = [TimeUnitAnalysis(instruction_durations)]
if scheduling_method in {'alap', 'as_late_as_possible'}:
_scheduling += [ALAPSchedule(instruction_durations)]
elif scheduling_method in {'asap', 'as_soon_as_possible'}:
_scheduling += [ASAPSchedule(instruction_durations)]
else:
raise TranspilerError("Invalid scheduling method %s." % scheduling_method)
# Build pass manager
pm0 = PassManager()
if coupling_map:
@ -156,5 +171,6 @@ def level_0_pass_manager(pass_manager_config: PassManagerConfig) -> PassManager:
if coupling_map and not coupling_map.is_symmetric:
pm0.append(_direction_check)
pm0.append(_direction, condition=_direction_condition)
if scheduling_method:
pm0.append(_scheduling)
return pm0

View File

@ -47,6 +47,9 @@ from qiskit.transpiler.passes import Layout2qDistance
from qiskit.transpiler.passes import Collect2qBlocks
from qiskit.transpiler.passes import ConsolidateBlocks
from qiskit.transpiler.passes import UnitarySynthesis
from qiskit.transpiler.passes import TimeUnitAnalysis
from qiskit.transpiler.passes import ALAPSchedule
from qiskit.transpiler.passes import ASAPSchedule
from qiskit.transpiler import TranspilerError
@ -82,6 +85,8 @@ def level_1_pass_manager(pass_manager_config: PassManagerConfig) -> PassManager:
layout_method = pass_manager_config.layout_method or 'dense'
routing_method = pass_manager_config.routing_method or 'stochastic'
translation_method = pass_manager_config.translation_method or 'translator'
scheduling_method = pass_manager_config.scheduling_method
instruction_durations = pass_manager_config.instruction_durations
seed_transpiler = pass_manager_config.seed_transpiler
backend_properties = pass_manager_config.backend_properties
@ -171,6 +176,16 @@ def level_1_pass_manager(pass_manager_config: PassManagerConfig) -> PassManager:
_opt = [Optimize1qGates(basis_gates), CXCancellation()]
# 10. Schedule the circuit only when scheduling_method is supplied
if scheduling_method:
_scheduling = [TimeUnitAnalysis(instruction_durations)]
if scheduling_method in {'alap', 'as_late_as_possible'}:
_scheduling += [ALAPSchedule(instruction_durations)]
elif scheduling_method in {'asap', 'as_soon_as_possible'}:
_scheduling += [ASAPSchedule(instruction_durations)]
else:
raise TranspilerError("Invalid scheduling method %s." % scheduling_method)
# Build pass manager
pm1 = PassManager()
if coupling_map:
@ -187,5 +202,7 @@ def level_1_pass_manager(pass_manager_config: PassManagerConfig) -> PassManager:
pm1.append(_direction, condition=_direction_condition)
pm1.append(_reset)
pm1.append(_depth_check + _opt, do_while=_opt_control)
if scheduling_method:
pm1.append(_scheduling)
return pm1

View File

@ -48,6 +48,9 @@ from qiskit.transpiler.passes import CheckCXDirection
from qiskit.transpiler.passes import Collect2qBlocks
from qiskit.transpiler.passes import ConsolidateBlocks
from qiskit.transpiler.passes import UnitarySynthesis
from qiskit.transpiler.passes import TimeUnitAnalysis
from qiskit.transpiler.passes import ALAPSchedule
from qiskit.transpiler.passes import ASAPSchedule
from qiskit.transpiler import TranspilerError
@ -85,6 +88,8 @@ def level_2_pass_manager(pass_manager_config: PassManagerConfig) -> PassManager:
layout_method = pass_manager_config.layout_method or 'dense'
routing_method = pass_manager_config.routing_method or 'stochastic'
translation_method = pass_manager_config.translation_method or 'translator'
scheduling_method = pass_manager_config.scheduling_method
instruction_durations = pass_manager_config.instruction_durations
seed_transpiler = pass_manager_config.seed_transpiler
backend_properties = pass_manager_config.backend_properties
@ -166,6 +171,16 @@ def level_2_pass_manager(pass_manager_config: PassManagerConfig) -> PassManager:
_opt = [Optimize1qGates(basis_gates), CommutativeCancellation()]
# 9. Schedule the circuit only when scheduling_method is supplied
if scheduling_method:
_scheduling = [TimeUnitAnalysis(instruction_durations)]
if scheduling_method in {'alap', 'as_late_as_possible'}:
_scheduling += [ALAPSchedule(instruction_durations)]
elif scheduling_method in {'asap', 'as_soon_as_possible'}:
_scheduling += [ASAPSchedule(instruction_durations)]
else:
raise TranspilerError("Invalid scheduling method %s." % scheduling_method)
# Build pass manager
pm2 = PassManager()
if coupling_map:
@ -182,5 +197,7 @@ def level_2_pass_manager(pass_manager_config: PassManagerConfig) -> PassManager:
pm2.append(_direction, condition=_direction_condition)
pm2.append(_reset)
pm2.append(_depth_check + _opt, do_while=_opt_control)
if scheduling_method:
pm2.append(_scheduling)
return pm2

View File

@ -51,6 +51,9 @@ from qiskit.transpiler.passes import ConsolidateBlocks
from qiskit.transpiler.passes import UnitarySynthesis
from qiskit.transpiler.passes import ApplyLayout
from qiskit.transpiler.passes import CheckCXDirection
from qiskit.transpiler.passes import TimeUnitAnalysis
from qiskit.transpiler.passes import ALAPSchedule
from qiskit.transpiler.passes import ASAPSchedule
from qiskit.transpiler import TranspilerError
@ -88,6 +91,8 @@ def level_3_pass_manager(pass_manager_config: PassManagerConfig) -> PassManager:
layout_method = pass_manager_config.layout_method or 'dense'
routing_method = pass_manager_config.routing_method or 'stochastic'
translation_method = pass_manager_config.translation_method or 'translator'
scheduling_method = pass_manager_config.scheduling_method
instruction_durations = pass_manager_config.instruction_durations
seed_transpiler = pass_manager_config.seed_transpiler
backend_properties = pass_manager_config.backend_properties
@ -177,6 +182,16 @@ def level_3_pass_manager(pass_manager_config: PassManagerConfig) -> PassManager:
CommutativeCancellation(),
]
# Schedule the circuit only when scheduling_method is supplied
if scheduling_method:
_scheduling = [TimeUnitAnalysis(instruction_durations)]
if scheduling_method in {'alap', 'as_late_as_possible'}:
_scheduling += [ALAPSchedule(instruction_durations)]
elif scheduling_method in {'asap', 'as_soon_as_possible'}:
_scheduling += [ASAPSchedule(instruction_durations)]
else:
raise TranspilerError("Invalid scheduling method %s." % scheduling_method)
# Build pass manager
pm3 = PassManager()
pm3.append(_unroll3q)
@ -194,5 +209,7 @@ def level_3_pass_manager(pass_manager_config: PassManagerConfig) -> PassManager:
pm3.append(_direction_check)
pm3.append(_direction, condition=_direction_condition)
pm3.append(_reset)
if scheduling_method:
pm3.append(_scheduling)
return pm3

View File

@ -154,3 +154,40 @@ def is_main_process():
and (sys.version_info[1] == 5 or sys.version_info[1] == 6)
and mp.current_process().name != 'MainProcess')
)
def apply_prefix(value: float, unit: str) -> float:
"""
Given a SI unit prefix and value, apply the prefix to convert to
standard SI unit.
Args:
value: The number to apply prefix to.
unit: String prefix.
Returns:
Converted value.
Raises:
Exception: If the units aren't recognized.
"""
downfactors = {
'p': 1e12,
'n': 1e9,
'u': 1e6,
'µ': 1e6,
'm': 1e3
}
upfactors = {
'k': 1e3,
'M': 1e6,
'G': 1e9
}
if not unit:
return value
if unit[0] in downfactors:
return value / downfactors[unit[0]]
elif unit[0] in upfactors:
return value * upfactors[unit[0]]
else:
raise Exception("Could not understand units: {u}".format(u=unit))

View File

@ -40,6 +40,7 @@ except ImportError:
HAS_PYLATEX = False
from qiskit.circuit import ControlledGate
from qiskit.circuit import Delay
from qiskit.visualization.qcstyle import DefaultStyle, BWStyle
from qiskit import user_config
from qiskit.circuit.tools.pi_check import pi_check
@ -902,6 +903,12 @@ class MatplotlibDrawer:
else:
self._multiqubit_gate(q_xy, fc=fc, ec=ec, gt=gt, sc=sc,
text=gate_text, subtext=vec)
elif isinstance(op.op, Delay):
param_text = "(%s)" % param
if op.op.unit:
param_text += "[%s]" % op.op.unit
self._gate(q_xy[0], fc=fc, ec=ec, gt=gt, sc=sc,
text=gate_text, subtext=param_text)
#
# draw single qubit gates
#

View File

@ -23,6 +23,7 @@ from qiskit.circuit import ControlledGate, Gate, Instruction
from qiskit.circuit import Reset as ResetInstruction
from qiskit.circuit import Measure as MeasureInstruction
from qiskit.circuit import Barrier as BarrierInstruction
from qiskit.circuit import Delay as DelayInstruction
from qiskit.circuit.library.standard_gates import IGate, RZZGate, SwapGate, SXGate, SXdgGate
from qiskit.extensions import UnitaryGate, HamiltonianGate, Snapshot
from qiskit.extensions.quantum_initializer.initializer import Initialize
@ -807,6 +808,9 @@ class TextDrawing():
params = TextDrawing.params_for_label(instruction)
if params:
if isinstance(instruction.op, DelayInstruction) and instruction.op.unit:
label += "(%s[%s])" % (params[0], instruction.op.unit)
else:
label += "(%s)" % ','.join(params)
return label

View File

@ -0,0 +1,65 @@
---
features:
- |
A new class :class:`~qiskit.circuit.Delay` for representing a delay instruction in a circuit has been added.
A new method :meth:`~qiskit.circuit.QuatumCircuit.delay` is now available for easily appending delays to circuits.
This makes it possible to describe timing-sensitive experiments (e.g. T1/T2 experiment) in the circuit level.
.. jupyter-execute::
from qiskit import QuantumCircuit
qc = QuantumCircuit(1, 1)
qc.delay(500, 0, unit='ns')
qc.measure(0, 0)
qc.draw()
- |
A new argument `scheduling_method` for :func:`~qiskit.compiler.transpile` has been added.
It is required when transpiling circuits with delays.
If `scheduling_method` is specified, the transpiler returns a scheduled circuit such that all idle times in it
are padded with delays (i.e. start time of each instruction is uniquely determined).
This makes it possible to see how scheduled instructions (gates) look in the circuit level.
.. jupyter-execute::
from qiskit import QuantumCircuit, transpile
from qiskit.test.mock.backends import FakeAthens
qc = QuantumCircuit(2)
qc.h(0)
qc.cx(0, 1)
scheduled_circuit = transpile(qc, backend=FakeAthens(), scheduling_method="alap")
print("Duration in dt:", scheduled_circuit.duration)
scheduled_circuit.draw(idle_wires=False)
Note that scheduled circuits can not be directly executed on real backends so far (not yet supported).
- |
A new fuction :func:`~qiskit.compiler.sequence` has been also added so that we can convert a scheduled circuit
into a :class:`~qiskit.pulse.Schedule` to make it executable on a pulse-enabled backend.
.. code-block:: python
from qiskit.compiler import sequence
sched = sequence(scheduled_circuit, pulse_enabled_backend)
upgrade:
- |
The :func:`~qiskit.compiler.schedule` has been updated so that it can schedule circuits with delays.
Now we have two paths to schedule a circuit with delay::
qc = QuantumCircuit(1, 1)
qc.h(0)
qc.delay(500, 0, unit='ns')
qc.h(0)
qc.measure(0, 0)
sched_path1 = schedule(qc.decompose(), backend)
sched_path2 = sequence(transpile(qc, backend, scheduling_method='alap'), backend)
assert pad(sched_path1) == sched_path2
See New Features section for the details of the second path with `transpile` and `sequence`.

View File

@ -0,0 +1,81 @@
# 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.
# pylint: disable=missing-function-docstring
"""Test delay instruction for quantum circuits."""
import numpy as np
from qiskit.circuit import Delay
from qiskit.circuit import QuantumCircuit
from qiskit.circuit.exceptions import CircuitError
from qiskit.test.base import QiskitTestCase
class TestDelayClass(QiskitTestCase):
"""Test delay instruction for quantum circuits."""
def test_keep_units_after_adding_delays_to_circuit(self):
qc = QuantumCircuit(1)
qc.h(0)
qc.delay(100, 0)
qc.delay(200, 0, unit='s')
qc.delay(300, 0, unit='ns')
qc.delay(400, 0, unit='dt')
self.assertEqual(qc.data[1][0].unit, 'dt')
self.assertEqual(qc.data[2][0].unit, 's')
self.assertEqual(qc.data[3][0].unit, 'ns')
self.assertEqual(qc.data[4][0].unit, 'dt')
def test_fail_if_non_integer_duration_with_dt_unit_is_supplied(self):
qc = QuantumCircuit(1)
with self.assertRaises(CircuitError):
qc.delay(0.5, 0, unit='dt')
def test_fail_if_unknown_unit_is_supplied(self):
qc = QuantumCircuit(1)
with self.assertRaises(CircuitError):
qc.delay(100, 0, unit='my_unit')
def test_add_delay_on_single_qubit_to_circuit(self):
qc = QuantumCircuit(1)
qc.h(0)
qc.delay(100, 0)
qc.delay(200, [0])
qc.delay(300, qc.qubits[0])
self.assertEqual(qc.data[1], (Delay(duration=100), qc.qubits, []))
self.assertEqual(qc.data[2], (Delay(duration=200), qc.qubits, []))
self.assertEqual(qc.data[3], (Delay(duration=300), qc.qubits, []))
def test_add_delay_on_multiple_qubits_to_circuit(self):
qc = QuantumCircuit(3)
qc.h(0)
qc.delay(100)
qc.delay(200, range(2))
qc.delay(300, qc.qubits[1:])
expected = QuantumCircuit(3)
expected.h(0)
expected.delay(100, 0)
expected.delay(100, 1)
expected.delay(100, 2)
expected.delay(200, 0)
expected.delay(200, 1)
expected.delay(300, 1)
expected.delay(300, 2)
self.assertEqual(qc, expected)
def test_to_matrix_return_identity_matrix(self):
actual = Delay(100).to_matrix()
expected = np.array([[1, 0],
[0, 1]], dtype=complex)
self.assertTrue(np.array_equal(actual, expected))

View File

@ -0,0 +1,231 @@
# 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.
# pylint: disable=missing-function-docstring
"""Test scheduled circuit (quantum circuit with duration)."""
from qiskit import QuantumCircuit, QiskitError
from qiskit import transpile, execute, assemble
from qiskit.test.mock.backends import FakeParis, FakeVigo
from qiskit.transpiler.exceptions import TranspilerError
from qiskit.transpiler.instruction_durations import InstructionDurations
from qiskit.test.base import QiskitTestCase
class TestScheduledCircuit(QiskitTestCase):
"""Test scheduled circuit (quantum circuit with duration)."""
def setUp(self):
super().setUp()
self.backend_with_dt = FakeParis()
self.backend_without_dt = FakeVigo()
self.dt = 2.2222222222222221e-10
def test_schedule_circuit_when_backend_tells_dt(self):
"""dt is known to transpiler by backend"""
qc = QuantumCircuit(2)
qc.delay(0.1, 0, unit='ms') # 450000[dt]
qc.delay(100, 0, unit='ns') # 450[dt]
qc.h(0) # 160[dt]
qc.h(1) # 160[dt]
sc = transpile(qc, self.backend_with_dt, scheduling_method='alap')
self.assertEqual(sc.duration, 450610)
self.assertEqual(sc.unit, 'dt')
self.assertEqual(sc.data[1][0].name, "delay")
self.assertEqual(sc.data[1][0].duration, 450)
self.assertEqual(sc.data[1][0].unit, 'dt')
self.assertEqual(sc.data[2][0].name, "u2")
self.assertEqual(sc.data[2][0].duration, 160)
self.assertEqual(sc.data[2][0].unit, 'dt')
self.assertEqual(sc.data[3][0].name, "delay")
self.assertEqual(sc.data[3][0].duration, 450450)
self.assertEqual(sc.data[3][0].unit, 'dt')
qobj = assemble(sc, self.backend_with_dt)
self.assertEqual(qobj.experiments[0].instructions[1].name, "delay")
self.assertEqual(qobj.experiments[0].instructions[1].params[0], 450)
self.assertEqual(qobj.experiments[0].instructions[3].name, "delay")
self.assertEqual(qobj.experiments[0].instructions[3].params[0], 450450)
def test_schedule_circuit_when_transpile_option_tells_dt(self):
"""dt is known to transpiler by transpile option"""
qc = QuantumCircuit(2)
qc.delay(0.1, 0, unit='ms') # 450000[dt]
qc.delay(100, 0, unit='ns') # 450[dt]
qc.h(0)
qc.h(1)
sc = transpile(qc, self.backend_without_dt, scheduling_method='alap', dt=self.dt)
self.assertEqual(sc.duration, 450610)
self.assertEqual(sc.unit, 'dt')
self.assertEqual(sc.data[1][0].name, "delay")
self.assertEqual(sc.data[1][0].duration, 450)
self.assertEqual(sc.data[1][0].unit, 'dt')
self.assertEqual(sc.data[2][0].name, "u2")
self.assertEqual(sc.data[2][0].duration, 160)
self.assertEqual(sc.data[2][0].unit, 'dt')
self.assertEqual(sc.data[3][0].name, "delay")
self.assertEqual(sc.data[3][0].duration, 450450)
self.assertEqual(sc.data[3][0].unit, 'dt')
def test_schedule_circuit_in_sec_when_no_one_tells_dt(self):
"""dt is unknown and all delays and gate times are in SI"""
qc = QuantumCircuit(2)
qc.delay(0.1, 0, unit='ms')
qc.delay(100, 0, unit='ns')
qc.h(0)
qc.h(1)
sc = transpile(qc, self.backend_without_dt, scheduling_method='alap')
self.assertAlmostEqual(sc.duration, 450610*self.dt)
self.assertEqual(sc.unit, 's')
self.assertEqual(sc.data[1][0].name, "delay")
self.assertAlmostEqual(sc.data[1][0].duration, 1.0e-7)
self.assertEqual(sc.data[1][0].unit, 's')
self.assertEqual(sc.data[2][0].name, "u2")
self.assertAlmostEqual(sc.data[2][0].duration, 160*self.dt)
self.assertEqual(sc.data[2][0].unit, 's')
self.assertEqual(sc.data[3][0].name, "delay")
self.assertAlmostEqual(sc.data[3][0].duration, 1.0e-4+1.0e-7)
self.assertEqual(sc.data[3][0].unit, 's')
with self.assertRaises(QiskitError):
assemble(sc, self.backend_without_dt)
def test_cannot_schedule_circuit_with_mixed_SI_and_dt_when_no_one_tells_dt(self):
"""dt is unknown but delays and gate times have a mix of SI and dt"""
qc = QuantumCircuit(2)
qc.delay(100, 0, unit='ns')
qc.delay(30, 0, unit='dt')
qc.h(0)
qc.h(1)
with self.assertRaises(QiskitError):
transpile(qc, self.backend_without_dt, scheduling_method='alap')
def test_cannot_execute_delay_circuit_when_schedule_circuit_off(self):
qc = QuantumCircuit(2)
qc.h(0)
qc.delay(500, 1)
qc.cx(0, 1)
with self.assertRaises(QiskitError):
execute(qc, backend=self.backend_with_dt, schedule_circuit=False)
def test_transpile_single_delay_circuit(self):
qc = QuantumCircuit(1)
qc.delay(1234, 0)
sc = transpile(qc, backend=self.backend_with_dt, scheduling_method='alap')
self.assertEqual(sc.duration, 1234)
self.assertEqual(sc.data[0][0].name, "delay")
self.assertEqual(sc.data[0][0].duration, 1234)
self.assertEqual(sc.data[0][0].unit, 'dt')
def test_transpile_t1_circuit(self):
qc = QuantumCircuit(1)
qc.x(0) # 320 [dt]
qc.delay(1000, 0, unit='ns') # 4500 [dt]
qc.measure_all() # 19200 [dt]
scheduled = transpile(qc, backend=self.backend_with_dt, scheduling_method='alap')
self.assertEqual(scheduled.duration, 24020)
def test_transpile_delay_circuit_with_backend(self):
qc = QuantumCircuit(2)
qc.h(0)
qc.delay(100, 1, unit='ns') # 450 [dt]
qc.cx(0, 1) # 1408 [dt]
scheduled = transpile(qc, backend=self.backend_with_dt, scheduling_method='alap')
self.assertEqual(scheduled.duration, 1858)
def test_transpile_delay_circuit_without_backend(self):
qc = QuantumCircuit(2)
qc.h(0)
qc.delay(500, 1)
qc.cx(0, 1)
scheduled = transpile(qc,
scheduling_method='alap',
instruction_durations=[('h', 0, 200), ('cx', [0, 1], 700)])
self.assertEqual(scheduled.duration, 1200)
def test_transpile_delay_circuit_without_scheduling_method_as_normal_circuit(self):
qc = QuantumCircuit(2)
qc.h(0)
qc.delay(500, 1)
qc.cx(0, 1)
transpiled = transpile(qc)
self.assertEqual(transpiled.duration, None)
def test_raise_error_if_transpile_with_scheduling_method_but_without_durations(self):
qc = QuantumCircuit(2)
qc.h(0)
qc.delay(500, 1)
qc.cx(0, 1)
with self.assertRaises(TranspilerError):
transpile(qc, scheduling_method="alap")
def test_invalidate_schedule_circuit_if_new_instruction_is_appended(self):
qc = QuantumCircuit(2)
qc.h(0)
qc.delay(500*self.dt, 1, 's')
qc.cx(0, 1)
scheduled = transpile(qc,
backend=self.backend_with_dt,
scheduling_method='alap')
# append a gate to a scheduled circuit
scheduled.h(0)
self.assertEqual(scheduled.duration, None)
def test_default_units_for_my_own_duration_users(self):
qc = QuantumCircuit(2)
qc.h(0)
qc.delay(500, 1)
qc.cx(0, 1)
# accept None for qubits
scheduled = transpile(qc,
basis_gates=['h', 'cx', 'delay'],
scheduling_method='alap',
instruction_durations=[('h', 0, 200),
('cx', None, 900)]
)
self.assertEqual(scheduled.duration, 1400)
# prioritize specified qubits over None
scheduled = transpile(qc,
basis_gates=['h', 'cx', 'delay'],
scheduling_method='alap',
instruction_durations=[('h', 0, 200),
('cx', None, 900),
('cx', [0, 1], 800)]
)
self.assertEqual(scheduled.duration, 1300)
def test_unit_seconds_for_users_who_uses_durations_given_by_backend(self):
qc = QuantumCircuit(2)
qc.h(0)
qc.delay(500*self.dt, 1, 's')
qc.cx(0, 1)
# usual case
scheduled = transpile(qc,
backend=self.backend_with_dt,
scheduling_method='alap'
)
self.assertEqual(scheduled.duration, 1908)
# update durations
scheduled = transpile(qc,
backend=self.backend_with_dt,
scheduling_method='alap',
instruction_durations=[('cx', [0, 1], 1000*self.dt, 's')]
)
self.assertEqual(scheduled.duration, 1500)
my_own_durations = InstructionDurations([('cx', [0, 1], 1000*self.dt, 's')])
scheduled = transpile(qc,
backend=self.backend_with_dt, # unit='s'
scheduling_method='alap',
instruction_durations=my_own_durations
)
self.assertEqual(scheduled.duration, 1500)

View File

@ -0,0 +1,65 @@
# 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.
# pylint: disable=missing-function-docstring
"""Tests basic functionality of the sequence function"""
import unittest
from qiskit import QuantumCircuit
from qiskit.compiler import sequence, transpile, schedule
from qiskit.pulse.transforms import pad
from qiskit.test.mock import FakeParis
from qiskit.test import QiskitTestCase
class TestSequence(QiskitTestCase):
"""Test sequence function."""
def setUp(self):
super().setUp()
self.backend = FakeParis()
def test_sequence_empty(self):
self.assertEqual(sequence([], self.backend), [])
def test_transpile_and_sequence_agree_with_schedule(self):
qc = QuantumCircuit(2, name="bell")
qc.h(0)
qc.cx(0, 1)
qc.measure_all()
sc = transpile(qc, self.backend, scheduling_method='alap')
actual = sequence(sc, self.backend)
expected = schedule(qc.decompose(), self.backend)
self.assertEqual(actual, pad(expected))
def test_transpile_and_sequence_agree_with_schedule_for_circuit_with_delay(self):
qc = QuantumCircuit(1, 1, name="t2")
qc.h(0)
qc.delay(500, 0, unit='ns')
qc.h(0)
qc.measure(0, 0)
sc = transpile(qc, self.backend, scheduling_method='alap')
actual = sequence(sc, self.backend)
expected = schedule(qc.decompose(), self.backend)
self.assertEqual(actual, pad(expected))
@unittest.skip("not yet determined if delays on ancilla should be removed or not")
def test_transpile_and_sequence_agree_with_schedule_for_circuits_without_measures(self):
qc = QuantumCircuit(2, name="bell_without_measurement")
qc.h(0)
qc.cx(0, 1)
sc = transpile(qc, self.backend, scheduling_method='alap')
actual = sequence(sc, self.backend)
expected = schedule(qc.decompose(), self.backend)
self.assertEqual(actual, pad(expected))

View File

@ -861,6 +861,21 @@ class TestTranspile(QiskitTestCase):
transpiled_circ = transpile(circ, FakeAlmaden())
self.assertEqual(set(transpiled_circ.count_ops().keys()), {'u2', 'mycustom', 'h'})
@data(0, 1, 2, 3)
def test_circuit_with_delay(self, optimization_level):
"""Verify a circuit with delay can transpile to a scheduled circuit."""
qc = QuantumCircuit(2)
qc.h(0)
qc.delay(500, 1)
qc.cx(0, 1)
out = transpile(qc, scheduling_method='alap', basis_gates=['h', 'cx'],
instruction_durations=[('h', 0, 200), ('cx', [0, 1], 700)],
optimization_level=optimization_level)
self.assertEqual(out.duration, 1200)
class StreamHandlerRaiseException(StreamHandler):
"""Handler class that will raise an exception on formatting errors."""

View File

@ -38,6 +38,7 @@ from qiskit.pulse.channels import (
MemorySlot,
RegisterSlot,
DriveChannel,
ControlChannel,
AcquireChannel,
SnapshotChannel,
MeasureChannel,
@ -955,6 +956,17 @@ class TestScheduleEquality(BaseTestSchedule):
self.assertEqual(Schedule(*instructions), Schedule(*reversed(instructions)))
def test_same_commands_on_two_channels_at_same_time_out_of_order(self):
"""Test that schedule with same commands on two channels at the same time equal
when out of order."""
sched1 = Schedule()
sched1 = sched1.append(Delay(100, DriveChannel(1)))
sched1 = sched1.append(Delay(100, ControlChannel(1)))
sched2 = Schedule()
sched2 = sched2.append(Delay(100, ControlChannel(1)))
sched2 = sched2.append(Delay(100, DriveChannel(1)))
self.assertEqual(sched1, sched2)
def test_different_name_equal(self):
"""Test that names are ignored when checking equality."""

View File

@ -0,0 +1,48 @@
# 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.
# pylint: disable=missing-function-docstring
"""Test InstructionDurations class."""
from qiskit.transpiler.exceptions import TranspilerError
from qiskit.transpiler.instruction_durations import InstructionDurations
from qiskit.test.base import QiskitTestCase
from qiskit.test.mock.backends import FakeParis, FakeVigo
class TestInstructionDurationsClass(QiskitTestCase):
"""Test Test InstructionDurations class."""
def test_empty(self):
durations = InstructionDurations()
self.assertEqual(durations.dt, None)
with self.assertRaises(TranspilerError):
durations.get('cx', [0, 1], 'dt')
def test_fail_if_invalid_dict_is_supplied_when_construction(self):
invalid_dic = [('cx', [0, 1])] # no duration
with self.assertRaises(TranspilerError):
InstructionDurations(invalid_dic)
def test_from_backend_for_backend_with_dt(self):
durations = InstructionDurations.from_backend(FakeParis())
self.assertGreater(durations.dt, 0)
self.assertGreater(durations.get('u2', 0), 0)
def test_from_backend_for_backend_without_dt(self):
durations = InstructionDurations.from_backend(FakeVigo())
self.assertIsNone(durations.dt)
self.assertGreater(durations.get('u2', 0, 's'), 0)
with self.assertRaises(TranspilerError):
durations.get('u2', 0)