aboutsummaryrefslogtreecommitdiff
path: root/src/argaze/DataFeatures.py
diff options
context:
space:
mode:
authorThéo de la Hogue2024-04-08 21:56:47 +0200
committerThéo de la Hogue2024-04-08 21:56:47 +0200
commitad25274a8493b63c40cd5d59f59889214fcc8aa2 (patch)
tree022900c5755691e41a98ab151449f4bbf1a275f8 /src/argaze/DataFeatures.py
parentea0500cc35c048ab86bb572ea9c80d93c24c073f (diff)
downloadargaze-ad25274a8493b63c40cd5d59f59889214fcc8aa2.zip
argaze-ad25274a8493b63c40cd5d59f59889214fcc8aa2.tar.gz
argaze-ad25274a8493b63c40cd5d59f59889214fcc8aa2.tar.bz2
argaze-ad25274a8493b63c40cd5d59f59889214fcc8aa2.tar.xz
Fixing coding style and types.
Diffstat (limited to 'src/argaze/DataFeatures.py')
-rw-r--r--src/argaze/DataFeatures.py47
1 files changed, 27 insertions, 20 deletions
diff --git a/src/argaze/DataFeatures.py b/src/argaze/DataFeatures.py
index a7b0a48..4e85aaf 100644
--- a/src/argaze/DataFeatures.py
+++ b/src/argaze/DataFeatures.py
@@ -330,7 +330,11 @@ class TimestampedObjectsList(list):
Timestamped objects are considered to be stored according to their coming time.
"""
- def __init__(self, ts_object_type: type, ts_objects: list = []):
+ # noinspection PyMissingConstructor
+ def __init__(self, ts_object_type: type, ts_objects=None):
+
+ if ts_objects is None:
+ ts_objects = []
self.__object_type = ts_object_type
self.__object_properties = properties(self.__object_type)
@@ -396,9 +400,12 @@ class TimestampedObjectsList(list):
return [tuple(as_dict(ts_object, filter=False).values()) for ts_object in self]
@classmethod
- def from_dataframe(cls, ts_object_type: type, dataframe: pandas.DataFrame, exclude=[]) -> Self:
+ def from_dataframe(cls, ts_object_type: type, dataframe: pandas.DataFrame, exclude=None) -> Self:
"""Create a TimestampedObjectsList from [Pandas DataFrame](https://pandas.pydata.org/docs/reference/api/pandas.DataFrame.html)."""
+ if exclude is None:
+ exclude = []
+
dataframe.drop(exclude, inplace=True, axis=True)
assert (dataframe.index.name == 'timestamp')
@@ -583,7 +590,7 @@ class SharedObject(TimestampedObject):
class TimestampedException(Exception, TimestampedObject):
"""Wrap exception to keep track of raising timestamp."""
- def __init__(self, exception=Exception, timestamp: int | float = math.nan):
+ def __init__(self, exception: Exception, timestamp: int | float = math.nan):
Exception.__init__(self, exception)
TimestampedObject.__init__(self, timestamp)
@@ -639,10 +646,10 @@ def PipelineStepInit(method):
def wrapper(self, **kwargs):
"""Wrap pipeline __init__ init method to update PipelineStepObject attributes with arguments after init call.
- Parameters:
- self:
- kwargs: any arguments defined by PipelineStepMethodInit.
- """
+ Parameters:
+ self:
+ kwargs: any arguments defined by PipelineStepMethodInit.
+ """
# Init pipeline step object attributes
PipelineStepObject.__init__(self)
@@ -811,9 +818,8 @@ def PipelineStepDraw(method):
# noinspection PyAttributeOutsideInit
class PipelineStepObject():
+ """Define class to assess pipeline step methods execution time and observe them.
"""
- Define class to assess pipeline step methods execution time and observe them.
- """
__initialized = False
@@ -939,9 +945,9 @@ class PipelineStepObject():
def as_dict(self) -> dict:
"""Export PipelineStepObject attributes as dictionary.
- Returns:
- object_data: dictionary with pipeline step object attributes values.
- """
+ Returns:
+ object_data: dictionary with pipeline step object attributes values.
+ """
return {
"name": self.__name,
"observers": self.__observers
@@ -964,12 +970,11 @@ class PipelineStepObject():
#json.dump(self, object_file, ensure_ascii=False, indent=4, cls=JsonEncoder)
def __str__(self) -> str:
+ """String representation of pipeline step object.
+
+ Returns:
+ String representation
"""
- String representation of pipeline step object.
-
- Returns:
- String representation
- """
logging.debug('%s.__str__ %s', get_class_path(self), self.name if self.name is not None else '')
@@ -1064,7 +1069,7 @@ class PipelineStepObject():
yield name, getattr(self, name)
@property
- def children(self) -> object:
+ def children(self):
"""Iterate over children pipeline step objects."""
for name, value in self.properties:
@@ -1098,7 +1103,8 @@ def PipelineStepMethod(method):
PipelineStepMethod must have a timestamp as first argument.
"""
- def wrapper(self, *args, timestamp: int | float = None, unwrap: bool = False, catch_exceptions: bool = True, **kwargs):
+ def wrapper(self, *args, timestamp: int | float = None, unwrap: bool = False, catch_exceptions: bool = True,
+ **kwargs):
"""Wrap pipeline step method to measure execution time.
Parameters:
@@ -1117,7 +1123,8 @@ def PipelineStepMethod(method):
else:
- 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 TimestampedObject subclass. You must pass a timestamp argument.',
+ get_class_path(self), method.__name__, type(args[0]).__name__)
if unwrap:
return method(self, *args, **kwargs)