diff options
Diffstat (limited to 'src/argaze/DataFeatures.py')
| -rw-r--r-- | src/argaze/DataFeatures.py | 222 |
1 files changed, 114 insertions, 108 deletions
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.""" |