从Python列表中选择子列表，以相同元素开始和结束

Question

我有一个（非常大的）类似于以下列表：

a = ['A', 'B', 'A', 'B', 'A', 'C', 'D', 'E', 'D', 'E', 'D', 'F', 'G', 'A', 'B']

我希望从中提取出一个列表，其格式如下：

result = [['A', 'B', 'A', 'B', 'A'], ['D', 'E', 'D', 'E', 'D']]

重复的模式可以不同，例如还可以有间隔，比如：

['A', 'B', 'C', 'A', 'D', 'E', 'A'] (with a 'jump' over two elements)

我已经写了一段非常简单的代码，似乎可以工作:

tolerance = 2
counter = 0
start, stop = 0, 0
for idx in range(len(a) - 1):
    if a[idx] == a[idx+1] and counter == 0:
        start = idx
        counter += 1
    elif a[idx] == a[idx+1] and counter != 0:
        if tolerance <= 0: 
            stop = idx
        tolerance = 2
    elif a[idx] != a[idx+1]:
        tolerance -= 1
    if start != 0 and stop != 0:
        result = [a[start::stop]]

但是，1）这个方法非常繁琐，2）我需要将其应用于非常大的列表，因此是否有一种更简洁、更快速的实现方式?

编辑：正如@Kasramvd正确指出的那样，我需要满足（起始和结束元素之间最多容忍跳数的）要求的最大集合，因此我采取：

['A', 'B', 'A', 'B', 'A'] instead of [ 'B', 'A', 'B' ]

因为前者包括后者。

如果代码可以选择一定容差范围内的元素，那就更好了。例如，如果容差（最大不等于起始或结束元素的元素数量）为2，则应返回以下集合：

['A', 'A', 'A', 'B', 'A', 'B', 'A', 'C', 'D', 'A']

使用公差为0、1和2。

- Qubix

---

您希望我解释一下预期输出背后的逻辑，以及为什么不计算诸如 ['B', 'A', 'B'] 这样的集合吗？ - Mazdak

为什么你的第一个例子没有返回 ABABACDEDEDFGA？另外，在你的最后一个例子中，当字符串有4个不同于A的字母（B、B、C、D）时，容差如何为2？ - Tamas Hegedus

ABABABABBA是否匹配容差为2，尽管如果您从第一个A跳过两个，则会落在B上？ - Davis Herring

@MatrixTai，我不明白你的问题。如果容差=0，则只能有AAAA，BBBBB等。容差为0时，您甚至无法检测到ABA，因为您跳过了B。 - Qubix

你的问题非常令人困惑。我会尝试总结一下，所以请告诉我我的理解是否正确：您从左到右迭代列表。您取出当前正在处理的元素（例如“A”），并检查是否在“容差”步骤内有另一个A。如果没有，则移动到列表中的下一个元素。如果是，则重复“是否有另一个A”的检查，直到找不到A为止。然后，您将此A->随机字母-> A的运行附加到结果中。听起来没错吧？ - Aran-Fey

Answer 1

不需要复制除子列表结果之外的任何列表，即可解决问题：

def sublists(a, tolerance):
    result = []
    index = 0

    while index < len(a):
        curr = a[index]

        for i in range(index, len(a)):
            if a[i] == curr:
                end = i
            elif i - end > tolerance:
                break

        if index != end:
            result.append(a[index:end+1])
        index += end - index + 1

    return result

使用方法如下：

a = ['A', 'B', 'A', 'B', 'A', 'C', 'D', 'E', 'D', 'E', 'D', 'F', 'G', 'A', 'B']

sublists(a, 0)  # []
sublists(a, 1)  # [['A', 'B', 'A', 'B', 'A'], ['D', 'E', 'D', 'E', 'D']]
sublists(a, 2)  # [['A', 'B', 'A', 'B', 'A'], ['D', 'E', 'D', 'E', 'D']]

根据评论中提出的额外要求，可能的解决方案如下：

if i > index and a[i] == a[i-1] == curr:
    end = i - 1
    break
elif a[i] == curr:
    end = i
elif i - end > tolerance:
    break

注意：我没有彻底测试过这个。

- ikkuh

---

似乎工作，但如果我输入真正的列表，它就会崩溃。这行代码:result.append(a[index:end+1])会引发一个列表索引超出范围的错误。既然您超越了len(a)，为什么这里不会发生这种情况呢？ - Qubix

@Qubix。你能在你的问题中提供一个触发错误的示例列表吗？ - Mad Physicist

@Qubix。在Python中，您可以相对自由地索引列表的末尾（某种程度上）。您肯定可以想办法重现错误条件吧？ - Mad Physicist

实际上，如果我用end替换end+1，它似乎可以工作，但我需要进行更多测试，以确保我没有搞砸什么。 - Qubix

我更新了答案，包括更新的if/elif语句以匹配新需求。不确定是否适用于所有情况。 - ikkuh

Answer 2

可能更容易递归地编写。

def rep_sublist(x):
    global thisrun, collection
    if len(x) == 0:
        return None
    try: # find the next value in x that is same as x[0]
        nextidx = x[1:].index(x[0])
    except ValueError: # not found, set nextidx to something larger than tol
        nextidx = tol + 1

    if nextidx <= tol: # there is repetition within tol, add to thisrun, restart at the next repetition
        thisrun += x[:nextidx+1]
        rep_sublist(x[nextidx+1:])
    else: # no rep within tol, add in the last element, restart afresh from the next element
        thisrun += x[0]
        if len(thisrun)>1:
            collection.append(thisrun)
        thisrun = []
        rep_sublist(x[1:])


tol = 2
collection = []
thisrun = []
x = ['A', 'B', 'A', 'B', 'A', 'C', 'D', 'E', 'D', 'E', 'D', 'F', 'G', 'A', 'B', 'A', 'A', 'A', 'B', 'A', 'B', 'A', 'C', 'D', 'A']
rep_sublist(x)
print(collection)

#[['A', 'B', 'A', 'B', 'A'], ['D', 'E', 'D', 'E', 'D'], ['A', 'B', 'A', 'A', 'A', 'B', 'A', 'B', 'A', 'C', 'D', 'A']]


tol = 1 # now change tolerance to 1
collection = []
thisrun = []
rep_sublist(x)
print(collection) # last sublist is shorter

#[['A', 'B', 'A', 'B', 'A'], ['D', 'E', 'D', 'E', 'D'], ['A', 'B', 'A', 'A', 'A', 'B', 'A', 'B', 'A']]

这里使用了全局变量，可以将其封装成一个函数

- jf328

---

使用thisrun.extend(...)代替thisrun += ...，并使用del thisrun[:]代替thisrun = []以避免需要全局声明。collection似乎可以保持不变（您只调用方法，但不重新分配）。 - Mad Physicist

我应该如何将上述内容转换为函数？我尝试过，但结果是一个空列表。 - Qubix

这里有很多子列表副本被抛出来 :) - Mad Physicist

Answer 3

list.index() 实际上可以接受最多 3 个参数，在这里可以大有用处。您只需使用 l.index(item, start + 1, start + tolerance + 2) 来查找下一个项，并捕获它引发的 ValueError。

l = list("aaa,..a/,a../a,.aaa.a,..a/,.aaa.,..aaa.,..a/.,a..a,./a.aaa.,a.a..a/.aa..a,.a/a.,a../.,a/..a..a/.a..,a/.,.a/a.")

def find_sublist(l, start, tol, found):
    # a is the value to check, i_l and i_r stand for "index_left" and "index_right", respectively
    a = l[start]
    i_l, i_r = start, start
    try:
        while True:
            i_r = l.index(a, i_r + 1, i_r + tol + 2)
    except ValueError:
        pass

    if i_l < i_r:
        found.append(l[i_l:i_r + 1])
    return i_r + 1

def my_split(l)
    found = []
    i = 0
    while i < len(l):
        i = find_sublist(l, i, 2, found)

print([ "".join(s) for s in my_split(l) ])

输出结果（结尾处的连接符仅用于说明目的 - 字符串比单个字符的列表更易读）：

['aaa', '..', 'a/,a', '..', 'a,.aaa.a', '..', 'aaa', '.,..', 'aaa', '.,..a/.,a..a,./a.', 'aaa.,a.a..a/.aa..a,.a/a.,a', '../.', '..a..a/.a..', '.,.', 'a/a']

对于您的样本输入（第一个块）和 tol = 2，它给出了以下结果：

['ABABA', 'DEDED']

主函数find_sublist需要10行（非空），使用my_split的部分需要4行。我不喜欢使用递归，因为普通循环可以完成任务。

- iBug

---

请删除行号，以便将您的答案粘贴到编辑器中。 - Mad Physicist

为什么你要在结尾处连接字符串？答案应该是由1个字符字符串组成的列表。 - Qubix

@Qubix 仅供说明目的。您可以轻松删除该连接。 - iBug

@MadPhysicist 完成了。下次在复制之前我会执行 :set nonumber 命令的 :)（是的，我暴露了我的编辑器）。 - iBug

Answer 4

你可以为此定义一个自定义迭代器。无需创建广泛的子列表。

这个想法很简单：

1.按步长（你称之为'jump'）切割列表。 2.遍历被切割的列表，并检查前一个元素是否等于当前元素： - 是：记住你目前在一个子列表中并继续。 - 否：检查你是否在一个子列表中： - 是：你在一个子列表的末尾，所以yield相应于此子列表的列表切片并继续。 - 否：继续寻找。

一些小复杂性：您需要对0到step之间的任何起始索引执行此过程，否则我们会错过重复模式形式l[x+i]==l[x+step+i]，其中0

因此，以下是该迭代器的外观：

def get_sec_it(a_list, step=1):                                                 
   for _start in range(step):  # this is the minor complication                                              
       prev_el = a_list[_start]  # as we compare previous and current element
       prev_idx = _start         # we store the first element here and iterate from the second on
       insec = False                                                     
       for idx in range(_start + step, len(a_list), step):  # iteration from the second element of the sliced list                   
           el = a_list[idx]  # get the element                                                  
           if el==prev_el:  # compare it with previous (step 2 first check)                                                     
               insec=True                                                      
               continue   
           # now we are in the first no of the 2. step, so 2. step - no                                                                       
           if insec:  # 2. step - no - yes:                                                         
               insec = False                                                   
               yield a_list[prev_idx: idx - step + 1]                          
           prev_el = el    # continue the iteration by                                                        
           prev_idx = idx  # updating the previous element                                                         
       if insec:  # at the very end of a slice we wont necessarily encounter an element different from the previous one                                                  
           yield a_list[prev_idx:idx+1]  # so in this case yield the sequence if we were in one.l

这是如何使用它的：

l =['A', 'B', 'A', 'B', 'A', 'C', 'D', 'E', 'D', 'E', 'D', 'F', 'G', 'A', 'B', 'G']
for sec in get_sec_it(l, 2):
    print(sec)

快速、内存高效、易于使用。 < p > < em > Le voilà，欢迎您！:)

- j-i-l

---

如果您使用的是Python 2.x，您可能需要将range替换为xrange。 - j-i-l

Answer 5

我认为这个实现了你想要的序列查找逻辑。我相当确定它可以改进，但希望它仍然有用。

a = ['A', 'B', 'A', 'B', 'A', 'C', 'D', 'E', 'D', 'E', 'D', 'F', 'G', 'A', 'B', 'A', 'A', 'A', 'B', 'A', 'B', 'A', 'C', 'D', 'A']
tol = 2
min_str_length = 2


a_str = ''.join(a)
split_char = a[0]
all_substrs = a_str.split(split_char)[1:] #First bit will be an empty string



strs_to_return = []
current_str = split_char

while len(all_substrs) != 0:
    substr = all_substrs.pop(0)
    if len(substr) <= tol and all_substrs != []:
        current_str =  current_str + substr + split_char
    elif len(substr) > tol:
        if len(current_str) > min_str_length:
            strs_to_return.append(current_str)
        #Setup the next round
        a_str = a_str[len(current_str):]
        split_char = a_str[0]
        all_substrs = a_str.split(split_char)[1:]
        current_str = split_char

if len(current_str) > min_str_length:
    strs_to_return.append(current_str)        

print(strs_to_return)

- RadtkeJCJ

Answer 6

有点类似于 @RadhikeJCJ-

a = ['A', 'B', 'A', 'B', 'A', 'C', 'D', 'E', 'D', 'E', 'D', 'F', 'G', 'A', 'B', 'A', 'A', 'A', 'B', 'A', 'B', 'A', 'C', 'D', 'A']
tol = 1
a_str = ''.join(a)

idx_to_split = 0
output = []
while idx_to_split < len(a_str):
    a_str = a_str[idx_to_split:]
    split_char = a_str[0]
    all_substrs = a_str.split(split_char)[1:]
    if len(all_substrs) == 1:
        idx_to_split = 1
        continue
    out = []
    for i in all_substrs:
        if i == '':
            out.append("")
        elif len(i) <= tol:
            out.append(i)
        else:
            break

    if out:
        final = split_char + '{0}'.format(split_char).join(out)
        if out[-1] != '':
            final = final + split_char
        idx_to_split = len(final)
        output.append(final)
    else:
        idx_to_split = 1

#For tolerance 2,
#output = ['ABABA', 'DEDED', 'ABAAABABACDA']

#For tolerance 1,
#output = ['ABABA', 'DEDED', 'ABAAABABA']

- Raunaq Jain

Answer 7

你需要在这里设置你想要的长度：如果 len(tmp) > 2：

如果你想要长度为5：

len(tmp) == 5 或者等等...

a = ['A', 'B', 'A', 'B', 'A', 'C', 'D', 'E', 'D', 'E', 'D', 'F', 'G', 'A', 'B']
start = -1
stop = -1
result = []
for i,c in enumerate(a):
    start = i
    for idx in range(i,len(a)-1,2):
        if c == a[idx]:
            stop = idx+1
        else:
            break
    tmp = a[start:stop]
    if len(tmp) == 5:
        result.append(tmp)
        print(tmp)
    start = -1
    stop = -1
print(result)
#[['A', 'B', 'A', 'B', 'A'], ['D', 'E', 'D', 'E', 'D']]

- wailinux

Answer 8

如果您的目标是速度，且您的数据容易分类，我建议使用numpy解决方案。

假设您有：

a = np.array(['A', 'B', 'A', 'B', 'A', 'C', 'D', 'E', 'D', 'E', 'D', 'F', 'G', 'A', 'B'])
tolerance = 1

为了检查任何元素是否完全相等，你可以进行类似于 diff 的操作，但是通过判断相等性来实现：

tolerance += 1
mask = a[:-tolerance] == a[tolerance:]

如果你将这个布尔掩码 tolerance 向右移动，每个连续的运行将是你感兴趣的元素。一种简短的方法是使用 np.lib.stride_tricks.as_strided：

def smear(mask, n):
    view = np.lib.stride_tricks.as_strided(mask, shape=(n + 1, mask.size - n),
                                           strides=mask.strides * 2)
    view[1:, view[0]] = True

你甚至可以将其转换为一行代码，因为它是原地操作的：

np.lib.stride_tricks.as_strided(mask, shape=(n + 1, mask.size - n),
                                strides=mask.strides * 2)[1:, mask[:-n]] = True

然后你应用它：

smear(mask, tolerance)

使用np.diff、np.flatnonzero和np.split（参考）的组合可以轻松找到并提取连续的运行：

result = np.split(a, np.flatnonzero(np.diff(m)) + 1)[1 - m[0]::2]

该解决方案唯一的不足之处在于它无法捕获彼此间距小于tolerance的匹配元素。为了解决这个问题，我们可以使用np.lib.stride_tricks.as_strided来创建一个考虑到公差的遮罩（使用np.any）：

b = np.lib.stride_tricks.as_strided(np.r_[a, np.zeros(tolerance, dtype=a.dtype)],
                                    shape=(tolerance + 1, a.size),
                                    strides=a.strides * 2)

b 现在是一个 3x15 的数组（其中 a 长度为 15），第二个维度只是跟在开头后面的字符。请记住，这只是对原始数据的一种视图。对于大型数组，该操作基本上是免费的。

现在您可以对第一个维度应用 np.any，以确定哪些字符在容差范围内重复：

mask = np.any(b[0] == b[1:], axis=0)

从这里开始，我们像以前一样继续。这使得函数相当小：

总结：

def find_patterns(a, tol):
    a = np.asanyarray(a)
    tol += 1
    b = np.lib.stride_tricks.as_strided(np.r_[a, np.zeros(tol, dtype=a.dtype)],
                                        shape=(tol + 1, a.size),
                                        strides=a.strides * 2)
    mask = np.any(b[0] == b[1:], axis=0)
    np.lib.stride_tricks.as_strided(mask, shape=(tol + 1, mask.size - tol),
                                    strides=mask.strides * 2)[1:, mask[:-tol]] = True
    return np.split(a, np.flatnonzero(np.diff(mask)) + 1)[1 - mask[0]::2]

>>> find_patterns(['A', 'B', 'A', 'B', 'A', 'C', 'D', 'E', 'D', 'E', 'D', 'F', 'G', 'A', 'B'], 1)
[array(['A', 'B', 'A', 'B', 'A'], dtype='<U1'),
 array(['D', 'E', 'D', 'E', 'D'], dtype='<U1')]
>>> find_patterns(['A', 'B', 'C', 'A', 'D', 'E', 'A'], 1)
[]
>>> find_patterns(['A', 'B', 'C', 'A', 'D', 'E', 'A'], 2)
[array(['A', 'B', 'C', 'A', 'D', 'E', 'A'], dtype='<U1')]

附录

如果您查看下面的参考资料，您会发现上面显示的涂抹掩码和查找掩码部分的方法是为简洁而选择的，而不是为了速度。从这里获取的更快的涂抹掩码的方法是：

def smear(mask, n):
    n += 1
    mask1 = mask.copy()
    len0, len1 = 1, 1
    while len0 + len1 < n:
        mask[len0:] |= mask1[:-len0]
        mask, mask1 = mask1, mask
        len0, len1 = len1, len0 + len1
    mask1[n - len0:] |= mask[:-n + len0]
    return mask1

同样，从数组中提取连续掩码区域的更快方法（取自此处）是：

def extract_masked(a, mask):
    mask = np.concatenate(([False], mask, [False]))
    idx = np.flatnonzero(mask[1:] != mask[:-1])
    return [a[idx[i]:idx[i + 1]] for i in range(0, len(idx), 2)]

---

参考资料

• 高效地将numpy掩码向右扩展n个单元以处理每个错误值
• 从掩码中高效地提取numpy子数组

- Mad Physicist