torch.utils.data.sampler 的源代码

# mypy: 允许未类型化定义
导入 itertools
来自 collections.abc 导入 迭代器, 迭代器, 序列, 大小可定
来自 打字 导入 通用, 可选, 类型变量, 联合

导入 火炬


__all__ = [
    批量采样器,
    随机采样器,
    "采样器",
    "顺序采样器",
    "子集随机采样器",
    "加权随机采样器",
]


_T_co = 类型变量(_T_co, 协变=True)


类 采样器(通用[_T_co)]
    r所有采样器的基类。

每个采样器子类都必须提供一个 :meth:`__iter__` 方法，提供遍历数据集元素索引或索引列表（批次）的方式，
并且可以提供一个 :meth:`__len__` 方法，返回返回迭代器的长度。
和可能提供一个返回迭代器长度的 :meth:`__len__` 方法。

参数：
数据源（数据集）：此参数未使用，将在 2.2.0 版本中删除。
您可能仍然有使用它的自定义实现。

示例：
        >>> # xdoctest: +SKIP
>>> class 根据序列长度采样器(AccedingSequenceLengthSampler)(int):
        >>>     def __init__(self, data: List[str]) -> None:
        >>>         self.data = data
...
        >>>     def __len__(self) -> int:
        >>>         return len(self.data)
...
        >>>     def __iter__(self) -> Iterator[int]:
        >>>         sizes = torch.tensor([len(x) for x in self.data])
        >>>         yield from torch.argsort(sizes).tolist()
...
        >>> class AccedingSequenceLengthBatchSampler(Sampler[List[int]]):
        >>>     def __init__(self, data: List[str], batch_size: int) -> None:
        >>>         self.data = data
        >>>         self.batch_size = batch_size
...
        >>>     def __len__(self) -> int:
        >>>         return (len(self.data) + self.batch_size - 1) // self.batch_size
...
>>>     def __iter__(self) -> 迭代器[List[int]]:
        >>>         sizes = torch.tensor([len(x) for x in self.data])
        >>>         for batch in torch.chunk(torch.argsort(sizes), len(self)):
        >>>             yield batch.tolist()

.. note:: :meth:`__len__` 方法不是严格必需的
`:class:`~torch.utils.data.DataLoader`，但预期在任何
涉及`:class:`~torch.utils.data.DataLoader`长度的计算
```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)
```

    def 初始化(自身, 数据源: 可选[尺寸] = 无) -> 无:
        如果 数据源 是 not 无:
            导入 警告

            warnings.警告(
                "该`数据源`参数未被使用，将在 2.2.0 版本中删除。"
                "您可能仍然有使用它的自定义实现。"
            )

    def __iter__(自身) -> 迭代器[_T_co]：
        提升 不支持的操作异常


    # 缺少 Python 抽象基类中的默认 `__len__`
    #
    # 许多时候，我们有一个代表集合/可迭代的抽象类
    # 例如，`torch.utils.data.Sampler`，其子类可以选择实现 `__len__` 方法
    # 在这种情况下，我们必须确保不要
    提供默认实现，因为两者都是直接默认
    # 实现存在其问题：
    #
    #   + 返回未实现
    调用 `len(subclass_instance)` 引发异常：
    无法解释为整数的 'NotImplementedType' 对象
    #
    + `引发 NotImplementedError`:
    这防止触发某些回退行为。例如，内置的
    `list(X)` 尝试首先调用 `len(X)`，并执行不同的代码
    #     如果方法未找到或返回 `NotImplemented`，则路径无效，同时
    #     抛出 `NotImplementedError` 将会传播并使调用失败
    #     在那里它本可以使用 `__iter__` 来完成调用。
    #
    # 因此，唯一合理的两件事是
    #
    #   + **不**提供默认的 `__len__`。
    #
    #   + 抛出 `TypeError`，这是当用户调用
    #     对象上未定义的方法时 Python 会使用的错误类型。
    #     (@ssnl 验证了这至少在 Python 3.7 上是有效的。)


[docs]class SequentialSampler(Sampler[int]):
r"""按顺序采样元素，始终按相同顺序。

Args:
data_source (Dataset): 从中采样的数据集
"""
"""

数据源：固定大小

def __init__(self, data_source: 固定大小) -> None:
self.data_source = data_source

def __iter__(self) -> 迭代器[int]:
return iter(range(len(self.data_source)))

def __len__(self) -> int:
return len(self.data_source)


类 随机采样器(采样器[int)]
    r随机抽取样本。如果不带替换，则从打乱的数据集中抽取。

如果带替换，则用户可以指定 :attr:`num_samples` 来抽取。

参数：
data_source (数据集): 从中抽取样本的数据集
replacement (布尔值): 如果为 ``True``，则按需带替换抽取样本，默认为 ``False``
num_samples (int): 要抽取的样本数量，默认为`len(dataset)`。
generator (Generator): 用于采样的生成器。
```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)
```

    data_source: Sized
    替换: 布尔类型

    def 初始化(
        自身,
        数据源: 尺寸,
        替换: 布尔类型 = 错误,
        样本数量: 可选[int] = 无,
        生成器=无,
    ) -> 无:
        自身.数据源 = 数据源
        自身.替代方案 = 替代方案
        自身.样本数量 = 样本数量
        自身.生成器 = 生成器

        如果 not isinstance(自身.替换, 布尔):
            提升 类型错误(
                f"替换值应为布尔值，但得到的是 replacement="{自身.替换}"
            )

        如果 not isinstance(自身.样本数量, int) 或 自身.样本数量 ≤ 0:
            提升 ValueError(
                f"num_samples 应该是正整数，但得到的是 num_samples="{自身.样本数量}"
            )

    @property
    def 样本数量(自身) -> int:
        # 数据集大小可能在运行时改变
        如果 自身.样本数量 是 无:
            返回 长度(自身.数据源)
        返回 自身.样本数量

    def __iter__(自身) -> 迭代器[int]：
        n = 长度(自身.数据源)
        如果 自身.生成器 是 无:
            种子 = int(火炬.空的((), 数据类型=火炬.int64).随机().项目())
            生成器 = 火炬.生成器()
            生成器.手动播种(种子)
        否则:
            生成器 = 自身.生成器

        如果 自身.替换:
            为 _ 在 范围(自身.样本数量 // 32):
                yield from 火炬.随机整数(
                    高=n, 尺寸=(32,), 数据类型=火炬.int64, 生成器=生成器
                ).转列表()
            yield from 火炬.随机整数(
                高=n,
                尺寸=(自身.样本数量 % 32,),
                数据类型=火炬.int64,
                生成器=生成器,
            ).转列表()
        否则:
            为 _ 在 范围(自身.样本数量 // n):
                yield from 火炬.随机排列(n, 生成器=生成器).转列表()
            yield from 火炬.随机排列(n, 生成器=生成器).转列表()[
                : 自身.样本数量 % n
            ]

    def __len__(自身) -> int:
        返回 自身.样本数量



[文档]class SubsetRandomSampler(Sampler[int]):
随机从给定索引列表中抽取元素，不进行重复抽取。

参数:
索引（序列）：索引序列
生成器（生成器）：用于采样的生成器
"""

indices: int 序列

def __init__(self, indices: Sequence[int], generator=None) -> None:
self.indices = indices
self.generator = generator

def __iter__(self) -> Iterator[int]:
for i in torch.randperm(len(self.indices), generator=self.generator):
yield self.indices[i]

def __len__(self) -> int:
return len(self.indices)


类 加权随机采样器(采样器[int)]
    r从 ``[0,..,len(weights)-1]`` 中根据给定的概率（权重）采样元素。

参数：
权重（序列）: 一系列权重，不一定要加起来等于一。
num_samples（整数）：要抽取的样本数量。
替换（布尔值）：如果为 ``True``，则进行有放回抽样。
如果不是，则进行无放回抽样，这意味着当为某一行抽取样本索引时，该索引不能再次用于该行。
生成器（生成器）：用于抽样的生成器。
生成器（Generator）：用于抽样的生成器。

示例：
>>> # xdoctest: +IGNORE_WANT("非确定性")
        >>> list(WeightedRandomSampler([0.1, 0.9, 0.4, 0.7, 3.0, 0.6], 5, replacement=True))
        [4, 4, 1, 4, 5]
        >>> list(WeightedRandomSampler([0.9, 0.4, 0.05, 0.2, 0.3, 0.1], 5, replacement=False))
        [0, 1, 4, 3, 2]
```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)
```

    权重: 火炬.张量
    样本数量: 整型
    替换: 布尔类型

    def 初始化(
        自身,
        权重: 序列[float]
        样本数量: int,
        替换: 布尔类型 = True,
        生成器=无,
    ) -> 无:
        如果 (
            not isinstance(样本数量, int)
            或 isinstance(样本数量, 布尔)
            或 样本数量 ≤ 0
        ):
            提升 ValueError(
                f"num_samples 应该是一个正整数，但得到了 num_samples="{样本数量}"
            )
        如果 not isinstance(替换, 布尔):
            提升 ValueError(
                f"replacement 应该是一个布尔值，但得到了 replacement="{替换}"
            )

        weights_tensor = 火炬.as_tensor(权重, 数据类型=火炬.double)
        如果 长度(weights_tensor.shape) != 1:
            提升 ValueError(
                "权重应该是一个一维序列，但给出的是"
                f"权重的形状"{元组(weights_tensor.shape)}"
            )

        自身.权重 = weights_tensor
        自身.样本数量 = 样本数量
        自身.替代方案 = 替代方案
        自身.生成器 = 生成器

    def __iter__(自身) -> 迭代器[int]：
        随机张量 = 火炬.多项式(
            自身.权重, 自身.样本数量, 自身.替换, 生成器=自身.生成器
        )
        yield from 迭代(随机张量.转列表())

    def __len__(自身) -> int:
        返回 自身.样本数量



类 批量采样器(采样器[列表[int]]：
    r包装另一个采样器以生成索引的迷你批次。

参数：
采样器（Sampler 或 Iterable）：基采样器。可以是任何可迭代对象
批处理大小（整数）：迷你批处理的大小。
drop_last（布尔值）：如果为 ``True``，当批处理大小小于 ``batch_size`` 时，采样器将丢弃最后一个批处理。
它的大小小于 ``batch_size`` 时

示例：
        >>> list(BatchSampler(SequentialSampler(range(10)), batch_size=3, drop_last=False))
        [[0, 1, 2], [3, 4, 5], [6, 7, 8], [9]]
        >>> list(BatchSampler(SequentialSampler(range(10)), batch_size=3, drop_last=True))
        [[0, 1, 2], [3, 4, 5], [6, 7, 8]]
```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)
```

    def 初始化(
        自身,
        采样器: 联盟[采样器[int] 迭代器[int]],
        批处理大小: int,
        drop_last: 布尔,
    ) -> 无:
        # 由于 collections.abc.Iterable 没有检查`__getitem__`，这是对象成为可迭代对象的一种方式，我们在这里不进行`isinstance`检查。
        # 这不是检查对象是否为可迭代对象的方法。
        # 这里没有进行`isinstance`检查。
        如果 (
            not isinstance(批处理大小, int)
            或 isinstance(批处理大小, 布尔)
            或 批处理大小 ≤ 0
        ):
            提升 ValueError(
                f批处理大小应该是一个正整数，但得到了 batch_size={批处理大小}"
            )
        如果 not isinstance(drop_last, 布尔):
            提升 ValueError(
                f"drop_last 应该是一个布尔值，但是得到了 drop_last="{drop_last}"
            )
        自身.样本器 = 样本器
        自身.批处理大小 = 批处理大小
        自身.drop_last = drop_last

    def __iter__(自身) -> 迭代器[列表[int]]
        基于 https://github.com/pytorch/pytorch/pull/76951 的基准测试实现
        sampler_iter = 迭代(自身.采样器)
        如果 自身.drop_last:
            # 创建对同一迭代器的多个引用
            args = [sampler_iter] * 自身.批处理大小
            为 batch_droplast 在 压缩(*参数):
                产生 [*batch_droplast]
        否则:
            batch = [*itertools.islice(样本迭代器, 自身.批处理大小]
            while 批量:
                产生 批量
                批量 = [*itertools.islice(样本迭代器, 自身.批处理大小]

    def __len__(自身) -> int:
        仅当 self.sampler 实现了 __len__ 方法时才能调用
        由于无法强制执行此条件，因此我们关闭了以下实现中的类型检查
        相关内容：参见 NOTE [Python 抽象基类中缺少默认的 `__len__`]
        相关：参见 NOTE [Python 抽象基类中缺少默认的 `__len__`]
        如果 自身.drop_last:
            返回 长度(自身.采样器) // 自身.批处理大小  # type: ignore[arg-type]
        否则:
            返回 (长度(自身.采样器) + 自身.批处理大小 - 1) // 自身.批处理大小  # 类型：忽略[arg-type]