lazy_set.py

#coding: utf-8

# Copyright (C) 2011 by Enrico Franchi  (tweaks by crap0101)
#
# This file is released under the terms of the MIT license
# http://www.opensource.org/licenses/mit-license.php

import itertools as it

def silence_generator_already_executing(generator):
    try:
        for element in generator:
            yield element
    except ValueError, e:
        pass

class lazy_set(object):
    """
    Trace of implementation of lazy sets in Python.
    """
    def __init__(self, input_=[]):
        self.input_ = iter(input_)
        self.seen = set()

    def __contains__(self, item):
        for i in self:
            if item == i:
                return True
        return False

    def __len__(self):
        return sum(1 for x in self)

    def _iter(self):
        for element in self.input_:
            if element in self.seen:
                continue
            else:
                self.seen.add(element)
                yield element

    def add(self, obj):
        self.seen.add(obj)

    def copy(self):
        return lazy_set(i for i in self)

    def isdisjoint(self, other):
        for item in other:
            if item in self:
                return False
        return True

    def __eq__(self, other):
        if not isinstance(other, (lazy_set, set, frozenset)):
            return False
        if len(self) != len(other):
            return False
        for item in self:
            if item not in other:
                return False
        return True

    def __ne__(self, other):
        return (self == other)^True

    def issubset(self, other):
        return len(self) <= sum(1 for x in self if x in set(other))

    def __le__(self, other):
        if not isinstance(other, (lazy_set, set, frozenset)):
            raise TypeError('can only compare to a set')
        return self.issubset(other)

    def __lt__(self, other):
        if not isinstance(other, (lazy_set, set, frozenset)):
            raise TypeError('can only compare to a set')
        return self.issubset(other) and self != other

    def issuperset(self, other):
        other = set(other)
        return len(other) <= sum(1 for x in other if x in self)

    def __ge__(self, other):
        if not isinstance(other, (lazy_set, set, frozenset)):
            raise TypeError('can only compare to a set')
        return self.issuperset(other)

    def __gt__(self, other):
        if not isinstance(other, (lazy_set, set, frozenset)):
            raise TypeError('can only compare to a set')
        return self.issuperset(other) and self != other

    def union(self, *others):
        new_lazy_set = self.copy()
        for other in others:
            new_lazy_set.update(other)
        return new_lazy_set

    def __or__(self, other):
        if not isinstance(other, (lazy_set, set, frozenset)):
            raise TypeError(
                "unsupported operand type(s) for |: '%s' and '%s'" %
                (type(self).__name__, type(other).__name__))
        new_lazy_set = self.copy()
        new_lazy_set.update(other)
        return new_lazy_set

    __ror__ = __or__

    def intersection(self, *others):
        raise NotImplementedError

    def __and__(self, other):
        return NotImplemented

    __rand__ = __and__

    def difference(self, *others):
        raise NotImplementedError

    def __sub__(self, other):
        return NotImplemented

    def symmetric_difference(self, *others):
        raise NotImplementedError

    def __xor__(self, other):
        return NotImplemented
    
    __rxor_ = __xor__

    def update(self, iterator):
        self.input_ = it.chain(self.input_,
            silence_generator_already_executing(iterator))

    def remove(self, item):
        if item not in self:
            raise KeyError(item)
        self.seen.remove(item)

    def discard(self, item):
        if item in self:
            self.seen.remove(item)

        
    def __iter__(self):
        return it.chain(self.seen, self._iter())


if __name__ == "__main__":
    import doctest
    doctest.testmod()

lazy_set_test.py

import itertools as it
import string
import unittest
import lazy_set


class LazySetTest(unittest.TestCase):
    def setUp(self):
        self.set_types = (lazy_set.lazy_set, set, frozenset)
        self.lazy_set = lazy_set.lazy_set(xrange(1, 4))
        self.starting_value = range(1, 4)

    def testCreation(self):
        self.assertEqual(
            sorted(self.lazy_set),
            self.starting_value
        )

    def testUpdate(self):
        self.lazy_set.update(2* x for x in range(4, 10))
        self.assertEqual(
            sorted(self.lazy_set),
            [1, 2, 3, 8, 10, 12, 14, 16, 18]
        )

    def testAdd(self):
        self.lazy_set.add(5)
        self.assertEqual(
            sorted(self.lazy_set),
            [1, 2, 3, 5]
        )

    def testUniqueness(self):
        self.lazy_set.update(xrange(5))
        self.lazy_set.update(xrange(3, 8))
        self.assertEqual(
            sorted(self.lazy_set),
            range(8)
        )

    def testSelfUpdateWIter(self):
        self.lazy_set.update(iter(self.lazy_set))
        self.assertEqual(
            sorted(self.lazy_set),
            self.starting_value
        )

    def testSelfUpdateWOIter(self):
        self.lazy_set.update(self.lazy_set)
        self.assertEqual(
            sorted(self.lazy_set),
            self.starting_value
        )

    def testSelfUpdateNotLats(self):
        self.lazy_set.update(iter(self.lazy_set))
        self.lazy_set.update(xrange(5, 8))
        self.assertEqual(
            sorted(self.lazy_set),
            range(1, 4) + range(5, 8)
        )

    def testLen(self):
       self.assertEqual(len(self.lazy_set), len(self.starting_value))
       self.lazy_set.update(self.starting_value)
       self.assertEqual(len(self.lazy_set), len(self.starting_value))
       strings = ['awert', 'i', '12', 'ferhjkl', 'cvbnm', 'bar', 'baz']
       lists = [range(x) for x in range(20)]
       for item in it.chain(strings, lists):
           lset = lazy_set.lazy_set(item)
           self.assertEqual(len(lset), len(item))
       numb1 = range(10)
       numb2 = range(11,20)
       lset = lazy_set.lazy_set(numb1)
       init_len = len(lset)
       for add, n in zip(it.count(1), numb2):
           lset.add(n)
           self.assertEqual(len(lset), init_len+add)

    def testCopy(self):
        new_lazy_set = self.lazy_set.copy()
        self.assertEqual(self.lazy_set, new_lazy_set)
        self.assertEqual(list(sorted(self.lazy_set)),
                         list(sorted(new_lazy_set)))
        other_lazy_set = new_lazy_set.copy()
        self.assertEqual(self.lazy_set, other_lazy_set)
        self.lazy_set.add(999)
        self.assertNotEqual(self.lazy_set, other_lazy_set)
        self.assertNotEqual(self.lazy_set, new_lazy_set)

    def testIsdisjoint(self):
        set_strings = list(lazy_set.lazy_set(string.letters[n:n+5])
                           for n in range(0, len(string.letters), 5))
        for s1, s2 in it.combinations(set_strings, 2):
            self.assertTrue(s1.isdisjoint(s2))
            self.assertTrue(s2.isdisjoint(s1))
            s1.update(s2)
            self.assertFalse(s1.isdisjoint(s2))

    def testEquality(self):
        new_lazy_set = lazy_set.lazy_set(i for i in self.lazy_set)
        self.assertEqual(self.lazy_set, new_lazy_set)
        self.assertFalse(self.lazy_set != new_lazy_set)
        new_lazy_set.update(list(i for i in self.lazy_set))
        self.assertEqual(self.lazy_set, new_lazy_set)
        self.assertFalse(self.lazy_set != new_lazy_set)
        self.lazy_set.update(range(100))
        new_lazy_set.update(range(100,200))
        self.assertNotEqual(self.lazy_set, new_lazy_set)
        foo_set = '1234567'
        new_lazy_set = lazy_set.lazy_set(foo_set)
        for settype in self.set_types:
            eq_set = settype(foo_set)
            self.assertEqual(new_lazy_set, eq_set)
        for othertype in (list, tuple, str):
            ne_set = othertype(foo_set)
            self.assertNotEqual(new_lazy_set, ne_set)
            
    def testIsSubSuperSet(self):
        def op_cmp_sub(s1,s2):
            return s1 <= s2
        def op_cmp_sup(s1,s2):
            return s1 >= s2
        other_set = self.lazy_set.copy()
        self.assertTrue(self.lazy_set.issubset(other_set))
        for i in range(20):
            other_set.add(i)
            self.assertTrue(self.lazy_set.issubset(other_set))
            self.assertTrue(self.lazy_set <= other_set)
            self.assertTrue(other_set.issuperset(self.lazy_set))
            self.assertTrue(other_set >= self.lazy_set)
        items = list(self.lazy_set)
        other_set = self.lazy_set.copy()
        for item in items:
            other_set.remove(item)
            self.assertFalse(self.lazy_set.issubset(other_set))
            self.assertFalse(self.lazy_set <= other_set)
            self.assertFalse(other_set.issuperset(self.lazy_set))
            self.assertFalse(other_set >= self.lazy_set)
        foo_set = '1234567'
        new_lazy_set = lazy_set.lazy_set(foo_set)
        for settype in self.set_types:
            sub_set = settype(foo_set)
            self.assertTrue(new_lazy_set.issubset(sub_set))
            self.assertTrue(sub_set.issuperset(new_lazy_set))
        for othertype in (list, tuple, str):
            no_set = othertype(foo_set)
            self.assertTrue(new_lazy_set.issubset(no_set))
            self.assertRaises(TypeError, op_cmp_sub, new_lazy_set, no_set)
            self.assertRaises(TypeError, op_cmp_sup, no_set, new_lazy_set)
            #with self.assertRaises(TypeError):
            #    new_lazy_set <= no_set

    def testUnion(self):
        def op_cmp_or(s, other):
            return s | other
        items = [range(n, n+10) for n in range(10)]
        sets = [lazy_set.lazy_set(item) for item in items]
        new_lazy_set = lazy_set.lazy_set()
        total_union = new_lazy_set.union(*sets)
        for settype in self.set_types:
            set2 = settype()
            set3 = settype()
            for s in sets:
                set2 = set2.union(s)
                set3 |= s
            self.assertEqual(total_union, set2)
            self.assertEqual(total_union, set3)
        no_set = '1234567'
        for othertype in (list, tuple, str):
            self.assertRaises(TypeError, op_cmp_or, total_union, no_set)

    def testRemoveAndDiscard(self):
        items = [range(10), 'qwerty']
        for item in items:
            new_set = lazy_set.lazy_set(item)
            orig_set = new_set.copy()
            for element in item:
                new_set.remove(element)
                self.assertTrue(new_set.issubset(orig_set))
            for element in item:
                self.assertRaises(KeyError, new_set.remove, element)
                new_set.discard(element)

                
if __name__ == '__main__':
    unittest.main()