diff options
Diffstat (limited to 'src')
| -rw-r--r-- | src/argaze.test/DataFeatures.py | 296 | ||||
| -rw-r--r-- | src/argaze.test/DataLog/FileWriter.py | 37 | ||||
| -rw-r--r-- | src/argaze.test/DataLog/__init__.py | 0 | ||||
| -rw-r--r-- | src/argaze.test/utils/ts_buffer.json | 1 | ||||
| -rw-r--r-- | src/argaze.test/utils/ts_data_file.json | 82 | ||||
| -rw-r--r-- | src/argaze/DataFeatures.py | 47 |
6 files changed, 282 insertions, 181 deletions
diff --git a/src/argaze.test/DataFeatures.py b/src/argaze.test/DataFeatures.py index 4d9c909..9543915 100644 --- a/src/argaze.test/DataFeatures.py +++ b/src/argaze.test/DataFeatures.py @@ -19,16 +19,21 @@ import numpy class BasicDataClass(DataFeatures.TimestampedObject): """Define a basic dataclass for testing purpose.""" - def __init__(self, value: tuple = (), **kwargs): + 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. @@ -42,7 +47,7 @@ def random_data_list(size): for i in range(0, size): # Edit data - random_data = BasicDataClass((i, random.random())) + random_data = BasicDataClass((i, random.random()), f'test_{i}') # Timestamp data random_data.timestamp = time.time() @@ -54,6 +59,14 @@ def random_data_list(size): return DataFeatures.TimestampedObjectsList(BasicDataClass, data_list) +# 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 TestTimestampedObjectsListClass(unittest.TestCase): """Test TimestampedObjectsList class.""" @@ -83,23 +96,24 @@ class TestTimestampedObjectsListClass(unittest.TestCase): # Check dataframe conversion self.assertEqual(data_frame.index.name, "timestamp") self.assertEqual(data_frame.index.size, 10) - self.assertEqual(data_frame.columns.size, 1) + 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 data exclusion option data_frame = random_data_list(10).as_dataframe(exclude=["value"]) self.assertEqual(data_frame.index.name, "timestamp") self.assertEqual(data_frame.index.size, 10) - self.assertEqual(data_frame.columns.size, 0) + 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, 2) + self.assertEqual(data_frame.columns.size, 3) self.assertEqual(data_frame["value_0"].dtype, 'int64') self.assertEqual(data_frame["value_1"].dtype, 'float64') @@ -112,210 +126,250 @@ class TestTimestampedObjectsListClass(unittest.TestCase): data_list = DataFeatures.TimestampedObjectsList.from_dataframe(BasicDataClass, data_frame) self.assertEqual(len(data_list), 10) - @unittest.skip("DEBUG") + def test_from_json(self): """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 TimestampedObjectsList from json file - ts_buffer = DataFeatures.TimestampedObjectsList.from_json(json_filepath) + data_list = DataFeatures.TimestampedObjectsList.from_json(BasicDataClass, json_filepath) - self.assertEqual(len(ts_buffer), 3) - @unittest.skip("DEBUG") + self.assertEqual(len(data_list), 10) + self.assertEqual(type(data_list[0]), BasicDataClass) + def test___repr__(self): """Test TimestampedObjectsList string representation.""" - self.assertEqual(repr(DataFeatures.TimestampedObjectsList()), "{}") - self.assertEqual(repr(DataFeatures.TimestampedObjectsList({0: ""})), "{\"0\": \"\"}") - self.assertEqual(repr(DataFeatures.TimestampedObjectsList({0.1: ""})), "{\"0.1\": \"\"}") - - data = BasicDataClass((123, 456)) - ts_buffer = DataFeatures.TimestampedObjectsList({0: data}) + self.assertEqual(repr(DataFeatures.TimestampedObjectsList(BasicDataClass)), "[]") - self.assertEqual(repr(ts_buffer), "{\"0\": {\"value\": [123, 456]}}") + data_list = [BasicDataClass((0, 0), 'test', timestamp=0)] + self.assertEqual(repr(DataFeatures.TimestampedObjectsList(BasicDataClass, data_list)), "[{\"value\": [0, 0], \"message\": \"test\", \"timestamp\": 0}]") - array = numpy.zeros(3) - ts_buffer = DataFeatures.TimestampedObjectsList({0: array}) - - self.assertEqual(repr(ts_buffer), "{\"0\": [0.0, 0.0, 0.0]}") - @unittest.skip("DEBUG") def test___str__(self): """Test TimestampedObjectsList string representation.""" - self.assertEqual(str(DataFeatures.TimestampedObjectsList()), "{}") - self.assertEqual(str(DataFeatures.TimestampedObjectsList({0: ""})), "{\"0\": \"\"}") - self.assertEqual(str(DataFeatures.TimestampedObjectsList({0.1: ""})), "{\"0.1\": \"\"}") - - data = BasicDataClass((123, 456)) - ts_buffer = DataFeatures.TimestampedObjectsList({0: data}) - - self.assertEqual(str(ts_buffer), "{\"0\": {\"value\": [123, 456]}}") + self.assertEqual(str(DataFeatures.TimestampedObjectsList(BasicDataClass)), "[]") - array = numpy.zeros(3) - ts_buffer = DataFeatures.TimestampedObjectsList({0: array}) + data_list = [BasicDataClass((0, 0), 'test', timestamp=0)] + self.assertEqual(str(DataFeatures.TimestampedObjectsList(BasicDataClass, data_list)), "[{\"value\": [0, 0], \"message\": \"test\", \"timestamp\": 0}]") - self.assertEqual(str(ts_buffer), "{\"0\": [0.0, 0.0, 0.0]}") - @unittest.skip("DEBUG") def test_append(self): """Test TimestampedObjectsList append method.""" - ts_buffer = DataFeatures.TimestampedObjectsList({0: "A", 1: "B"}) - ts_buffer_next = DataFeatures.TimestampedObjectsList({2: "C", 3: "D"}) + data_list = DataFeatures.TimestampedObjectsList(BasicDataClass) + next_data = BasicDataClass((0, 0), 'test', timestamp=0) - self.assertEqual(len(ts_buffer.append(ts_buffer_next)), 4) - self.assertEqual(list(ts_buffer.append(ts_buffer_next).keys()), [0, 1, 2, 3]) - @unittest.skip("DEBUG") - def test_first(self): - """Test TimestampedObjectsList first property.""" + data_list.append(next_data) - self.assertEqual(DataFeatures.TimestampedObjectsList({0: "A", 1: "B"}).first, (0, "A")) - - # Check that accessing to first item of an empty TimestampedObjectsList fails - with self.assertRaises(IndexError): - - DataFeatures.TimestampedObjectsList().first - @unittest.skip("DEBUG") + self.assertEqual(len(data_list), 1) + self.assertEqual(data_list.timestamps(), [0]) + def test_pop_first(self): """Test TimestampedObjectsList pop_first method.""" - ts_buffer = DataFeatures.TimestampedObjectsList({0: "A", 1: "B"}) + data_list = DataFeatures.TimestampedObjectsList(BasicDataClass, \ + [ + BasicDataClass(message="A", timestamp=0), + BasicDataClass(message="B", timestamp=1) + ]) + + first = data_list.pop(0) + + 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")) - @unittest.skip("DEBUG") def test_pop_last_until(self): """Test TimestampedObjectsList pop_last_until method.""" - ts_buffer = DataFeatures.TimestampedObjectsList({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.TimestampedObjectsList({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")) - @unittest.skip("DEBUG") def test_pop_last_before(self): """Test TimestampedObjectsList pop_last_before method.""" - ts_buffer = DataFeatures.TimestampedObjectsList({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.TimestampedObjectsList({0: "A", 1: "B", 2: "C", 3: "D"}) - - first_until_1dot5 = ts_buffer.pop_last_before(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) + ]) + + 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) - self.assertEqual(first_until_1dot5, (1, "B")) - self.assertEqual(len(ts_buffer), 2) - self.assertEqual(ts_buffer.first, (2, "C")) - @unittest.skip("DEBUG") - def test_last(self): - """Test TimestampedObjectsList last property.""" - - self.assertEqual(DataFeatures.TimestampedObjectsList({0: "A", 1: "B"}).last, (1, "B")) - - # Check that accessing to last item of an empty TimestampedObjectsList fails - with self.assertRaises(IndexError): - - DataFeatures.TimestampedObjectsList().last - @unittest.skip("DEBUG") def test_pop_last(self): """Test TimestampedObjectsList pop_last method.""" - ts_buffer = DataFeatures.TimestampedObjectsList({0: "A", 1: "B"}) + data_list = DataFeatures.TimestampedObjectsList(BasicDataClass, \ + [ + BasicDataClass(message="A", timestamp=0), + BasicDataClass(message="B", timestamp=1) + ]) + + last = data_list.pop(-1) + + 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) - self.assertEqual(ts_buffer.pop_last(), (1, "B")) - self.assertEqual(len(ts_buffer), 1) - self.assertEqual(ts_buffer.last, (0, "A")) - @unittest.skip("DEBUG") def test_get_first_from(self): """Test TimestampedObjectsList get_first_from method.""" - ts_buffer = DataFeatures.TimestampedObjectsList({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) - @unittest.skip("DEBUG") + data_list.get_first_from(4) + def test_get_last_before(self): """Test TimestampedObjectsList get_last_before method.""" - ts_buffer = DataFeatures.TimestampedObjectsList({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) - @unittest.skip("DEBUG") + data_list.get_last_before(-1) + def test_get_last_until(self): """Test TimestampedObjectsList get_last_until method.""" - ts_buffer = DataFeatures.TimestampedObjectsList({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) + data_list.get_last_until(-1) if __name__ == '__main__': diff --git a/src/argaze.test/DataLog/FileWriter.py b/src/argaze.test/DataLog/FileWriter.py deleted file mode 100644 index 648385c..0000000 --- a/src/argaze.test/DataLog/FileWriter.py +++ /dev/null @@ -1,37 +0,0 @@ -#!/usr/bin/env python - -""" """ - -__author__ = "Théo de la Hogue" -__credits__ = [] -__copyright__ = "Copyright 2023, Ecole Nationale de l'Aviation Civile (ENAC)" -__license__ = "BSD" - -import unittest -import os - -from argaze import DataFeatures -from argaze.DataLog import FileWriter -from argaze.utils import UtilsFeatures - -DataFeaturesTest = UtilsFeatures.importFromTestPackage('DataFeatures') - -class TestTimeStampedDataLogger(unittest.TestCase): - """Test DataLogger class.""" - - def test_creation(self): - """Test logger creation.""" - - file_writer = FileWriter.TimeStampedDataLogger(path='./_export/logs/data.txt', separator=',') - - # Check file creation - self.assertEqual(os.path.exists('./_export/logs/data.txt'), True) - - # Write into file - file_writer.emit(0, 'A') - file_writer.emit(1, 'B') - file_writer.emit(2, 'C') - -if __name__ == '__main__': - - unittest.main()
\ No newline at end of file diff --git a/src/argaze.test/DataLog/__init__.py b/src/argaze.test/DataLog/__init__.py deleted file mode 100644 index e69de29..0000000 --- a/src/argaze.test/DataLog/__init__.py +++ /dev/null diff --git a/src/argaze.test/utils/ts_buffer.json b/src/argaze.test/utils/ts_buffer.json deleted file mode 100644 index b1433de..0000000 --- a/src/argaze.test/utils/ts_buffer.json +++ /dev/null @@ -1 +0,0 @@ -{"0":{"timestamp":0,"int_value":1,"float_value":0.000001,"string_value":"A","list_value":[[0,0],[0,1],[1,1],[1,0]],"tuple_value":[0,0],"nan_value":1.0,"json_value":{"0":"A","1":"B","2":"C","3":"D"}},"1":{"timestamp":1,"int_value":2,"float_value":0.000002,"string_value":"B","list_value":[[0,0],[0,2],[2,2],[2,0]],"tuple_value":[1,1],"nan_value":null,"json_value":{"0":"A","1":"B","2":"C","3":"D"}},"2":{"timestamp":2,"int_value":3,"float_value":0.000003,"string_value":"C","list_value":[[0,0],[0,3],[3,3],[3,0]],"tuple_value":[2,2],"nan_value":1.0,"json_value":{"0":"A","1":"B","2":"C","3":"D"}}}
\ No newline at end of file diff --git a/src/argaze.test/utils/ts_data_file.json b/src/argaze.test/utils/ts_data_file.json new file mode 100644 index 0000000..d69dd77 --- /dev/null +++ b/src/argaze.test/utils/ts_data_file.json @@ -0,0 +1,82 @@ +[ + { + "value": [ + 0, + 0.04245991513702008 + ], + "message": "test_0", + "timestamp": 1709160858.387703 + }, + { + "value": [ + 1, + 0.6761490271896192 + ], + "message": "test_1", + "timestamp": 1709160858.38784 + }, + { + "value": [ + 2, + 0.046407850274610474 + ], + "message": "test_2", + "timestamp": 1709160858.387973 + }, + { + "value": [ + 3, + 0.9378067398496651 + ], + "message": "test_3", + "timestamp": 1709160858.388105 + }, + { + "value": [ + 4, + 0.4197936347606107 + ], + "message": "test_4", + "timestamp": 1709160858.388236 + }, + { + "value": [ + 5, + 0.26937423401632943 + ], + "message": "test_5", + "timestamp": 1709160858.3883672 + }, + { + "value": [ + 6, + 0.9478731386524537 + ], + "message": "test_6", + "timestamp": 1709160858.3884978 + }, + { + "value": [ + 7, + 0.39010865778889914 + ], + "message": "test_7", + "timestamp": 1709160858.388629 + }, + { + "value": [ + 8, + 0.4100480317631575 + ], + "message": "test_8", + "timestamp": 1709160858.388763 + }, + { + "value": [ + 9, + 0.5900791904864906 + ], + "message": "test_9", + "timestamp": 1709160858.388895 + } +]
\ No newline at end of file diff --git a/src/argaze/DataFeatures.py b/src/argaze/DataFeatures.py index b8acb2e..6909222 100644 --- a/src/argaze/DataFeatures.py +++ b/src/argaze/DataFeatures.py @@ -285,6 +285,23 @@ class TimestampedObjectsList(list): return df + @classmethod + def from_json(self, ts_object_type: type, json_filepath: str) -> TimestampedObjectsListType: + """Create a TimestampedObjectsList from .json file.""" + + with open(json_filepath, encoding='utf-8') as ts_objects_file: + + json_ts_objects = json.load(ts_objects_file) + + return TimestampedObjectsList(ts_object_type, [ts_object_type(**ts_object_dict) for ts_object_dict in json_ts_objects]) + + def to_json(self, json_filepath: str): + """Save a TimestampedObjectsList to .json file.""" + + with open(json_filepath, 'w', encoding='utf-8') as ts_objects_file: + + json.dump(self, ts_objects_file, ensure_ascii=False, default=(lambda obj: as_dict(obj)), indent=' ') + def __repr__(self): """String representation""" return json.dumps([as_dict(ts_object) for ts_object in self], ensure_ascii=False,) @@ -322,7 +339,8 @@ class TimestampedObjectsList(list): def get_first_from(self, timestamp: TimeStampType) -> TimestampedObjectType: """Retreive first item timestamp from a given timestamp value.""" - first_from_index = bisect.bisect_left(self.timestamps(), timestamp) + ts_list = self.timestamps() + first_from_index = bisect.bisect_left(ts_list, timestamp) if first_from_index < len(self): @@ -335,7 +353,8 @@ class TimestampedObjectsList(list): def get_last_before(self, timestamp: TimeStampType) -> TimestampedObjectType: """Retreive last item timestamp before a given timestamp value.""" - last_before_index = bisect.bisect_left(self.timestamps(), timestamp) - 1 + ts_list = self.timestamps() + last_before_index = bisect.bisect_left(ts_list, timestamp) - 1 if last_before_index >= 0: @@ -343,12 +362,13 @@ class TimestampedObjectsList(list): else: - raise KeyError(f'No data stored before {ts} timestamp.') + raise KeyError(f'No data stored before {timestamp} timestamp.') def get_last_until(self, timestamp: TimeStampType) -> TimestampedObjectType: """Retreive last item timestamp until a given timestamp value.""" - last_until_index = bisect.bisect_right(self.timestamps(), timestamp) - 1 + ts_list = self.timestamps() + last_until_index = bisect.bisect_right(ts_list, timestamp) - 1 if last_until_index >= 0: @@ -356,24 +376,7 @@ class TimestampedObjectsList(list): else: - raise KeyError(f'No data stored until {ts} timestamp.') - - @classmethod - def from_json(self, json_filepath: str) -> TimestampedObjectsListType: - """Create a TimestampedObjectsList from .json file.""" - - with open(json_filepath, encoding='utf-8') as ts_objects_file: - - json_ts_objects = json.load(ts_objects_file) - - return TimestampedObjectsList([ast.literal_eval(ts_object) for ts_object in json_ts_objects]) - - def to_json(self, json_filepath: str): - """Save a TimeStampedBuffer to .json file.""" - - with open(json_filepath, 'w', encoding='utf-8') as ts_buffer_file: - - json.dump(self, ts_buffer_file, ensure_ascii=False, cls=JsonEncoder) + raise KeyError(f'No data stored until {timestamp} timestamp.') def plot(self, names=[], colors=[], split={}, samples=None) -> list: """Plot as [matplotlib](https://matplotlib.org/) time chart.""" |