# mypy: 允许未类型化定义
导入
集合
导入 importlib.machinery
导入
输入/输出
导入
linecache 模块
导入
操作系统
导入 pickletools
导入
平台
导入
类型
from 集合
导入
默认字典, OrderedDict
from collections.abc 导入
序列
from dataclasses 导入
数据类
from 枚举
导入
枚举
from importlib.machinery 导入
源文件加载器
from pathlib 导入 Path
from 打字
导入
任何,
可调用,
角色, IO,
可选,
联合
导入
火炬
from torch 序列化
导入
位置标签, normalize_storage_type
from torch.types 导入
文件类,
存储
from torch.utils.hooks 导入
可移除句柄
from ._digraph 导入
图 DiGraph
from ._导入 lib
导入
_规范化路径
from ._名称修改
导入 demangle, is_mangled
from ._package_pickler 导入 create_pickler
from 标准库
导入
是否为标准库模块
from .查找文件依赖
导入
查找源文件依赖
from .glob_group 导入
全局组,
全局模式
from .导入器
导入
导入器,
排序导入器,
sys 导入器
全部 = [
打包错误原因,
空匹配错误,
打包错误,
包导出器,
]
_gate_torchscript 序列化 =
真实
动作钩子 =
可调用[[
包导出器,
字符串
]
无]
类
_模块提供者动作(
枚举):
表示：class:`PackageExporter`可以对模块执行的动作之一。
请参阅 :meth:`PackageExporter.extern` 及相关内容以了解这些操作的功能。
"文档"
内部 = 1
外部 = 2
模拟 = 3
拒绝 = 4
# 当一个模块被模拟时，PackageExporter 会将我们的模拟存根写入一个
# `_mock`模块。如果我们重新打包代码，我们可能会遇到来自原始包的
# `_mock`模块。如果我们遇到这种情况，我们可能会遇到
`_mock`模块
重新打包的模拟模块 = 5
特殊情况：PackageImporter 添加了一个假模块
(`torch_package_importer`) 允许打包的代码访问它。不要
# 导出此内容。
跳过 = 6
类
包装错误原因(
枚举):
列出依赖可能无法打包的不同原因。
此枚举用于在出现错误时提供良好的错误信息。
抛出了 :class:`PackagingError` 异常。
"文档"
定义 __repr__(
我):
返回 f
"<"{
我.
类.__name__}.{
我.
名称}
>
IS_EXTENSION_MODULE = (
"模块是一个 C 扩展模块。torch.package 只支持 Python 模块。"
)
NO_DUNDER_FILE = "模块未定义 __file__。"
找不到源文件 = (
"模块定义了 __file__，但我们在文件系统中找不到它。"
)
依赖关系解析失败 =
"依赖解析失败。"
无操作 = (
"模块与任何操作模式不匹配。外部、模拟或内部它。"
)
被拒绝 =
"模块被模式拒绝。"
MOCKED_BUT_STILL_USED = (
"模块被模拟，但仍在包中使用。"
"如果对象应该在 " 请内部或外部调用模拟模块。"
"该包。"
)
@dataclass
类
模式信息:
"包含有关如何执行与 :class:`PackageExporter` 相关的匹配操作的具体信息。"
# 对匹配此模式的模块应采取什么操作。
操作:
模块提供者操作
用户指定模式时给出的 `allow_empty` 的值。
allow_empty: 布尔
# 在打包过程中此模式是否已匹配。
匹配成功:
布尔
定义 __init__(
我,
动作,
允许为空):
我.
动作 =
行动
我.
允许为空 =
允许为空
我.
已匹配 =
假
[文档]class EmptyMatchError(Exception):
"""这是一个在将模拟或外部标记为
``allow_empty=False``，并且在打包过程中没有与任何模块匹配时抛出的异常。
"""
[文档]
类
包装错误(
异常):
当导出包出现问题时，将抛出此异常。
``PackageExporter`` 将尝试收集所有错误，并一次性向您展示。
一次展示给您。
"文档"
定义 __init__(
我,
依赖图:
有向图,
调试=
错误):
按原因分组错误。
破碎的:
字典[
包装错误原因,
列表[
字符串]] =
默认字典(
列表)
for 模块名称, attrs
在
依赖图.
节点.
项目():
错误 =
属性.
获取(
错误)
如果
错误
是
无:
continue
如果
错误 ==
包装错误原因.
无动作:
断言
动作
不
在 attrs
破碎[
错误].append(
模块名称)
消息 =
输入/输出.StringIO()
消息.
写("
输入文本翻译为简体中文为：\n")
for 原因,
模块名称
在
破碎.
项目():
消息.
写(f
"*"{
原因.
值}
输入文本翻译为简体中文为：\n")
for 模块名称
在
模块名称:
消息.
写(f" {
模块名称}
输入文本翻译为简体中文为：\n")
# 如果提供额外上下文，则打印。
错误上下文 =
依赖图.
节点[
模块名称].
获取(
"错误上下文")
如果
错误上下文
是
不
无:
消息.
写(f
"上下文："{
错误上下文}
输入文本翻译为简体中文为：\n")
如果
模块名称
在
禁止使用的模块:
消息.
写(
注意：虽然我们通常使用来自本地环境的 python 标准库模块
f`从{
模块名称}
系统有很多
"级别访问因此可能存在安全风险。我们强烈建议“
f"从您的打包代码中删除`"{
模块名称}
"如果不可能，可以通过调用“
"将其添加到 extern 列表中。"
f'PackageExporter.extern("'{
模块名称}
`")'
输入文本翻译为简体中文为：\n'
)
如果
调试:
模块路径 =
依赖图.
首路径(
模块名称)
消息.
写(
f" 一个路径到{
模块名称}: {' -> '.
连接(
模块路径)}
输入文本翻译为简体中文为：\n"
)
如果
不
调试:
消息.
写("
输入文本翻译为简体中文为：\n")
消息.
写(
"调用 PackageExporter 时设置 debug=True，以便可视化“
"损坏的模块来自何方！
输入文本翻译为简体中文为：\n"
)
# 保存依赖图，以便工具可以访问。
我.
依赖图 =
依赖图
超级().__init__(
消息.
获取值())
[文档]
类
包导出器:
导出器允许您编写代码包、pickle 的 Python 数据和
将任意二进制和文本资源打包成一个自包含的包。
导入可以以密封的方式加载此代码，这样代码是从包中加载的
而不是通过正常的 Python 导入系统。这允许
将 PyTorch 模型代码和数据打包，以便可以运行
在服务器上或在未来的迁移学习中使用。
包含在包中的代码在创建时是逐文件从原始源复制过来的。
文件格式是特别组织的压缩文件。未来的包用户可以解压包，并编辑代码。
包含在包中的代码在创建时是逐文件从原始源复制过来的，文件格式是特别组织的压缩文件。未来的包用户可以解压包，并编辑代码。
为了对其进行自定义修改。
包导入器确保模块中的代码只能从包中加载。
除了使用 :meth:`extern` 明确列出的外部模块外，模块只能在包内加载。
压缩包中的 `extern_modules` 文件列出了包外部依赖的所有模块。
这防止了“隐式”依赖，因为该软件包在本地运行因为它正在导入
本地安装的包，但随后在将包复制到另一台机器时失败。
当源代码添加到包中时，导出器可以可选地扫描它
为进一步代码依赖（`dependencies=True`）。它查找导入语句，
解析相对引用到限定模块名称，并执行用户指定的操作
（参见：:meth:`extern`，:meth:`mock` 和 :meth:`intern`）。
"文档"
“一个导入器，它将被搜索以找到其他模块或被序列化对象引用的模块。”
默认模块环境仅使用 sys_importer，它搜索 Python 环境。
"文档"
导入器:
导入器
[文档]
定义 __init__(
我,
f: 文件类,
导入器:
联盟[
导入器,
序列[
导入器]] =
sys 导入器,
调试:
布尔 =
错误,
) -> 无:
""
创建导出器。
参数：
f: 导出位置。可以是包含文件名的 ``string``/``Path`` 对象
或二进制输入/输出对象。
导入器：如果传递单个导入器，则使用该导入器搜索模块。
如果传递一系列导入器，则将使用这些导入器构建一个 `OrderedImporter`。
调试：如果设置为 True，则将损坏的模块的路径添加到 PackagingErrors。
"文档"
火把._C._log_api_usage_once("torch.package.PackageExporter")
我.debug = debug
如果 isinstance(f, (
字符串, os.PathLike)):
f = os.文件系统路径(f)
我.
缓冲区:
可选[IO[
字节]] =
无
else: # 是一个字节数组
我.
缓冲区 = f
我.
压缩文件 =
火把._C.
PyTorch 文件写入器(f)
我.
zip 文件.
设置最小版本(6)
我.
已写入文件:
设置[
字符串] =
设置()
我.
序列化减少:
字典[
整数,
任何] = {}
# 跟踪所有模块和添加到本地的 pickle 对象的一个图
包及其之间的依赖关系。
- 每个节点是一个模块名称（或看起来像 '' 的 pickle 名称）。
- 每个有向边（u, v）表示 u 依赖于 v。
- 节点可能包含描述如何将内容写入 zip 文件的元数据。
我.
依赖图 =
Di 图()
我.
脚本模块序列化器 =
火把._C.
脚本模块序列化器类(
我.
zip 文件)
我.
存储上下文 =
我.
脚本模块序列化器.
存储上下文()
# 这些是用于与 RemovableHandle 兼容的 OrderedDict。
# 通用 OrderedDict 类型注解直到 3.7 版本才存在。
# 实际的类型签名是 OrderedDict[int, Callable[[PackageExporter, str], None]]
我.
外部钩子: OrderedDict =
有序字典()
我.
模拟钩子: OrderedDict =
有序字典()
我.
内部钩子: OrderedDict =
有序字典()
如果 isinstance(
导入器,
导入器):
我.
导入器 =
导入器
else:
如果
不 isinstance(
导入器,
集合.abc.
序列):
raise 类型错误(
"导入器参数应该是导入器或导入器序列，"
f"得到"{
类型(
导入器)}
。
)
我.
导入器 =
排序导入器(*
导入器)
我.
模式:
字典[GlobGroup,
模式信息] = {}
我.
_唯一标识符 = 0
[文档]
定义
保存源文件(
我,
模块名称:
字符串,
文件或目录:
字符串,
依赖项=
真实
):
将本地文件系统 `文件或目录` 添加到源包中，以提供 `module_name` 的代码
。
参数：
模块名称（字符串）：例如 ``"my_package.my_subpackage"``, 代码将被保存以提供此包的代码。
文件或目录（字符串）：代码文件或目录的路径。当为目录时，目录中所有 Python 文件
递归地使用 :meth:`save_source_file` 复制。如果一个文件名为 ``"/__init__.py"``，则代码被视为
作为一个包。
dependencies (bool, 可选): 如果为 ``True``，我们将扫描源代码中的依赖项。
"文档"
路径 =
路径(
文件或目录)
如果
路径.
是否为目录():
保存 =
输入文本为空，请提供需要翻译的文本
# 保存_source_string 参数的元组列表
模块路径 =
模块名称.
替换(
“。”,
根目录)
for 文件名
在
路径.
球("**/*.py"):
相对路径 =
文件名.
相对于(
路径).
转换为 POSIX 路径()
存档名称 =
模块路径 +
根目录 +
相对路径
子模块名称 =
无
如果
文件名.
名称 == "__init__.py":
子模块名称 =
存档名称
[ -
长度("/__init__.py")].
替换(
根目录, "."
)
是否为包 =
真实
else:
子模块名称 =
存档名称
[ -
长度(".py")].
替换(
根目录,
“。”)
是否为包 =
假
# 我们延迟调用 save_source_string 以记录所有源文件
在尝试解决依赖关系之前，由这个目录结构提供
在源文件上。这确保我们不会尝试复制那些会被这个目录块覆盖的模块
被覆盖的模块
保存到.append(
(
子模块名称,
读取文件(
字符串(
文件名)),
软件包,
依赖项,
)
)
for 项目
在
保存:
我.
保存源字符串(*
项目)
else:
软件包 =
路径.
名称 == "__init__.py"
我.
保存源字符串(
模块名称,
读取文件(
文件或目录),
是否为包,
依赖项,
)
[文档] def get_unique_id(self) -> str:
"""获取一个 ID。此 ID 保证对于此包只发放一次。"""
ret = str(self._unique_id)
self._unique_id += 1
返回 ret
定义 _get_dependencies(
我,
源:
字符串,
模块名称:
字符串, is_package:
布尔
) -> 列表[
字符串
]
"""返回所有该源代码所依赖的模块。
依赖项通过扫描源代码中的导入类似语句来查找。
参数：
src: 要分析依赖项的 Python 源代码。
module_name: 与 src 对应的模块名称。
is_package: 是否应将此模块视为包。
请参阅：py:meth:`save_source_string` 获取更多信息。
返回：
包含检测到的直接依赖模块的列表
``src``。列表中的项目保证是唯一的。
"文档"
包名 = (
模块名称
如果
包含
否则
模块名称.
右分割(
“。”,
最大分割数=1
)0]
)
尝试:
依赖对 =
查找依赖的源文件(
源,
包名)
除了
异常
是 e:
我.
依赖图.
添加节点(
模块名称,
错误=
包装错误原因.
依赖解析失败,
错误上下文=
字符串(e),
)
返回
输入文本为空，请提供需要翻译的文本
# Use a dict to get uniquing but also deterministic order
依赖 = {}
for 依赖模块名称,
依赖模块对象
在
依赖对:
处理某人像 `from pack import sub` 这样的情况
其中 `sub` 是一个子模块。在这种情况下，我们不需要保存 `pack`，只需保存 `sub` 即可。
这样可以确保我们不会在 `pack` 上选择额外的依赖。
然而，在 `sub` 不是一个子模块而是对象的情况下，我们就需要这样做。
保存包。
如果
模块对象
是
不
无:
可能的子模块 = f"{
模块名称}.{dep_module_obj}"
如果
我._module_exists(possible_submodule):
dependencies[可能的子模块] =
真实
我们不需要保存 `pack`
continue
如果
我.
_模块存在(
依赖模块名称):
依赖关系[
模块名称] =
真实
返回
列表(
依赖关系.
键())
[文档] def 保存源字符串(
self,
module_name: 字符串,
src: 字符串,
is_package: 布尔型 = False,
dependencies: 布尔 = True,
):
"""将 ``src`` 添加为 ``module_name`` 在导出包中的源代码。
Args: 参数：
模块名称（字符串）：例如 ``my_package.my_subpackage``，代码将被保存以提供该包的代码。
src（字符串）：为该包保存的 Python 源代码。
is_package（布尔值，可选）：如果为 ``True``，则此模块被视为包。包可以有子模块（例如 ``my_package.my_subpackage.my_subsubpackage``），并且可以在其中保存资源。默认为 ``False``。
（例如 ``my_package.my_subpackage.my_subsubpackage``），默认为 ``False``。
dependencies (bool, 可选): 如果为 ``True``，我们将扫描源代码中的依赖项。
"""
self.dependency_graph.add_node(
模块名称，
source=源,
is_package=是否为包,
provided=True,
action=_ModuleProviderAction.INTERN,
)
if dependencies:
deps = self._get_dependencies(src, module_name, is_package)
for dep in deps:
self.dependency_graph.add_edge(module_name, dep)
self.add_dependency(dep)
定义
写入源字符串(
我,
模块名称:
字符串,
源:
字符串,
包含:
布尔 =
错误,
):
将 ``src`` 写作 zip 文件夹中 ``module_name`` 的源代码。
参数与 :meth:`save_source_string` 相同。
"文档"
扩展名 = "/__init__.py"
如果
是包
否则 ".py"
文件名 =
模块名称.
替换(
“。”,
根目录) +
扩展
我.
_写入(
文件名,
源)
定义
导入模块(
我,
模块名称:
字符串):
尝试:
返回
我.
导入器.
导入模块(
模块名称)
除了
模块未找到错误:
如果
不
已混淆(
模块名称):
raise
msg = (
f"模块未找到: '"{
模块名称}
'. 请确保 "PackageImporter" 包已安装'
创建此模块存在于`self.importer`中
)
raise 模块未找到错误(
信息) from
无
定义
模块存在(
我,
模块名称:
字符串) ->
布尔:
尝试:
我.
导入模块(
模块名称)
返回
真实
除了
异常:
返回
假
定义
获取模块的源(
我,
模块:
类型.
模块类型) ->
可选[
字符串
]
文件名 =
无
规范 = getattr(
模块, "__spec__",
无)
如果
规范
是
不
无:
加载器 = getattr(
规格,
"加载器",
无)
如果
加载器
是
不
无
并且 isinstance(
加载器,
源文件加载器):
尝试:
文件名 =
加载器.
获取文件名(
模块.__name__)
除了
导入错误:
通过
如果
文件名
是
无:
文件名 = getattr(
模块,
"__文件__",
无)
如果 isinstance(
文件名,
字符串)
并且
文件名.
以...结尾(".py"):
返回
输入文本翻译为简体中文为："".
连接(linecache.
获取行(
文件名,
模块.
字典))
返回
无
[文档]
定义
添加依赖(
我,
模块名称:
字符串,
依赖项=
是):
给定一个模块，根据用户指定的模式将其添加到依赖图中。
。
"文档"
如果 (
模块名称
在
我.
依赖图
并且
我.
依赖图.
节点[
模块名称].
获取(
提供)
是
真实
):
返回
# 特殊情况：PackageImporter 提供了一个名为
`torch_package_importer` 允许打包模块引用
他们的 PackageImporter。我们不想重新导出这个。
如果
模块名称 ==
torch_package_importer:
我.
依赖图.
添加节点(
模块名称,
动作=
模块提供者操作.
跳过,
提供=
是,
)
返回
如果
模块名称 ==
_模拟:
我.
依赖图.
添加节点(
模块名称,
动作=
_模块提供者动作.
重新打包的模拟模块,
提供=
是,
)
返回
如果
我.
可以隐式导出(
模块名称):
我.
依赖图.
添加节点(
模块名称,
动作=
_模块提供者动作.
外部,
提供=
真实
)
返回
for 模式,
模式信息
在
我.
模式.
项目():
如果
模式.
匹配(
模块名称):
模式信息.
已匹配 =
真实
我.
依赖图.
添加节点(
模块名称,
动作=
模式信息.
行动,
提供=
真实
)
如果
模式信息.
行动 ==
模块提供者操作.
拒绝:
# 仅要求拒绝的模块会在图中添加一个错误。
我.
依赖图.
添加节点(
模块名称,
错误=
包装错误原因.
被拒绝
)
# 如果我们要实习这个模块，我们需要检索其
# 依赖并将其打包
如果
模式信息.
动作 ==
_模块提供者动作.
国际:
我.
内置模块(
模块名称,
依赖项)
返回
无匹配模式。请明确添加此错误。
我.
依赖图.
添加节点(
模块名称,
错误=
包装错误原因.
无动作
)
[文档] def save_module(self, module_name: str, dependencies=True): 保存模块(self, 模块名称: 字符串, 依赖关系=True)
将代码模块的代码保存到包中。模块的代码通过使用`importers`路径来解析
模块对象，然后使用其 `__file__` 属性来查找源代码。
参数:
模块名称（字符串）：例如 ``my_package.my_subpackage``，代码将被保存以提供代码
用于此包。
依赖项（布尔值，可选）：如果为 ``True``，我们将扫描源代码以查找依赖项。
"""
如果不是字符串类型 module_name:
raise TypeError(
"save_module() 期望字符串输入，你可能是想传递 `__name__` 吧？"
)
self._intern_module(module_name, dependencies)
定义 _intern_module(
我,
模块名称:
字符串,
依赖项:
布尔,
):
将模块添加到依赖图作为内部模块，并
在序列化时将其写入 zip 文件的任何所需元数据。
"文档"
模块对象 =
我.
导入模块(
模块名称)
如果上述导入成功，则可能是：
1. 模块名称未被混淆，这只是一个常规导入，或者
2. 模块名称被混淆，但其中一个导入器能够
识别混淆并导入它。
# 不论哪种方式，现在都可以安全地取消名称的混淆，这样我们就不需要
# 将混淆后的版本序列化到包中。
模块名称 =
# 取消混淆(
模块名称)
# 查找此模块的依赖项并将它们也一起要求。
软件包 =
有属性(
模块对象,
"__路径__")
源 =
我.
_获取模块源代码(module_obj)
如果
源
是
无:
# 没有找到源！将其添加到我们的依赖图作为损坏项
# 并继续。
文件名 = getattr(
模块对象,
"__文件__",
无)
错误上下文 =
无
如果
文件名
是
无:
打包错误 =
包装错误原因.
无双下划线文件
elif 文件名.
以...结尾(
元组(importlib.
机械.
扩展后缀)):
打包错误 =
包装错误原因.
扩展模块
else:
包装错误 =
包装错误原因.
源文件未找到
错误上下文 = f
文件名:{
文件名}"
我.
依赖图.
添加节点(
模块名称,
动作=
_模块提供者动作.
国际,
是否为包=
是包,
错误=
包装错误,
错误上下文=
错误上下文,
提供=
是,
)
返回
我.
依赖图.
添加节点(
模块名称,
动作=
_模块提供者动作.INTERN,
是否为包=
软件包,
源=
源,
提供=
是,
)
如果
依赖项:
依赖 (缩写) =
我.
获取依赖项(
源,
模块名称,
是否为包)
for 依赖
在
依赖项列表:
我.
依赖图.
添加边(
模块名称, dep)
我.
添加依赖(
沉浸式翻译)
[文档]
定义
保存 pickle(
我,
包:
字符串,
资源:
字符串,
对象:
任何,
依赖项:
布尔 =
是,
pickle 协议:
整型 = 3,
):
将 Python 对象保存到归档中，使用 pickle。相当于：func:`torch.save`，但保存到
归档文件而不是独立文件。标准 pickle 不保存代码，只保存对象。
如果 `dependencies` 为真，此方法还将扫描需要哪些模块的已冻结对象
重建它们并保存相关代码。
能够保存一个对象，其中 `type(obj).__name__` 为 `my_module.MyObject`
`my_module.MyObject` 必须根据 `importer` 顺序解析为对象的类。当保存对象时，
之前已打包，导入器的 `import_module` 方法需要存在于 `importer` 列表中
为了使这可行。
参数：
package (str): 该资源应放入的模块包名称（例如："my_package.my_subpackage"）。
resource (str): 资源的唯一名称，用于标识加载的资源。
obj (Any): 要保存的对象，必须是可序列化的。
dependencies (bool, optional): 如果为 ``True``，我们将扫描源以查找依赖项。
"文档"
断言 (
pickle 协议 == 4)
或者 (
pickle 协议 == 3
), "torch.package 仅支持 pickle 协议 3 和 4"
文件名 =
我.
_文件名(
包,
资源)
# 写入`obj`的 pickle 数据
数据缓冲区 =
输入/输出.BytesIO()
序列化工具 =
创建 pickler(
数据缓冲区,
我.
导入器,
协议=
pickle 协议)
序列化器.
持久化 ID =
我.
__持久化 ID__
序列化器.
导出(
对象)
数据值 =
数据缓冲区.
获取值()
模拟模块 =
默认字典(
列表)
依赖图中的名称 = f
"<"{
软件包}.{
资源}
>
我.
依赖图.
添加节点(
依赖图中的名称,
动作=
_模块提供者动作.INTERN,
提供=
是,
是 pickle=
是,
)
定义
检查模拟错误(
模块:
可选[
字符串
]
字段:
可选[
字符串
)]
""
检查一个对象（字段）是否来自模拟模块，然后将该对添加到包含模拟模块及其
模拟对象列表的 mocked_modules 中。
检查 pickle 中存在的模拟对象列表。
我们还坚持一个不变的原则，即应用于模块的第一个用户定义的规则是我们所使用的。
对模块适用的第一个用户定义的规则就是我们使用的规则。
"文档"
断言 isinstance(
模块,
字符串)
断言 isinstance(
字段,
字符串)
如果
我.
_隐式_外部_(
模块):
返回
for 模式,
模式信息
在
我.
模式.
项目():
如果
模式.
匹配(
模块):
如果
模式信息.
动作 ==
模块提供者操作.
模拟:
被模拟模块[
模块].append(
字段)
返回
如果
依赖项:
所有依赖项 =
输入文本为空，请提供需要翻译的文本
模块 =
无
字段 =
无
描述:
默认字典[
整数,
字符串] =
默认字典(
无)
纪要数量 = 0
# pickletools.dis(data_value)
for 指令码, arg,
_位置
在 pickletools.
生成操作(
数据值):
如果
Pickle 协议 == 4:
如果 (
指令码.
名称 ==
"简短二进制 Unicode"
或者
指令码.
名称 ==
"二进制 Unicode"
或者
指令码.
名称 ==
"二进制 Unicode8"
):
断言 isinstance(arg,
字符串)
模块 =
字段
字段 =
参数
描述[
备忘录计数] =
参数
elif (
指令码.
名称 ==
长柄
或者
指令码.
名称 ==
柄
或者
指令码.
名称 ==
取
):
断言 isinstance(arg,
整数)
模块 =
字段
字段 =
描述.
获取(arg,
无)
elif 指令码.
名称 == "MEMOIZE":
备忘数 += 1
elif 指令码.
名称 ==
"全局栈":
如果
模块
是
无:
如果没有在之前的条目中传递模块，则继续。
continue
断言 isinstance(
模块,
字符串)
如果
模块
不
在
所有依赖项:
所有依赖项.append(
模块)
_检查模拟错误(
模块,
字段)
elif (
pickle 协议 == 3
并且
指令码.
名称 ==
全球
): # 全球引用
断言 isinstance(arg,
字符串)
模块,
字段 = arg.
分割(
输入文本为空，请提供需要翻译的文本)
如果
模块
不
在
所有依赖项:
所有依赖项.append(
模块)
_检查模拟错误(
模块,
字段)
for 模块名称
在
所有依赖项:
我.
依赖图.
添加边(
依赖图中名称,
模块名称)
如果对象恰好来自模拟模块，那么我们收集这些错误并将它们输出
将其他由包导出器发现的错误排除。
"文档"
如果
模块名称
在
模拟模块:
断言 isinstance(
模块名称,
字符串)
字段 =
模拟模块[
模块名称]
我.
依赖图.
添加节点(
模块名称,
动作=
_模块提供者动作.MOCK,
错误=
包装错误原因.
模拟但仍然使用,
错误上下文=f
"对象（们） '"{
字段}
'来自模块 `'{
模块名称}
`在打包过程中被模拟出来 "
f但是正在被用于资源 -{
资源}
` 在包 `{
包}
`. "`,
提供=
是,
)
else:
我.
添加依赖(
模块名称)
我.
_写入(
文件名,
数据值)
[文档] def 保存文本(self, 包: str, 资源: str, 文本: str):
"""保存文本数据到包。
Args:
包名（str）：资源应归属的模块包名（例如："my_package.my_subpackage"）。
资源（str）：用于标识资源的唯一名称，用于加载时识别。
文本（str）：要保存的内容。
"""
return self.save_binary(package, resource, text.encode("utf-8"))
[docs] def save_binary(self, package, resource, binary: bytes):
"""保存原始字节到包中。
Args:
包名（str）：资源应归属的模块包名（例如："my_package.my_subpackage"）。
资源（str）：用于标识资源的唯一名称，用于加载时识别。
二进制数据（str）：要保存的数据。
"""
filename = self._filename(包, 资源)
self._write(文件名, 二进制)
[文档] def register_extern_hook(self, hook: 外部钩子动作) -> 可移除句柄:
"""在导出器上注册外部钩子。"""
每当模块与 :meth:`extern` 模式匹配时，都会调用此钩子。
它应该具有以下签名::
hook(exporter: PackageExporter, module_name: str) -> None
钩子将按照注册顺序调用。
返回：
`:class:`torch.utils.hooks.可移除句柄`:
一个可以通过调用来移除已添加钩子的句柄
处理.remove().
"""
handle = RemovableHandle(self._extern_hooks)
self._extern_hooks[handle.id] = hook
return handle
[文档] def register_mock_hook(self, hook: ActionHook) -> RemovableHandle:
"""注册导出器的模拟钩子。
每当模块与 :meth:`mock` 模式匹配时，将调用此钩子。
它应该具有以下签名::
hook(exporter: PackageExporter, module_name: str) -> None
插件将按照注册顺序被调用。
返回：
class:`torch.utils.hooks.RemovableHandle`:
可以用来通过调用 handle.remove()移除已添加的钩子的句柄
``handle.remove()``.
"""
handle = RemovableHandle(self._mock_hooks)
self._mock_hooks[handle.id] = hook
return handle
[文档] def register_intern_hook(self, hook: ActionHook) -> RemovableHandle:
"""注册导出器上的内部钩子。
每当模块与 :meth:`intern` 模式匹配时，都会调用此钩子。
它应该具有以下签名::
hook(exporter: PackageExporter, module_name: str) -> None
钩子将按照注册顺序调用。
返回：
`:class:`torch.utils.hooks.可移除句柄`:
一个可以通过调用来移除已添加钩子的句柄
处理.remove().
"""
handle = RemovableHandle(self._intern_hooks)
self._intern_hooks[handle.id] = hook
return handle
[文档] def intern(
self,
include: "GlobPattern",
*,
排除: "GlobPattern" = ()
允许空值: bool = True,
):
指定应打包的模块。一个模块必须匹配某些 ``intern`` 模式才能
包含在内，并递归处理其依赖项。
Args:
include (Union[List[str], str]): 字符串，例如 "my_package.my_subpackage"，或字符串列表
用于外部化模块名称。这也可以是 glob 样式模式，如 :meth:`mock` 中所述。
排除（Union[List[str], str]）：一个可选的模式，用于排除与包含字符串匹配的一些模式。
allow_empty（bool）：一个可选的标志，指定是否必须匹配某些在打包期间由此调用指定的 intern 模块。
如果使用 allow_empty=False 添加了 intern 模块的 glob 模式，并且调用了:meth:`close`（无论是显式调用还是通过`__exit__`）
如果添加了 intern 模块的 glob 模式且 allow_empty=False，并且在调用 close（无论是显式调用还是通过`__exit__`）后
在任何模块匹配该模式之前，会抛出异常。如果 `allow_empty=True`，则不会抛出此类异常。
"""
self.patterns[GlobGroup(include, exclude=exclude)] = _PatternInfo(
_ModuleProviderAction.INTERN, allow_empty
)
[文档] def 模拟(
self,
include: "全局模式",
*,
exclude: "GlobPattern" = 空集合
allow_empty: bool = True, 允许空值: 布尔类型 = True
):
将一些必需的模块替换为模拟实现。模拟模块将返回一个假的
对象用于从它访问的任何属性。因为我们逐文件复制，依赖关系解析有时
查找由模型文件导入但从未使用其功能的文件
（例如自定义序列化代码或训练辅助工具）。
使用此功能来模拟此功能，而无需修改原始代码。
Args:
include (Union[List[str], str]): 一个字符串，例如 ``"my_package.my_subpackage"``, 或字符串列表
用于模拟的模块名称。字符串也可以是 glob 样式的模式
字符串可能匹配多个模块。任何匹配此模式的所需依赖项
字符串将被自动模拟。
示例：
torch.** 匹配 torch 及其所有子模块，例如 torch.nn
`torch.nn.functional`
`torch.*` -- 匹配 `torch.nn` 或 `torch.functional`，但不匹配
`torch.nn.functional`
exclude (Union[List[str], str]): 一个可选的模式，用于排除与 include 字符串匹配的一些模式。
例如 ``include='torch.**', exclude='torch.foo'`` 将模拟除 ``'torch.foo'`` 之外的所有 torch 包，
默认：是 ``[]``。
allow_empty (bool)：一个可选标志，指定通过此调用指定的模拟实现是否必须在打包期间匹配到某个模块。
如果通过 :meth:`mock` 方法添加了模拟，则必须匹配到某个模块。
``allow_empty=False``，并且调用了 :meth:`close` 方法（无论是显式调用还是通过 ``__exit__``）且该模拟未与导出包中使用的模块匹配时，将抛出异常。
如果 ``allow_empty=True``，则不会抛出此类异常。
如果 ``allow_empty=True``，则不会抛出此类异常。
"""
self.patterns[GlobGroup(include, exclude=exclude)] = _PatternInfo(
_ModuleProviderAction.MOCK, 允许为空
)
[文档] def extern(
self,
include: "GlobPattern",
*,
exclude: "GlobPattern" = ()
allow_empty: 布尔 = True,
):
将 ``module`` 包含在包可以导入的外部模块列表中。
这将防止依赖发现时保存
将其放入包中。导入器将直接从标准导入系统加载外部模块。
代码对于外部模块也必须在加载包的过程中存在。
参数:
包含（Union[List[str], str]）：字符串，例如 ``"my_package.my_subpackage"``, 或字符串列表
指定要外部化的模块名称。这也可以是一个 glob 样式的模式，如：:meth:`mock`中所述。
排除（Union[List[str], str]）：一个可选的模式，用于排除与包含字符串匹配的一些模式。
exclude（Union[List[str], str]）：一个可选的模式，用于排除与包含字符串匹配的一些模式。
exclude（Union[List[str], str]）：一个可选的模式，用于排除与包含字符串匹配的一些模式。
允许空（布尔值）：一个可选标志，指定此调用指定的外部模块是否允许为空
在打包过程中，必须将“extern”方法匹配到某个模块。如果 extern 模块 glob
模式添加了 `allow_empty=False`，并调用了 :meth:`close`（无论是显式调用还是通过
在匹配任何模块之前抛出异常。如果 `allow_empty=True`，
没有抛出这样的异常。
"""
self.patterns[GlobGroup(include, exclude=exclude)] = _PatternInfo(
_ModuleProviderAction.EXTERN, 允许为空
)
[文档] def deny(self, include: "GlobPattern", *, exclude: "GlobPattern" = ()):
"""阻止匹配给定 glob 模式的模块名称从包可以导入的模块列表中。
如果发现任何匹配的包的依赖项，将引发 :class:`PackagingError`。
Args:
包含（Union[List[str], str]）：一个字符串，例如 ``"my_package.my_subpackage"``, 或者字符串列表
用于指定要导出的模块名称。这也可以是一个 glob 样式的模式，如 :meth:`mock` 中所述。
exclude（Union[List[str], str]）：一个可选的模式，用于排除与 include 字符串匹配的一些模式。
"""
self.patterns[GlobGroup(include, exclude=exclude)] = _PatternInfo(
_ModuleProviderAction.DENY, allow_empty=True
)
定义
持久化 ID(
我,
对象):
如果
火把.is_storage(
对象)
或者 isinstance(
对象,
火把.
存储.
类型化存储):
存储:
存储
如果 isinstance(
对象,
火把.
存储.
类型化存储):
一旦我们决定中断序列化功能，我们就可以
删除此情况
未类型化存储 =
对象.
_未指定存储
存储类型字符串 =
对象.
Pickle 存储类型()
存储类型 = getattr(
火把,
存储类型字符串)
存储 =
角色(
存储,
未类型化存储)
存储 numel =
对象.
尺寸()
elif isinstance(对象,
火把.
未类型化存储):
未类型化存储 =
对象
存储 =
角色(
存储,
未类型化存储)
存储类型 =
标准化存储类型(
类型(
存储))
存储 numel =
存储.
字节数()
else:
raise 运行时错误(f
"未识别的存储类型："{
类型(
对象)}")
位置 =
位置标签(
存储)
如果尚未写入，则序列化存储
存储存在 =
我.
存储上下文.
有存储(
存储)
存储 ID =
我.
存储上下文.
获取或添加存储(
存储)
如果
不
存储存在:
如果
存储.
设备.
类型 != "cpu":
存储 =
存储.cpu()
字节数 =
存储.
字节数()
我.
zip 文件.
记录写入(
f".数据/"{
存储 ID}
.存储,
存储,
字节数
)
返回 (
存储,
存储类型,
存储 ID,
位置,
存储数目)
如果
有属性(
对象, "__reduce_package__"):
如果 _gate_torchscript_serialization
并且 isinstance(
对象,
火把.
算子.RecursiveScriptModule
):
raise 异常(
# 无需注意：TRY002
将 ScriptModules 直接序列化到包中是 Beta 功能。
"使用时，请设置全局 "
"将 `torch.package.package_exporter._gate_torchscript_serialization` 设置为 `False`。"
)
如果
我.
序列化减少.
获取(id(
对象))
是
无:
我.
序列化减少[id(
对象)] = (
减少包,
id(对象),
*对象.__reduce_package__(
我),
)
返回
我.
序列化减少[id(
对象)]
返回
无
定义
__进入__(
我):
返回 self
定义
__退出__(
我,
异常类型, exc_value,
跟踪回溯):
如果 __exit__ 被调用是因为抛出了异常，我们不需要
尝试最终化包。相反，控制权返回到
# 调用者继续抛出异常。
如果
异常类型
是
不
无:
# 仅做最基本的工作以使打开的缓冲区处于有效状态。
我.
_完成压缩()
返回
我.
关闭()
定义
_写入(
我,
文件名,
字符串或字节):
如果
文件名
在
我.
已写入文件:
raise 断言错误(
f尝试写入文件 '{
文件名}
，但该文件已存在于此存档中。 "
"请提交一个错误报告。"
)
我.
_写入的文件.
添加(
文件名)
如果
已混淆(
文件名):
raise 断言错误(
f"尝试将一个 torch.package'd 模块保存为"{
文件名}
。
"直接保存 torch.package 模块是不允许的。"
)
如果 isinstance(
字符串或字节,
字符串):
字符串或字节 =
字符串或字节.
编码(
utf-8)
我.
zip 文件.
记录写入(
文件名,
字符串或字节,
长度(
字符串或字节))
定义
验证依赖图(
我):
# 1. 检查在依赖分析过程中是否插入任何错误。
for attrs 在
我.
依赖图.
节点.
值():
如果
"错误"
在
属性:
raise 包装错误(
我.
依赖图,
调试=
我.
调试)
# 2. 确保所有允许_empty=False 的模式的匹配至少发生一次。
for 模式,
模式信息
在
我.
模式.
项目():
如果
不
模式信息.allow_empty
并且
不
模式信息.
匹配成功:
raise 空匹配错误(
f导出器没有匹配到任何模块，{
模式}
，该模块被标记为 allow_empty=False
)
定义
写入模拟文件(
我):
如果 "_mock.py"
不
在
我.
已写入文件:
模拟文件 =
字符串(
路径(__file__).
父级 / "_mock.py")
我.
写入源字符串(
_模拟,
读取文件(
模拟文件),
是否为包=
错误)
定义
_执行依赖图(
我):
"""接受一个描述如何打包所有
模块并执行它，写入 ZIP 存档。
"文档"
我.
_验证依赖图()
外部模块 =
输入文本为空，请提供需要翻译的文本
for 模块名称, attrs
在
我.
依赖图.
节点.
项目():
动作 =
属性[
"动作"]
如果
动作 ==
_模块提供者动作.
外部:
for 钩子
在
我.
外部钩子.
值():
hook(我,
模块名称)
外部模块.append(
模块名称)
elif 动作 ==
_模块提供者动作.
模拟:
for 钩子
在
我.
_模拟钩子.
值():
hook(我,
模块名称)
我.
写入模拟文件()
是包 =
有属性(
我.
导入模块(
模块名称),
"__路径__")
我.
写入源字符串(
模块名称,
模拟实现,
是否为包)
elif 动作 ==
_模块提供者动作.
国际:
for 钩子
在
我.
_内部钩子.
值():
hook(我,
模块名称)
# 依赖图中节点包含的元数据告诉我们
# 如何内部化模块。
如果
"提供"
不
在
属性:
raise 断言错误(
f"模块被标记为`intern`但未提供："{
模块名称}"
)
如果
属性.
获取("is_pickle")
是
是:
# 此节点来自 save_pickle，我们不需要为其编写任何源代码。
continue
是包 =
属性["is_package"]
源 =
属性[
"源"]
我.
写入源字符串(
模块名称,
源,
是否为包)
elif 动作 ==
_模块提供者动作.
重新打包的模拟模块:
我.
_写入模拟文件()
elif 动作 ==
_模块提供者动作.
跳过:
continue
else:
raise 断言错误(
f"无效的动作："{
模块名称}, {
动作}
请向 PyTorch 报告一个错误。"
)
外部文件内容 = "
输入文本翻译为简体中文为：\n".
连接(
外部模块) + "
输入文本翻译为简体中文为：\n"
我.
_写入(
".data/外部模块",
外部文件内容)
定义
_写入 Python 版本(
我):
"""将创建包时使用的 Python 版本写入到.data/python_version 文件中"""
我.
_写入(".data/python_version",
平台.
python 版本())
[文档] def close(self):
“将包写入文件系统。在调用 :meth:`close` 之后，任何调用都无效。
建议使用资源保护语法代替::
使用 PackageExporter("file.zip") as e:
...
"""
self._execute_dependency_graph()
self._write_python_version()
self.script_module_serializer.write_files()
self._finalize_zip()
定义 _finalize_zip(
我):
在打包的最后阶段调用，以确保 zip 文件处于关闭但有效的状态。
删除
我.
zip 文件
如果
我.
缓冲区:
我.
缓冲区.
清空()
定义 _filename(
我,
包,
资源):
包路径 =
包.
替换(
“。”,
根目录)
资源 = _normalize_path(
资源)
返回 f"{
包路径}/{
资源}"
定义
_可以隐式外部化(
我,
模块名称:
字符串):
顶级包名 =
模块名称.partition(
“。”
)0]
返回
顶级包名 ==
torch
或者 (
顶级包名
不
在
禁止导入的模块
并且
stdlib 模块(
最高级包名)
)
[文档] def dependency_graph_string(self) -> str:
"""返回包中依赖的 digraph 字符串表示形式。
返回：
包中依赖关系的字符串表示。
```python
# 输入文本
input_text = '"""'
# 翻译函数（此处为示例，实际翻译功能需调用真实的翻译 API）
def translate_to_simplified_chinese(text):
# 假设的翻译结果
return text
# 输出翻译结果
translated_text = translate_to_simplified_chinese(input_text)
print(translated_text)
```
返回 self.dependency_graph.to_dot()
定义
具有动作类型的节点(
我,
动作:
可选[
_模块提供者动作]
) -> 列表[
字符串
]
结果 =
输入文本为空，请提供需要翻译的文本
for 名称,
节点字典
在
我.
依赖图.
节点.
项目():
节点动作 =
节点字典.
获取(
动作,
无)
如果
节点动作 ==
动作
并且
是 pickle
不
在 node_dict:
结果.append(
名称)
结果.
排序()
返回
结果
[文档] def externed_modules(self) -> list[str]:
"""返回所有当前已外部的模块。
返回值：
包含将要在此包中公开的模块名称的列表
在此包中公开的模块。
"""
返回 self._nodes_with_action_type(_ModuleProviderAction.EXTERN)
[文档] def interned_modules(self) -> list[str]:
返回所有当前已内部化的模块。
返回：
模块名称列表
包含在此包中。
"""
返回 self._nodes_with_action_type(_ModuleProviderAction.INTERN)
[文档] def mocked_modules(self) -> list[str]:
返回所有当前被模拟的模块。
返回：
一个包含将在本包中模拟的模块名称的列表。
本包中将模拟的模块。
"""
返回 self._nodes_with_action_type(_ModuleProviderAction.MOCK)
[文档] def denied_modules(self) -> list[str]:
"""返回所有当前被拒绝的模块。
返回值：
包含将被此包拒绝的模块名称的列表。
该包将拒绝的模块。
"""
return self._nodes_with_action_type(_ModuleProviderAction.DENY)
[文档] def get_rdeps(self, module_name: str) -> list[str]:
"""返回所有依赖于模块 ``module_name`` 的模块列表。
返回：
包含依赖于 ``module_name`` 的模块名称的列表。
"""
如果 module_name 在 self.dependency_graph._pred.keys() 中：
返回 list(self.dependency_graph._pred[module_name].keys())
else:
返回 []
[文档] def all_paths(self, src: str, dst: str) -> str:
返回子图的点表示
具有从源到目的地的所有路径。
返回：
包含从源到目的地的所有路径的点表示。
(https://graphviz.org/doc/info/lang.html)
"""
返回 self.dependency_graph.all_paths(src, dst)
即使它们在标准库中，我们也不允许它们自动外部化
因为它们提供了大量的系统级访问
禁止模块 = ["sys", "io"]
_MOCK_IMPL = """\
从 _mock 导入 MockedObject
def __getattr__(attr: str):
返回 MockedObject(__name__ + '.' + attr, _suppress_err=True)
""
定义
读取文件(
文件名:
字符串) ->
字符串:
与
打开(
文件名,
rb)
是 f:
b = f.阅读()
返回 b.
解码(
utf-8)
