Browse Source

Add DBInterval and associated tests. Clean up other interval tests as

well, and fix missing coverage on IntervalSet.str()


git-svn-id: https://bucket.mit.edu/svn/nilm/nilmdb@10890 ddd99763-3ecb-0310-9145-efcb8ce7c51f
tags/bxinterval-last
Jim Paris 10 years ago
parent
commit
8a7b3b5f95
4 changed files with 114 additions and 20 deletions
  1. +51
    -0
      nilmdb/interval.py
  2. +2
    -2
      nilmdb/nilmdb.py
  3. +1
    -1
      setup.cfg
  4. +60
    -17
      tests/test_interval.py

+ 51
- 0
nilmdb/interval.py View File

@@ -49,6 +49,54 @@ class Interval(bxintersect.Interval):
raise IntervalError("not a subset")
return Interval(start, end)

class DBInterval(Interval):
"""
Like Interval, but also tracks corresponding start/end times and
positions within the database. These are not currently modified
when subsets are taken, but can be used later to help zero in on
database positions.

The actual 'start' and 'end' will always fall within the database
start and end, e.g.:
db_start = 100, db_startpos = 10000
start = 123
end = 150
db_end = 200, db_endpos = 20000
"""
def __init__(self, start, end,
db_start, db_end,
db_startpos, db_endpos):
"""
'db_start' and 'db_end' are arbitrary floats that represent
time. They must be a strict superset of the time interval
covered by 'start' and 'end'. The 'db_startpos' and
'db_endpos' are arbitrary database position indicators that
correspond to those points.
"""
Interval.__init__(self, start, end)
self.db_start = db_start
self.db_end = db_end
self.db_startpos = db_startpos
self.db_endpos = db_endpos
if db_start > start or db_end < end:
raise IntervalError("database times must span the interval times")

def __repr__(self):
s = repr(self.start) + ", " + repr(self.end)
s += ", " + repr(self.db_start) + ", " + repr(self.db_end)
s += ", " + repr(self.db_startpos) + ", " + repr(self.db_endpos)
return self.__class__.__name__ + "(" + s + ")"

def subset(self, start, end):
"""
Return a new DBInterval that is a subset of this one
"""
if start < self.start or end > self.end:
raise IntervalError("not a subset")
return DBInterval(start, end,
self.db_start, self.db_end,
self.db_startpos, self.db_endpos)

class IntervalSet(object):
"""
A non-intersecting set of intervals.
@@ -174,6 +222,9 @@ class IntervalSet(object):
intersection between `self` and the provided interval.
Returns a generator that yields each of these intervals
in turn.

Output intervals are built as subsets of the intervals in the
first argument (self).
"""
for i in self.tree.find(interval.start, interval.end):
if i.start > interval.start and i.end < interval.end:


+ 2
- 2
nilmdb/nilmdb.py View File

@@ -22,7 +22,7 @@ import errno

import pyximport
pyximport.install()
from nilmdb.interval import Interval, IntervalSet, IntervalError
from nilmdb.interval import Interval, DBInterval, IntervalSet, IntervalError

# Note about performance and transactions:
#
@@ -384,12 +384,12 @@ class NilmDB(object):
and a new request with a start time of 'restart' will fetch
the next block of data.
"""
# TODO: FIX
stream_id = self._stream_id(path)
intervals = self._get_intervals(stream_id)
requested = Interval(start or 0, end or 1e12)
result = []
for n, i in enumerate(intervals.intersection(requested)):
# TODO: fix
if n >= self.max_results:
restart = i.start
break


+ 1
- 1
setup.cfg View File

@@ -11,7 +11,7 @@ cover-erase=
stop=
verbosity=2
#tests=tests/test_cmdline.py
tests=tests/test_layout.py
#tests=tests/test_layout.py
#tests=tests/test_interval.py
#tests=tests/test_client.py
#tests=tests/test_timestamper.py


+ 60
- 17
tests/test_interval.py View File

@@ -8,7 +8,7 @@ from nose.tools import *
from nose.tools import assert_raises
import itertools

from nilmdb.interval import Interval, IntervalSet, IntervalError
from nilmdb.interval import Interval, DBInterval, IntervalSet, IntervalError

from test_helpers import *
import unittest
@@ -46,8 +46,8 @@ class TestInterval:
# basic construction
i = Interval(d1, d1)
i = Interval(d1, d3)
assert(i.start == d1)
assert(i.end == d3)
eq_(i.start, d1)
eq_(i.end, d3)

# assignment is allowed, but not verified
i.start = d2
@@ -101,8 +101,8 @@ class TestInterval:
try:
i1 = Interval(a, b)
i2 = Interval(c, d)
assert(i1.intersects(i2) == i2.intersects(i1))
assert(i in should_intersect[i1.intersects(i2)])
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])
@@ -140,31 +140,32 @@ class TestInterval:
x = IntervalSet([1, 2])

iset = IntervalSet(isetb) # test iterator
assert(iset == isetb)
assert(len(iset) == 2)
assert(len(IntervalSet()) == 0)
eq_(iset, isetb)
eq_(len(iset), 2)
eq_(len(IntervalSet()), 0)

# Test adding
iset = IntervalSet(a)
iset += IntervalSet(b)
assert(iset == IntervalSet([a, b]))
eq_(iset, IntervalSet([a, b]))
iset = IntervalSet(a)
iset += b
assert(iset == IntervalSet([a, b]))
eq_(iset, IntervalSet([a, b]))
iset = IntervalSet(a) + IntervalSet(b)
assert(iset == IntervalSet([a, b]))
eq_(iset, IntervalSet([a, b]))
iset = IntervalSet(b) + a
assert(iset == IntervalSet([a, b]))
eq_(iset, IntervalSet([a, b]))

# A set consisting of [0-1],[1-2] should match a set consisting of [0-2]
assert(IntervalSet([a,b]) == IntervalSet([c]))
eq_(IntervalSet([a,b]), IntervalSet([c]))
# Etc
assert(IntervalSet([a,d]) != IntervalSet([c]))
assert(IntervalSet([c]) != IntervalSet([a,d]))
assert(IntervalSet([c,d]) != IntervalSet([b,d]))
ne_(IntervalSet([a,d]), IntervalSet([c]))
ne_(IntervalSet([c]), IntervalSet([a,d]))
ne_(IntervalSet([c,d]), IntervalSet([b,d]))

# misc
assert(repr(iset) == repr(eval(repr(iset))))
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
@@ -212,6 +213,48 @@ class TestInterval:
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 TestIntervalSpeed:
@unittest.skip("this is slow")
def test_interval_speed(self):


Loading…
Cancel
Save