diff options
Diffstat (limited to 'src/argaze.test/DataFeatures.py')
| -rw-r--r-- | src/argaze.test/DataFeatures.py | 464 |
1 files changed, 245 insertions, 219 deletions
diff --git a/src/argaze.test/DataFeatures.py b/src/argaze.test/DataFeatures.py index b30c560..c64ad4c 100644 --- a/src/argaze.test/DataFeatures.py +++ b/src/argaze.test/DataFeatures.py @@ -16,334 +16,360 @@ from argaze import DataFeatures import pandas import numpy -def random_data_buffer(size, data_keys): - """ Generate a random TimeStampedBuffer for testing purpose. +class BasicDataClass(DataFeatures.TimestampedObject): + """Define a basic dataclass for testing purpose.""" + + def __init__(self, value: tuple = (), message: str = None, **kwargs): + + DataFeatures.TimestampedObject.__init__(self, **kwargs) + + self.__value = value + self.__message = message + + @property + def value(self): + return self.__value + + @property + def message(self): + return self.__message + +def random_data_list(size): + """ Generate a random TimestampedObjectsList for testing purpose. Timestamps are current time. Values are tuples containing an expected value and a random value. """ - import random import time - ts_buffer = DataFeatures.TimeStampedBuffer() + data_list = [] for i in range(0, size): # Edit data - random_data = {} - for key in data_keys: - random_data[key] = (i, random.random()) + random_data = BasicDataClass((i, random.random()), f'test_{i}') + + # Timestamp data + random_data.timestamp = time.time() # Store data - ts_buffer[time.time()] = random_data + data_list.append(random_data) time.sleep(0.0001) - return ts_buffer + return DataFeatures.TimestampedObjectsList(BasicDataClass, data_list) -@dataclass() -class BasicDataClass(): - """Define a basic dataclass for testing purpose.""" - - value: tuple +# DEBUG +''' +print('test_as_dataframe: export ts_data_file.json') +current_directory = os.path.dirname(os.path.abspath(__file__)) +json_filepath = os.path.join(current_directory, 'utils/ts_data_file.json') +random_data_list(10).to_json(json_filepath) +''' -class TestTimeStampedBufferClass(unittest.TestCase): - """Test TimeStampedBuffer class.""" +class TestTimestampedObjectsListClass(unittest.TestCase): + """Test TimestampedObjectsList class.""" def test_new(self): - """Test TimeStampedBuffer creation.""" + """Test TimestampedObjectsList creation.""" + + # Check TimestampedObjectsList length after creation + self.assertEqual(len(DataFeatures.TimestampedObjectsList(BasicDataClass)), 0) + + # Check TimestampedObjectsList timestamps after creation + self.assertEqual(DataFeatures.TimestampedObjectsList(BasicDataClass).timestamps(), []) + + # Check TimestampedObjectsList items after creation + self.assertEqual(DataFeatures.TimestampedObjectsList(BasicDataClass), []) + + # Check that TimestampedObjectsList creation fails when data are not timestamped + with self.assertRaises(ValueError): + + data_list = [BasicDataClass((0, 0))] + DataFeatures.TimestampedObjectsList(BasicDataClass, data_list) - # Check TimeStampedBuffer length after creation - self.assertEqual(len(DataFeatures.TimeStampedBuffer()), 0) - self.assertEqual(len(DataFeatures.TimeStampedBuffer({0: ""})), 1) - self.assertEqual(len(DataFeatures.TimeStampedBuffer({0.1: ""})), 1) - self.assertEqual(len(DataFeatures.TimeStampedBuffer({0: "A", 1: "B"})), 2) + def test_as_dataframe(self): + """Test TimestampedObjectsList as_dataframe method.""" - # Check TimeStampedBuffer keys after creation - self.assertEqual(list(DataFeatures.TimeStampedBuffer().keys()), []) - self.assertEqual(list(DataFeatures.TimeStampedBuffer({0: ""}).keys()), [0]) - self.assertEqual(list(DataFeatures.TimeStampedBuffer({0.1: ""}).keys()), [0.1]) - self.assertEqual(list(DataFeatures.TimeStampedBuffer({0: "A", 1: "B"}).keys()), [0, 1]) + data_frame = random_data_list(10).as_dataframe() - # Check TimeStampedBuffer items after creation - self.assertEqual(list(DataFeatures.TimeStampedBuffer().items()), []) - self.assertEqual(list(DataFeatures.TimeStampedBuffer({0: ""}).items()), [(0, "")]) - self.assertEqual(list(DataFeatures.TimeStampedBuffer({0.1: ""}).items()), [(0.1, "")]) - self.assertEqual(list(DataFeatures.TimeStampedBuffer({0: "A", 1: "B"}).items()), [(0, "A"), (1, "B")]) + # Check dataframe conversion + self.assertEqual(data_frame.index.name, "timestamp") + self.assertEqual(data_frame.index.size, 10) + self.assertEqual(data_frame.columns.size, 2) + self.assertEqual(data_frame.index.dtype, 'float64') + self.assertEqual(data_frame["value"].dtype, 'object') + self.assertEqual(data_frame["message"].dtype, 'object') - # Check that TimeStampedBuffer creation fails when keys are not numbers - with self.assertRaises(AssertionError): + # Check data exclusion option + data_frame = random_data_list(10).as_dataframe(exclude=["value"]) - DataFeatures.TimeStampedBuffer({"first": ""}) + self.assertEqual(data_frame.index.name, "timestamp") + self.assertEqual(data_frame.index.size, 10) + self.assertEqual(data_frame.columns.size, 1) + # Check dataframe split option + data_frame = random_data_list(10).as_dataframe(split={"value": ["value_0", "value_1"]}) + + self.assertEqual(data_frame.index.name, "timestamp") + self.assertEqual(data_frame.index.size, 10) + self.assertEqual(data_frame.columns.size, 3) + self.assertEqual(data_frame["value_0"].dtype, 'int64') + self.assertEqual(data_frame["value_1"].dtype, 'float64') + def test_from_dataframe(self): - """Test TimeStampedBuffer creation from pandas dataframe.""" + """Test TimestampedObjectsList creation from pandas dataframe.""" - ts_buffer = random_data_buffer(10, ["data_A", "data_B", "data_C"]) + data_frame = random_data_list(10).as_dataframe() # Check dataframe conversion - ts_buffer_from_df = DataFeatures.TimeStampedBuffer.from_dataframe(ts_buffer.as_dataframe()) - - self.assertEqual(len(ts_buffer_from_df), 10) + data_list = DataFeatures.TimestampedObjectsList.from_dataframe(BasicDataClass, data_frame) + self.assertEqual(len(data_list), 10) + def test_from_json(self): - """Test TimeStampedBuffer creation from json file.""" + """Test TimestampedObjectsList 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') + json_filepath = os.path.join(current_directory, 'utils/ts_data_file.json') - # Load TimeStampedBuffer from json file - ts_buffer = DataFeatures.TimeStampedBuffer.from_json(json_filepath) - - self.assertEqual(len(ts_buffer), 3) + # Load TimestampedObjectsList from json file + data_list = DataFeatures.TimestampedObjectsList.from_json(BasicDataClass, json_filepath) + self.assertEqual(len(data_list), 10) + self.assertEqual(type(data_list[0]), BasicDataClass) + def test___repr__(self): - """Test TimeStampedBuffer string representation.""" + """Test TimestampedObjectsList string representation.""" - self.assertEqual(repr(DataFeatures.TimeStampedBuffer()), "{}") - self.assertEqual(repr(DataFeatures.TimeStampedBuffer({0: ""})), "{\"0\": \"\"}") - self.assertEqual(repr(DataFeatures.TimeStampedBuffer({0.1: ""})), "{\"0.1\": \"\"}") + self.assertEqual(repr(DataFeatures.TimestampedObjectsList(BasicDataClass)), "[]") - data = BasicDataClass((123, 456)) - ts_buffer = DataFeatures.TimeStampedBuffer({0: data}) - - self.assertEqual(repr(ts_buffer), "{\"0\": {\"value\": [123, 456]}}") - - array = numpy.zeros(3) - ts_buffer = DataFeatures.TimeStampedBuffer({0: array}) - - self.assertEqual(repr(ts_buffer), "{\"0\": [0.0, 0.0, 0.0]}") + data_list = [BasicDataClass((0, 0), 'test', timestamp=0)] + self.assertEqual(repr(DataFeatures.TimestampedObjectsList(BasicDataClass, data_list)), "[{\"value\": [0, 0], \"message\": \"test\", \"timestamp\": 0}]") def test___str__(self): - """Test TimeStampedBuffer string representation.""" - - self.assertEqual(str(DataFeatures.TimeStampedBuffer()), "{}") - self.assertEqual(str(DataFeatures.TimeStampedBuffer({0: ""})), "{\"0\": \"\"}") - self.assertEqual(str(DataFeatures.TimeStampedBuffer({0.1: ""})), "{\"0.1\": \"\"}") + """Test TimestampedObjectsList string representation.""" - data = BasicDataClass((123, 456)) - ts_buffer = DataFeatures.TimeStampedBuffer({0: data}) + self.assertEqual(str(DataFeatures.TimestampedObjectsList(BasicDataClass)), "[]") - self.assertEqual(str(ts_buffer), "{\"0\": {\"value\": [123, 456]}}") - - array = numpy.zeros(3) - ts_buffer = DataFeatures.TimeStampedBuffer({0: array}) - - self.assertEqual(str(ts_buffer), "{\"0\": [0.0, 0.0, 0.0]}") + data_list = [BasicDataClass((0, 0), 'test', timestamp=0)] + self.assertEqual(str(DataFeatures.TimestampedObjectsList(BasicDataClass, data_list)), "[{\"value\": [0, 0], \"message\": \"test\", \"timestamp\": 0}]") def test_append(self): - """Test TimeStampedBuffer append method.""" - - ts_buffer = DataFeatures.TimeStampedBuffer({0: "A", 1: "B"}) - ts_buffer_next = DataFeatures.TimeStampedBuffer({2: "C", 3: "D"}) - - self.assertEqual(len(ts_buffer.append(ts_buffer_next)), 4) - self.assertEqual(list(ts_buffer.append(ts_buffer_next).keys()), [0, 1, 2, 3]) + """Test TimestampedObjectsList append method.""" - def test_first(self): - """Test TimeStampedBuffer first property.""" + data_list = DataFeatures.TimestampedObjectsList(BasicDataClass) + next_data = BasicDataClass((0, 0), 'test', timestamp=0) - self.assertEqual(DataFeatures.TimeStampedBuffer({0: "A", 1: "B"}).first, (0, "A")) + data_list.append(next_data) - # Check that accessing to first item of an empty TimeStampedBuffer fails - with self.assertRaises(IndexError): + self.assertEqual(len(data_list), 1) + self.assertEqual(data_list.timestamps(), [0]) + + def test_pop_first(self): + """Test TimestampedObjectsList pop_first method.""" - DataFeatures.TimeStampedBuffer().first + data_list = DataFeatures.TimestampedObjectsList(BasicDataClass, \ + [ + BasicDataClass(message="A", timestamp=0), + BasicDataClass(message="B", timestamp=1) + ]) - def test_pop_first(self): - """Test TimeStampedBuffer pop_first method.""" + first = data_list.pop(0) - ts_buffer = DataFeatures.TimeStampedBuffer({0: "A", 1: "B"}) + self.assertEqual(len(data_list), 1) + self.assertEqual(first.message, "A") + self.assertEqual(first.timestamp, 0) + self.assertEqual(data_list[0].message, "B") + self.assertEqual(data_list[0].timestamp, 1) - 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.""" + """Test TimestampedObjectsList pop_last_until method.""" - ts_buffer = DataFeatures.TimeStampedBuffer({0: "A", 1: "B", 2: "C", 3: "D"}) + data_list = DataFeatures.TimestampedObjectsList(BasicDataClass, \ + [ + BasicDataClass(message="A", timestamp=0), + BasicDataClass(message="B", timestamp=1), + BasicDataClass(message="C", timestamp=2), + BasicDataClass(message="D", timestamp=3) + ]) # Check pop until an existing timestamp - pop_last_until_2 = ts_buffer.pop_last_until(2) + pop_last_until_2 = data_list.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")) + self.assertEqual(pop_last_until_2.message, "C") + self.assertEqual(pop_last_until_2.timestamp, 2) + self.assertEqual(len(data_list), 2) + self.assertEqual(data_list[0].message, "C") + self.assertEqual(data_list[0].timestamp, 2) # Check first until an none existing timestamp - ts_buffer = DataFeatures.TimeStampedBuffer({0: "A", 1: "B", 2: "C", 3: "D"}) - - pop_last_until_1dot5 = ts_buffer.pop_last_until(1.5) + data_list = DataFeatures.TimestampedObjectsList(BasicDataClass, \ + [ + BasicDataClass(message="A", timestamp=0), + BasicDataClass(message="B", timestamp=1), + BasicDataClass(message="C", timestamp=2), + BasicDataClass(message="D", timestamp=3) + ]) + + pop_last_until_1dot5 = data_list.pop_last_until(1.5) + + self.assertEqual(pop_last_until_1dot5.message, "B") + self.assertEqual(pop_last_until_1dot5.timestamp, 1) + self.assertEqual(len(data_list), 3) + self.assertEqual(data_list[0].message, "B") + self.assertEqual(data_list[0].timestamp, 1) - 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.""" + """Test TimestampedObjectsList pop_last_before method.""" - ts_buffer = DataFeatures.TimeStampedBuffer({0: "A", 1: "B", 2: "C", 3: "D"}) + data_list = DataFeatures.TimestampedObjectsList(BasicDataClass, \ + [ + BasicDataClass(message="A", timestamp=0), + BasicDataClass(message="B", timestamp=1), + BasicDataClass(message="C", timestamp=2), + BasicDataClass(message="D", timestamp=3) + ]) # Check pop until an existing timestamp - last_before_2 = ts_buffer.pop_last_before(2) + last_before_2 = data_list.pop_last_before(2) - self.assertEqual(last_before_2, (1, "B")) - self.assertEqual(len(ts_buffer), 2) - self.assertEqual(ts_buffer.first, (2, "C")) + self.assertEqual(last_before_2.message, "B") + self.assertEqual(last_before_2.timestamp, 1) + self.assertEqual(len(data_list), 2) + self.assertEqual(data_list[0].message, "C") + self.assertEqual(data_list[0].timestamp, 2) # Check pop until an none existing timestamp - ts_buffer = DataFeatures.TimeStampedBuffer({0: "A", 1: "B", 2: "C", 3: "D"}) - - first_until_1dot5 = ts_buffer.pop_last_before(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(DataFeatures.TimeStampedBuffer({0: "A", 1: "B"}).last, (1, "B")) - - # Check that accessing to last item of an empty TimeStampedBuffer fails - with self.assertRaises(IndexError): - - DataFeatures.TimeStampedBuffer().last + data_list = DataFeatures.TimestampedObjectsList(BasicDataClass, \ + [ + BasicDataClass(message="A", timestamp=0), + BasicDataClass(message="B", timestamp=1), + BasicDataClass(message="C", timestamp=2), + BasicDataClass(message="D", timestamp=3) + ]) + + first_until_1dot5 = data_list.pop_last_before(1.5) + + self.assertEqual(first_until_1dot5.message, "B") + self.assertEqual(first_until_1dot5.timestamp, 1) + self.assertEqual(len(data_list), 2) + self.assertEqual(data_list[0].message, "C") + self.assertEqual(data_list[0].timestamp, 2) def test_pop_last(self): - """Test TimeStampedBuffer pop_last method.""" + """Test TimestampedObjectsList pop_last method.""" + + data_list = DataFeatures.TimestampedObjectsList(BasicDataClass, \ + [ + BasicDataClass(message="A", timestamp=0), + BasicDataClass(message="B", timestamp=1) + ]) - ts_buffer = DataFeatures.TimeStampedBuffer({0: "A", 1: "B"}) + last = data_list.pop(-1) - self.assertEqual(ts_buffer.pop_last(), (1, "B")) - self.assertEqual(len(ts_buffer), 1) - self.assertEqual(ts_buffer.last, (0, "A")) + self.assertEqual(len(data_list), 1) + self.assertEqual(last.message, "B") + self.assertEqual(last.timestamp, 1) + self.assertEqual(data_list[0].message, "A") + self.assertEqual(data_list[0].timestamp, 0) def test_get_first_from(self): - """Test TimeStampedBuffer get_first_from method.""" + """Test TimestampedObjectsList get_first_from method.""" - ts_buffer = DataFeatures.TimeStampedBuffer({0: "A", 1: "B", 2: "C", 3: "D"}) + data_list = DataFeatures.TimestampedObjectsList(BasicDataClass, \ + [ + BasicDataClass(message="A", timestamp=0), + BasicDataClass(message="B", timestamp=1), + BasicDataClass(message="C", timestamp=2), + BasicDataClass(message="D", timestamp=3) + ]) - get_first_from_1 = ts_buffer.get_first_from(1) + get_first_from_1 = data_list.get_first_from(1) - self.assertEqual(get_first_from_1, (1, "B")) - self.assertEqual(len(ts_buffer), 4) + self.assertEqual(get_first_from_1.message, "B") + self.assertEqual(get_first_from_1.timestamp, 1) + self.assertEqual(len(data_list), 4) - get_first_from_1dot5 = ts_buffer.get_first_from(1.5) + get_first_from_1dot5 = data_list.get_first_from(1.5) - self.assertEqual(get_first_from_1dot5, (2, "C")) + self.assertEqual(get_first_from_1dot5.message, "C") + self.assertEqual(get_first_from_1dot5.timestamp, 2) - get_first_from_0 = ts_buffer.get_first_from(0) + get_first_from_0 = data_list.get_first_from(0) - self.assertEqual(get_first_from_0, (0, "A")) + self.assertEqual(get_first_from_0.message, "A") + self.assertEqual(get_first_from_0.timestamp, 0) # Check that accessing to lately timestamp fails with self.assertRaises(KeyError): - ts_buffer.get_first_from(4) + data_list.get_first_from(4) def test_get_last_before(self): - """Test TimeStampedBuffer get_last_before method.""" + """Test TimestampedObjectsList get_last_before method.""" - ts_buffer = DataFeatures.TimeStampedBuffer({0: "A", 1: "B", 2: "C", 3: "D"}) + data_list = DataFeatures.TimestampedObjectsList(BasicDataClass, \ + [ + BasicDataClass(message="A", timestamp=0), + BasicDataClass(message="B", timestamp=1), + BasicDataClass(message="C", timestamp=2), + BasicDataClass(message="D", timestamp=3) + ]) - get_last_before_2 = ts_buffer.get_last_before(2) + get_last_before_2 = data_list.get_last_before(2) - self.assertEqual(get_last_before_2, (1, "B")) - self.assertEqual(len(ts_buffer), 4) + self.assertEqual(get_last_before_2.message, "B") + self.assertEqual(get_last_before_2.timestamp, 1) + self.assertEqual(len(data_list), 4) - get_last_before_1dot5 = ts_buffer.get_last_before(1.5) + get_last_before_1dot5 = data_list.get_last_before(1.5) - self.assertEqual(get_last_before_1dot5, (1, "B")) + self.assertEqual(get_last_before_1dot5.message,"B") + self.assertEqual(get_last_before_1dot5.timestamp, 1) - get_last_before_4 = ts_buffer.get_last_before(4) + get_last_before_4 = data_list.get_last_before(4) - self.assertEqual(get_last_before_4, (3, "D")) + self.assertEqual(get_last_before_4.message, "D") + self.assertEqual(get_last_before_4.timestamp, 3) # Check that accessing to early timestamp fails with self.assertRaises(KeyError): - ts_buffer.get_last_before(-1) - - + data_list.get_last_before(-1) + def test_get_last_until(self): - """Test TimeStampedBuffer get_last_until method.""" + """Test TimestampedObjectsList get_last_until method.""" - ts_buffer = DataFeatures.TimeStampedBuffer({0: "A", 1: "B", 2: "C", 3: "D"}) + data_list = DataFeatures.TimestampedObjectsList(BasicDataClass, \ + [ + BasicDataClass(message="A", timestamp=0), + BasicDataClass(message="B", timestamp=1), + BasicDataClass(message="C", timestamp=2), + BasicDataClass(message="D", timestamp=3) + ]) - get_last_until_2 = ts_buffer.get_last_until(2) + get_last_until_2 = data_list.get_last_until(2) - self.assertEqual(get_last_until_2, (2, "C")) - self.assertEqual(len(ts_buffer), 4) + self.assertEqual(get_last_until_2.message, "C") + self.assertEqual(get_last_until_2.timestamp, 2) + self.assertEqual(len(data_list), 4) - get_last_until_1dot5 = ts_buffer.get_last_until(1.5) + get_last_until_1dot5 = data_list.get_last_until(1.5) - self.assertEqual(get_last_until_1dot5, (1, "B")) + self.assertEqual(get_last_until_1dot5.message, "B") + self.assertEqual(get_last_until_1dot5.timestamp, 1) - get_last_until_4 = ts_buffer.get_last_until(4) + get_last_until_4 = data_list.get_last_until(4) - self.assertEqual(get_last_until_4, (3, "D")) + self.assertEqual(get_last_until_4.message, "D") + self.assertEqual(get_last_until_4.timestamp, 3) # 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.""" - - 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") - - self.assertEqual(ts_buffer_dataframe.index.dtype, 'float64') - self.assertEqual(ts_buffer_dataframe["data_A"].dtype, 'object') - self.assertEqual(ts_buffer_dataframe["data_B"].dtype, 'object') - self.assertEqual(ts_buffer_dataframe["data_C"].dtype, 'object') - - # 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") - - # Check dataframe conversion with dataclass - data = BasicDataClass((123, 456)) - ts_buffer_dataframe = DataFeatures.TimeStampedBuffer({0: data}).as_dataframe() - - self.assertEqual(ts_buffer_dataframe.index.name, "timestamp") - self.assertEqual(ts_buffer_dataframe.index.size, 1) - - self.assertEqual(ts_buffer_dataframe.columns.size, 1) - self.assertEqual(ts_buffer_dataframe.columns[0], "value") - + data_list.get_last_until(-1) if __name__ == '__main__': |