aboutsummaryrefslogtreecommitdiff
diff options
context:
space:
mode:
authorThéo de la Hogue2022-12-07 16:24:21 +0100
committerThéo de la Hogue2022-12-07 16:24:21 +0100
commit70fa9dc083f36819f9d030f0c9dac434d66fa964 (patch)
tree29fbcc9ad234b711a76ae4605b59b299275b696b
parent7f71390d1a0fc51222c5ac202c483c5847873697 (diff)
downloadargaze-70fa9dc083f36819f9d030f0c9dac434d66fa964.zip
argaze-70fa9dc083f36819f9d030f0c9dac434d66fa964.tar.gz
argaze-70fa9dc083f36819f9d030f0c9dac434d66fa964.tar.bz2
argaze-70fa9dc083f36819f9d030f0c9dac434d66fa964.tar.xz
Testing dataframe conversion, json encoding, repr and str string representations.
-rw-r--r--src/argaze.test/DataStructures.py125
1 files changed, 121 insertions, 4 deletions
diff --git a/src/argaze.test/DataStructures.py b/src/argaze.test/DataStructures.py
index c1b78d5..1d34740 100644
--- a/src/argaze.test/DataStructures.py
+++ b/src/argaze.test/DataStructures.py
@@ -2,9 +2,13 @@
import unittest
from dataclasses import dataclass, field
+import os
from argaze import DataStructures
+import pandas
+import numpy
+
def random_data_buffer(size, data_keys):
""" Generate a random TimeStampedBuffer for testing purpose.
Timestamps are current time.
@@ -26,6 +30,8 @@ def random_data_buffer(size, data_keys):
# Store data
ts_buffer[time.time()] = random_data
+ time.sleep(0.0001)
+
return ts_buffer
@dataclass()
@@ -63,6 +69,28 @@ class TestTimeStampedBufferClass(unittest.TestCase):
DataStructures.TimeStampedBuffer({"first": ""})
+ def test_from_dataframe(self):
+ """Test TimeStampedBuffer creation from pandas dataframe."""
+
+ ts_buffer = random_data_buffer(10, ["data_A", "data_B", "data_C"])
+
+ # Check dataframe conversion
+ ts_buffer_from_df = DataStructures.TimeStampedBuffer.from_dataframe(ts_buffer.as_dataframe())
+
+ self.assertEqual(len(ts_buffer_from_df), 10)
+
+ def test_from_json(self):
+ """Test TimeStampedBuffer creation from json file."""
+
+ # Edit dataframe csv file path
+ current_directory = os.path.dirname(os.path.abspath(__file__))
+ json_filepath = os.path.join(current_directory, 'utils/ts_buffer.json')
+
+ # Load TimeStampedBuffer from json file
+ ts_buffer = DataStructures.TimeStampedBuffer.from_json(json_filepath)
+
+ self.assertEqual(len(ts_buffer), 3)
+
def test___repr__(self):
"""Test TimeStampedBuffer string representation."""
@@ -75,6 +103,28 @@ class TestTimeStampedBufferClass(unittest.TestCase):
self.assertEqual(repr(ts_buffer), "{\"0\": {\"value\": [123, 456]}}")
+ array = numpy.zeros(3)
+ ts_buffer = DataStructures.TimeStampedBuffer({0: array})
+
+ self.assertEqual(repr(ts_buffer), "{\"0\": [0.0, 0.0, 0.0]}")
+
+ def test___str__(self):
+ """Test TimeStampedBuffer string representation."""
+
+ self.assertEqual(str(DataStructures.TimeStampedBuffer()), "{}")
+ self.assertEqual(str(DataStructures.TimeStampedBuffer({0: ""})), "{\"0\": \"\"}")
+ self.assertEqual(str(DataStructures.TimeStampedBuffer({0.1: ""})), "{\"0.1\": \"\"}")
+
+ data = BasicDataClass((123, 456))
+ ts_buffer = DataStructures.TimeStampedBuffer({0: data})
+
+ self.assertEqual(str(ts_buffer), "{\"0\": {\"value\": [123, 456]}}")
+
+ array = numpy.zeros(3)
+ ts_buffer = DataStructures.TimeStampedBuffer({0: array})
+
+ self.assertEqual(str(ts_buffer), "{\"0\": [0.0, 0.0, 0.0]}")
+
def test_append(self):
"""Test TimeStampedBuffer append method."""
@@ -102,14 +152,35 @@ class TestTimeStampedBufferClass(unittest.TestCase):
self.assertEqual(ts_buffer.pop_first(), (0, "A"))
self.assertEqual(len(ts_buffer), 1)
self.assertEqual(ts_buffer.first, (1, "B"))
+
+ def test_pop_last_until(self):
+ """Test TimeStampedBuffer pop_last_until method."""
- 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
+ pop_last_until_2 = ts_buffer.pop_last_until(2)
+
+ self.assertEqual(pop_last_until_2, (2, "C"))
+ self.assertEqual(len(ts_buffer), 2)
+ self.assertEqual(ts_buffer.first, (2, "C"))
+
+ # Check first until an none existing timestamp
+ ts_buffer = DataStructures.TimeStampedBuffer({0: "A", 1: "B", 2: "C", 3: "D"})
+
+ pop_last_until_1dot5 = ts_buffer.pop_last_until(1.5)
+
+ self.assertEqual(pop_last_until_1dot5, (1, "B"))
+ self.assertEqual(len(ts_buffer), 3)
+ self.assertEqual(ts_buffer.first, (1, "B"))
+
+ def test_pop_last_before(self):
+ """Test TimeStampedBuffer pop_last_before 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)
+ first_until_2 = ts_buffer.pop_last_before(2)
self.assertEqual(first_until_2, (1, "B"))
self.assertEqual(len(ts_buffer), 2)
@@ -118,7 +189,7 @@ class TestTimeStampedBufferClass(unittest.TestCase):
# 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)
+ first_until_1dot5 = ts_buffer.pop_last_before(1.5)
self.assertEqual(first_until_1dot5, (1, "B"))
self.assertEqual(len(ts_buffer), 2)
@@ -143,6 +214,29 @@ class TestTimeStampedBufferClass(unittest.TestCase):
self.assertEqual(len(ts_buffer), 1)
self.assertEqual(ts_buffer.last, (0, "A"))
+ def test_get_first_from(self):
+ """Test TimeStampedBuffer get_first_from method."""
+
+ ts_buffer = DataStructures.TimeStampedBuffer({0: "A", 1: "B", 2: "C", 3: "D"})
+
+ get_first_from_1 = ts_buffer.get_first_from(1)
+
+ self.assertEqual(get_first_from_1, (1, "B"))
+ self.assertEqual(len(ts_buffer), 4)
+
+ get_first_from_1dot5 = ts_buffer.get_first_from(1.5)
+
+ self.assertEqual(get_first_from_1dot5, (2, "C"))
+
+ get_first_from_0 = ts_buffer.get_first_from(0)
+
+ self.assertEqual(get_first_from_0, (0, "A"))
+
+ # Check that accessing to lately timestamp fails
+ with self.assertRaises(KeyError):
+
+ ts_buffer.get_first_from(4)
+
def test_get_last_before(self):
"""Test TimeStampedBuffer get_last_before method."""
@@ -166,6 +260,29 @@ class TestTimeStampedBufferClass(unittest.TestCase):
ts_buffer.get_last_before(-1)
+ def test_get_last_until(self):
+ """Test TimeStampedBuffer get_last_until method."""
+
+ ts_buffer = DataStructures.TimeStampedBuffer({0: "A", 1: "B", 2: "C", 3: "D"})
+
+ get_last_until_2 = ts_buffer.get_last_until(2)
+
+ self.assertEqual(get_last_until_2, (2, "C"))
+ self.assertEqual(len(ts_buffer), 4)
+
+ get_last_until_1dot5 = ts_buffer.get_last_until(1.5)
+
+ self.assertEqual(get_last_until_1dot5, (1, "B"))
+
+ get_last_until_4 = ts_buffer.get_last_until(4)
+
+ self.assertEqual(get_last_until_4, (3, "D"))
+
+ # Check that accessing to early timestamp fails
+ with self.assertRaises(KeyError):
+
+ ts_buffer.get_last_until(-1)
+
def test_as_dataframe(self):
"""Test TimeStampedBuffer as_dataframe method."""