Produits Entr'ouvert » Chrono

# chrono - agendas system

# Copyright (C) 2016  Entr'ouvert

#

# This program is free software: you can redistribute it and/or modify it

# under the terms of the GNU Affero General Public License as published

# by the Free Software Foundation, either version 3 of the License, or

# (at your option) any later version.

#

# This program is distributed in the hope that it will be useful,

# but WITHOUT ANY WARRANTY; without even the implied warranty of

# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the

# GNU Affero General Public License for more details.

#

# You should have received a copy of the GNU Affero General Public License

# along with this program.  If not, see <http://www.gnu.org/licenses/>.

import bisect

class Interval(object):

    __slots__ = ['begin', 'end', 'data']

    def __init__(self, begin, end, data=None):

        assert begin < end

        self.begin = begin

        self.end = end

        self.data = data

    def overlap(self, begin, end):

        if end <= self.begin:

            return False

        if begin >= self.end:

            return False

        return True

    def intersect(self, interval):

        return self.overlap(interval.begin, interval.end)

    def __lt__(self, interval):

        return self.begin < interval.begin or (self.begin == interval.begin and

            self.end < interval.end)

    def __repr__(self):

        return '<Interval [%s, %s] %s>' % (self.begin, self.end, self.data or '')

class Intervals(object):

    "Maintain a list of mostly non overlapping intervals, allow removing overlap"

    def __init__(self, sequence=None):

        self.points = []

        self.container = []

        if sequence:

            d = {}

            for interval in sequence:

                l = d.setdefault(interval.begin, [])

                l.append(interval)

                l = d.setdefault(interval.end, [])

                l.append(interval)

            for point, l in sorted(d.iteritems()):

                self.points.append(point)

                self.container.append(l)

    def __insert_point(self, point, interval):

        i = bisect.bisect_left(self.points, point)

        if i >= len(self.container) or self.points[i] != point:

            self.points.insert(i, point)

            self.container.insert(i, [])

        self.container[i].append(interval)

    def add(self, begin, end, data=None):

        'Add an interval'

        self.add_interval(Interval(begin, end, data))

    def add_interval(self, interval):

        'Add an interval object'

        self.__insert_point(interval.begin, interval)

        self.__insert_point(interval.end, interval)

    def __iter_interval(self, begin, end, modify=False):

         i = bisect.bisect_left(self.points, begin)

         while i < len(self.points) and self.points[i] <= end:

             container = self.container[i]

             if modify:

                 container = list(container)

             for interval in container:

                 yield self.points[i], interval

             i += 1

    def remove_overlap(self, begin, end):

        'Remove all overlapping intervals'

        for point, interval in self.__iter_interval(begin, end, modify=True):

            if interval.overlap(begin, end):

                self.__remove_interval(interval)

    def overlap(self, begin, end):

        'Test if some intervals overlap'

        for point, interval in self.__iter_interval(begin, end):

            if interval.overlap(begin, end):

                return True

        return False

    def search(self, begin, end):

        'Search overlapping intervals'

        for point, interval in self.__iter_interval(begin, end):

            if interval.overlap(begin, end):

                 # prevent returning the same interval two times

                 if interval.begin < begin or interval.begin == point:

                     yield interval

    def search_data(self, begin, end):

        'Search data elements of overlapping intervals'

        for interval in self.search(begin, end):

            yield interval.data

    def iter(self):

        'Iterate intervals'

        if not self.points:

            return []

        return self.search(self.points[0], self.points[-1])

    def iter_data(self):

        'Iterate data element attached to intervals'

        for interval in self.iter():

            yield interval.data

    def __remove_interval(self, interval):

        self.__remove_point_interval(interval.begin, interval)

        self.__remove_point_interval(interval.end, interval)

    def __remove_point_interval(self, point, interval):

        i = bisect.bisect_left(self.points, point)

        assert self.points[i] == point

        self.container[i].remove(interval)

    def iter_merge(self):

        'Returns the union of all contained intervals'

        if not self.points:

            return

        begin = None

        end = None

        for i in range(len(self.points)):

            if begin is None and self.container[i]:

                begin = self.points[i]

            for interval in self.container[i]:

                if end is None:

                    end = interval.end

                else:

                    end = max(end, interval.end)

            assert end is None or (not end < self.points[i])

            if end == self.points[i]:

                yield Interval(begin, end)

                begin = None

                end = None

    def remove_overlap_non_overlapping_ordered_intervals(self, intervals):

        '''Remove overlapping intervals with a strictly

           ordered sequence of intervals such as returned

           by iter_merge().

        '''

        i = 0

        l = len(self.points)

        interval = None

        points = []

        containers = []

        for interval in intervals:

            while i < l and self.points[i] <= interval.end:

                container = self.container[i]

                for b in list(container):

                    if b.intersect(interval):

                        container.remove(b)

                i += 1

            if i >= l:

                break

        while i < l and interval is not None:

            container = self.container[i]

            for b in list(container):

                if b.intersect(interval):

                    container.remove(b)

            i += 1

    @classmethod

    def from_ordered_intervals(self, sequence):

        'Build a new Intervals from a strictly ordered sequence of intervals'

        d = {}

        points = []

        container = []

        def add_point(point, interval):

            if point in d:

                i = d[point]

            else:

                i = d[point] = len(points)

                points.append(point)

                container.append([])

            container[i].append(interval)

        last = None

        for interval in sequence:

            assert last is None or last < interval, (last, interval)

            add_point(interval.begin, interval)

            add_point(interval.end, interval)

            last = interval

        intervals = Intervals()

        intervals.points = points

        intervals.container = container

        return intervals

from chrono.interval import Interval, Intervals

def test_interval_overlap():

    a = Interval(1, 4)

    assert not a.overlap(0, 1)

    assert a.overlap(0, 2)

    assert a.overlap(1, 4)

    assert a.overlap(2, 3)

    assert a.overlap(3, 5)

    assert not a.overlap(5, 6)

def test_intervals():

    intervals = Intervals()

    for i in range(10):

        intervals.add(i, i + 1, 1)

    for i in range(10, 20):

        intervals.add(i, i + 1, 2)

    for i in range(5, 15):

        intervals.add(i, i + 1, 3)

    assert len(list(intervals.search(0, 5))) == 5

    assert len(list(intervals.search(0, 10))) == 15

    assert len(list(intervals.search(5, 15))) == 20

    assert len(list(intervals.search(10, 20))) == 15

    assert len(list(intervals.search(15, 20))) == 5

    assert set(intervals.search_data(0, 5)) == {1}

    assert set(intervals.search_data(0, 10)) == {1, 3}

    assert set(intervals.search_data(5, 15)) == {1, 2, 3}

    assert set(intervals.search_data(10, 20)) == {2, 3}

    assert set(intervals.search_data(15, 20)) == {2}

    for i in range(20):

        assert intervals.overlap(i, i + 1)

    assert len(list(intervals.iter_merge())) == 1

    l = list(intervals.iter_merge())

    assert l[0].begin == 0

    assert l[0].end == 20

    intervals.remove_overlap(5, 15)

    assert set(intervals.search_data(0, 20)) == {1, 2}

    for i in range(5):

        assert intervals.overlap(i, i + 1)

    for i in range(5, 15):

        assert not intervals.overlap(i, i + 1)

    for i in range(15, 20):

        assert intervals.overlap(i, i + 1)

    assert len(list(intervals.iter_merge())) == 2

    a, b = list(intervals.iter_merge())

    assert a.begin == 0

    assert a.end == 5

    assert b.begin == 15

    assert b.end == 20

COUNT = 10000

def g():

    for i in range(COUNT):

        yield Interval(3 * i, 3 * i + 1, 2)

        yield Interval(3 * i, 3 * i + 3, 1)

def amortized():

    def g():

        for i in range(COUNT):

            yield Interval(3 * i, 3 * i + 1, 2)

            yield Interval(3 * i, 3 * i + 3, 1)

    intervals = Intervals.from_ordered_intervals(g())

    def h():

        for i in range(COUNT):

            yield Interval(3 * i + 2, 3 * i + 3)

    intervals.remove_overlap_non_overlapping_ordered_intervals(

        Interval(3 * i + 2, 3 * i + 3) for i in range(COUNT))

def amortized2():

    intervals = Intervals.from_ordered_intervals(g())

    for i in range(COUNT):

        intervals.remove_overlap(3 * i + 2, 3 * i + 3)

def amortized3():

    from intervaltree import IntervalTree

    intervals = IntervalTree()

    for i in range(COUNT):

        intervals.addi(3 * i, 3 * i + 1, 2)

        intervals.addi(3 * i, 3 * i + 3, 1)

    for i in range(COUNT):

        intervals.remove_overlap(3 * i + 2, 3 * i + 3)

def test_amortized(benchmark):

    benchmark(amortized)

def test_amortized2(benchmark):

    benchmark(amortized2)

def test_amortized3(benchmark):

    benchmark(amortized3)

  pytest-benchmark

  intervaltree