aboutsummaryrefslogtreecommitdiff
path: root/src/argaze/DataFeatures.py
diff options
context:
space:
mode:
Diffstat (limited to 'src/argaze/DataFeatures.py')
-rw-r--r--src/argaze/DataFeatures.py216
1 files changed, 134 insertions, 82 deletions
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)
generated by cgit v1.1 at 2024-07-03 03:38:46 +0000