/
time_unit_conversion.py
135 lines (111 loc) · 4.9 KB
/
time_unit_conversion.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
# This code is part of Qiskit.
#
# (C) Copyright IBM 2021.
#
# This code is licensed under the Apache License, Version 2.0. You may
# obtain a copy of this license in the LICENSE.txt file in the root directory
# of this source tree or at http://www.apache.org/licenses/LICENSE-2.0.
#
# Any modifications or derivative works of this code must retain this
# copyright notice, and modified files need to carry a notice indicating
# that they have been altered from the originals.
"""Unify time unit in circuit for scheduling and following passes."""
from typing import Set
from qiskit.circuit import Delay
from qiskit.dagcircuit import DAGCircuit
from qiskit.transpiler.basepasses import TransformationPass
from qiskit.transpiler.exceptions import TranspilerError
from qiskit.transpiler.instruction_durations import InstructionDurations
from qiskit.transpiler.target import Target
class TimeUnitConversion(TransformationPass):
"""Choose a time unit to be used in the following time-aware passes,
and make all circuit time units consistent with that.
This pass will add a :attr:`.Instruction.duration` metadata to each op whose duration is known
which will be used by subsequent scheduling passes for scheduling.
If ``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 = None, target: Target = None):
"""TimeUnitAnalysis initializer.
Args:
inst_durations (InstructionDurations): A dictionary of durations of instructions.
target: The :class:`~.Target` representing the target backend, if both
``inst_durations`` and ``target`` are specified then this argument will take
precedence and ``inst_durations`` will be ignored.
"""
super().__init__()
self.inst_durations = inst_durations or InstructionDurations()
if target is not None:
self.inst_durations = target.durations()
def run(self, dag: DAGCircuit):
"""Run the TimeUnitAnalysis pass on `dag`.
Args:
dag (DAGCircuit): DAG to be checked.
Returns:
DAGCircuit: DAG with consistent timing and op nodes annotated with duration.
Raises:
TranspilerError: if the units are not unifiable
"""
# Choose unit
if self.inst_durations.dt is not None:
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":
time_unit = "s"
elif unified_unit == "dt":
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."
)
# Make units consistent
for node in dag.op_nodes():
try:
duration = self.inst_durations.get(
node.op, [dag.find_bit(qarg).index for qarg in node.qargs], unit=time_unit
)
except TranspilerError:
continue
node.op = node.op.to_mutable()
node.op.duration = duration
node.op.unit = time_unit
self.property_set["time_unit"] = time_unit
return dag
@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"