• t2vec 辅助笔记：data_utils

    1 argsort

    1. '''
    2. 目的是对一个列表 seq 中的子列表或数组按其长度进行排序，并返回排序后的索引。
    3. 具体来说，它会按长度的降序排列，这意味着最长的子列表/数组的索引会在前面。
    4. '''
    5. def argsort(seq):
    6.     """
    7.     sort by length in reverse order
    8.     ---
    9.     seq (list[array[int32]])
    10.     """
    11.     return [x for x,y in sorted(enumerate(seq),
    12.                                 key = lambda x: len(x[1]),
    13.                                 reverse=True)]
    • `enumerate(seq)` 返回一个枚举对象，其中包含 `seq` 中每个项的索引和值。
    • `key = lambda x: len(x[1])` 是排序的关键。它告诉 `sorted` 函数按照长度 (`len(x[1])`) 进行排序。这里的 `x[1]` 表示 `enumerate` 枚举对象的值，即 `seq` 中的子列表/数组。
    • `reverse=True` 指定排序应按降序进行。
    • 最后，列表解析 `[x for x,y in ...]` 提取了排序后的索引值，这些索引值随后返回。

    1.1 举例

    1. seq = [[1, 2], [1, 2, 3, 4], [1], [1, 2, 3]]
    2. result = argsort(seq)
    3. print(result)
    4. #[1, 3, 0, 2]

    2 pad_array相关

    2.1 pad_array

    1. '''
    2. 对给定的数组 a 进行填充，使其长度达到指定的 max_length。填充的值是 PAD
    3. '''
    4.  
    5. def pad_array(a, max_length, PAD=constants.PAD):
    6.     """
    7.     a (array[int32])
    8.     """
    9.     return np.concatenate((a, [PAD]*(max_length - len(a))))
    •  `[PAD]*(max_length - len(a))`：这部分代码会创建一个新的列表，其长度为 `(max_length - len(a))`，并且所有的元素都是 `PAD`
    • `np.concatenate((a, ...))`：这将原始数组 `a` 和新生成的 `PAD` 值列表连接起来，从而产生一个新的、长度为 `max_length` 的数组。

    2.2 pad_arrays

    1. '''
    2. 接受一个数组的列表（或序列的序列）a，并确保所有这些数组都填充到与列表中最长的数组相同的长度
    3. '''
    4. def pad_arrays(a):
    5.     max_length = max(map(len, a)) 
    6.     #确定 a 中的所有数组中的最大长度
    7.     a = [pad_array(a[i], max_length) for i in range(len(a))]
    8.     #遍历 a 中的每个数组，然后使用先前定义的 pad_array 函数将每个数组填充到 max_length
    9.     a = np.stack(a).astype(np.int)
    10.     #将填充后的数组堆叠成一个二维数组，确保所有的值都是整数类型
    11.     return torch.LongTensor(a)

    2.3 pad_arrays_pair

    处理源-目标（src-trg）序列对，使源/目标序列分别填充到相同的长度

    1. def pad_arrays_pair(src, trg, keep_invp=False):
    2.     """
    3.     输入：
    5.         src (list[array[int32]])：一个包含源序列的列表。
    6.         trg (list[array[int32]])：一个包含目标序列的列表。
    7.         keep_invp：一个布尔值，用来决定是否保留反向排列的信息。
    9.     输出：
    11.         src (seq_len1, batch)：填充后的源序列的张量，形状为 (序列长度, 批量大小)。
    12.         trg (seq_len2, batch)：填充后的目标序列的张量，形状为 (序列长度, 批量大小)。
    13.         lengths (1, batch)：一个张量，包含源序列的原始长度。
    14.         invp (batch,)：一个张量，包含反向排列的信息，只有在 keep_invp 为 True 时才返回。
    15.     """
    16.     TD = namedtuple('TD', ['src', 'lengths', 'trg', 'invp'])
    17.  
    18.     assert len(src) == len(trg), "source and target should have the same length"
    19.     #确保源序列和目标序列的长度是相同的。
    20.  
    21.     idx = argsort(src)
    22.     src = list(np.array(src)[idx])
    23.     trg = list(np.array(trg)[idx])
    24.     '''
    25.     使用 `argsort` 函数来获取按长度降序排序后的序列的索引。
    26.     然后，使用这些索引来对 `src` 和 `trg` 进行重新排序。
    27.     '''
    28.  
    29.     lengths = list(map(len, src))
    30.     lengths = torch.LongTensor(lengths)
    31.     #计算每个源序列的长度并将其转换为一个 PyTorch 张量。
    32.  
    33.     src = pad_arrays(src)
    34.     trg = pad_arrays(trg)
    35.     #使用之前定义的 `pad_arrays` 函数来对 `src` 和 `trg` 中的每个序列进行填充。
    36.     #(batch,seq_len1) (batch,seq_len2)
    37.  
    38.     if keep_invp == True:
    39.         invp = torch.LongTensor(invpermute(idx))
    40.  
    41.         # (batch, seq_len) => (seq_len, batch)
    42.         return TD(src=src.t().contiguous(), 
    43.                     lengths=lengths.view(1, -1), 
    44.                     trg=trg.t().contiguous(), 
    45.                     invp=invp)
    46.     else:
    47.         # (batch, seq_len) => (seq_len, batch)
    48.         return TD(src=src.t().contiguous(), 
    49.                 lengths=lengths.view(1, -1), 
    50.                 trg=trg.t().contiguous(), 
    51.                 invp=[])

     2.4 pad_arrays_keep_invp

    1. '''
    2. 对输入的序列src进行填充，以便它们在批处理中具有相同的长度。
    3. 同时，它还保存并返回一个逆排列，这样你可以使用这个逆排列来恢复原始的顺序
    4. '''
    5. def pad_arrays_keep_invp(src):
    6.     """
    7.     Input:
    8.     src (list[array[int32]])
    9.     ---
    10.     Output:
    11.     src (seq_len, batch)
    12.     lengths (1, batch)
    13.     invp (batch,): inverse permutation, src.t()[invp] gets original order
    14.     """
    15.  
    16.     idx = argsort(src)
    17.     #对src中的序列按长度进行排序,返回一个排列索引
    18.     src = list(np.array(src)[idx])
    19.     #通过索引idx从src中提取排序后的序列
    20.  
    21.     lengths = list(map(len, src))
    22.     lengths = torch.LongTensor(lengths)
    23.     #获取每个序列的长度
    24.  
    25.     src = pad_arrays(src)
    26.     #序列进行填充，使所有序列具有相同的长度
    27.  
    28.     invp = torch.LongTensor(invpermute(idx))
    29.     #获得idx的逆排列，之后可以使用这个逆排列来恢复原始的顺序
    30.  
    31.     return src.t().contiguous(), lengths.view(1, -1), invp

    3 invpermute

    1. '''
    2. 计算一个排列的逆排列
    4. 给定一个排列 p，它的逆排列是一个排列 invp，满足 invp[p[i]] = i 和 p[invp[i]] = i 对所有 i 成立
    5. '''
    6. def invpermute(p):
    7.     """
    8.     inverse permutation
    9.     """
    10.     p = np.asarray(p)
    11.     invp = np.empty_like(p)
    12.     for i in range(p.size):
    13.         invp[p[i]] = i
    14.     #对于每个 i，设置 invp 在 p[i] 的位置上的值为 i
    15.     return invp

    3.1 举例

    1. p = [2, 0, 1]
    2. invp = invpermute(p)
    3. invp
    4. #[1, 2, 0]
    • invp的第p[0]个位置（第2个位置）值为0
    • invp的第p[1]个位置（第0个位置）值为1
    • invp的第p[2]个位置（第1个位置）值为2

    4 random_subseq

    1. '''
    2. 根据给定的 rate 从输入的数组 a 中随机丢弃一些元素，但始终保留首尾两个元素
    3. '''
    4. def random_subseq(a, rate):
    5.  
    6.     idx = np.random.rand(len(a)) < rate
    7.     #生成一个与 a 同长度的随机数组，其值在 [0, 1) 之间。
    8.     #然后，该随机数组中小于 rate 的值被标记为 True，其他的被标记为 False
    9.  
    10.     idx[0], idx[-1] = True, True
    11.     '''
    12.     无论先前的随机操作的结果如何，这行代码确保数组 idx 的首元素和尾元素始终为 True
    13.     这意味着在最终结果中，a 的首尾元素永远不会被丢弃
    14.     '''
    15.     return a[idx]

    5 DataLoader

    5.1 init

    1. class DataLoader():
    2.     """
    3.     srcfile: source file name
    4.     trgfile: target file name
    5.     batch: batch size
    6.     validate: if validate = True return batch orderly otherwise return
    7.         batch randomly
    8.     """
    9.     def __init__(self, srcfile, trgfile, mtafile, batch, bucketsize, validate=False):
    10.         self.srcfile = srcfile
    11.         self.trgfile = trgfile
    12.         self.mtafile = mtafile
    13.         $源、目标、元文件文件名
    14.  
    15.         self.batch = batch
    16.         self.validate = validate
    17.         #如果为True，则按照顺序返回batch，否则随即返回
    18.         #self.bucketsize = [(30, 30), (30, 50), (50, 50), (50, 70), (70, 70)]
    19.         self.bucketsize = bucketsize

    5.1.1 桶策略

    • 桶是一种策略，用于在序列长度可变的情况下处理数据。
    • 例如，假设我们有不同长度的句子，我们可能不希望为每个句子单独创建一个批处理，因为这会非常低效。
    • 桶策略是将句子放入一个接近它们长度的桶中，以尽量减少填充。 

    5.2 insert

    1. '''
    2. 将一个源序列（s）、一个目标序列（t）和一个元数据序列（m）插入到适当的桶中
    3. '''
    4. def insert(self, s, t, m):
    5.         for i in range(len(self.bucketsize)):
    6.             #遍历每个桶的大小
    7.             if len(s) <= self.bucketsize[i][0] and len(t) <= self.bucketsize[i][1]:
    8.                 #检查序列 s 和 t 的长度是否匹配当前桶的大小
    9.                 self.srcdata[i].append(np.array(s, dtype=np.int32))
    10.                 self.trgdata[i].append(np.array(t, dtype=np.int32))
    11.                 self.mtadata[i].append(np.array(m, dtype=np.float32))
    12.                 return 1
    13.         return 0
    14. #如果循环结束后，数据没有被插入到任何桶中，则返回 0，表示数据没有被插入

    5.3 load

    注：self.srcdata = list(map(np.array, self.srcdata)) 这几行需要用老版本的numpy（我是用1.23.5，有warning不报错），新版本的numpy，需要srcdata长度相同

    1. '''
    2. 从指定的源文件、目标文件和元数据文件中加载数据，并按照之前定义的桶大小进行组织
    3. '''
    4. def load(self, max_num_line=0):
    5.         self.srcdata = [[] for _ in range(len(self.bucketsize))]
    6.         self.trgdata = [[] for _ in range(len(self.bucketsize))]
    7.         self.mtadata = [[] for _ in range(len(self.bucketsize))]
    8.         #根据桶的数量为 self.srcdata, self.trgdata 和 self.mtadata 初始化空列表。
    9.  
    10.         srcstream, trgstream, mtastream = open(self.srcfile, 'r'), open(self.trgfile, 'r'), open(self.mtafile, 'r')
    11.         #从源、目标和元数据文件中读取数据
    12.  
    13.         num_line = 0
    14.         for (s, t, m) in zip(srcstream, trgstream, mtastream):
    15.             #每次读取srcstream, trgstream, mtastream的各一行
    16.  
    17.             s = [int(x) for x in s.split()]
    18.             #源数据直接转为整数列表
    19.             t = [constants.BOS] + [int(x) for x in t.split()] + [constants.EOS]
    20.             #目标数据前后添加起始（BOS）和结束（EOS）标记
    21.             m = [float(x) for x in m.split()]
    22.  
    23.             num_line += self.insert(s, t, m)
    24.             #使用 insert 方法将数据插入到适当的桶中
    25.             #如果插入成功，那么num_line加一
    26.  
    27.             if num_line >= max_num_line and max_num_line > 0: break
    28.             #如果达到最大行数 max_num_line（如果设置了的话），则停止读取
    29.  
    30.             if num_line % 500000 == 0:
    31.                 print("Read line {}".format(num_line))
    32.             #每读取 500,000 行，打印一条消息表示进度
    33.  
    34.         ## 如果处于验证模式，合并所有桶:
    35.         if self.validate == True:
    36.             self.srcdata = np.array(merge(*self.srcdata))
    37.             self.trgdata = np.array(merge(*self.trgdata))
    38.             self.mtadata = np.array(merge(*self.mtadata))
    39.  
    40.             self.start = 0
    41.             self.size = len(self.srcdata)
    42.         else:
    43.             #否则，将数据转为 NumPy 数组，并计算桶的分配比例:
    44.             self.srcdata = list(map(np.array, self.srcdata))
    45.             self.trgdata = list(map(np.array, self.trgdata))
    46.             self.mtadata = list(map(np.array, self.mtadata))
    47.  
    48.             self.allocation = list(map(len, self.srcdata))
    49.             #计算每个桶中的数据量，并存储在 self.allocation 中。
    50.             self.p = np.array(self.allocation) / sum(self.allocation)
    51.             #计算每个桶的相对大小，并存储在 self.p 中。
    52.         srcstream.close(), trgstream.close(), mtastream.close()
    53.         #关闭源、目标和元数据文件的文件流。

    5.4 getbatch_one

    1. '''
    2. 获取一个批次的数据
    3. '''
    4. def getbatch_one(self):
    5.         if self.validate == True:
    6.             #验证模式
    7.             src = self.srcdata[self.start:self.start+self.batch]
    8.             trg = self.trgdata[self.start:self.start+self.batch]
    9.             mta = self.mtadata[self.start:self.start+self.batch]
    10.             #从当前的start位置开始，连续选取batch大小的数据
    11.  
    12.             self.start += self.batch
    13.             #更新start以便下一次获取批次时从新的位置开始。
    14.             if self.start >= self.size:
    15.                 self.start = 0
    16.             #如果start加上batch的大小超过了数据的大小，那么start将被重置为0
    17.             #这意味着验证数据是循环的
    18.             return list(src), list(trg), list(mta)
    19.             #返回这个批次的源数据、目标数据和元数据
    20.         else:
    21.             ## 非验证模式
    22.             sample = np.random.multinomial(1, self.p)
    23.             #使用多项分布从self.p中抽取一个样本，表示抽取哪一个桶。
    24.             #这里的self.p是一个数组，表示每个"bucket"被选中的概率
    25.             '''
    26.             如果self.p = [0.2, 0.5, 0.3]，那么返回的数组可能是[0, 1, 0]，表示第二个"bucket"被选中
    27.             '''
    28.  
    29.             bucket = np.nonzero(sample)[0][0]
    30.             '''
    31.             np.nonzero(sample) 返回一个元组，其中包含了sample中所有非零元素的索引
    32.             因为sample中只有一个元素是1（即被选中的"bucket"），所以这个函数会返回一个只有一个元素的数组
    33.             [0][0] 提取这个数组的第一个元素，即被选中的"bucket"的索引
    34.             '''
    35.        
    36.             idx = np.random.choice(len(self.srcdata[bucket]), self.batch)
    37.             #从所选"bucket"中随机选择batch大小的数据。
    38.  
    39.             src = self.srcdata[bucket][idx]
    40.             trg = self.trgdata[bucket][idx]
    41.             mta = self.mtadata[bucket][idx]
    42.             return list(src), list(trg), list(mta)
    43.             #返回这个批次的源数据、目标数据和元数据。

    5.4.1 选择桶的两行举例

    1. import numpy as np
    2. sample=np.random.multinomial(1, np.array([0.2,0.3,0.5]))
    3. sample
    4. #array([1, 0, 0])
    5.  
    6. bucket = np.nonzero(sample)
    7. bucket
    8. #(array([0]),)
    9.  
    10. bucket[0][0]
    11. #0

    5.5 getbatch_generative

    1. '''
    2. 从数据加载器中获取一个批次的数据
    3. '''
    4. def getbatch_generative(self):
    5.         src, trg, _ = self.getbatch_one()
    6.         #调用getbatch_one函数，该函数返回一批数据。它返回三个值：源数据、目标数据和元数据
    7.         
    8.         return pad_arrays_pair(src, trg, keep_invp=False)
    9.         '''
    10.         填充src和trg，使得他们分别是相同长度
    11.         返回值一个名为TD的namedtuple，其中包含填充和排序后的src和trg，以及源数据的长度
    12.         '''

    5.6 getbatch_discriminative_cross

    1. '''
    2. 获取三个batch的数据，a、p和n[锚点（anchor）、正例（positive）和负例（negative）]
    3. '''
    4. def getbatch_discriminative_cross(self):
    5.         def distance(x, y):
    6.             return np.linalg.norm(x - y)
    7.         #定义了一个简单的欧几里得距离函数来计算两个向量之间的距离。
    8.  
    9.         a_src, a_trg, a_mta = self.getbatch_one()
    10.         p_src, p_trg, p_mta = self.getbatch_one()
    11.         n_src, n_trg, n_mta = self.getbatch_one()
    12.         '''
    13.         连续三次调用getbatch_one方法，获取三个不同的批次数据：a、p和n
    14.         代表了锚点（anchor）、正例（positive）和负例（negative）。
    15.         '''
    16.  
    17.         for i in range(len(a_src)):
    18.             if distance(a_mta[i], p_mta[i]) > distance(a_mta[i], n_mta[i]):
    19.                 p_src[i], n_src[i] = n_src[i], p_src[i]
    20.                 p_trg[i], n_trg[i] = n_trg[i], p_trg[i]
    21.                 p_mta[i], n_mta[i] = n_mta[i], p_mta[i]
    22.         '''
    23.         如果a和p之间的距离大于a和n之间的距离，我们交换p和n。
    24.         这是为了确保p是与a更接近的点，而n是更远离的点
    25.         '''
    26.  
    27.         a = pad_arrays_pair(a_src, a_trg, keep_invp=True)
    28.         p = pad_arrays_pair(p_src, p_trg, keep_invp=True)
    29.         n = pad_arrays_pair(n_src, n_trg, keep_invp=True)
    30.         '''
    31.         对于每个数据批次，调用pad_arrays_pair进行填充和排序。
    32.         这里保持了逆排列信息，因为在后续的处理中需要它
    33.         '''
    34.         return a, p, n

    5.7 getbatch_discriminative_inner

    1. '''
    2. 从给定的输入数据中生成a(锚点)、p(正例)、和n(负例)的批次，类似于上一个函数。
    4. 但在这里，所有这些批次数据都是从同一个初始数据的不同部分产生的。
    5. '''
    6. def getbatch_discriminative_inner(self):
    7.         a_src, a_trg = [], []
    8.         p_src, p_trg = [], []
    9.         n_src, n_trg = [], []
    10.  
    11.         _, trgs, _ = self.getbatch_one()
    12.         #从getbatch_one方法中仅获取一个batch的目标数据trgs
    13.         #回忆：getbatch_one返回的东西还没有pad
    14.  
    15.         for i in range(len(trgs)):
    16.             trg = trgs[i][1:-1]
    17.             #目标轨迹掐头去尾（BOS、EOS）
    18.  
    19.             if len(trg) < 10: continue
    20.             #如果目标轨迹长度小于10，那么不考虑这条轨迹
    21.  
    22.             a1, a3, a5 = 0, len(trg)//2, len(trg)
    23.             a2, a4 = (a1 + a3)//2, (a3 + a5)//2
    24.             #将trg数据(不考虑开头和结尾)分为三部分：a1到a5
    25.  
    26.             '''
    27.             根据一个随机的rate，使用random_subseq函数来从这三部分中随机选取子序列
    28.             '''
    29.             rate = np.random.choice([0.5, 0.6, 0.8])
    30.             if np.random.rand() > 0.5:
    31.                 a_src.append(random_subseq(trg[a1:a4], rate))
    32.                 a_trg.append(np.r_[constants.BOS, trg[a1:a4], constants.EOS])
    33.                 p_src.append(random_subseq(trg[a2:a5], rate))
    34.                 p_trg.append(np.r_[constants.BOS, trg[a2:a5], constants.EOS])
    35.                 n_src.append(random_subseq(trg[a3:a5], rate))
    36.                 n_trg.append(np.r_[constants.BOS, trg[a3:a5], constants.EOS])
    37.                 '''
    38.                 如果np.random.rand() > 0.5，那么：
    39.                 取a1到a4为锚点、a2到a5为正例、a3到a5为负例
    40.                 '''
    41.             else:
    42.                 a_src.append(random_subseq(trg[a2:a5], rate))
    43.                 a_trg.append(np.r_[constants.BOS, trg[a2:a5], constants.EOS])
    44.                 p_src.append(random_subseq(trg[a1:a4], rate))
    45.                 p_trg.append(np.r_[constants.BOS, trg[a1:a4], constants.EOS])
    46.                 n_src.append(random_subseq(trg[a1:a3], rate))
    47.                 n_trg.append(np.r_[constants.BOS, trg[a1:a3], constants.EOS])
    48.  
    49.         a = pad_arrays_pair(a_src, a_trg, keep_invp=True)
    50.         p = pad_arrays_pair(p_src, p_trg, keep_invp=True)
    51.         n = pad_arrays_pair(n_src, n_trg, keep_invp=True)
    52.         '''
    53.         使用pad_arrays_pair函数，我们将a_src, a_trg, p_src, p_trg, n_src, 和n_trg填充并排序
    54.         '''
    55.         return a, p, n

    6 DataOrderScaner

    1. '''
    2. 从一个源文件中扫描和加载数据，并以批量方式返回这些数据
    3. '''
    4. class DataOrderScaner():
    5.     def __init__(self, srcfile, batch):
    6.         self.srcfile = srcfile
    7.         self.batch = batch
    8.         self.srcdata = []
    9.         self.start = 0
    10.         # 初始化函数，设置源文件、批处理大小，并初始化存放数据的列表
    11.  
    12.     '''
    13.     从指定的源文件self.srcfile中加载数据，并将其存储到self.srcdata列表中
    14.     '''
    15.     def load(self, max_num_line=0):
    16.         num_line = 0
    17.         with open(self.srcfile, 'r') as srcstream:
    18.             for s in srcstream:
    19.                 s = [int(x) for x in s.split()]
    20.                 self.srcdata.append(np.array(s, dtype=np.int32))
    21.                 #从指定的源文件self.srcfile中加载数据，并将其存储到self.srcdata列表中
    22.  
    23.                 num_line += 1
    24.                 if max_num_line > 0 and num_line >= max_num_line:
    25.                     break
    26.         self.size = len(self.srcdata)
    27.         self.start = 0
    28.  
    29.     '''
    30.     从加载的数据中按批次获取数据，并进行适当的格式化处理
    31.     '''
    32.     def getbatch(self):
    33.         """
    34.         Output:
    35.         src (seq_len, batch)
    36.         lengths (1, batch)
    37.         invp (batch,): inverse permutation, src.t()[invp] gets original order
    38.         """
    39.  
    40.         if self.start >= self.size:
    41.             return None, None, None
    42.         '''
    43.         首先，检查self.start（当前批次的开始位置）是否已经超过或等于self.size（总的数据大小）。
    44.         如果是这样，表示所有数据都已经被提取过了，所以直接返回None
    45.         '''
    46.  
    47.         src = self.srcdata[self.start:self.start+self.batch]
    48.         #从self.srcdata中提取当前批次的数据
    49. =
    50.         self.start += self.batch
    51.         return pad_arrays_keep_invp(src)
    52.         #对当前批次的数据进行处理，以获得适当的格式和反向排列

  • 相关阅读:
    软件系统的测试方法
    HarmonyOS开发(四)：UIAbility组件
    html5 web 按钮跳转方法（及其相关）
    VBA技术资料MF163：获取所点击按钮的名称和时间
    测试人员Bug书写规范
    Redis 哨兵模式的原理及其搭建
    【大型电商项目开发】JWT介绍以及SpringSecurity基于JWT实现Token的处理-51
    什么是产品主导的增长以及为什么它对 API 优先的公司至关重要
    快解析：轻松实现共享上网
    【限时免费】20天拿下华为OD笔试之 【位运算】2023B-分苹果【欧弟算法】全网注释最详细分类最全的华为OD真题题解
  • 原文地址：https://blog.csdn.net/qq_40206371/article/details/134041676