aboutsummaryrefslogtreecommitdiff
path: root/src
diff options
context:
space:
mode:
authorThéo de la Hogue2022-12-07 16:13:25 +0100
committerThéo de la Hogue2022-12-07 16:13:25 +0100
commita4c222f3c249b7fa2865f77516d9f46e52a6a655 (patch)
treee086690628d17355176d30b092c26c8f5bc395f8 /src
parentba181dc9a240bd5a589abe49a76ea9bad2416c7e (diff)
downloadargaze-a4c222f3c249b7fa2865f77516d9f46e52a6a655.zip
argaze-a4c222f3c249b7fa2865f77516d9f46e52a6a655.tar.gz
argaze-a4c222f3c249b7fa2865f77516d9f46e52a6a655.tar.bz2
argaze-a4c222f3c249b7fa2865f77516d9f46e52a6a655.tar.xz
Encoding any data to json format. Adding pop_last_until, pop_last_before, get_last_until methods.
Diffstat (limited to 'src')
-rw-r--r--src/argaze/DataStructures.py110
1 files changed, 107 insertions, 3 deletions
diff --git a/src/argaze/DataStructures.py b/src/argaze/DataStructures.py
index 5bbd47e..7fb8cd8 100644
--- a/src/argaze/DataStructures.py
+++ b/src/argaze/DataStructures.py
@@ -3,9 +3,11 @@
from typing import TypeVar, Tuple
import collections
import json
+import ast
import bisect
import pandas
+import numpy
import matplotlib.pyplot as mpyplot
import matplotlib.patches as mpatches
@@ -18,6 +20,30 @@ DataType = TypeVar('Data')
TimeStampedBufferType = TypeVar('TimeStampedBuffer', bound="TimeStampedBuffer")
# Type definition for type annotation convenience
+class JsonEncoder(json.JSONEncoder):
+
+ def default(self, obj):
+
+ # numpy cases
+ if isinstance(obj, numpy.integer):
+ return int(obj)
+
+ if isinstance(obj, numpy.floating):
+ return float(obj)
+
+ if isinstance(obj, numpy.ndarray):
+ return obj.tolist()
+
+ # default case
+ try:
+
+ return json.JSONEncoder.default(self, obj)
+
+ # class case
+ except:
+
+ return obj.__dict__
+
class TimeStampedBuffer(collections.OrderedDict):
"""Ordered dictionary to handle timestamped data.
```
@@ -49,7 +75,12 @@ class TimeStampedBuffer(collections.OrderedDict):
def __repr__(self):
"""String representation"""
- return json.dumps(self, ensure_ascii = False, default=vars)
+ return json.dumps(self, ensure_ascii=False, cls=JsonEncoder)
+
+ def __str__(self):
+ """String representation"""
+
+ return json.dumps(self, ensure_ascii=False, cls=JsonEncoder)
def append(self, timestamped_buffer: TimeStampedBufferType) -> TimeStampedBufferType:
"""Append a timestamped buffer."""
@@ -70,8 +101,22 @@ class TimeStampedBuffer(collections.OrderedDict):
return self.popitem(last=False)
- def pop_first_until(self, ts: TimeStampType) -> Tuple[TimeStampType, DataType]:
- """Pop all item until a given timestamped value and return the last poped item."""
+ def pop_last_until(self, ts: TimeStampType) -> Tuple[TimeStampType, DataType]:
+ """Pop all item until a given timestamped value and return the first after."""
+
+ # get last item before given timestamp
+ earliest_ts, earliest_value = self.get_last_until(ts)
+
+ first_ts, first_value = self.first
+
+ while first_ts < earliest_ts:
+ self.pop_first()
+ first_ts, first_value = self.first
+
+ return first_ts, first_value
+
+ def pop_last_before(self, ts: TimeStampType) -> Tuple[TimeStampType, DataType]:
+ """Pop all item before a given timestamped value and return the last one."""
# get last item before given timestamp
earliest_ts, earliest_value = self.get_last_before(ts)
@@ -94,6 +139,22 @@ class TimeStampedBuffer(collections.OrderedDict):
return self.popitem(last=True)
+ def get_first_from(self, ts) -> Tuple[TimeStampType, DataType]:
+ """Retreive first item timestamp from a given timestamp value."""
+
+ ts_list = list(self.keys())
+ first_from_index = bisect.bisect_left(ts_list, ts)
+
+ if first_from_index < len(self):
+
+ first_from_ts = ts_list[first_from_index]
+
+ return first_from_ts, self[first_from_ts]
+
+ else:
+
+ raise KeyError(f'No data stored after {ts} timestamp.')
+
def get_last_before(self, ts) -> Tuple[TimeStampType, DataType]:
"""Retreive last item timestamp before a given timestamp value."""
@@ -110,6 +171,49 @@ class TimeStampedBuffer(collections.OrderedDict):
raise KeyError(f'No data stored before {ts} timestamp.')
+ def get_last_until(self, ts) -> Tuple[TimeStampType, DataType]:
+ """Retreive last item timestamp until a given timestamp value."""
+
+ ts_list = list(self.keys())
+ last_until_index = bisect.bisect_right(ts_list, ts) - 1
+
+ if last_until_index >= 0:
+
+ last_until_ts = ts_list[last_until_index]
+
+ return last_until_ts, self[last_until_ts]
+
+ else:
+
+ raise KeyError(f'No data stored until {ts} timestamp.')
+
+ @classmethod
+ def from_json(self, json_filepath: str) -> TimeStampedBufferType:
+ """Create a TimeStampedBuffer from .json file."""
+
+ with open(json_filepath, encoding='utf-8') as ts_buffer_file:
+
+ json_buffer = json.load(ts_buffer_file)
+
+ return TimeStampedBuffer({ast.literal_eval(ts_str): json_buffer[ts_str] for ts_str in json_buffer})
+
+ def to_json(self, json_filepath: str):
+ """Save a TimeStampedBuffer to .json file."""
+
+ with open(json_filepath, 'w', encoding='utf-8') as ts_buffer_file:
+
+ 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/frame.html)."""
+
+ dataframe.drop(exclude, inplace=True, axis=True)
+
+ assert(dataframe.index.name == 'timestamp')
+
+ return TimeStampedBuffer(dataframe.to_dict('index'))
+
def as_dataframe(self, exclude=[], split={}) -> pandas.DataFrame:
"""Convert as [pandas dataframe](https://pandas.pydata.org/docs/reference/frame.html).