Python program that deals with overlapping intervals

Question

This is related to my previous question, this is the core logic of that script that I wanted you to review.

Basically, I have two very large lists collectively containing literally over one million items, each item can be simplified to a triplet, in which the first and second element are integers, and the third element is some arbitrary data.

In each triplet the second integer is always no less than the first integer, the two integers represent integer ranges that includes both ends. Basically I have two extremely big lists containing integer ranges with some data associated with each range.

And here is the problem I wanted to address: the ranges often overlap, and sub-ranges can also have sub-ranges. So I intended to split the overlapping ranges into discrete ranges.

And there are two kinds of problematic situations:

• 1, the sub-ranges can have the same data as their parent ranges, this causes unnecessary nesting so I wanted the sub-ranges to be ignored and only the parent ranges need to be processed in this situation.

• 2, if the start of a range is the end of the previous range plus one, they sometimes have the same data, in this situation they need to be merged.

I intend to split the ranges into discrete ranges, so that any gap is filled with data from the immediate parent range, and no adjacent ranges share the same data.

My previous post didn't get reviewed, the code was too long and more importantly the solution was very inefficient.

I had spent many hours today trying to find a better solution, and I have found it, I have achieved doing everything I mentioned in one for loop, so that the code is much more efficient. But the code is ugly so I wanted it to be reviewed.

---

Code

class Merging_List:
    def __init__(self):
        self.rows = []
        self.last = [False]*3
    
    def append(self, row):
        if row[0] != self.last[1] + 1 or row[2] != self.last[2]:
            self.rows.append(row)
            self.last = row
        else:
            self.last[1] = row[1]

def process(rows):
    last = rows[0]
    pos = last[0]
    stack = [last]
    processed = Merging_List()
    for row in rows[1:]:
        if stack and row[2] == stack[-1][2]:
            continue
        if last[0] > pos and stack:
            processed.append([pos, last[0] - 1, stack[-1][2]])
        if row[0] > last[1]:
            processed.append(last)
            pos = last[1] + 1
        elif last not in stack:
            stack.append(last)
        if stack and row[0] > stack[-1][1]:
            if pos < stack[-1][1]:
                processed.append([pos, stack[-1][1], stack[-1][2]])
            pos = stack[-1][1] + 1
            stack.pop(-1)
        if row[0] >= last[1] or row[2] != last[2]:
            last = row

    if stack and last[2] != stack[-1][2]:
        if last[0] > pos:
            processed.append([pos, last[0] - 1, stack[-1][2]])
        processed.append(last)
        pos = last[1] + 1

    while stack:
        row = stack.pop(-1)
        if pos < row[1]:
            processed.append([pos, row[1], row[2]])
            pos = row[1] + 1

    return processed.rows

I haven't tested it on the full dataset, so I am not sure if it is completely working, but from what I have tested so far the code seems to be working perfectly.

The ranges are always sorted in such a way that ranges with lower indexes have smaller starts, and if two adjacent ranges share the same start, the range with larger end is ordered first. And I have verified that if two ranges overlap, then the smaller range is always completely contained within the larger range, with no exceptions, in other words if the start of range A is no greater than range B, the end of range A will never be greater than range B.

So for two adjacent ranges A and B, B comes after A, if the start of B is larger than the end of A, it is guaranteed that they don't overlap, else B must be a sub-range of A.

---

Test data

rows = [
    [16777216, 33554431, 0], [16777216, 16777471, 1], [16777472, 16777727, 2], [16777728, 16778239, 2],
    [16778240, 16779263, 3], [16778496, 16778751, 3], [16779264, 16781311, 2], [16781312, 16785407, 4],
    [16781312, 16781567, 5], [16785408, 16793599, 2], [16785408, 16785663, 6], [16793600, 16809983, 7],
    [16809984, 16842751, 8], [16809984, 16826367, 8], [16809984, 16818175, 8], [16809984, 16810239, 8],
    [16810240, 16810495, 8], [16810496, 16811007, 8], [16811008, 16811263, 8], [16811264, 16811519, 8],
    [16812032, 16812287, 8], [16812288, 16812543, 8], [16812544, 16812799, 8], [16812800, 16813055, 8],
    [16813312, 16813567, 8], [16814080, 16818175, 8], [16818176, 16826367, 8], [16818176, 16819199, 8],
    [16819200, 16819455, 8], [16819456, 16819711, 8], [16819712, 16819967, 8], [16819968, 16820223, 8],
    [16820224, 16820479, 8], [16820480, 16820735, 8], [16820736, 16820991, 8], [16820992, 16821247, 8],
    [16821248, 16822271, 8], [16822272, 16822527, 8], [16822528, 16822783, 8], [16822784, 16823039, 8],
    [16823040, 16823295, 8], [16823808, 16824063, 8], [16825600, 16825855, 8], [16825856, 16826111, 8],
    [16826112, 16826367, 8], [16826368, 16842751, 8], [16826368, 16834559, 8], [16826368, 16826623, 8],
    [16826624, 16826879, 8], [16826880, 16827135, 8], [16827136, 16827391, 8], [16827392, 16827647, 8],
    [16827648, 16827903, 8], [16827904, 16828159, 8], [16828160, 16828415, 8], [16828416, 16828671, 8],
    [16828672, 16828927, 8], [16828928, 16829439, 8], [16829440, 16829695, 8], [16829696, 16829951, 8],
    [16829952, 16830207, 8], [16830208, 16830463, 8], [16830464, 16830975, 8], [16830976, 16831487, 8]
]

---

Test result

[[16777216, 16777471, 1],
 [16777472, 16778239, 2],
 [16778240, 16779263, 3],
 [16779264, 16781311, 2],
 [16781312, 16781567, 5],
 [16781568, 16785407, 4],
 [16785408, 16785663, 6],
 [16785664, 16793599, 2],
 [16793600, 16809983, 7],
 [16809984, 16842751, 8],
 [16842752, 33554431, 0]]

I have verified the result to be indeed correct.

I know my code is ugly and this is exactly why I want it to be reviewed, it is just prototype code. The function is way too long and not elegant, but it does get the job done. I want the function to be split into smaller functions, but I don't know where to start, I need everything to be done in exactly one loop, because I literally have millions of rows to process.

So what is a better method to achieve exactly the same thing I have done, but with smaller function size, more concise and elegant code?

---

Update

I couldn't respond to the comments and update the question in time, because my ISP cut off my network connection, so it took me some time to fix it and change modem settings.

That said I spent sometime to make some really crude graphics to illustrate the point better.

The input:

The output:

Obviously the images are not to scale, because the numbers are so large it is impossible to draw them to scale. The first image captures the nesting structure of the networks. The rectangles each represent a network, the color represent the data of the network, and if a rectangle is within another, then the network it represents is the subnet of the network represented by the larger rectangle, and vice versa.

The larger the rectangle is, the more subnets the corresponding network has.

The criteria to merge ranges is really simple:

• 1, if two networks overlap, say they are network A and network B, it always will be the case that one network is fully contained in another. Say network A is larger, then the data is (start_A, end_A, attribute_A), (start_B, end_B, attribute_B), and this will always hold: start_A <= start_B <= end_B <= end_A. The networks need to be merged if and only if the share the same data, here it means if attribute_B == attribute_A. If that is the case, network B needs to be ignored.

• 2, if two networks don't overlap, but they are adjacent to each other, again, say they are networks A and B, network B comes after A, they are adjacent if and only if the start of B comes immediately after the end of A, or in code: start_B == end_A + 1, and they need to be merged if and only if they share the same attribute, the merged result would be: (start_A, end_B, attribute_A).

I hope the output graphics is self-explanatory enough, I don't need to clarify anything further.

I didn't claim my code to be completely working, but from what I have tested, in my use case, it is working. And it is prototype code. And since there is already an answer I am not allowed to edit the code.

---

Update 2

No, that is not a bug, that is exactly the intended behavior.

Like I already wrote, I want to split the overlapping ranges into discrete non-overlapping ranges. If two ranges overlap, then one range is completely inside the other.

Again, say they are range A and B, B comes after A, in all cases B is a su-brange of A. And if B and A have different data, they need to be split into discrete ranges, this means the portion of A that is B will be deleted so that the numbers inside range B will only have data from B.

I did write they are number ranges, right? The rule is very simple, each range assigns its attribute to all numbers within the corresponding range, and like Python assignment, later assignment should always overwrite everything assigned before. This means the sub-range always wins and the for each number there will only be one value to it.

Here is a simple example to show what I mean.

Data:

[(0, 10, 'A'), (0, 1, 'B'), (2, 5, 'C'), (3, 4, 'C'), (6, 7, 'C'), (8, 8, 'D')]

What the data actually means:

[
    {0: 'A', 1: 'A', 2: 'A', 3: 'A', 4: 'A', 5: 'A', 6: 'A', 7: 'A', 8: 'A', 9: 'A', 10: 'A'},
    {0: 'B', 1: 'B'},
    {2: 'C', 3: 'C', 4: 'C', 5: 'C'},
    {3: 'C', 4: 'C'},
    {6: 'C', 7: 'C'},
    {8: 'D'}
]

My intended process is equivalent to simply processing them in order and update the dictionary according to the current item one by one.

Processed data:

{0: 'B', 1: 'B', 2: 'C', 3: 'C', 4: 'C', 5: 'C', 6: 'C', 7: 'C', 8: 'D', 9: 'A', 10: 'A'}

Output:

[(0, 1, 'B'), (2, 7, 'C'), (8, 8, 'D'), (9, 10, 'A')]

And I know I can just expand all the ranges and merge them one by one, and that is fool-proof and guaranteed to work, but that is incredibly stupid and terribly inefficient, as you can see the numbers in the data are so big and there are millions of rows I really have no idea how long it will take.

I wrote the completely working first version, the first smart approach I came up with, it was really completely working and included in the question I linked.

But it was still inefficient. This is my second attempt at a smart solution. This does everything in one for loop, in one go, therefore extremely efficient. But it isn't completely working and ugly. However the output of this algorithm on any subset of the gigantic dataset I have is correct. I always started with the first item though.

---

Update 3.0

I have updated my code yet again and this time it seems to be working correctly, at least it gives correct output for all the three examples given so far: the original example included in the question, [[10, 15, 9], [16, 20, 9]] from a comment, and my second example [[0, 10, 'A'], [0, 1, 'B'], [2, 5, 'C'], [3, 4, 'C'], [6, 7, 'C'],[8, 8, 'D']].

As you can see the code is of poor quality, it needs to be split into smaller methods.

Before you ask why did I share code as a screenshot, first see if there is existing answer to the question. As there is already an answer posted, I don't know if I am allowed to edit the code, lest it be rolled back.

The changes I made were quite simple, first I duplicated the last element in the data by appending the last element to the list. This is to ensure every row will be processed, because my code works by comparing the current element to the last element, and only the last element will be appended to the result.

Second I got rid of last element assignment, or assigning the current element to last at the end of the loop to memorize what was the last element, by simply zipping the data starting at the zeroth index with the data starting at the first index, so I have two variables that keep track of what comes before the latter.

Third I eliminated the first and last if checks inside the for loop.

Finally and most importantly, this is what made it work for all inputs (that I have tested anyway, again there might still be edge cases), the trick is extremely simple, I changed the two conditions that look like this: if pos < number: to this: if pos <= number:. That is right, I merely changed less than operator to less than or equal to operator and it made a huge difference.

Answer 1

Start with a clearer understanding of the current code's drawbacks. You described your current implementation as prototype code. While criticizing it, you expressed a desire for an implementation "with smaller function size, more concise and elegant code" and conveyed the idea that "it needs to be split into smaller methods". Perhaps those things are true. But I would suggest something different. The current code is plenty concise. Its primary drawback is its algorithmic complexity and its lack of any mechanisms to help the reader understand what's going on. There are no code comments, for example, to guide the reader. The logic rests substantially on opaque list indexes, which carry no meaningful information. To illustrate the last point with one line of code, consider what something as simple as a few constants could do to enhance readability:

# Current code: making the reader work hard.
if row[0] >= last[1] or row[2] != last[2]:
    ...

# Add a few constants to give meaning to those indexes.
BEGIN, END, DATA = (0, 1, 2)
...
if row[BEGIN] >= last[END] or row[DATA] != last[DATA]:
    ...

Smarter data, simpler code. As with your password program, you are under-investing in data and over-investing in algorithm. As Fred Brooks, Rob Pike, and other computing giants have emphasized, the key to programming is data. If you select the right data structures, and create the right data objects, algorithms tend to simplify and code tends to become more readable. Your current data structures are bare bones (a list of triples) -- hence the need to rely on numeric indexes rather than meaningful and self-documenting attributes. One good move would be to define a simple data object to represent one of those intervals/ranges/triples.

from dataclasses import dataclass

@dataclass(order = True)
class Interval:
    begin: int
    end: int
    data: object

Speaking of data structures, consider using an interval tree. This problem seems well-suited for a data structure specialized to answer questions about overlapping intervals. The intervaltree library seems to be active and reasonably maintained. Here's an illustration of how you might use it. Compared to your current implementation, the code is a little bit shorter and, much more important, a lot more readable (even without the comments). But notice the role that the comments play in providing a narrative to guide the reader through the process, clarifying intent, and explaining anything that might not be directly evident from the code.

from intervaltree import IntervalTree, Interval as TreeInterval

def merge_rows(rows):
    # Convert the row data to an IntervalTree.
    t = IntervalTree()
    for begin, end, data in rows:
        # Add the row to the tree.
        iv = TreeInterval(begin, end + 1, data)
        t.add(iv)

        # Slice all intervals in the tree at the current interval's endpoints.
        t.slice(iv.begin)
        t.slice(iv.end)

        # After those slices, intervals in the tree that overlap the current
        # interval can be replaced by new intervals holding the current data.
        for ov in t.overlap(iv):
            new = TreeInterval(ov.begin, ov.end, iv.data)
            t.remove(ov)
            t.add(new)

    # Merge abutting intervals having the same data value.
    # Because TreeInterval instances are immutable, we will
    # convert them to our own Interval instances.
    ivs = []
    prev = None
    for iv in sorted(Interval(*iv) for iv in t):
        if prev and iv.data == prev.data and prev.end == iv.begin:
            prev.end = iv.end
        else:
            ivs.append(iv)
            prev = iv

    # Convert back to the "row" data format that we started with.
    # Or not: perhaps the rest of your program would benefit from
    # using Interval instances rather than raw triples.
    return [
        [iv.begin, iv.end - 1, iv.data]
        for iv in ivs
    ]

Answer 2

Merging_List should be MergingList by PEP8.

If you're going object-oriented, fine; but you should go all the way: MergingList should hold your output state, and process should be a method on it that consumes input and adds to that state.

Do not initialise last to False. You're adding to those elements, and it doesn't make sense to add to a boolean. Initialise it to all 0 and type-hint it.

You're mutating your input, and that's very problematic. A naive caller would be surprised when their input doesn't look the same after it's been processed. Your method should also be able to accept immutable sequences (tuples), but cannot due to the way that it's designed.

It's good that you've provided test input and output; you should convert that to an actual unit test.

Your data are better-represented as hex than as decimal. You say that these are IP addresses, and showing them in hex makes it more obvious which octets are 0.

A first pass covering these items can look like

from typing import Sequence


class MergingList:
    def __init__(self) -> None:
        self.rows: list[Sequence[int]] = []
        self.last: list[int] = [0] * 3

    def append(self, row: list[int]) -> None:
        if row[0] != self.last[1] + 1 or row[2] != self.last[2]:
            self.rows.append(row)
            self.last = row  # list(row)
        else:
            self.last[1] = row[1]

    def process(self, rows: Sequence[list[int]]) -> None:
        last = rows[0]
        pos = last[0]
        stack = [last]
        for row in rows[1:]:
            if stack and row[2] == stack[-1][2]:
                continue
            if last[0] > pos and stack:
                self.append([pos, last[0] - 1, stack[-1][2]])
            if row[0] > last[1]:
                self.append(last)
                pos = last[1] + 1
            elif last not in stack:
                stack.append(last)
            if stack and row[0] > stack[-1][1]:
                if pos < stack[-1][1]:
                    self.append([pos, stack[-1][1], stack[-1][2]])
                pos = stack[-1][1] + 1
                stack.pop(-1)
            if row[0] >= last[1] or row[2] != last[2]:
                last = row

        if stack and last[2] != stack[-1][2]:
            if last[0] > pos:
                self.append([pos, last[0] - 1, stack[-1][2]])
            self.append(last)
            pos = last[1] + 1

        while stack:
            row = stack.pop(-1)
            if pos < row[1]:
                self.append([pos, row[1], row[2]])
                pos = row[1] + 1


def repr_as_hex(rows: Sequence[Sequence[int]], group_size: int = 4) -> None:
    for i_group in range(0, len(rows), group_size):
        for a, b, c in rows[i_group: i_group+group_size]:
            print(f'(0x{a:08X}, 0x{b:08X}, {c}),', end=' ')
        print()


def test() -> None:
    rows = (
        (0x01000000, 0x01FFFFFF, 0), (0x01000000, 0x010000FF, 1), (0x01000100, 0x010001FF, 2), (0x01000200, 0x010003FF, 2),
        (0x01000400, 0x010007FF, 3), (0x01000500, 0x010005FF, 3), (0x01000800, 0x01000FFF, 2), (0x01001000, 0x01001FFF, 4),
        (0x01001000, 0x010010FF, 5), (0x01002000, 0x01003FFF, 2), (0x01002000, 0x010020FF, 6), (0x01004000, 0x01007FFF, 7),
        (0x01008000, 0x0100FFFF, 8), (0x01008000, 0x0100BFFF, 8), (0x01008000, 0x01009FFF, 8), (0x01008000, 0x010080FF, 8),
        (0x01008100, 0x010081FF, 8), (0x01008200, 0x010083FF, 8), (0x01008400, 0x010084FF, 8), (0x01008500, 0x010085FF, 8),
        (0x01008800, 0x010088FF, 8), (0x01008900, 0x010089FF, 8), (0x01008A00, 0x01008AFF, 8), (0x01008B00, 0x01008BFF, 8),
        (0x01008D00, 0x01008DFF, 8), (0x01009000, 0x01009FFF, 8), (0x0100A000, 0x0100BFFF, 8), (0x0100A000, 0x0100A3FF, 8),
        (0x0100A400, 0x0100A4FF, 8), (0x0100A500, 0x0100A5FF, 8), (0x0100A600, 0x0100A6FF, 8), (0x0100A700, 0x0100A7FF, 8),
        (0x0100A800, 0x0100A8FF, 8), (0x0100A900, 0x0100A9FF, 8), (0x0100AA00, 0x0100AAFF, 8), (0x0100AB00, 0x0100ABFF, 8),
        (0x0100AC00, 0x0100AFFF, 8), (0x0100B000, 0x0100B0FF, 8), (0x0100B100, 0x0100B1FF, 8), (0x0100B200, 0x0100B2FF, 8),
        (0x0100B300, 0x0100B3FF, 8), (0x0100B600, 0x0100B6FF, 8), (0x0100BD00, 0x0100BDFF, 8), (0x0100BE00, 0x0100BEFF, 8),
        (0x0100BF00, 0x0100BFFF, 8), (0x0100C000, 0x0100FFFF, 8), (0x0100C000, 0x0100DFFF, 8), (0x0100C000, 0x0100C0FF, 8),
        (0x0100C100, 0x0100C1FF, 8), (0x0100C200, 0x0100C2FF, 8), (0x0100C300, 0x0100C3FF, 8), (0x0100C400, 0x0100C4FF, 8),
        (0x0100C500, 0x0100C5FF, 8), (0x0100C600, 0x0100C6FF, 8), (0x0100C700, 0x0100C7FF, 8), (0x0100C800, 0x0100C8FF, 8),
        (0x0100C900, 0x0100C9FF, 8), (0x0100CA00, 0x0100CBFF, 8), (0x0100CC00, 0x0100CCFF, 8), (0x0100CD00, 0x0100CDFF, 8),
        (0x0100CE00, 0x0100CEFF, 8), (0x0100CF00, 0x0100CFFF, 8), (0x0100D000, 0x0100D1FF, 8), (0x0100D200, 0x0100D3FF, 8),
    )
    # process() mutates input; yikes
    rows = tuple(list(row) for row in rows)

    expected = (
        (0x01000000, 0x010000FF, 1), (0x01000100, 0x010003FF, 2),
        (0x01000400, 0x010007FF, 3), (0x01000800, 0x01000FFF, 2),
        (0x01001000, 0x010010FF, 5), (0x01001100, 0x01001FFF, 4),
        (0x01002000, 0x010020FF, 6), (0x01002100, 0x01003FFF, 2),
        (0x01004000, 0x01007FFF, 7), (0x01008000, 0x0100FFFF, 8),
        (0x01010000, 0x01FFFFFF, 0),
    )
    merger = MergingList()
    merger.process(rows)

    assert len(merger.rows) == len(expected)
    for row, expected_row in zip(merger.rows, expected):
        assert tuple(row) == expected_row


if __name__ == '__main__':
    test()

More to follow.