aboutsummaryrefslogtreecommitdiff
path: root/src/argaze.test/DataStructures.py
diff options
context:
space:
mode:
Diffstat (limited to 'src/argaze.test/DataStructures.py')
-rw-r--r--src/argaze.test/DataStructures.py143
1 files changed, 143 insertions, 0 deletions
diff --git a/src/argaze.test/DataStructures.py b/src/argaze.test/DataStructures.py
index cab3a31..5f0a4ac 100644
--- a/src/argaze.test/DataStructures.py
+++ b/src/argaze.test/DataStructures.py
@@ -4,6 +4,29 @@ import unittest
from argaze import DataStructures
+def random_data_buffer(size, data_keys):
+ """ Generate a random TimeStampedBuffer for testing purpose.
+ Timestamps are current time.
+ Values are tuples containing an expected value and a random value.
+ """
+
+ import random
+ import time
+
+ ts_buffer = DataStructures.TimeStampedBuffer()
+
+ for i in range(0, size):
+
+ # Edit data
+ random_data = {}
+ for key in data_keys:
+ random_data[key] = (i, random.random())
+
+ # Store data
+ ts_buffer[time.time()] = random_data
+
+ return ts_buffer
+
class TestTimeStampedBufferClass(unittest.TestCase):
"""Test TimeStampedBuffer class."""
@@ -49,6 +72,126 @@ class TestTimeStampedBufferClass(unittest.TestCase):
self.assertEqual(len(ts_buffer.append(ts_buffer_next)), 4)
self.assertEqual(list(ts_buffer.append(ts_buffer_next).keys()), [0, 1, 2, 3])
+ def test_first(self):
+ """Test TimeStampedBuffer first property."""
+
+ self.assertEqual(DataStructures.TimeStampedBuffer({0: "A", 1: "B"}).first, (0, "A"))
+
+ # Check that accessing to first item of an empty TimeStampedBuffer fails
+ with self.assertRaises(IndexError):
+
+ DataStructures.TimeStampedBuffer().first
+
+ def test_pop_first(self):
+ """Test TimeStampedBuffer pop_first method."""
+
+ ts_buffer = DataStructures.TimeStampedBuffer({0: "A", 1: "B"})
+
+ self.assertEqual(ts_buffer.pop_first(), (0, "A"))
+ self.assertEqual(len(ts_buffer), 1)
+ self.assertEqual(ts_buffer.first, (1, "B"))
+
+ def test_pop_first_until(self):
+ """Test TimeStampedBuffer pop_first_until method."""
+
+ ts_buffer = DataStructures.TimeStampedBuffer({0: "A", 1: "B", 2: "C", 3: "D"})
+
+ # Check pop until an existing timestamp
+ first_until_2 = ts_buffer.pop_first_until(2)
+
+ self.assertEqual(first_until_2, (1, "B"))
+ self.assertEqual(len(ts_buffer), 2)
+ self.assertEqual(ts_buffer.first, (2, "C"))
+
+ # Check pop until an none existing timestamp
+ ts_buffer = DataStructures.TimeStampedBuffer({0: "A", 1: "B", 2: "C", 3: "D"})
+
+ first_until_1dot5 = ts_buffer.pop_first_until(1.5)
+
+ self.assertEqual(first_until_1dot5, (1, "B"))
+ self.assertEqual(len(ts_buffer), 2)
+ self.assertEqual(ts_buffer.first, (2, "C"))
+
+ def test_last(self):
+ """Test TimeStampedBuffer last property."""
+
+ self.assertEqual(DataStructures.TimeStampedBuffer({0: "A", 1: "B"}).last, (1, "B"))
+
+ # Check that accessing to last item of an empty TimeStampedBuffer fails
+ with self.assertRaises(IndexError):
+
+ DataStructures.TimeStampedBuffer().last
+
+ def test_pop_last(self):
+ """Test TimeStampedBuffer pop_last method."""
+
+ ts_buffer = DataStructures.TimeStampedBuffer({0: "A", 1: "B"})
+
+ self.assertEqual(ts_buffer.pop_last(), (1, "B"))
+ self.assertEqual(len(ts_buffer), 1)
+ self.assertEqual(ts_buffer.last, (0, "A"))
+
+ def test_get_last_before(self):
+ """Test TimeStampedBuffer get_last_before method."""
+
+ ts_buffer = DataStructures.TimeStampedBuffer({0: "A", 1: "B", 2: "C", 3: "D"})
+
+ get_last_before_2 = ts_buffer.get_last_before(2)
+
+ self.assertEqual(get_last_before_2, (1, "B"))
+ self.assertEqual(len(ts_buffer), 4)
+
+ get_last_before_1dot5 = ts_buffer.get_last_before(1.5)
+
+ self.assertEqual(get_last_before_1dot5, (1, "B"))
+
+ get_last_before_4 = ts_buffer.get_last_before(4)
+
+ self.assertEqual(get_last_before_4, (3, "D"))
+
+ # Check that accessing to early timestamp fails
+ with self.assertRaises(KeyError):
+
+ ts_buffer.get_last_before(-1)
+
+ def test_as_dataframe(self):
+ """Test TimeStampedBuffer as_dataframe method."""
+
+ ts_buffer = random_data_buffer(10, ["data_A", "data_B", "data_C"])
+
+ # Check dataframe conversion
+ ts_buffer_dataframe = ts_buffer.as_dataframe()
+
+ self.assertEqual(ts_buffer_dataframe.index.name, "timestamp")
+ self.assertEqual(ts_buffer_dataframe.index.size, 10)
+
+ self.assertEqual(ts_buffer_dataframe.columns.size, 3)
+ self.assertEqual(ts_buffer_dataframe.columns[0], "data_A")
+ self.assertEqual(ts_buffer_dataframe.columns[1], "data_B")
+ self.assertEqual(ts_buffer_dataframe.columns[2], "data_C")
+
+ # Check data exclusion option
+ ts_buffer_dataframe = ts_buffer.as_dataframe(exclude=["data_B"])
+
+ self.assertEqual(ts_buffer_dataframe.index.name, "timestamp")
+ self.assertEqual(ts_buffer_dataframe.index.size, 10)
+
+ self.assertEqual(ts_buffer_dataframe.columns.size, 2)
+ self.assertEqual(ts_buffer_dataframe.columns[0], "data_A")
+ self.assertEqual(ts_buffer_dataframe.columns[1], "data_C")
+
+ # Check dataframe split option
+ ts_buffer_dataframe = ts_buffer.as_dataframe(split={"data_B": ["data_B0", "data_B1"]})
+
+ self.assertEqual(ts_buffer_dataframe.index.name, "timestamp")
+ self.assertEqual(ts_buffer_dataframe.index.size, 10)
+
+ self.assertEqual(ts_buffer_dataframe.columns.size, 4)
+ self.assertEqual(ts_buffer_dataframe.columns[0], "data_A")
+ self.assertEqual(ts_buffer_dataframe.columns[1], "data_B0")
+ self.assertEqual(ts_buffer_dataframe.columns[2], "data_B1")
+ self.assertEqual(ts_buffer_dataframe.columns[3], "data_C")
+
if __name__ == '__main__':
unittest.main() \ No newline at end of file