# mypy: 忽略错误
导入
集合
导入
复制
导入 dis
导入
枚举
导入
检查
导入
记录日志
导入
操作符
导入
系统
from 集合
导入 OrderedDict
from collections.abc 导入
迭代器
from dataclasses 导入
字段,
是数据类
from 打字
导入
任何,
可调用,
可选
导入
火炬
导入 torch.fx.traceback
是 fx_traceback
from torch._C 导入 _fx_map_aggregate
是 map_aggregate, _fx_map_arg
是 map_arg
from torch.utils._traceback 导入
捕获的 Traceback
from 兼容性
导入
兼容性
from .graph 导入
图,
魔法方法,
反射魔法方法
from .不可变集合
导入
不可变字典,
不可变列表
from .节点
导入
参数,
基础类型,
节点,
目标
from .操作符模式
导入
检查可变操作
全部 = [
TracerBase,
GraphAppendingTracer,
TraceError,
代理,
元代理,
属性,
参数代理,
范围,
作用域上下文管理器,
]
日志 =
记录.
获取日志记录器(__name__)
@compatibility(兼容旧版本=
错误)
类
范围:
作用域对象,记录模块路径和模块类型
模块的作用域。作用域用于跟踪模块的信息
包含 GraphModule 图中的节点。例如:
class Sub(torch.nn.Module):
def forward(self, x):
# 这将在 GraphModule 中成为调用方法节点,
# 此作用域为(module_path="sub", module_type=Sub)
return x 转置(1, 2)
class M(torch.nn.Module):
def __init__(self) -> None:
self.sub = Sub()
def forward(self, x):
# This will be a call_method Node as well,
# 范围为 (module_path="", None)
x = x.transpose(1, 2)
x = self.sub(x)
return x
""
def __init__(self, 模块路径:
字符串,
模块类型:
任何):
超级().__init__()
self.模块路径 =
模块路径
self.模块类型 =
模块类型
@compatibility(兼容旧版本=
错误)
类
范围上下文管理器:
“一个用于在符号跟踪期间跟踪节点范围的上下文管理器。
当进入模块的前向函数时,我们将更新当前模块的作用域信息,
当退出时,我们将恢复之前的作用域信息。
""
def __init__(
self,
范围:
作用域,
current_scope: 范围,
):
超级().__init__()
# 退出时恢复前一个范围的副本
self._prev_scope = 复制.
复制(
范围)
# 更新范围为当前范围
范围.
模块路径 =
当前作用域.
模块路径
范围.
模块类型 =
当前作用域.
模块类型
# 保存一个引用,以便我们可以恢复它
self._作用域 =
范围
def __进入__(self):
返回 self.
范围
def __退出__(self, *
参数):
self.范围.
模块路径 = self._prev_scope.
模块路径
self.范围.
模块类型 = self.
_prev_range.
模块类型
返回
_复制元字段 = [
神经网络模块栈,
torch_fn,
"source_fn_stack",
"原始_aten",
"重新计算",
"ac_graph_id",
"有反向钩子",
from_node,
"量化标签",
# TODO 已弃用
"_数值调试句柄",
# TODO 已弃用
"自定义",
"分区标签",
"参数关键字值",
]
@compatibility(兼容旧版本=
是)
类 TracerBase:
graph: 图
记录堆栈跟踪:
布尔值 =
假
# 可变模式检查功能标志
# 默认在 1.12 中启用
检查可变操作:
布尔值 =
假
# 功能标志用于断言跟踪
trace_asserts: 布尔值 =
假
# 功能标志用于代理缓冲区值的访问
proxy_buffer_attributes: 布尔值 =
假
跟踪的函数名称。它仅在
``root`` 是一个 ``nn.Module`` 的实例
跟踪函数名:
字符串 =
向前
将包含模块的名称映射到操作符名称
范围:
范围
记录模块调用堆栈
模块栈:
有序字典[
字符串,
元组[
字符串,
任何]]
节点名到模块作用域的映射
node_name_to_scope: 字典[
字符串,
元组[
字符串,
类型]]
@compatibility(兼容旧版本=
是)
def 创建节点(
self,
类型:
字符串,
目标:
目标,
参数:
元组[
参数, ...
]
kwargs: 字典[
字符串,
参数
]
名称:
可选[
字符串] =
无,
type_expr: 可选[
任何] =
无,
) 翻译
节点:
""
插入一个图节点,给定目标、参数、关键字参数和名称。
此方法可以被重写以进行额外的检查、验证或修改用于节点创建的值。例如,可能希望禁止记录原地操作。
修改。例如,可能希望禁止记录原地操作。
。例如,可能希望禁止记录原地操作。
""
如果
仁慈 ==
调用函数
和 self.
检查可变操作:
检查可变操作是否存在(
目标,
参数, kwargs)
节点 = self.graph.
创建节点(
类型,
目标,
参数, kwargs,
名称, type_expr)
# TODO node_name_to_scope 将会被废弃,以 node.meta['nn_module_stack'] 代替
# node.meta['nn_module_stack']
self.节点名称到作用域映射[
节点.
名称] = (
self.范围.
模块路径,
self.范围.
模块类型,
)
可选地设置创建的节点堆栈跟踪以进行调试
如果 fx_traceback.
保留节点元数据():
当前元数据:
字典[
字符串,
任何] = fx_traceback.
获取当前元数据()
堆栈跟踪 =
当前元数据.
获取(
堆栈跟踪)
如果
堆栈跟踪:
节点.
堆栈跟踪 =
堆栈跟踪
# 在节点.meta 上显式设置 stack_trace、nn_module_stack 和 source_fn
# 如果需要其他元字段,可以在此处添加
为
字段
在
复制元字段:
如果
字段
在
当前元信息:
节点.
元数据[
字段] =
复制.
复制(
当前元信息[
字段])
# 这里减去以考虑序列号
# 在追踪这个降低的 aten op 时,编号已增加。
新序列号 =
火炬.
自动微分.
获取序列号() - 1
# 每次创建新的 autograd 节点时,sequence_nr 都会增加。在正向传播过程中,我们存储 sequence_nr
# # 是创建。在正向传播过程中,我们存储 sequence_nr
# 对应于在此 fx 上创建的最后一个 autograd 节点
# 节点的元信息。单个 aten 操作可以创建多个 autograd 节点
# 如 in-place foreach 操作那样,可以创建多个节点。在 BWD 传递过程中,我们检索存储在当前
# 节点上的 sequence_nr
执行自动微分节点。参见[序列号]注释。
如果
当前元数据.
获取(
"输入梯度函数", 0) > 0:
新序列号 =
当前元数据[
"梯度函数序列号"
]
[-1]
节点.
元数据[
"序列号"] =
新序列号
如果...否则 self.
模块栈:
节点.
元数据[
神经网络模块栈] =
复制.
复制(self.
模块栈)
日志.
调试(
"创建节点"%s",
节点)
返回
节点
@compatibility(兼容旧版本=
是)
def 代理(self,
节点:
节点)
翻译
"代理":
返回
代理(
节点, self)
@compatibility(兼容旧版本=
是)
def 创建代理(
self,
类型:
字符串,
目标:
目标,
参数:
元组[
任何, ...
]
kwargs: 字典[
字符串,
任何
]
名称:
可选[
字符串] =
无,
type_expr: 可选[
任何] =
无,
当更新 bc 测试时修复 # fix noqa when updating bc tests
代理工厂函数:
可调用[[
节点
]
代理] =
无, # noqa: RUF013
):
""
从给定的参数创建一个节点,然后返回该节点
并将其包装在一个代理对象中。
如果 kind = '占位符',那么我们正在创建一个节点,
表示函数的参数。如果需要编码
默认参数,我们使用 `args` 元组。`args` 是
否则对于 `placeholder` 节点为空。
""
args_ = self.create_arg(参数)
kwargs_ = self.create_arg(kwargs)
断言 isinstance(args_,
元组)
断言 isinstance(kwargs_,
字典)
节点 = self.
创建节点(
类型,
目标, args_, kwargs_,
名称, type_expr)
如果
不是 proxy_factory_fn:
代理 = self.
代理(
节点)
else:
代理 = proxy_factory_fn(
节点)
如果 self.
记录堆栈跟踪
和
不是
代理.
节点.
堆栈跟踪:
代理.
节点.
堆栈跟踪 =
输入文本翻译为简体中文为:"".
连接(
捕获的 Traceback.
提取().
格式())
返回
代理
def _find_user_frame(self):
""
在符号跟踪期间查找执行用户代码的 Python 堆栈帧。
这里我们需要跳一支小舞。基本上,沿着调用栈向上走。
""
# 我们在这里需要做一些小动作。基本上,沿着调用栈向上走。
记录不在 PyTorch 源中首次出现的帧。这是执行中的帧。
追踪期间的用户代码。
框架 =
检查.currentframe()
pt_files = [
"torch/fx/proxy.py",
"torch/fx/_symbolic_trace.py",
"torch/fx/experimental/proxy_tensor.py",
"torch/_ops.py",
"torch/_tensor.py",
"torch/utils/_python_dispatch.py",
"torch/_prims_common/wrappers.py",
"torch/_refs/__init__.py",
"torch/_refs/nn/functional/__init__.py",
torch/utils/_stats.py,
]
当 frame:
框架 = frame.f_back
如果
框架
和
所有(
不是 frame.
代码文件.co_filename.
以...结尾(
文件)
为
文件
在 pt_files
):
断开
如果
不是 frame:
返回
无
返回
框架
@compatibility(兼容旧版本=
是)
def 创建参数(self, a:
任何)
翻译
参数:
""
一种将符号计算中视为参数的对象降低到
可以存储在 IR 中的参数类型的方法。
可以被覆盖以支持更多特定于跟踪的类型。
""
# 重要:你是来这里尝试代理一个新的类型吗?
# 图?请请请联系 PyTorch 编译器团队中的某人
# 考虑因素微妙。
#
当您添加新类型时,所有下游消费者和传递编写器
需要处理新类型。torch.fx 旨在易于编写
# 通过,因此我们将抵制新类型。
# 2) 在 torch.compile 的 IR 中,只有特定的操作可以进行
将 Tensor 操作放入图中。
但是非 Tensor 操作不应该。这意味着构造函数
对于新类型*不应*成为 FX 图中的节点。
处理器 = _create_arg_bypass.
获取(
类型(a))
如果
处理器 is
不是
无:
这只是一个性能优化,如果需要可以移除
对于常见类型,我们有一个快速路径来避免 isinstance()的开销
这并不会移除下面的检查,因为我们还需要处理子类
返回
处理器(self, a)
如果 isinstance(a,
代理):
返回 a.
节点
# 最常见的参数类型排在前面
如果...否则
有属性(a, "__fx_create_arg__"):
返回 a.__fx_create_arg__(self)
# 聚合
如果...否则 isinstance(a,
元组):
如果
有属性(a,
"_字段"):
# 命名元组构造函数似乎不喜欢从生成器获取
# 将表达式作为参数传递给它们的构造函数,因此构建这个
# 中间元组并将其解包到命名元组构造函数中
args = [self.创建参数(
元素)
为
元素
在 a]
返回
类型(a)(*
参数) # type: ignore[arg-type]
返回
类型(a)([self.
创建参数(
元素)
为
元素
在 a])
如果...否则 isinstance(a,
列表):
返回 [self.
创建参数(
元素)
为
元素
在 a]
如果...否则 isinstance(a,
字典):
返回
创建参数字典(self, a)
如果...否则 isinstance(a,
切片):
返回
切片(
self.创建参数(a.
开始),
self.创建参数(a.
停止),
self.创建参数(a.
步长),
)
如果...否则 isinstance(a,
范围):
返回
范围(
self.创建参数(a.
开始),
self.创建参数(a.
停止),
self.创建参数(a.
步长),
)
如果...否则 isinstance(a, (
火炬.
操作符.
操作符重载,
火炬.
操作符.
高阶运算符)):
返回 a
如果...否则
是数据类(a):
kwargs = {
字段.
名称: self.
创建参数(getattr(a,
字段.
名称))
为
字段
在
字段(a)
}
返回 self.
创建节点(
调用函数, a.
类, (), kwargs)
如果...否则 isinstance(a, (*
基础类型,
枚举.
枚举))
或者 a is
无
或者 a is ...:
返回 a
提升
不支持的操作异常(f
"类型参数的"{
类型(a)}")
@compatibility(兼容旧版本=
是)
def to_bool(self, 对象:
"代理")
翻译
布尔:
"当代理对象被转换为布尔值时调用,例如"
当用于控制流时。通常我们不知道该做什么。
我们不知道代理的值,但自定义跟踪器可以附加更多
使用 create_node 向图节点添加信息,可以选择返回一个值。
""
提升
追踪错误(
"符号追踪的变量不能用作控制流的输入"
)
@compatibility(兼容旧版本=
是)
def 迭代(self,
对象:
"代理")
翻译
迭代器:
"当代理对象被迭代时调用,例如"
在控制流中使用时。通常我们不知道该怎么办,因为
我们不知道代理的值,但自定义跟踪器可以附加更多
信息到图节点,并可以选择返回一个迭代器。
""
提升
跟踪错误(
"代理对象不能迭代。这可能是 "
"当代理在循环中使用时尝试"
"作为 *args 或 **kwargs 函数参数。"
"请参阅 pytorch.org 上的 torch.fx 文档以获取更详细的解释。"
"有关类型等更多信息。"
控制流程可追踪,并查看
代理文档字符串,用于帮助故障排除
代理迭代错误
)
@compatibility(兼容旧版本=
是)
def 键(self,
对象:
代理)
翻译
任何:
当代理对象调用 keys()方法时被调用。
当在代理上调用**时会发生这种情况。这应该返回一个
迭代器 it **应该在您的自定义跟踪器中工作。
""
返回
属性(
对象,
"键")()
# 在不进行跟踪的情况下向图追加时用于代理对象。
@compatibility(兼容旧版本=
是)
类
图追加追踪器(
追踪器基类):
def __init__(self, graph: 图):
超级().__init__()
self.图 =
图
self.范围 =
范围(
输入文本翻译为简体中文为:"",
无)
self.模块堆栈 =
集合.
有序字典()
self.节点名称到范围映射 = {}
@compatibility(兼容旧版本=
错误)
def 断言函数(x):
断言 x
@compatibility(兼容旧版本=
是)
类 TraceError(ValueError):
通过
[文档]@compatibility(
兼容旧版本=
是)
类
代理:
""
``代理``对象是``节点``包装器,在符号跟踪期间流经程序
记录它们所接触的所有操作(``torch``函数调用、方法调用、运算符)
(``torch``函数调用、方法调用、运算符)的操作
进入不断发展的 FX 图形。
如果您正在进行图变换,您可以将自己的“Proxy”包装起来
绕过原始“Node”方法,以便您可以使用重载
操作员用于向“图”添加额外内容。
代理对象不能迭代。换句话说,如果在一个循环或作为 *args/**kwargs 函数参数中使用代理,跟踪器将抛出错误。
有两种主要方法可以解决这个问题:
``*args``/``**kwargs`` 函数参数。
有两种主要方法可以解决这个问题:
将无法追踪的逻辑提取到一个顶级函数中并
使用 `fx.wrap` 对其进行包装。
2. 如果控制流是静态的(即循环迭代次数是
基于某些超参数),代码可以保持其原始状态
位置和重构为类似:
for i in range(self.some_hyperparameter):
indexed_item = proxied_value[i]
想要更详细地了解代理内部结构,请查看
torch/fx/README.md 中的“Proxy”部分
""
@compatibility(兼容旧版本=
是)
def __init__(self, 节点:
节点,
追踪器:
"可选[追踪器基类]" =
无):
如果
追踪器 is
无:
# 这允许你围绕原始节点创建代理对象
追踪器 =
图追加追踪器(
节点.graph)
self.追踪器 =
追踪器
self.节点 =
节点
def __repr__(self) 翻译
字符串:
返回 f
"代理("{self.
节点.
名称})"
def __getattr__(self, k) 翻译
属性:
# note: 还未添加到图中,如果这是一个方法调用
# 我们进行窥孔优化以方法调用
返回
属性(self, k)
def __getstate__(self) 翻译
字典:
返回 self.
字典
def 深拷贝(self,
描述)
翻译
字典:
# 我们必须显式重写此方法,否则 deepcopy
# 将进入 __getattr__(self, "__deepcopy__") 并返回一个
# __deepcopy__ 属性,在某些情况下可能会进入无限循环。
导入
复制
新字典 = {}
为 k, v
在 self.
字典.
项目():
尝试:
新对象 =
复制.
深拷贝(v,
描述)
除了
异常:
日志.
警告(
浅拷贝%s
由于无法深度复制,因此为代理创建了浅拷贝。
代理是为节点创建的%s",
k,
self.节点.
名称,
)
新对象 =
复制.
复制(v)
新字典[k] =
新对象
断言
节点
在
新字典
断言
跟踪器
在
新字典
新代理 =
代理(
新字典[
节点
]
新字典[
跟踪器])
为 k, v
在
新字典.
项目():
新代理.
字典[k] = v
返回
新代理
def __setstate__(self, d):
# 这是在被反序列化/加载时调用的。
self.字典 = d
def __调用__(self, *
参数, **kwargs)
翻译
代理:
返回 self.
追踪器.
创建代理(
调用方法,
__call__, (self,) +
参数, kwargs
)
def __iter__(self) 翻译
迭代器[
代理
]
框架 =
检查.currentframe()
断言
框架 is
不是
无
调用帧 = frame.f_back
断言
调用帧 is
不是
无
inst_list
实例列表 = 列表(
对不起,您提供的文本 "dis" 缺乏上下文,无法进行准确的翻译。请提供完整的句子或段落以便进行翻译.
获取说明(
调用框架.
代码文件))
如果
系统模块.version_info
≥ (3, 11):
from 二分查找
导入
二分查找左边界
实例索引 =
二分查找左边界(
实例列表,
调用帧.f_lasti,
键=lambda x: x.
偏移
)
else:
inst_idx = 调用帧.f_lasti // 2
实例 =
实例列表[
实例索引]
如果
实例.
操作名称 ==
解包序列:
返回 (self[i]
为 i
在
范围(
实例.argval))
忽略索引
返回 self.
追踪器.
迭代(self)
def 绝对值(self):
返回 self.
追踪器.
创建代理(
调用函数,
操作符.
绝对值, (self,), {})
def __bool__(self) 翻译
布尔:
如果 self.
追踪器.
跟踪断言:
# 检查这个布尔值是否在断言中使用,断言的字节码模式
# 对于 Python 3.7--3.9 来说相当稳定
框架 =
检查.currentframe()
断言
框架 is
不是
无
调用框架 = frame.f_back
断言
调用框架 is
不是
无
指令 =
列表(
禁止.
获取说明(
调用帧.
代码文件))
如果
系统模块.version_info
≥ (3, 11):
from 二分查找
导入
二分查找左边界
cur = bisect_left(insts, calling_frame.f_lasti, 键=lambda x: x.
偏移量)
else:
当前 =
调用帧.f_lasti // 2
实例 =
指令集[cur]
如果
实例.
操作名称 ==
"跳转如果为真":
第一 =
指令[cur + 1]
断言
实例.
参数 is
不是
无
最后一个 = insts[
实例.
参数 // 2 - 1]
starts_with_assert = (
第一.
操作名称 == "LOAD_GLOBAL"
和
第一.argval ==
AssertionError
或者
第一.
操作名称 ==
LOAD_ASSERTION_ERROR
)
如果 starts_with_assert
和
最后.
操作名称 ==
RAISE_VARARGS:
self.追踪器.
创建代理(
调用函数, assert_fn, (self,), {})
返回
真实
返回 self.
追踪器.to_bool(self)
@compatibility(兼容旧版本=
是)
def 键(self):
返回 self.
追踪器.
键(self)
def __len__(self):
提升
运行时错误(
"'len' 默认情况下不支持在符号跟踪中。如果您想记录这个调用,请调用 torch.fx.wrap('len') "
"在此处,以便记录这个调用,请调用 torch.fx.wrap('len') "
模块作用域
)
@classmethod
def __torch_function__(类,
原始方法,
类型,
参数=
无, kwargs=
无):
args = args 如果 args
否则 ()
kwargs = kwargs 如果 kwargs
否则 {}
跟踪器:
字典[
任何,
无] = {}
def find_tracer(a):
如果 isinstance(a,
类):
跟踪器[a.
追踪器] =
无
map_aggregate(参数,
查找跟踪器)
map_aggregate(kwargs, 查找跟踪器)
如果
长度(
跟踪器) > 1:
提升
运行时错误(
f"发现多个不同的追踪器"{
列表(
追踪器.
键())}
在尝试追踪操作时
f"的引号中"{
原始方法}"
)
追踪器 =
下一(
迭代(
追踪器.
键()))
如果 isinstance(
原始方法,
火炬._C.
脚本方法):
args = (原始方法.
拥有者,) + args
返回
追踪器.
创建代理(
调用方法,
原始方法.
名称,
参数, kwargs)
如果
火炬.
覆盖.is_tensor_method_or_property(
原始方法):
返回
追踪器.
创建代理(
调用方法,
原始方法.__name__,
参数, kwargs
)
else:
如果 isinstance(
原始方法,
火炬.
操作符.
高阶运算符):
# TODO: 定义如何符号化追踪高阶算子
提升
运行时错误(
"无法符号化追踪高阶算子")
返回
追踪器.
创建代理(
调用函数,
原始方法,
参数,
kwargs,
名称=
追踪器.graph.
目标_to_str(
原始方法.__name__),
)
@compatibility(兼容旧版本=
错误)
类
元数据代理(
代理):
""
一个在图追踪期间传播元数据(meta['val'])的代理子类。
""
def __init__(
self, 节点:
节点,
追踪器:
可选[TracerBase] =
无,
模拟模式=
无
):
超级().__init__(
节点,
追踪器)
self.模拟模式 =
模拟模式
def __repr__(self) 翻译
字符串:
返回 f
MetaProxy({self.
节点.
名称})"
@classmethod
def __torch_function__(类,
原始方法,
类型,
参数=
无, kwargs=
无):
args = args 如果 args
否则 ()
kwargs = kwargs 如果 kwargs
否则 {}
元代理 =
无
为
参数
在
参数:
如果 isinstance(arg, MetaProxy):
元代理 =
参数
断开
断言 (
元数据代理 is
不是
无
), "在参数中未找到 MetaProxy,但预期有一个。"
代理 =
超级().__torch_function__(
原始方法,
类型,
参数, kwargs)
与
元数据代理.
模拟模式:
代理.
节点.
元数据[
值] =
原始方法(
*[a.节点.
元数据[
值]
如果 isinstance(a,
代理)
否则 a
为 a
在
参数
]
**kwargs,
)
返回 MetaProxy(
代理.
节点,
代理.
追踪器,
元数据代理.
模拟模式)
@compatibility(兼容旧版本=
是)
类
属性(
代理):
@compatibility(兼容旧版本=
是)
def __init__(self, 根:
代理,
属性:
字符串):
self.根 =
根
self.属性 =
属性
self.追踪器 =
根.
追踪器
self._节点:
可选[
节点] =
无
@property
def 节点(self):
# 该节点的属性是延迟添加的,因为大多数情况下都只是方法调用
# 这些调用不依赖于 getitem 方法
如果 self.
_节点 is
无:
self.节点 = self.
追踪器.
创建代理(
调用函数, getattr, (self.
根, self.
属性), {}
).节点
返回 self.
节点
def __调用__(self, *
参数, **kwargs):
返回 self.
追踪器.
创建代理(
调用方法, self.
属性, (self.
根,) +
参数, kwargs
)
@compatibility(兼容旧版本=
错误)
类
参数代理(
代理):
""
一种特殊的代理,允许“形状”、“大小”、“dim”和其他一些属性访问通过底层模块参数对象
通过到底层模块参数对象,以便属性访问
因此在跟踪期间对这些属性的条件测试不会抛出异常
""
def __init__(self, 追踪器: TracerBase,
节点:
节点,
名称,
参数):
超级().__init__(
节点,
追踪器)
断言 isinstance(
参数,
火炬.
神经网络.
参数)
self.参数 =
参数
self.名称 =
名称
def __repr__(self) 翻译
字符串:
返回 f
参数代理({self.
名称})"
@property
def 形状(self):
返回 self.
参数.
形状
def 尺寸(self):
返回 self.
参数.
尺寸()
def 暗(self):
返回 self.
参数.
暗()
@property
def 维数(self):
返回 self.
参数.
维数
def 元素数量(self):
返回 self.
参数.
元素数量()
def 元素个数(self):
返回 self.
参数.
元素个数()
为
方法
在
魔法方法:
def 范围(
方法):
def 实现(*
参数, **kwargs):
追踪器 =
参数[0].
追踪器
目标 = getattr(
操作符,
方法)
返回
追踪器.
创建代理(
调用函数,
目标,
参数, kwargs)
实现.__name__ =
方法
如魔法般 = f
'__'{
方法.strip(
“_”)}
'__'
setattr(代理,
魔法般,
实现)
范围(
方法)
def 定义可反射(
原始方法名):
方法名 = f
'__r'{
原方法名.strip(
“_”)}__'
def 实现(self,
右侧):
目标 = getattr(
操作符,
原方法名)
返回 self.
追踪器.
创建代理(
调用函数,
目标, (rhs, self), {})
实现.__name__ =
方法名
实现.__qualname__ =
方法名
setattr(代理,
方法名称,
实现)
为
原方法名
在
“可反射的魔法方法”:
_定义可反射(
原方法名)
def _无节点错误(arg):
提升
运行时错误(
字典用作参数的键不能包含一个
f节点。获取到键:{arg}"
)
def 创建参数字典(self, a):
r = {}
为 k, v
在 a.
项目():
如果
不是 isinstance(k,
字符串):
# 检查无效的字典键。我们不希望代理出现在
# 键中的任何位置。由于键可以是集合类型,
# 我们使用 map_arg 遍历键
k = self.创建参数(k)
map_arg(k, 无节点错误)
r[k] = self.创建参数(v)
返回 r
创建参数绕过 = {
t: lambda self, a: a
为 t
在 [
*基础类型,
类型(
无),
类型(...),
火炬.
操作符.
操作符重载,
火炬.
操作符.
高阶运算符,
]
}
创建参数绕过[
代理] = lambda self, a: a.
节点
创建参数绕过[
元组] = lambda self, a:
元组
[self.
创建参数(
元素)
为
元素
在 a])
创建参数绕过[
列表] = lambda self, a: [self.
创建参数(
元素)
为
元素
在 a]
_创建_arg_bypass[
字典] =
_创建_arg_dict
_创建_arg_bypass[
不可变列表] =
_创建_arg_bypass[
列表]
_创建_arg_bypass[
不可变字典] =
_创建_arg_bypass[
字典]