aboutsummaryrefslogtreecommitdiff
path: root/src
diff options
context:
space:
mode:
Diffstat (limited to 'src')
-rw-r--r--src/argaze/DataFeatures.py217
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():
"""