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
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
|
# mypy: ignore-errors
import operator
from typing import Dict, List, TYPE_CHECKING
import torch
from torch._dynamo.source import AttrSource, GetItemSource
from .. import variables
from ..exc import unimplemented
from ..utils import common_constant_types, istype, np
from .base import typestr, VariableTracker
if TYPE_CHECKING:
from torch._dynamo.symbolic_convert import InstructionTranslator
class ConstantVariable(VariableTracker):
@staticmethod
def create(value, **kwargs) -> VariableTracker:
"""
Create a `ConstantVariable` based on the given value, and supports
automatic routing for collection types like `tuple` (in which case we'd
create `ConstantVariable` for the leaf items).
NOTE: the caller must install the proper guards if needed; most often
the guard will be `CONSTANT_MATCH`.
"""
source = kwargs.get("source", None)
# Routing for supported collection literals.
if isinstance(value, set):
items = [ConstantVariable.create(x) for x in value]
return variables.SetVariable(items, **kwargs)
elif isinstance(value, frozenset):
items = [ConstantVariable.create(x) for x in value]
return variables.FrozensetVariable(items, **kwargs)
elif isinstance(value, (list, tuple)):
items = []
for i, x in enumerate(value):
item_source = GetItemSource(source, i) if source else None
items.append(
ConstantVariable.create(
x,
source=item_source,
)
)
return variables.BaseListVariable.cls_for(type(value))(items, **kwargs)
return ConstantVariable(value, **kwargs)
def __init__(self, value, **kwargs) -> None:
super().__init__(**kwargs)
assert ConstantVariable.is_base_literal(
value
), f"""
Cannot construct `ConstantVariable` for value of type {type(value)}.
This failure likely due to PyTorch-internal use of `ConstantVariable` on
non-literal python values, please try using `VariableTracker.build` instead. If
you believe it's a necessary and legitimate use case (the value is immutable and
can't easily be represented with another `VariableTracker` class), please add
its type to `common_constant_types`.
"""
if np is not None and isinstance(value, np.number):
self.value = value.item()
else:
self.value = value
def as_proxy(self):
return self.value
def __repr__(self) -> str:
return f"ConstantVariable({type(self.value).__name__}: {repr(self.value)})"
def as_python_constant(self):
return self.value
def is_python_constant(self):
return True
@property
def items(self):
"""
Need this when adding a BaseListVariable and a ConstantVariable together.
Happens in detectron2.
"""
return self.unpack_var_sequence(tx=None)
def getitem_const(self, tx: "InstructionTranslator", arg: VariableTracker):
return ConstantVariable.create(
self.value[arg.as_python_constant()],
)
@staticmethod
def is_base_literal(obj):
return type(obj) in common_constant_types
@staticmethod
def is_literal(obj):
if type(obj) in (list, tuple, set, frozenset, torch.Size):
return all(ConstantVariable.is_literal(x) for x in obj)
return ConstantVariable.is_base_literal(obj)
def unpack_var_sequence(self, tx):
try:
return [ConstantVariable.create(x) for x in self.as_python_constant()]
except TypeError as e:
raise NotImplementedError from e
def const_getattr(self, tx: "InstructionTranslator", name):
member = getattr(self.value, name)
if callable(member):
raise NotImplementedError
return member
def call_method(
self,
tx,
name,
args: "List[VariableTracker]",
kwargs: "Dict[str, VariableTracker]",
) -> "VariableTracker":
from .tensor import SymNodeVariable
if name == "format" and istype(self.value, str):
return variables.BuiltinVariable(str.format).call_function(
tx, [self, *args], kwargs
)
elif name == "join" and istype(self.value, str):
assert len(args) == 1 and len(kwargs) == 0
arg_unpacked = args[0].force_unpack_var_sequence(tx)
try:
arg_const = [x.as_python_constant() for x in arg_unpacked]
return ConstantVariable.create(self.value.join(arg_const))
except NotImplementedError:
return super().call_method(tx, name, args, kwargs)
if any(isinstance(x, SymNodeVariable) for x in args):
# Promote to SymNodeVariable for operations involving dynamic shapes.
return variables.SymNodeVariable(self.as_proxy(), self.value).call_method(
tx, name, args, kwargs
)
try:
const_args = [a.as_python_constant() for a in args]
const_kwargs = {k: v.as_python_constant() for k, v in kwargs.items()}
except NotImplementedError:
return super().call_method(tx, name, args, kwargs)
if isinstance(self.value, str) and name in str.__dict__.keys():
method = getattr(self.value, name)
return ConstantVariable.create(method(*const_args, **const_kwargs))
elif isinstance(self.value, (float, int)):
if not (args or kwargs):
return ConstantVariable.create(getattr(self.value, name)())
if (
hasattr(operator, name)
and len(args) == 1
and args[0].is_python_constant()
):
add_target = const_args[0]
op = getattr(operator, name)
if isinstance(
add_target, (torch.SymBool, torch.SymFloat, torch.SymInt)
):
# Addition between a non sym and sym makes a sym
proxy = tx.output.create_proxy(
"call_function", op, (self.value, add_target), {}
)
return SymNodeVariable.create(tx, proxy, add_target)
else:
return ConstantVariable.create(op(self.value, add_target))
elif isinstance(self.value, bytes) and name == "decode":
method = getattr(self.value, name)
return ConstantVariable.create(method(*const_args, **const_kwargs))
if name == "__len__" and not (args or kwargs):
return ConstantVariable.create(len(self.value))
elif name == "__round__" and len(args) == 1 and args[0].is_python_constant():
return ConstantVariable.create(
round(self.value, args[0].is_python_constant())
)
elif name == "__contains__" and len(args) == 1 and args[0].is_python_constant():
assert not kwargs
search = args[0].as_python_constant()
result = search in self.value
return ConstantVariable.create(result)
unimplemented(f"const method call {typestr(self.value)}.{name}")
def call_hasattr(self, tx: "InstructionTranslator", name: str) -> "VariableTracker":
result = hasattr(self.value, name)
return variables.ConstantVariable.create(result)
class EnumVariable(VariableTracker):
def __init__(self, value, **kwargs) -> None:
super().__init__(**kwargs)
self.value = value
@classmethod
def create(cls, cls_type, value_vt, options):
if isinstance(value_vt, variables.ConstantVariable):
for member in list(cls_type):
if member.value == value_vt.as_python_constant():
return cls(member, **options)
unimplemented("Enum variable is constructed with non constant values")
def as_proxy(self):
if isinstance(self.value, int):
return int(self.value) # convert IntEnum to a normal int
return self.value
def __repr__(self) -> str:
return f"EnumVariable({type(self.value)})"
def as_python_constant(self):
return self.value
def var_getattr(self, tx: "InstructionTranslator", name):
member = getattr(self.value, name)
source = self.source and AttrSource(self.source, name)
return VariableTracker.build(tx, member, source=source)
|
