# -*- coding: utf-8 -*- import nilmdb from nilmdb.printf import * import datetime_tz from nose.tools import * from nose.tools import assert_raises import itertools from nilmdb.interval import Interval, DBInterval, IntervalSet, IntervalError from test_helpers import * import unittest def makeset(string): """Build an IntervalSet from a string, for testing purposes Each character is 1 second [ = interval start | = interval end + adjacent start ] = interval end anything else is ignored """ iset = IntervalSet() for i, c in enumerate(string): day = i + 10000 if (c == "["): start = day elif (c == "|"): iset += Interval(start, day) start = day elif (c == "]"): iset += Interval(start, day) del start return iset class TestInterval: def test_interval(self): # Test Interval class os.environ['TZ'] = "America/New_York" datetime_tz._localtz = None (d1, d2, d3) = [ datetime_tz.datetime_tz.smartparse(x).totimestamp() for x in [ "03/24/2012", "03/25/2012", "03/26/2012" ] ] # basic construction i = Interval(d1, d1) i = Interval(d1, d3) eq_(i.start, d1) eq_(i.end, d3) # assignment is allowed, but not verified i.start = d2 #with assert_raises(IntervalError): # i.end = d1 i.start = d1 i.end = d2 # end before start with assert_raises(IntervalError): i = Interval(d3, d1) # compare assert(Interval(d1, d2) == Interval(d1, d2)) assert(Interval(d1, d2) < Interval(d1, d3)) assert(Interval(d1, d3) > Interval(d1, d2)) assert(Interval(d1, d2) < Interval(d2, d3)) assert(Interval(d1, d3) < Interval(d2, d3)) assert(Interval(d2, d2) > Interval(d1, d3)) assert(Interval(d3, d3) == Interval(d3, d3)) with assert_raises(AttributeError): x = (i == 123) # subset assert(Interval(d1, d3).subset(d1, d2) == Interval(d1, d2)) with assert_raises(IntervalError): x = Interval(d2, d3).subset(d1, d2) # big integers and floats x = Interval(5000111222, 6000111222) eq_(str(x), "[5000111222.0 -> 6000111222.0]") x = Interval(123.45, 234.56) eq_(str(x), "[123.45 -> 234.56]") # misc i = Interval(d1, d2) eq_(repr(i), repr(eval(repr(i)))) eq_(str(i), "[1332561600.0 -> 1332648000.0]") def test_interval_intersect(self): # Test Interval intersections dates = [ 100, 200, 300, 400 ] perm = list(itertools.permutations(dates, 2)) prod = list(itertools.product(perm, perm)) should_intersect = { False: [4, 5, 8, 20, 48, 56, 60, 96, 97, 100], True: [0, 1, 2, 12, 13, 14, 16, 17, 24, 25, 26, 28, 29, 32, 49, 50, 52, 53, 61, 62, 64, 65, 68, 98, 101, 104] } for i,((a,b),(c,d)) in enumerate(prod): try: i1 = Interval(a, b) i2 = Interval(c, d) eq_(i1.intersects(i2), i2.intersects(i1)) in_(i, should_intersect[i1.intersects(i2)]) except IntervalError: assert(i not in should_intersect[True] and i not in should_intersect[False]) with assert_raises(AttributeError): x = i1.intersects(1234) def test_intervalset_construct(self): # Test IntervalSet construction dates = [ 100, 200, 300, 400 ] a = Interval(dates[0], dates[1]) b = Interval(dates[1], dates[2]) c = Interval(dates[0], dates[2]) d = Interval(dates[2], dates[3]) iseta = IntervalSet(a) isetb = IntervalSet([a, b]) isetc = IntervalSet([a]) ne_(iseta, isetb) eq_(iseta, isetc) with assert_raises(TypeError): x = iseta != 3 ne_(IntervalSet(a), IntervalSet(b)) # test iterator for interval in iseta: pass # overlap with assert_raises(IntervalError): x = IntervalSet([a, b, c]) # bad types with assert_raises(Exception): x = IntervalSet([1, 2]) iset = IntervalSet(isetb) # test iterator eq_(iset, isetb) eq_(len(iset), 2) eq_(len(IntervalSet()), 0) # Test adding iset = IntervalSet(a) iset += IntervalSet(b) eq_(iset, IntervalSet([a, b])) iset = IntervalSet(a) iset += b eq_(iset, IntervalSet([a, b])) iset = IntervalSet(a) + IntervalSet(b) eq_(iset, IntervalSet([a, b])) iset = IntervalSet(b) + a eq_(iset, IntervalSet([a, b])) # A set consisting of [0-1],[1-2] should match a set consisting of [0-2] eq_(IntervalSet([a,b]), IntervalSet([c])) # Etc ne_(IntervalSet([a,d]), IntervalSet([c])) ne_(IntervalSet([c]), IntervalSet([a,d])) ne_(IntervalSet([c,d]), IntervalSet([b,d])) # misc eq_(repr(iset), repr(eval(repr(iset)))) eq_(str(iset), "[[100.0 -> 200.0], [200.0 -> 300.0]]") def test_intervalset_geniset(self): # Test basic iset construction assert(makeset(" [----] ") == makeset(" [-|--] ")) assert(makeset("[] [--] ") + makeset(" [] [--]") == makeset("[|] [-----]")) assert(makeset(" [-------]") == makeset(" [-|-----|")) def test_intervalset_intersect(self): # Test intersection (&) with assert_raises(AttributeError): x = makeset("[--]") & 1234 assert(makeset("[---------]") & makeset(" [---] ") == makeset(" [---] ")) assert(makeset(" [---] ") & makeset("[---------]") == makeset(" [---] ")) assert(makeset(" [-----]") & makeset(" [-----] ") == makeset(" [--] ")) assert(makeset(" [---]") & makeset(" [--] ") == makeset(" ")) assert(makeset(" [-|---]") & makeset(" [-----|-] ") == makeset(" [----] ")) assert(makeset(" [-|-] ") & makeset(" [-|--|--] ") == makeset(" [---] ")) assert(makeset(" [----][--]") & makeset("[-] [--] []") == makeset(" [] [-] []")) def test_dbinterval(self): # Test DBInterval class i = DBInterval(100, 200, 100, 200, 10000, 20000) eq_(i.start, 100) eq_(i.end, 200) eq_(i.db_start, 100) eq_(i.db_end, 200) eq_(i.db_startpos, 10000) eq_(i.db_endpos, 20000) eq_(repr(i), repr(eval(repr(i)))) # end before start with assert_raises(IntervalError): i = DBInterval(200, 100, 100, 200, 10000, 20000) # db_start too late with assert_raises(IntervalError): i = DBInterval(100, 200, 150, 200, 10000, 20000) # db_end too soon with assert_raises(IntervalError): i = DBInterval(100, 200, 100, 150, 10000, 20000) # actual start, end can be a subset a = DBInterval(150, 200, 100, 200, 10000, 20000) b = DBInterval(100, 150, 100, 200, 10000, 20000) c = DBInterval(150, 150, 100, 200, 10000, 20000) # Make a set of DBIntervals iseta = IntervalSet([a, b]) isetc = IntervalSet(c) assert(iseta.intersects(a)) assert(iseta.intersects(b)) # Test subset with assert_raises(IntervalError): x = a.subset(150, 250) # Subset of those IntervalSets should still contain DBIntervals for i in IntervalSet(iseta.intersection(Interval(125,250))): assert(isinstance(i, DBInterval)) class TestIntervalShape: def test_interval_shape(self): import random random.seed(1234) # make a set of 500 intervals iset = IntervalSet() j = 500 for i in random.sample(xrange(j),j): interval = Interval(i, i+1) iset += interval # Plot it import renderdot r = renderdot.Renderer(lambda node: node.cleft, lambda node: node.cright, lambda node: False, lambda node: node.start, lambda node: node.end, iset.tree.emptynode()) r.render_dot_live(iset.tree.rootnode(), "Random") # make a set of 500 intervals, inserted in order iset = IntervalSet() j = 500 for i in xrange(j): interval = Interval(i, i+1) iset += interval # Plot it import renderdot r = renderdot.Renderer(lambda node: node.cleft, lambda node: node.cright, lambda node: False, lambda node: node.start, lambda node: node.end, iset.tree.emptynode()) r.render_dot_live(iset.tree.rootnode(), "In-order") assert(False) class TestIntervalSpeed: #@unittest.skip("this is slow") def test_interval_speed(self): import yappi import time import aplotter print yappi.start() speeds = {} for j in [ 2**x for x in range(5,22) ]: start = time.time() iset = IntervalSet() for i in xrange(j): interval = Interval(i, i+1) iset += interval speed = (time.time() - start) * 1000000.0 printf("%d: %g μs (%g μs each)\n", j, speed, speed/j) speeds[j] = speed aplotter.plot(speeds.keys(), speeds.values(), plot_slope=True) yappi.stop() yappi.print_stats(sort_type=yappi.SORTTYPE_TTOT, limit=10)