Testing dataframe conversion, json encoding, repr and str string representations.

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."""