diff options
Diffstat (limited to 'src')
| -rw-r--r-- | src/argaze/DataFeatures.py | 217 |
1 files changed, 128 insertions, 89 deletions
diff --git a/src/argaze/DataFeatures.py b/src/argaze/DataFeatures.py index edbf8e9..931c21d 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.""" -DataType = TypeVar('Data') -"""Type definition for data to store anything in time.""" +TimeStampedObjectType = TypeVar('TimeStampedObject', bound="TimeStampedObject") +# Type definition for type annotation convenience -TimeStampedBufferType = TypeVar('TimeStampedBuffer', bound="TimeStampedBuffer") +TimeStampedObjectsListType = TypeVar('TimeStampedObjectsList', bound="TimeStampedObjectsList") # Type definition for type annotation convenience def module_path(obj) -> str: @@ -45,6 +45,39 @@ def module_path(obj) -> str: """ return obj.__class__.__module__ +def properties(cls) -> list: + """get class properties name.""" + + properties = [name for name, item in cls.__dict__.items() if isinstance(item, property)] + + for base in cls.__bases__: + + for name, item in base.__dict__.items(): + + if isinstance(item, property): + + properties.append(name) + + return properties + +def as_dict(obj, filter: bool=True) -> dict: + """Export object as dictionary. + + Parameters: + filter: remove None attribute values. + """ + _dict = {} + + for p in properties(obj.__class__): + + v = getattr(obj, p) + + if not filter or v is not None: + + _dict[p] = v + + return _dict + class JsonEncoder(json.JSONEncoder): """Specific ArGaze JSON Encoder.""" @@ -90,67 +123,63 @@ class JsonEncoder(json.JSONEncoder): return public_dict -class TimeStampedBuffer(collections.OrderedDict): - """Ordered dictionary to handle timestamped data. - ``` - { - timestamp1: data1, - timestamp2: data2, - ... - } - ``` - - !!! warning - - Timestamps must be numbers. +class TimeStampedObjectsList(list): + """Handle timestamped object into a list. - !!! warning "Timestamps are not sorted internally" + !!! warning "Timestamped objects are not sorted internally" - Data are considered to be stored according at their coming time. + Timestamped objects are considered to be stored according at their coming time. """ - def __new__(cls, args = None): - """Inheritance""" + def __init__(self, ts_object_type: type, ts_objects: list = []): - return super(TimeStampedBuffer, cls).__new__(cls) + super().__init__() + self.__object_type = ts_object_type + self.__object_properties = properties(self.__object_type) - def __setitem__(self, ts: TimeStampType, data: DataType): - """Store data at given timestamp.""" + for ts_object in ts_objects: - assert(type(ts) == int or type(ts) == float) + self.append(ts_object) - super().__setitem__(ts, data) + @property + def object_type(self): + """Get object type handled by the list.""" + return self.__object_type - def __repr__(self): - """String representation""" + def append(self, ts_object: TimeStampedObjectType|dict): + """Append timestamped object.""" - return json.dumps(self, ensure_ascii=False, cls=JsonEncoder) + # Convert dict into GazePosition + if type(ts_object) == dict: - def __str__(self): - """String representation""" + ts_object = self.__object_type.from_dict(ts_object) - return json.dumps(self, ensure_ascii=False, cls=JsonEncoder) + # Check object type + if type(ts_object) != self.__object_type: - def append(self, timestamped_buffer: TimeStampedBufferType) -> TimeStampedBufferType: - """Append a timestamped buffer.""" + raise TypeError(f'object type have to be {self.__object_type} not {type(ts_object)}') - for ts, value in timestamped_buffer.items(): - self[ts] = value + assert(ts_object.is_timestamped()) - return self + super().append(ts_object) - @property - def first(self) -> Tuple[TimeStampType, DataType]: - """Easing access to first item.""" + def timestamps(self): + """Get all timestamps in list.""" + return [ts_object.timestamp for ts_object in self] - return list(self.items())[0] + def tuples(self) -> list: + """Get all timestamped objects as list of tuple.""" + return [tuple(as_dict(ts_object, filter=False).values()) for ts_object in self] - def pop_first(self) -> Tuple[TimeStampType, DataType]: - """Easing FIFO access mode.""" + def __repr__(self): + """String representation""" + return json.dumps([as_dict(ts_object) for ts_object in self], ensure_ascii=False,) - return self.popitem(last=False) + def __str__(self): + """String representation""" + return json.dumps([as_dict(ts_object) for ts_object in self], ensure_ascii=False,) - def pop_last_until(self, ts: TimeStampType) -> Tuple[TimeStampType, DataType]: + def pop_last_until(self, ts: TimeStampType) -> TimeStampedObjectType: """Pop all item until a given timestamped value and return the first after.""" # get last item before given timestamp @@ -164,7 +193,7 @@ class TimeStampedBuffer(collections.OrderedDict): return first_ts, first_value - def pop_last_before(self, ts: TimeStampType) -> Tuple[TimeStampType, DataType]: + def pop_last_before(self, ts: TimeStampType) -> TimeStampedObjectType: """Pop all item before a given timestamped value and return the last one.""" # get last item before given timestamp @@ -177,18 +206,7 @@ class TimeStampedBuffer(collections.OrderedDict): return popep_ts, poped_value - @property - def last(self) -> Tuple[TimeStampType, DataType]: - """Easing access to last item.""" - - return list(self.items())[-1] - - def pop_last(self) -> Tuple[TimeStampType, DataType]: - """Easing FIFO access mode.""" - - return self.popitem(last=True) - - def get_first_from(self, ts) -> Tuple[TimeStampType, DataType]: + def get_first_from(self, ts) -> TimeStampedObjectType: """Retreive first item timestamp from a given timestamp value.""" ts_list = list(self.keys()) @@ -204,7 +222,7 @@ class TimeStampedBuffer(collections.OrderedDict): raise KeyError(f'No data stored after {ts} timestamp.') - def get_last_before(self, ts) -> Tuple[TimeStampType, DataType]: + def get_last_before(self, ts) -> TimeStampedObjectType: """Retreive last item timestamp before a given timestamp value.""" ts_list = list(self.keys()) @@ -220,7 +238,7 @@ class TimeStampedBuffer(collections.OrderedDict): raise KeyError(f'No data stored before {ts} timestamp.') - def get_last_until(self, ts) -> Tuple[TimeStampType, DataType]: + def get_last_until(self, ts) -> TimeStampedObjectType: """Retreive last item timestamp until a given timestamp value.""" ts_list = list(self.keys()) @@ -237,14 +255,14 @@ class TimeStampedBuffer(collections.OrderedDict): raise KeyError(f'No data stored until {ts} timestamp.') @classmethod - def from_json(self, json_filepath: str) -> TimeStampedBufferType: - """Create a TimeStampedBuffer 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_buffer_file: + with open(json_filepath, encoding='utf-8') as ts_objects_file: - json_buffer = json.load(ts_buffer_file) + json_ts_objects = json.load(ts_objects_file) - return TimeStampedBuffer({ast.literal_eval(ts_str): json_buffer[ts_str] for ts_str in json_buffer}) + 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.""" @@ -254,14 +272,14 @@ class TimeStampedBuffer(collections.OrderedDict): json.dump(self, ts_buffer_file, ensure_ascii=False, cls=JsonEncoder) @classmethod - def from_dataframe(self, dataframe: pandas.DataFrame, exclude=[]) -> TimeStampedBufferType: - """Create a TimeStampedBuffer from [Pandas DataFrame](https://pandas.pydata.org/docs/reference/api/pandas.DataFrame.html).""" + 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 TimeStampedBuffer(dataframe.to_dict('index')) + 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). @@ -278,7 +296,7 @@ class TimeStampedBuffer(collections.OrderedDict): Timestamps are stored as index column called 'timestamp'. """ - df = pandas.DataFrame.from_dict(self.values()) + df = pandas.DataFrame(self.tuples(), columns=self.__object_properties) # Exclude columns df.drop(exclude, inplace=True, axis=True) @@ -307,7 +325,7 @@ class TimeStampedBuffer(collections.OrderedDict): df = df[splited_columns] # Append timestamps as index column - df['timestamp'] = self.keys() + df['timestamp'] = self.timestamps() df.set_index('timestamp', inplace=True) return df @@ -344,13 +362,43 @@ class DataDictionary(dict): __setattr__ = dict.__setitem__ __delattr__ = dict.__delitem__ -class SharedObject(): +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.""" def __init__(self): + super().__init__() self._lock = threading.Lock() - self._timestamp = math.nan self._execution_times = {} self._exceptions = {} @@ -362,33 +410,24 @@ class SharedObject(): @property def timestamp(self) -> int|float: """Get shared object timestamp.""" - self._lock.acquire() - timestamp = self._timestamp - self._lock.release() - - return timestamp + with self._lock: + return super().timestamp @timestamp.setter def timestamp(self, timestamp: int|float): """Set shared object timestamp.""" - self._lock.acquire() - self._timestamp = timestamp - self._lock.release() + with self._lock: + super().timestamp = timestamp def untimestamp(self): """Reset shared object timestamp.""" - self._lock.acquire() - self._timestamp = math.nan - self._lock.release() + with self._lock: + self.timestamp = math.nan - @property - def timestamped(self) -> bool: + def is_timestamped(self) -> bool: """Is the object timestamped?""" - self._lock.acquire() - timestamped = not math.isnan(self._timestamp) - self._lock.release() - - return timestamped + with self._lock: + return super().is_timestamped() class PipelineStepObject(): """ |