diff options
-rw-r--r-- | src/argaze.test/DataFeatures.py | 5 | ||||
-rw-r--r-- | src/argaze/ArFeatures.py | 2 | ||||
-rw-r--r-- | src/argaze/DataFeatures.py | 216 | ||||
-rw-r--r-- | src/argaze/GazeFeatures.py | 41 | ||||
-rw-r--r-- | src/argaze/PupilFeatures.py | 2 |
5 files changed, 149 insertions, 117 deletions
diff --git a/src/argaze.test/DataFeatures.py b/src/argaze.test/DataFeatures.py index 04dfe9a..0635f3e 100644 --- a/src/argaze.test/DataFeatures.py +++ b/src/argaze.test/DataFeatures.py @@ -26,13 +26,12 @@ from argaze import DataFeatures import pandas import numpy -class BasicDataClass(DataFeatures.TimestampedObject): +@DataFeatures.timestamp +class BasicDataClass(): """Define a basic dataclass for testing purpose.""" def __init__(self, value: tuple = (), message: str = None, **kwargs): - DataFeatures.TimestampedObject.__init__(self, **kwargs) - self.__value = value self.__message = message diff --git a/src/argaze/ArFeatures.py b/src/argaze/ArFeatures.py index c478207..f838db0 100644 --- a/src/argaze/ArFeatures.py +++ b/src/argaze/ArFeatures.py @@ -608,7 +608,7 @@ class ArFrame(DataFeatures.SharedObject, DataFeatures.PipelineStepObject): if background.size != self.size: # Resize image to frame size - self.__background = DataFeatures.TimestampedImage(cv2.resize(background, dsize=self.size, interpolation=cv2.INTER_CUBIC), background.timestamp) + self.__background = DataFeatures.TimestampedImage(cv2.resize(background, dsize=self.size, interpolation=cv2.INTER_CUBIC), timestamp=background.timestamp) else: diff --git a/src/argaze/DataFeatures.py b/src/argaze/DataFeatures.py index b4874d3..5fb0e84 100644 --- a/src/argaze/DataFeatures.py +++ b/src/argaze/DataFeatures.py @@ -26,6 +26,7 @@ import sys import threading import time from typing import Self +from dataclasses import dataclass, is_dataclass import cv2 import matplotlib.patches as mpatches @@ -65,12 +66,12 @@ def set_working_directory(working_directory: str): def get_class(class_path: str) -> type: """Get class object from 'path.to.class' string. - Parameters: - class_path: a 'path.to.class' string. + Parameters: + class_path: a 'path.to.class' string. - Returns: - class: a 'path.to.class' class. - """ + Returns: + class: a 'path.to.class' class. + """ parts = class_path.split('.') module = ".".join(parts[:-1]) @@ -85,12 +86,12 @@ def get_class(class_path: str) -> type: def get_class_path(o: object) -> str: """Get 'path.to.class' class path from object. - Parameters: - o: any object instance. + Parameters: + o: any object instance. - Returns: - class_path: object 'path.to.class' class. - """ + Returns: + class_path: object 'path.to.class' class. + """ c = o.__class__ m = c.__module__ @@ -103,13 +104,18 @@ def get_class_path(o: object) -> str: def get_class_properties(cls: type) -> dict: """Get class properties dictionary. - Parameters: - cls: class to consider. + Parameters: + cls: class to consider. - Returns: - properties: dict of properties stored by names + Returns: + properties: dict of properties stored by names """ + # Dataclass case + if is_dataclass(cls): + + return cls.__dataclass_fields__ + # Stop recursion when reaching core objects if cls is not object and cls is not PipelineStepObject and cls is not SharedObject: @@ -309,44 +315,86 @@ class DataDictionary(dict): __setattr__ = dict.__setitem__ __delattr__ = dict.__delitem__ +def metrics(cls): + """Decorate a class to make it a dataclass.""" + return dataclass(cls) -class TimestampedObject(): - """Abstract class to enable timestamp management.""" +def timestamp(cls): + """Decorate a class to enable timestamp management.""" - def __init__(self, timestamp: int | float = math.nan): - """Initialize TimestampedObject.""" - self._timestamp = timestamp + class_init = cls.__init__ + class_repr = cls.__repr__ - def __repr__(self): + def __init__(self, *args, **kwargs): + """Initialize timestamped object.""" + + try: + + self._timestamp = kwargs.pop('timestamp') + + except KeyError: + + self._timestamp = math.nan + + class_init(self, *args, **kwargs) + + def __repr__(self) -> str: """String representation.""" - return json.dumps(as_dict(self)) - @property - def timestamp(self) -> int | float: - """Get object timestamp.""" - return self._timestamp + return str(self._timestamp) + ': ' + class_repr(self) + + if issubclass(cls, TimestampedObjectsList): - @timestamp.setter - def timestamp(self, timestamp: int | float): - """Set object timestamp.""" - self._timestamp = timestamp + def get_timestamp(self) -> int | float: + """Get first position timestamp.""" + if self: + return self[0].timestamp + + def set_timestamp(self, timestamp: int | float): + """Block timestamp setting.""" + raise ('TimestampedObjectsList timestamp is not settable.') + + def del_timestamp(self): + """Block timestamp resetting.""" + raise ('TimestampedObjectsList timestamp cannot be deleted.') + + def is_timestamped(self) -> bool: + """Is the object timestamped?""" + return bool(self) + + else: - def untimestamp(self): - """Reset object timestamp.""" - self._timestamp = math.nan + def get_timestamp(self) -> int | float: + """Get object timestamp.""" + return self._timestamp - def is_timestamped(self) -> bool: - """Is the object timestamped?""" - return not math.isnan(self._timestamp) + def set_timestamp(self, timestamp: int | float): + """Set object timestamp.""" + self._timestamp = timestamp + def del_timestamp(self): + """Reset object timestamp.""" + self._timestamp = math.nan + + def is_timestamped(self) -> bool: + """Is the object timestamped?""" + return not math.isnan(self._timestamp) + + cls.__init__ = __init__ + cls.__repr__ = __repr__ + + setattr(cls, "timestamp", property(get_timestamp, set_timestamp, del_timestamp, """Object timestamp.""")) + setattr(cls, "is_timestamped", is_timestamped) + + return cls class TimestampedObjectsList(list): """Handle timestamped object into a list. - !!! warning "Timestamped objects are not sorted internally" - - Timestamped objects are considered to be stored according to their coming time. - """ + !!! warning "Timestamped objects are not sorted internally" + + Timestamped objects are considered to be stored according to their coming time. + """ # noinspection PyMissingConstructor def __init__(self, ts_object_type: type, ts_objects=None): @@ -365,7 +413,7 @@ class TimestampedObjectsList(list): """Get object type handled by the list.""" return self.__object_type - def append(self, ts_object: TimestampedObject | dict): + def append(self, ts_object: object | dict): """Append timestamped object.""" # Convert dict into object @@ -383,7 +431,7 @@ class TimestampedObjectsList(list): super().append(ts_object) - def look_for(self, timestamp: int | float) -> TimestampedObject: + def look_for(self, timestamp: int | float) -> object: """Look for object at given timestamp.""" for ts_object in self: @@ -429,24 +477,24 @@ class TimestampedObjectsList(list): assert (dataframe.index.name == 'timestamp') object_list = [ts_object_type(timestamp=timestamp, **object_dict) for timestamp, object_dict in - dataframe.to_dict('index').items()] + 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"]} + + 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. + !!! warning "Values must be dictionaries" + + Each key is stored as a column name. - !!! note + !!! note - Timestamps are stored as index column called 'timestamp'. - """ + Timestamps are stored as index column called 'timestamp'. + """ df = pandas.DataFrame(self.tuples(), columns=self.__object_properties_names) @@ -486,15 +534,16 @@ class TimestampedObjectsList(list): """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]) - + 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): @@ -505,7 +554,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: int | float) -> TimestampedObject: + def pop_last_until(self, timestamp: int | float) -> object: """Pop all item until a given timestamped value and return the first after.""" # get last item before given timestamp @@ -516,7 +565,7 @@ class TimestampedObjectsList(list): return self[0] - def pop_last_before(self, timestamp: int | float) -> TimestampedObject: + def pop_last_before(self, timestamp: int | float) -> object: """Pop all item before a given timestamped value and return the last one.""" # get last item before given timestamp @@ -529,7 +578,7 @@ class TimestampedObjectsList(list): return popped_value - def get_first_from(self, timestamp: int | float) -> TimestampedObject: + def get_first_from(self, timestamp: int | float) -> object: """Retrieve first item timestamp from a given timestamp value.""" ts_list = self.timestamps() @@ -543,7 +592,7 @@ class TimestampedObjectsList(list): raise KeyError(f'No data stored after {timestamp} timestamp.') - def get_last_before(self, timestamp: int | float) -> TimestampedObject: + def get_last_before(self, timestamp: int | float) -> object: """Retrieve last item timestamp before a given timestamp value.""" ts_list = self.timestamps() @@ -557,7 +606,7 @@ class TimestampedObjectsList(list): raise KeyError(f'No data stored before {timestamp} timestamp.') - def get_last_until(self, timestamp: int | float) -> TimestampedObject: + def get_last_until(self, timestamp: int | float) -> object: """Retrieve last item timestamp until a given timestamp value.""" ts_list = self.timestamps() @@ -571,10 +620,14 @@ class TimestampedObjectsList(list): raise KeyError(f'No data stored until {timestamp} timestamp.') - def plot(self, names=[], colors=[], split={}, samples=None) -> list: + def plot(self, names=[], colors=[], split=None, samples=None) -> list: """Plot as [matplotlib](https://matplotlib.org/) time chart.""" + if split is None: + split = {} + df = self.as_dataframe(split=split) + legend_patches = [] # decimate data @@ -595,22 +648,22 @@ class TimestampedObjectsList(list): return legend_patches -class SharedObject(TimestampedObject): - """Abstract class to enable multiple threads sharing for timestamped object.""" +class SharedObject(): + """Enable multiple threads sharing.""" + + def __init__(self): - def __init__(self, timestamp: int | float = math.nan): - TimestampedObject.__init__(self, timestamp) self._lock = threading.Lock() self._execution_times = {} self._exceptions = {} +@timestamp +class TimestampedException(Exception,): + """Enable timestamp management for exception.""" -class TimestampedException(Exception, TimestampedObject): - """Wrap exception to keep track of raising timestamp.""" + def __init__(self, exception: Exception): - def __init__(self, exception: Exception, timestamp: int | float = math.nan): Exception.__init__(self, exception) - TimestampedObject.__init__(self, timestamp) class TimestampedExceptions(TimestampedObjectsList): @@ -626,21 +679,22 @@ class TimestampedExceptions(TimestampedObjectsList): class PipelineStepLoadingFailed(Exception): """ - Exception raised when pipeline step object loading fails. - """ + Exception raised when pipeline step object loading fails. + """ def __init__(self, message): super().__init__(message) - -class TimestampedImage(numpy.ndarray, TimestampedObject): +@timestamp +class TimestampedImage(numpy.ndarray): """Wrap numpy.array to timestamp image.""" - def __new__(cls, array: numpy.array, timestamp: int | float = math.nan): + def __new__(cls, array: numpy.array, **kwargs): + return numpy.ndarray.__new__(cls, array.shape, dtype=array.dtype, buffer=array) - def __init__(self, array: numpy.array, timestamp: int | float = math.nan): - TimestampedObject.__init__(self, timestamp) + def __init__(self, array: numpy.array, **kwargs): + pass def __array_finalize__(self, obj): pass @@ -896,7 +950,7 @@ class PipelineStepObject(): if hasattr(self, key): logging.debug('%s.update_attributes > update %s with %s value', get_class_path(self), key, - type(value).__name__) + type(value).__name__) setattr(self, key, value) @@ -1160,20 +1214,18 @@ def PipelineStepMethod(method): Parameters: self: args: any arguments defined by PipelineStepMethod. - timestamp: optional method call timestamp (unit doesn't matter) if first args parameter is not a - TimestampedObject instance. + timestamp: optional method call timestamp (unit doesn't matter) if first args parameter is not a TimestampedObject instance. unwrap: extra arguments used in wrapper function to call wrapped method directly. """ if timestamp is None and len(args) > 0: - if issubclass(type(args[0]), TimestampedObject): + try: timestamp = args[0].timestamp - else: + except: - logging.error('%s.%s: %s is not a TimestampedObject subclass. You must pass a timestamp argument.', - get_class_path(self), method.__name__, type(args[0]).__name__) + logging.error('%s.%s: %s is not a timestamped class. Use @DataFeatures.timestamp decorator.', get_class_path(self), method.__name__, type(args[0]).__name__) if unwrap: return method(self, *args, **kwargs) diff --git a/src/argaze/GazeFeatures.py b/src/argaze/GazeFeatures.py index 32b7de7..5ef3c32 100644 --- a/src/argaze/GazeFeatures.py +++ b/src/argaze/GazeFeatures.py @@ -28,8 +28,8 @@ import pandas from argaze import DataFeatures from argaze.AreaOfInterest import AOIFeatures - -class GazePosition(tuple, DataFeatures.TimestampedObject): +@DataFeatures.timestamp +class GazePosition(tuple): """Define gaze position as a tuple of coordinates with precision. Parameters: @@ -37,15 +37,12 @@ class GazePosition(tuple, DataFeatures.TimestampedObject): message: a string to describe why the position is what it is. """ - def __new__(cls, position: tuple = (), precision: int | float = None, message: str = None, - timestamp: int | float = math.nan): + def __new__(cls, position: tuple = (), **kwargs): return tuple.__new__(cls, position) - def __init__(self, position: tuple = (), precision: int | float = None, message: str = None, - timestamp: int | float = math.nan): + def __init__(self, position: tuple = (), precision: int | float = None, message: str = None): - DataFeatures.TimestampedObject.__init__(self, timestamp) self.__precision = precision self.__message = message @@ -392,8 +389,8 @@ class GazePositionCalibrator(DataFeatures.PipelineStepObject): raise NotImplementedError('ready getter not implemented') - -class GazeMovement(TimeStampedGazePositions, DataFeatures.TimestampedObject): +@DataFeatures.timestamp +class GazeMovement(TimeStampedGazePositions): """Define abstract gaze movement class as timestamped gaze positions list. !!! note @@ -410,30 +407,14 @@ class GazeMovement(TimeStampedGazePositions, DataFeatures.TimestampedObject): # noinspection PyArgumentList return TimeStampedGazePositions.__new__(cls, positions) - def __init__(self, positions: TimeStampedGazePositions = None, finished: bool = False, message: str = None, timestamp: int | float = math.nan): + def __init__(self, positions: TimeStampedGazePositions = None, finished: bool = False, message: str = None): """Initialize GazeMovement""" TimeStampedGazePositions.__init__(self, positions) - DataFeatures.TimestampedObject.__init__(self, timestamp) self.__finished = finished self.__message = message - @property - def timestamp(self) -> int | float: - """Get first position timestamp.""" - if self: - return self[0].timestamp - - def is_timestamped(self) -> bool: - """If first position exist, the movement is timestamped.""" - return bool(self) - - @timestamp.setter - def timestamp(self, timestamp: int | float): - """Block gaze movement timestamp setting.""" - raise ('GazeMovement timestamp is first position timestamp.') - def is_finished(self) -> bool: """Is the movement finished?""" return self.__finished @@ -552,8 +533,8 @@ class TimeStampedGazeMovements(DataFeatures.TimestampedObjectsList): def __init__(self, gaze_movements: list = []): DataFeatures.TimestampedObjectsList.__init__(self, GazeMovement, gaze_movements) - -class GazeStatus(list, DataFeatures.TimestampedObject): +@DataFeatures.timestamp +class GazeStatus(list): """Define gaze status as a list of 1 or 2 (index, GazeMovement) tuples. Parameters: @@ -561,8 +542,8 @@ class GazeStatus(list, DataFeatures.TimestampedObject): """ def __init__(self, position: GazePosition): - DataFeatures.TimestampedObject.__init__(self, timestamp=position.timestamp) - + + self.timestamp = position.timestamp self.__position = position @property diff --git a/src/argaze/PupilFeatures.py b/src/argaze/PupilFeatures.py index be13dbb..a4d42e6 100644 --- a/src/argaze/PupilFeatures.py +++ b/src/argaze/PupilFeatures.py @@ -21,7 +21,7 @@ import math from argaze import DataFeatures - +@DataFeatures.timestamp class PupilDiameter(float, DataFeatures.TimestampedObject): """Define pupil diameter as a single float value. |