diff options
| -rw-r--r-- | src/argaze.test/DataFeatures.py | 278 | ||||
| -rw-r--r-- | src/argaze/DataFeatures.py | 222 | ||||
| -rw-r--r-- | src/argaze/GazeFeatures.py | 10 |
3 files changed, 244 insertions, 266 deletions
diff --git a/src/argaze.test/DataFeatures.py b/src/argaze.test/DataFeatures.py index b30c560..4d9c909 100644 --- a/src/argaze.test/DataFeatures.py +++ b/src/argaze.test/DataFeatures.py @@ -16,154 +16,181 @@ 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 = (), **kwargs): + + DataFeatures.TimestampedObject.__init__(self, **kwargs) + + self.__value = value + + @property + def value(self): + return self.__value + +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())) + + # 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 - -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 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) - - # 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]) - - # 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 that TimeStampedBuffer creation fails when keys are not numbers + # 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(AssertionError): - DataFeatures.TimeStampedBuffer({"first": ""}) + data_list = [BasicDataClass((0, 0))] + DataFeatures.TimestampedObjectsList(BasicDataClass, data_list) + + def test_as_dataframe(self): + """Test TimestampedObjectsList as_dataframe method.""" + + data_frame = random_data_list(10).as_dataframe() + + # 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.index.dtype, 'float64') + self.assertEqual(data_frame["value"].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) + # 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["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) + @unittest.skip("DEBUG") 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') - # Load TimeStampedBuffer from json file - ts_buffer = DataFeatures.TimeStampedBuffer.from_json(json_filepath) + # Load TimestampedObjectsList from json file + ts_buffer = DataFeatures.TimestampedObjectsList.from_json(json_filepath) self.assertEqual(len(ts_buffer), 3) - + @unittest.skip("DEBUG") 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()), "{}") + self.assertEqual(repr(DataFeatures.TimestampedObjectsList({0: ""})), "{\"0\": \"\"}") + self.assertEqual(repr(DataFeatures.TimestampedObjectsList({0.1: ""})), "{\"0.1\": \"\"}") data = BasicDataClass((123, 456)) - ts_buffer = DataFeatures.TimeStampedBuffer({0: data}) + ts_buffer = DataFeatures.TimestampedObjectsList({0: data}) self.assertEqual(repr(ts_buffer), "{\"0\": {\"value\": [123, 456]}}") array = numpy.zeros(3) - ts_buffer = DataFeatures.TimeStampedBuffer({0: array}) + 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 TimeStampedBuffer string representation.""" + """Test TimestampedObjectsList string representation.""" - self.assertEqual(str(DataFeatures.TimeStampedBuffer()), "{}") - self.assertEqual(str(DataFeatures.TimeStampedBuffer({0: ""})), "{\"0\": \"\"}") - self.assertEqual(str(DataFeatures.TimeStampedBuffer({0.1: ""})), "{\"0.1\": \"\"}") + 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.TimeStampedBuffer({0: data}) + ts_buffer = DataFeatures.TimestampedObjectsList({0: data}) self.assertEqual(str(ts_buffer), "{\"0\": {\"value\": [123, 456]}}") array = numpy.zeros(3) - ts_buffer = DataFeatures.TimeStampedBuffer({0: array}) + ts_buffer = DataFeatures.TimestampedObjectsList({0: array}) self.assertEqual(str(ts_buffer), "{\"0\": [0.0, 0.0, 0.0]}") - + @unittest.skip("DEBUG") def test_append(self): - """Test TimeStampedBuffer append method.""" + """Test TimestampedObjectsList append method.""" - ts_buffer = DataFeatures.TimeStampedBuffer({0: "A", 1: "B"}) - ts_buffer_next = DataFeatures.TimeStampedBuffer({2: "C", 3: "D"}) + ts_buffer = DataFeatures.TimestampedObjectsList({0: "A", 1: "B"}) + ts_buffer_next = DataFeatures.TimestampedObjectsList({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]) - + @unittest.skip("DEBUG") def test_first(self): - """Test TimeStampedBuffer first property.""" + """Test TimestampedObjectsList first property.""" - self.assertEqual(DataFeatures.TimeStampedBuffer({0: "A", 1: "B"}).first, (0, "A")) + self.assertEqual(DataFeatures.TimestampedObjectsList({0: "A", 1: "B"}).first, (0, "A")) - # Check that accessing to first item of an empty TimeStampedBuffer fails + # Check that accessing to first item of an empty TimestampedObjectsList fails with self.assertRaises(IndexError): - DataFeatures.TimeStampedBuffer().first - + DataFeatures.TimestampedObjectsList().first + @unittest.skip("DEBUG") def test_pop_first(self): - """Test TimeStampedBuffer pop_first method.""" + """Test TimestampedObjectsList pop_first method.""" - ts_buffer = DataFeatures.TimeStampedBuffer({0: "A", 1: "B"}) + ts_buffer = DataFeatures.TimestampedObjectsList({0: "A", 1: "B"}) 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 TimeStampedBuffer pop_last_until method.""" + """Test TimestampedObjectsList pop_last_until method.""" - ts_buffer = DataFeatures.TimeStampedBuffer({0: "A", 1: "B", 2: "C", 3: "D"}) + ts_buffer = DataFeatures.TimestampedObjectsList({0: "A", 1: "B", 2: "C", 3: "D"}) # Check pop until an existing timestamp pop_last_until_2 = ts_buffer.pop_last_until(2) @@ -173,18 +200,18 @@ class TestTimeStampedBufferClass(unittest.TestCase): self.assertEqual(ts_buffer.first, (2, "C")) # Check first until an none existing timestamp - ts_buffer = DataFeatures.TimeStampedBuffer({0: "A", 1: "B", 2: "C", 3: "D"}) + ts_buffer = DataFeatures.TimestampedObjectsList({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")) - + @unittest.skip("DEBUG") 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"}) + ts_buffer = DataFeatures.TimestampedObjectsList({0: "A", 1: "B", 2: "C", 3: "D"}) # Check pop until an existing timestamp last_before_2 = ts_buffer.pop_last_before(2) @@ -194,37 +221,37 @@ class TestTimeStampedBufferClass(unittest.TestCase): self.assertEqual(ts_buffer.first, (2, "C")) # Check pop until an none existing timestamp - ts_buffer = DataFeatures.TimeStampedBuffer({0: "A", 1: "B", 2: "C", 3: "D"}) + ts_buffer = DataFeatures.TimestampedObjectsList({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")) - + @unittest.skip("DEBUG") def test_last(self): - """Test TimeStampedBuffer last property.""" + """Test TimestampedObjectsList last property.""" - self.assertEqual(DataFeatures.TimeStampedBuffer({0: "A", 1: "B"}).last, (1, "B")) + self.assertEqual(DataFeatures.TimestampedObjectsList({0: "A", 1: "B"}).last, (1, "B")) - # Check that accessing to last item of an empty TimeStampedBuffer fails + # Check that accessing to last item of an empty TimestampedObjectsList fails with self.assertRaises(IndexError): - DataFeatures.TimeStampedBuffer().last - + DataFeatures.TimestampedObjectsList().last + @unittest.skip("DEBUG") def test_pop_last(self): - """Test TimeStampedBuffer pop_last method.""" + """Test TimestampedObjectsList pop_last method.""" - ts_buffer = DataFeatures.TimeStampedBuffer({0: "A", 1: "B"}) + ts_buffer = DataFeatures.TimestampedObjectsList({0: "A", 1: "B"}) 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 TimeStampedBuffer get_first_from method.""" + """Test TimestampedObjectsList get_first_from method.""" - ts_buffer = DataFeatures.TimeStampedBuffer({0: "A", 1: "B", 2: "C", 3: "D"}) + ts_buffer = DataFeatures.TimestampedObjectsList({0: "A", 1: "B", 2: "C", 3: "D"}) get_first_from_1 = ts_buffer.get_first_from(1) @@ -243,11 +270,11 @@ class TestTimeStampedBufferClass(unittest.TestCase): with self.assertRaises(KeyError): ts_buffer.get_first_from(4) - + @unittest.skip("DEBUG") 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"}) + ts_buffer = DataFeatures.TimestampedObjectsList({0: "A", 1: "B", 2: "C", 3: "D"}) get_last_before_2 = ts_buffer.get_last_before(2) @@ -265,13 +292,12 @@ class TestTimeStampedBufferClass(unittest.TestCase): # Check that accessing to early timestamp fails with self.assertRaises(KeyError): - ts_buffer.get_last_before(-1) - - + ts_buffer.get_last_before(-1) + @unittest.skip("DEBUG") 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"}) + ts_buffer = DataFeatures.TimestampedObjectsList({0: "A", 1: "B", 2: "C", 3: "D"}) get_last_until_2 = ts_buffer.get_last_until(2) @@ -291,60 +317,6 @@ class TestTimeStampedBufferClass(unittest.TestCase): 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") - - if __name__ == '__main__': unittest.main()
\ No newline at end of file diff --git a/src/argaze/DataFeatures.py b/src/argaze/DataFeatures.py index ce3ce52..b8acb2e 100644 --- a/src/argaze/DataFeatures.py +++ b/src/argaze/DataFeatures.py @@ -30,10 +30,10 @@ from colorama import Style, Fore TimeStampType = TypeVar('TimeStamp', int, float) """Type definition for timestamp as integer or float values.""" -TimeStampedObjectType = TypeVar('TimeStampedObject', bound="TimeStampedObject") +TimestampedObjectType = TypeVar('TimestampedObject', bound="TimestampedObject") # Type definition for type annotation convenience -TimeStampedObjectsListType = TypeVar('TimeStampedObjectsList', bound="TimeStampedObjectsList") +TimestampedObjectsListType = TypeVar('TimestampedObjectsList', bound="TimestampedObjectsList") # Type definition for type annotation convenience def module_path(obj) -> str: @@ -123,7 +123,48 @@ class JsonEncoder(json.JSONEncoder): return public_dict -class TimeStampedObjectsList(list): +class DataDictionary(dict): + """Enable dot.notation access to dictionary attributes""" + + __getattr__ = dict.get + __setattr__ = dict.__setitem__ + __delattr__ = dict.__delitem__ + +class TimestampedObject(): + """Abstract class to enable timestamp management.""" + + def __init__(self, timestamp: int|float = math.nan): + """Initialize TimestampedObject.""" + self._timestamp = timestamp + + def __repr__(self): + """String representation.""" + return json.dumps(as_dict(self)) + + @property + def timestamp(self) -> int|float: + """Get object timestamp.""" + return self._timestamp + + @timestamp.setter + def timestamp(self, timestamp: int|float): + """Set object timestamp.""" + + assert(type(timestamp) == int or type(timestamp) == float) + + self._timestamp = timestamp + + def untimestamp(self): + """Reset object timestamp.""" + self.timestamp = math.nan + + def is_timestamped(self) -> bool: + """Is the object timestamped?""" + timestamped = not math.isnan(self.timestamp) + + return timestamped + +class TimestampedObjectsList(list): """Handle timestamped object into a list. !!! warning "Timestamped objects are not sorted internally" @@ -145,7 +186,7 @@ class TimeStampedObjectsList(list): """Get object type handled by the list.""" return self.__object_type - def append(self, ts_object: TimeStampedObjectType|dict): + def append(self, ts_object: TimestampedObjectType|dict): """Append timestamped object.""" # Convert dict into GazePosition @@ -183,6 +224,67 @@ class TimeStampedObjectsList(list): """Get all timestamped objects as list of tuple.""" return [tuple(as_dict(ts_object, filter=False).values()) for ts_object in self] + @classmethod + def from_dataframe(self, ts_object_type: type, dataframe: pandas.DataFrame, exclude=[]) -> TimestampedObjectsListType: + """Create a TimestampedObjectsList from [Pandas DataFrame](https://pandas.pydata.org/docs/reference/api/pandas.DataFrame.html).""" + + dataframe.drop(exclude, inplace=True, axis=True) + + assert(dataframe.index.name == 'timestamp') + + object_list = [ts_object_type(timestamp=timestamp, **object_dict) for timestamp, object_dict in dataframe.to_dict('index').items()] + + return TimestampedObjectsList(ts_object_type, object_list) + + def as_dataframe(self, exclude=[], split={}) -> pandas.DataFrame: + """Convert as [Pandas DataFrame](https://pandas.pydata.org/docs/reference/api/pandas.DataFrame.html). + + The optional *split* argument allows tuple values to be stored in dedicated columns. + For example: to convert {"point": (0, 0)} data as two separated "x" and "y" columns, use split={"point": ["x", "y"]} + + !!! warning "Values must be dictionaries" + + Each key is stored as a column name. + + !!! note + + Timestamps are stored as index column called 'timestamp'. + """ + + df = pandas.DataFrame(self.tuples(), columns=self.__object_properties) + + # Exclude columns + df.drop(exclude, inplace=True, axis=True) + + # Split columns + if len(split) > 0: + + splited_columns = [] + + for column in df.columns: + + if column in split.keys(): + + df[split[column]] = pandas.DataFrame(df[column].tolist(), index=df.index) + df.drop(column, inplace=True, axis=True) + + for new_column in split[column]: + + splited_columns.append(new_column) + + else: + + splited_columns.append(column) + + # Reorder splited columns + df = df[splited_columns] + + # Append timestamps as index column + df['timestamp'] = self.timestamps() + df.set_index('timestamp', inplace=True) + + return df + def __repr__(self): """String representation""" return json.dumps([as_dict(ts_object) for ts_object in self], ensure_ascii=False,) @@ -191,7 +293,7 @@ class TimeStampedObjectsList(list): """String representation""" return json.dumps([as_dict(ts_object) for ts_object in self], ensure_ascii=False,) - def pop_last_until(self, timestamp: TimeStampType) -> TimeStampedObjectType: + def pop_last_until(self, timestamp: TimeStampType) -> TimestampedObjectType: """Pop all item until a given timestamped value and return the first after.""" # get last item before given timestamp @@ -203,7 +305,7 @@ class TimeStampedObjectsList(list): return self[0] - def pop_last_before(self, timestamp: TimeStampType) -> TimeStampedObjectType: + def pop_last_before(self, timestamp: TimeStampType) -> TimestampedObjectType: """Pop all item before a given timestamped value and return the last one.""" # get last item before given timestamp @@ -217,7 +319,7 @@ class TimeStampedObjectsList(list): return poped_value - def get_first_from(self, timestamp: TimeStampType) -> TimeStampedObjectType: + 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) @@ -230,7 +332,7 @@ class TimeStampedObjectsList(list): raise KeyError(f'No data stored after {timestamp} timestamp.') - def get_last_before(self, timestamp: TimeStampType) -> TimeStampedObjectType: + 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 @@ -243,7 +345,7 @@ class TimeStampedObjectsList(list): raise KeyError(f'No data stored before {ts} timestamp.') - def get_last_until(self, timestamp: TimeStampType) -> TimeStampedObjectType: + 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 @@ -257,14 +359,14 @@ class TimeStampedObjectsList(list): raise KeyError(f'No data stored until {ts} timestamp.') @classmethod - def from_json(self, json_filepath: str) -> TimeStampedObjectsListType: - """Create a TimeStampedObjectsList from .json file.""" + 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]) + 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.""" @@ -273,65 +375,6 @@ class TimeStampedObjectsList(list): json.dump(self, ts_buffer_file, ensure_ascii=False, cls=JsonEncoder) - @classmethod - def from_dataframe(self, dataframe: pandas.DataFrame, exclude=[]) -> TimeStampedObjectsListType: - """Create a TimeStampedObjectsList from [Pandas DataFrame](https://pandas.pydata.org/docs/reference/api/pandas.DataFrame.html).""" - - dataframe.drop(exclude, inplace=True, axis=True) - - assert(dataframe.index.name == 'timestamp') - - return TimeStampedObjectsList(dataframe.to_dict('index')) - - def as_dataframe(self, exclude=[], split={}) -> pandas.DataFrame: - """Convert as [Pandas DataFrame](https://pandas.pydata.org/docs/reference/api/pandas.DataFrame.html). - - The optional *split* argument allows tuple values to be stored in dedicated columns. - For example: to convert {"point": (0, 0)} data as two separated "x" and "y" columns, use split={"point": ["x", "y"]} - - !!! warning "Values must be dictionaries" - - Each key is stored as a column name. - - !!! note - - Timestamps are stored as index column called 'timestamp'. - """ - - df = pandas.DataFrame(self.tuples(), columns=self.__object_properties) - - # Exclude columns - df.drop(exclude, inplace=True, axis=True) - - # Split columns - if len(split) > 0: - - splited_columns = [] - - for column in df.columns: - - if column in split.keys(): - - df[split[column]] = pandas.DataFrame(df[column].tolist(), index=df.index) - df.drop(column, inplace=True, axis=True) - - for new_column in split[column]: - - splited_columns.append(new_column) - - else: - - splited_columns.append(column) - - # Reorder splited columns - df = df[splited_columns] - - # Append timestamps as index column - df['timestamp'] = self.timestamps() - df.set_index('timestamp', inplace=True) - - return df - def plot(self, names=[], colors=[], split={}, samples=None) -> list: """Plot as [matplotlib](https://matplotlib.org/) time chart.""" @@ -357,43 +400,6 @@ class TimeStampedObjectsList(list): return legend_patches -class DataDictionary(dict): - """Enable dot.notation access to dictionary attributes""" - - __getattr__ = dict.get - __setattr__ = dict.__setitem__ - __delattr__ = dict.__delitem__ - -class TimestampedObject(): - """Abstract class to enable timestamp management.""" - - def __init__(self, timestamp: int|float = math.nan): - - self._timestamp = timestamp - - @property - def timestamp(self) -> int|float: - """Get object timestamp.""" - return self._timestamp - - @timestamp.setter - def timestamp(self, timestamp: int|float): - """Set object timestamp.""" - - assert(type(timestamp) == int or type(timestamp) == float) - - self._timestamp = timestamp - - def untimestamp(self): - """Reset object timestamp.""" - self.timestamp = math.nan - - def is_timestamped(self) -> bool: - """Is the object timestamped?""" - timestamped = not math.isnan(self.timestamp) - - return timestamped - class SharedObject(TimestampedObject): """Abstract class to enable multiple threads sharing and timestamp management.""" diff --git a/src/argaze/GazeFeatures.py b/src/argaze/GazeFeatures.py index 2f83703..fb3dceb 100644 --- a/src/argaze/GazeFeatures.py +++ b/src/argaze/GazeFeatures.py @@ -177,12 +177,12 @@ class GazePosition(tuple, DataFeatures.TimestampedObject): TimeStampedGazePositionsType = TypeVar('TimeStampedGazePositions', bound="TimeStampedGazePositions") # Type definition for type annotation convenience -class TimeStampedGazePositions(DataFeatures.TimeStampedObjectsList): +class TimeStampedGazePositions(DataFeatures.TimestampedObjectsList): """Handle timestamped gaze positions into a list""" def __init__(self, gaze_positions: list = []): - DataFeatures.TimeStampedObjectsList.__init__(self, GazePosition, gaze_positions) + DataFeatures.TimestampedObjectsList.__init__(self, GazePosition, gaze_positions) def values(self) -> list: """Get all timestamped position values as list of tuple.""" @@ -510,12 +510,12 @@ def is_saccade(gaze_movement): TimeStampedGazeMovementsType = TypeVar('TimeStampedGazeMovements', bound="TimeStampedGazeMovements") # Type definition for type annotation convenience -class TimeStampedGazeMovements(DataFeatures.TimeStampedObjectsList): +class TimeStampedGazeMovements(DataFeatures.TimestampedObjectsList): """Handle timestamped gaze movements into a list""" def __init__(self, gaze_movements: list = []): - DataFeatures.TimeStampedObjectsList.__init__(self, GazeMovement, gaze_movements) + DataFeatures.TimestampedObjectsList.__init__(self, GazeMovement, gaze_movements) GazeStatusType = TypeVar('GazeStatus', bound="GazeStatus") # Type definition for type annotation convenience @@ -539,7 +539,7 @@ class GazeStatus(GazePosition): TimeStampedGazeStatusType = TypeVar('TimeStampedGazeStatus', bound="TimeStampedGazeStatus") # Type definition for type annotation convenience -class TimeStampedGazeStatus(DataFeatures.TimeStampedObjectsList): +class TimeStampedGazeStatus(DataFeatures.TimestampedObjectsList): """Handle timestamped gaze movements into a list !!! note |