# mypy: 允许未类型化定义
r包含由 _BaseDataLoaderIter 工作者使用的方法的定义。
这些方法用于将数据集获取的样本收集到张量（Tensor）中。
这些 **需求** 必须在全局范围内，因为 Py2 不支持序列化
静态方法。
`default_collate` 和 `default_convert` 通过 'dataloader.py' 对用户公开。
""
导入
集合
导入 contextlib
导入
复制
导入
正则表达式
来自
打字
导入
可调用,
可选,
联合
导入
火炬
np_str_obj_array_pattern = 正则表达式.
编译(r"[SaUO]")
[文档]def
默认转换(
数据):
r""
将每个 NumPy 数组元素转换为 :class:`torch.Tensor`。
如果输入是 `Sequence`、`Collection` 或 `Mapping`，则尝试将内部每个元素转换为 :class:`torch.Tensor`。
如果输入不是 NumPy 数组，则保持不变。
当在 :class:`~torch.utils.data.DataLoader` 中未定义 `batch_sampler` 和 `batch_size` 时，此函数用作默认的排序函数。
输入类型到输出类型的映射与以下类似。
通用输入类型到输出类型的映射与以下类似。
有关详细信息，请参阅该函数的描述：:func:`~torch.utils.data.default_collate`。
参数：
data：要转换的单个数据点。
示例：
>>> # xdoctest: +SKIP
>>> # 示例使用 `int`
>>> default_convert(0)
0
>>> # NumPy 数组示例
>>> 默认转换 np.array([0, 1])
tensor([0, 1])
>>> # NamedTuple 示例
>>> Point = namedtuple('Point', ['x', 'y'])
>>> 默认转换(Point(0, 0))
Point(x=0, y=0)
>>> 默认转换(Point(np.array(0), np.array(0)))
Point(x=tensor(0), y=tensor(0))
>>> # 示例与列表
>>> default_convert([np.array([0, 1]), np.array([2, 3])])
[tensor([0, 1]), tensor([2, 3])]
```python
# 假设输入文本为：
input_text = '"""'
# 翻译函数（此处仅为示例，实际翻译功能需要调用真实的翻译 API）
def translate_to_simplified_chinese(text):
# 这里应该调用真实的翻译 API 进行翻译
# 由于示例中不使用真实的 API，以下为模拟翻译结果
return text
# 输出翻译结果
translated_text = translate_to_simplified_chinese(input_text)
print(translated_text)
```
elem_type = 类型(
数据)
如果 isinstance(
数据,
火炬.
张量):
返回
数据
elif (
元素类型.__module__ ==
numpy
和
元素类型.__name__ !=
str_
和
元素类型.__name__ !=
"字符串_"
):
# 字符串类数组与对象
如果 (
元素类型.__name__ ==
ndarray
和 np_str_obj_array_pattern.
搜索(
数据.
数据类型.
字符串)
是 not None
):
返回
数据
返回
火炬.as_tensor(
数据)
elif isinstance(数据,
集合.abc.
映射):
try:
如果 isinstance(
数据,
集合.abc.
可变映射):
映射类型可能有额外属性，因此我们不能仅仅
使用 `type(data)(...)` 来创建新的映射。
创建一个克隆并在映射类型可变时更新它。
克隆 =
复制.
复制(
数据)
克隆.
更新({
键:
默认转换(
数据[
键])
为 key
在
数据})
返回
克隆
否则:
返回 elem_type({
键:
默认转换(
数据[
键])
为 key
在
数据})
除了
类型错误:
# 映射类型可能不支持 `copy()` / `update(mapping)`
# 或 `__init__(iterable)`。
返回 {
键:
默认转换(
数据[
键])
为 key
在
数据}
elif isinstance(数据,
元组)
和
有属性(
数据,
"_字段"):
命名元组
返回
元素类型(*(
默认转换(d)
为 d
在
数据))
elif isinstance(数据,
元组):
返回 [
默认转换(d)
为 d
在
数据]
# 向后兼容。
elif isinstance(数据,
集合.abc.
序列)
和 not isinstance(
数据, (
字符串,
字节)
):
try:
如果 isinstance(
数据,
集合.abc.
可变序列):
# 序列类型可能有额外属性，所以我们不能仅仅
# 使用 `type(data)(...)` 来创建新的序列。
创建一个克隆并在序列类型可变时更新它。
克隆 =
复制.
复制(
数据) # type: ignore[arg-type]
为 i, d
在
列举(
数据):
克隆[i] =
默认转换(d)
返回
克隆
否则:
返回 elem_type([
默认转换(d)
为 d
在
数据])
除了
类型错误:
# 序列类型可能不支持 `copy()` / `__setitem__(index, item)`
# 或 `__init__(iterable)`（例如 `range`）。
返回 [
默认转换(d)
为 d
在
数据]
否则:
返回
数据
默认排序错误信息格式 = (
"默认排序：批次必须包含张量、numpy 数组、数字、"
"字典或列表；发现"{}"
)
[文档]def
汇总(
批量,
*,
collate_fn_map: 可选[
字典[
联盟[
类型,
元组[
类型, ...]],
可调用]] =
无,
):
r""
通用归一化函数，用于处理每个批次中元素的集合类型。
此函数还打开函数注册表以处理特定元素类型。`default_collate_fn_map`
为张量、NumPy 数组、数字和字符串提供默认归一化函数。
参数：
批次：要归档的单个批次
collate_fn_map：从元素类型到相应归档函数的可选字典映射。
如果元素类型不在此字典中，
此函数将按插入顺序遍历字典中的每个键。
如果元素类型是键的子类，则调用相应的 collate 函数。
示例：
>>> def collate_tensor_fn(batch, *, collate_fn_map):
... # 扩展此函数以处理张量批次
... return torch.stack(batch, 0)
>>> def custom_collate(batch):
... collate_map = {torch.Tensor: collate_tensor_fn}
... return collate(batch, collate_fn_map=collate_map)
>>> # 通过原地修改 `default_collate_fn_map` 来扩展 `default_collate`
>>> 默认合并函数映射更新为 {torch.Tensor: collate_tensor_fn}
注意：
每个归并函数需要一个用于批次的定位参数和一个用于关键字的参数
为合并函数字典作为 `collate_fn_map`。
```python
# 假设输入文本为：
input_text = '"""'
# 翻译函数（此处仅为示例，实际翻译功能需要调用真实的翻译 API）
def translate_to_simplified_chinese(text):
# 这里应该调用真实的翻译 API 进行翻译
# 由于示例中不使用真实的 API，以下为模拟翻译结果
return text
# 输出翻译结果
translated_text = translate_to_simplified_chinese(input_text)
print(translated_text)
```
元素 =
批量[0]
元素类型 =
类型(
元素)
如果 collate_fn_map
是 not
无:
如果 elem_type
在 collate_fn_map:
返回 collate_fn_map[
元素类型
]
批量,
排序函数映射=
排序函数映射)
为
排序类型
在
合并函数映射:
如果 isinstance(
元素, collate_type):
返回
合并函数映射[collate_type
[ ](
批量,
合并函数映射=collate_fn_map
)
如果 isinstance(
元素,
集合.abc.
映射):
try:
如果 isinstance(
元素,
集合.abc.
可变映射):
映射类型可能包含额外的属性，因此我们不能只是
使用 `type(data)(...)` 来创建新的映射。
如果映射类型是可变的，则创建一个克隆并更新它。
克隆 =
复制.
复制(
元素)
克隆.
更新(
{
键:
汇总(
[d[键]
为 d
在
批量
]
合并函数映射=collate_fn_map
)
为 key
在
元素
}
)
返回
克隆
否则:
返回
元素类型(
{
键:
汇总(
[d[键]
为 d
在
批量
]
排序函数映射=
排序函数映射
)
为 key
在
元素
}
)
除了
类型错误:
映射类型可能不支持 `copy()` / `update(mapping)`
或 `__init__(iterable)`。
返回 {
键:
汇总([d[
键]
为 d
在
批量
] collate_fn_map=collate_fn_map)
为 key
在
元素
}
elif isinstance(元素,
元组)
和
有属性(
元素,
"_字段"):
命名元组
返回
元素类型(
*(
汇总(
样本,
合并函数映射=collate_fn_map)
为
样本
在
压缩(*
批量)
)
)
elif isinstance(元素,
集合.abc.
序列):
# 检查批次中的元素是否具有一致的大小
它 =
迭代(
批量)
元素大小 =
长度(
下一(
它))
如果 not
所有(
长度(
元素) ==
元素大小
为
元素
在
它):
抛出异常
运行时错误(
批次列表中的每个元素大小应相等)
交换 =
列表(
压缩(*
批量))
它可能被访问两次，所以我们使用列表。
如果 isinstance(
元素,
元组):
返回 [
汇总(
样本, collate_fn_map=collate_fn_map)
为
样本
在
交换
] # 向后兼容。
否则:
try:
如果 isinstance(
元素,
集合.abc.
可变序列):
# 序列类型可能有额外属性，所以我们不能仅仅
# 使用 `type(data)(...)` 来创建新的序列。
创建一个克隆并在序列类型可变时更新它。
克隆 =
复制.
复制(
元素) # type: ignore[arg-type]
为 i,
样本
在
列举(
交换):
克隆[i] =
汇总(
样本, collate_fn_map=collate_fn_map)
返回 clone
否则:
返回 elem_type(
[
汇总(samples, collate_fn_map=collate_fn_map)
为
样本
在
交换
]
)
除了
类型错误:
序列类型可能不支持 `copy()` / `__setitem__(索引, 项目)`
或 `__init__(可迭代对象)`（例如 `range`）。
返回 [
汇总(
样本, collate_fn_map=collate_fn_map)
为
样本
在
交换
]
抛出异常
类型错误(
默认合并错误信息格式.
格式(
元素类型))
def 合并张量函数(
批量,
*,
合并函数映射:
可选[
字典[
联盟[
类型,
元组[
类型, ...]],
可调用]] =
无,
):
元素 =
批量[0]
out = None
如果
元素.
是嵌套的:
抛出异常
运行时错误(
"默认的 collate_fn 目前不支持嵌套张量的批次；"
"请提供一个自定义的 collate_fn 来适当处理它们。"
)
如果
元素.
布局
在 {
火炬.
稀疏 COO,
火炬.
稀疏压缩存储格式,
火炬.
稀疏_bsr,
火炬.
稀疏压缩存储格式,
火炬.
稀疏矩阵,
}:
抛出异常
运行时错误(
"默认的 collate_fn 目前不支持稀疏张量的批次；"
"请提供一个自定义的 collate_fn 来适当处理它们。"
)
如果
火炬.
工具.
数据.
获取工人信息()
是 not
无:
如果我们在后台进程中，则直接连接到共享内存张量以避免额外的复制
# shared memory tensor to avoid an extra copy
numel = 总和(x.
元素数量()
为 x
在
批量)
存储 =
元素._typed_storage().
新共享(
元素数量,
设备=
元素.
设备)
out = 元素.
新(
存储).
调整大小(
长度(
批量), *
列表(
元素.
尺寸()))
返回
火炬.
栈(
批量, 0,
输出=
输出)
def collate_numpy_array_fn(
批量,
*,
collate_fn_map: 可选[
字典[
联盟[
类型,
元组[
类型, ...]],
可调用]] =
无,
):
元素 =
批量[0]
# 数组字符串类和对象
如果 np_str_obj_array_pattern.
搜索(
元素.
数据类型.
字符串)
是 not
无:
抛出异常
类型错误(
默认排序错误信息格式.
格式(
元素.
数据类型))
返回
汇总([
火炬.as_tensor(b)
为 b
在
批量
]
排序函数映射=
排序函数映射)
def 排序 NumPy 标量函数(
批量,
*,
collate_fn_map: 可选[
字典[
联盟[
类型,
元组[
类型, ...]],
可调用]] =
无,
):
返回
火炬.as_tensor(
批量)
def collate_float_fn(
批量,
*,
collate_fn_map: 可选[
字典[
联盟[
类型,
元组[
类型, ...]],
可调用]] =
无,
):
返回
火炬.
张量(
批量,
数据类型=
火炬.float64)
def collate_int_fn(
批量,
*,
collate_fn_map: 可选[
字典[
联盟[
类型,
元组[
类型, ...]],
可调用]] =
无,
):
返回
火炬.
张量(
批量)
def collate 字符串函数(
批量,
*,
collate 函数映射:
可选[
字典[
联盟[
类型,
元组[
类型, ...]],
可调用]] =
无,
):
返回 batch
默认 collate 函数映射:
字典[
联盟[
类型,
元组[
类型, ...]],
可调用] = {
火炬.
张量:
collate 张量函数
}
与 contextlib.
抑制(
导入错误):
导入 numpy
作为 np
对于 ndarray 和 memmap（ndarray 的子类）
默认的 collate_fn 映射[
numpy.
纳维数组] =
collate_numpy_array_fn 函数
查看标量层次结构：https://numpy.org/doc/stable/reference/arrays.scalars.html
# Skip string scalars
默认排序函数映射
[
numpy.
布尔_,
numpy.
数字,
numpy.
对象
] =
排序 NumPy 标量函数
默认合并函数映射[float] =
合并浮点数函数
默认合并函数映射[int] =
合并整数函数
默认合并函数映射[
字符串] =
字符串合并函数
默认合并函数映射[
字节] =
字符串合并函数
[文档]def
默认的 collate(
批量):
r""
接收一批数据，并将批次中的元素放入一个具有额外外维度的张量中 - 批量大小。
确切的输出类型可以是 :class:`torch.Tensor`，一个 `Sequence` 的 :class:`torch.Tensor`，或者是一个 :class:`torch.Tensor` 的集合，或者根据输入类型保持不变。
这取决于输入类型，可以是 :class:`torch.Tensor` 的集合，或者保持不变。
这被用作默认的收集函数，用于 ...
`batch_size` 或 `batch_sampler` 定义在 :class:`~torch.utils.data.DataLoader` 中。
这里是输入类型（基于批内元素的类型）到输出类型的一般映射：
* :class:`torch.Tensor` -> :class:`torch.Tensor`（添加了批大小外层维度）
* NumPy 数组 -> :class:`torch.Tensor`
* `float` -> :class:`torch.Tensor`
* `int` -> :class:`torch.Tensor`
* `str` -> `str` (未改变)
* `bytes` -> `bytes` (未改变)
* `映射[K, V_i]` -> `映射[K, default_collate([V_1, V_2, ...])]`
* `命名元组[V1_i, V2_i, ...]` -> `命名元组[default_collate([V1_1, V1_2, ...]),
default_collate([V2_1, V2_2, ...]), ...]`
* `序列[V1_i, V2_i, ...]` -> `序列[default_collate([V1_1, V1_2, ...]),
default_collate([V2_1, V2_2, ...]), ...]
参数：
batch: 要归档的单个批次
示例：
>>> # xdoctest: +SKIP
>>> # 示例：使用 `int` 的批次：
>>> default_collate([0, 1, 2, 3])
tensor([0, 1, 2, 3])
>>> # 示例：使用一个字符串批次的例子
>>> default_collate(['a', 'b', 'c'])
['a', 'b', 'c']
>>> # 示例：`Map` 在批处理内部
>>> default_collate([{'A': 0, 'B': 1}, {'A': 100, 'B': 100}])
{'A': tensor([ 0, 100]), 'B': tensor([ 1, 100])}
>>> # 示例：`NamedTuple` 在批处理内部
>>> Point = namedtuple('Point', ['x', 'y'])
>>> default_collate([Point(0, 0), Point(1, 1)])
Point(x=tensor([0, 1]), y=tensor([0, 1]))
>>> # Example with `Tuple` inside the batch:
>>> default_collate([(0, 1), (2, 3)])
[tensor([0, 2]), tensor([1, 3])]
>>> 示例中包含 `List` 的批处理：
>>> 默认的 collate 函数：[[0, 1], [2, 3]]
[tensor([0, 2]), tensor([1, 3])]
>>> # 两种方法扩展 `default_collate` 以处理特定类型
>>> # 方法 1：编写自定义的 collate 函数并调用 `default_collate`
>>> def custom_collate(batch):
... elem = batch[0]
... 如果是 CustomType 类型的实例：# 某个自定义条件
... 返回 ...
... 否则：# 回退到 `default_collate`
... 返回 default_collate(batch)
>>> # 选项 2：就地修改 `default_collate_fn_map`
>>> 定义 collate_customtype_fn 函数
...
>>> default_collate_fn_map.update(CustomType, collate_customtype_fn)
>>> 默认合并批次 # 自动处理 `CustomType`
```python
# 假设输入文本为：
input_text = '"""'
# 翻译函数（此处仅为示例，实际翻译功能需要调用真实的翻译 API）
def translate_to_simplified_chinese(text):
# 这里应该调用真实的翻译 API 进行翻译
# 由于示例中不使用真实的 API，以下为模拟翻译结果
return text
# 输出翻译结果
translated_text = translate_to_simplified_chinese(input_text)
print(translated_text)
```
返回
汇总(
批量,
合并函数映射=
默认合并函数映射)
