Encoding any data to json format. Adding pop_last_until, pop_last_before, get_last_until methods.

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).