From 70fa9dc083f36819f9d030f0c9dac434d66fa964 Mon Sep 17 00:00:00 2001 From: Théo de la Hogue Date: Wed, 7 Dec 2022 16:24:21 +0100 Subject: Testing dataframe conversion, json encoding, repr and str string representations. --- src/argaze.test/DataStructures.py | 125 ++++++++++++++++++++++++++++++++++++-- 1 file changed, 121 insertions(+), 4 deletions(-) (limited to 'src/argaze.test') 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.""" -- cgit v1.1