Select sublists from python list, beginning and ending on the same element

Viewed 3887

I have a (very large) list similar to:

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

and I want to extract from it a list of lists like:

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

The repeating patterns can be different, for example there can also be intervals such as:

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

I have written a very simple code that seems to work:

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]]

But 1) it is very cumbersome, 2) I need to apply this to very large lists, so is there a more concise and faster way of implementing it?

EDIT: As @Kasramvd correctly pointed out, I need the largest set that satisfies the requirement of (at most a tolerance number of jumps between the start and end elements), so I take:

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

because the former includes the latter.

Also it would be good if the code can select elements UP TO the certain tolerance, for example if the tolerance (maximum number of elements not equal to the start or end element) is 2, it should also return sets as:

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

with tolerances 0, 1 and 2.

8 Answers

Solution without any extra copying of lists other than the sublist results:

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

Usage is simply as follows:

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']]

Possible solution to extra requirement as specified in the comments:

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

Note: I've not tested this thoroughly.

Probably easier to write recursively.

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']]

This uses global variables, it's easy wrap it into a function

list.index() actually accepts up to 3 arguments, which can make a lot of use here. You just use l.index(item, start + 1, start + tolerance + 2) to find the next item, and catch the ValueError that it raises.

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) ])

Output (the join at the end is intended for illustration purposes - strings are easier to read than lists of single chars):

['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']

For your sample input (first block) with tol = 2, it gives the following result:

['ABABA', 'DEDED']

10 lines (non-blank) of main function find_sublist and 4 lines of usage my_split. I dislike recursion when plain looping does the job.

You can define a custom iterator for this. There is no need for extensive sublist creation.

The idea is simple:

  1. Slice the list by the step size (you referred to this as 'jump').
  2. Run through the sliced list and check if the previous element is equal to the current one:
    • yes: remember that you currently are in a sublist and continue
    • no: check if you were in a sublist:
      • yes: you are at the end of a sublist, so yield the slice of the list corresponding to this sublist and continue.
      • no: just move on and keep looking

Some minor complication: You need to do this procedure for any starting index between 0 and step, as otherwise we miss repetition patterns of the form l[x+i]==l[x+step+i] for any 0 < i <step.

So here is how that iterator looks like:

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

And here is how to user it:

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)

Fast. Memory efficient. Easy to use.

Le voilĂ , you are welcome! :)

I think this implements the sequence finding logic you want. I'm fairly sure it could be improved upon, but hopefully it's useful anyway.

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)

A bit similar to @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']

you need to set the length that you want here: if len(tmp) > 2:

if you want to lenght is 5:

len(tmp) == 5 or etc...

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']]

If your goal is speed and your data is easily categorizable, I would recommend a numpy solution.

Let's say you have

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

To check if any elements are equal at exactly tolerance, you can do a diff-like operation, but with equality:

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

If you smear this boolean mask tolerance elements to the right, each contiguous run will be the elements you are interested in. One short way to do this is using 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

You could even make this into a one-liner, since it operates in-place:

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

Then you apply it:

smear(mask, tolerance)

Contiguous runs are easy to find and extract using a combination of np.diff, np.flatnonzero and np.split (ref):

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

The only thing that is lacking in this solution is that it will not pick up matching elements that occur less than tolerance away of each-other. To do this, we can use np.lib.stride_tricks.as_strided to make our mask in a way that takes the tolerance into account (using 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 is now a 3x15 array (where a has length 15), with the second dimension just being the characters following the start. Remember that this is just a view into the original data. For a large array, this operation is basically free.

Now you can apply np.any to the first dimension to figure out which characters are repeated within tolerance of each other:

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

From here, we continue as before. This makes for a fairly small function:

TL;DR

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')]

Addendum

If you look through the references below, you will find that the methods for smearing a mask and finding the masked portions of the array shown above here are were chosen for conciseness, not speed. A faster way to smear a mask, taken from here, is:

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

Similarly, a faster way to extract the contiguous masked regions from the array (taken from here) is:

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)]

References

Related