test data

python/biopro/devlib/data.py

@@ -6,6 +6,8 @@ from datetime import datetime
 
 from biopro.recording import RecordingData, RecordingFileDataFormat
 
+import numpy as np
+
 # from biopro.util.console import hex_line
 
 T = TypeVar('T')

python/biopro/recording/file.py

@@ -20,6 +20,11 @@ from statistics import mean
 import random
 # from numba import jit
 
+from .lttb import largest_triangle_three_buckets
+import math
+
+from scipy import signal
+
 from copy import copy
 
 def json_stringify(o) -> str:
@@ -1015,7 +1020,7 @@ class RecordingFileWriter:
                  '_mini_scale_list', '_time_real_time', '_data_rl', '_data_db',
                  '_raw_save', '_mini_save', '_data_time_ch', '_data_value_ch_for_rl',
                  '_data_time_ch_for_rl', '_device_id', '_send_data', '_data_mqtt_ch', '_id_db_save', '_raw_create_not_done',
-                 '_mini_create_not_done')
+                 '_mini_create_not_done', '_data_all')
 
     def __init__(self, meta: RecordingMetaFile, device_id, database = None):
         self._meta = meta
@@ -1075,6 +1080,8 @@ class RecordingFileWriter:
         self._data_time_ch = {}
         self._id_db_save = {}
 
+        self._data_all = {}
+
         # mini data
         self._data_mini_ch = {}
 
@@ -1227,17 +1234,17 @@ class RecordingFileWriter:
     def get_data_iter(self, d, mqtt_thread):
         # print('****d size', d.data_size)
         for t, c, v in d.entry_iter():
+            # print(t,c,v)
             if c in self._data_db:
+                self._data_all[c].append([int(t), v])
                 ### send real-time
                 if len(self._data_rl[c]) > 0 and self._send_data[c]:
                     self._data_rl[c].append(str(int(t)))
                     mes = ' '.join(self._data_rl[c])
-                    # self._data_mqtt_ch[c] = self._data_rl[c].copy()
-                    # self._data_mqtt_ch[c].append(str(int(t)))
+                    # print('message1', mes)
                     mqtt_thread[c].on_message(mes)
                     self._data_rl[c].clear()
                     self._send_data[c] = False
-                ### send real-time
                 
                 sample_rate_rl = 1
                 # rec data
@@ -1250,40 +1257,35 @@ class RecordingFileWriter:
                     self._data_value_ch[c].append(int(v))
                     self._data_value_ch_for_rl[c].append(int(v))
                     self._data_time_ch_for_rl[c].append(int(t))
-                    # self._data_time_ch[c].append(int(t))
+                    # print('self._data_value_ch[c]', self._data_value_ch[c])
+                    # print('self._data_value_ch_for_rl[c]', self._data_value_ch_for_rl[c])
+                    # print('self._data_time_ch_for_rl[c]', self._data_time_ch_for_rl[c])
                     for scale in self._mini_scale_list:
                         if self._data_mini_ch[c][str(scale)]['start_time'] is None:
                             self._data_mini_ch[c][str(scale)]['start_time'] = str(int(t))
                     
+                    # print('len(self._data_value_ch_for_rl[c])', len(self._data_value_ch_for_rl[c]))
+                    # print('sample_rate_rl:', sample_rate_rl)
                     if len(self._data_value_ch_for_rl[c]) >= sample_rate_rl:
                         if len(self._data_value_ch_for_rl[c]) == 1:
                             if len(self._data_rl[c]) == 0:
                                 self._data_rl[c].append(str(int(t)))
-                            # self._data_rl[c].append(str(self._data_time_ch_for_rl[c][0]))
                             self._data_rl[c].append(str(self._data_value_ch_for_rl[c][0]))
                         else:
                             if len(self._data_rl[c]) == 0:
                                 self._data_rl[c].append(str(int(t)))
-                            # self._data_rl[c].append(str(self._data_time_ch_for_rl[c][0]))
-                            # self._data_rl[c].append(str(self._data_value_ch_for_rl[c][0]))
                             _max = max(self._data_value_ch_for_rl[c])
                             _max_index = self._data_value_ch_for_rl[c].index(_max)
                             _min = min(self._data_value_ch_for_rl[c])
                             _min_index = self._data_value_ch_for_rl[c].index(_min)
-                            # _mean = mean(self._data_value_ch_for_rl[c])
 
-                            # _first_time = self._data_time_ch_for_rl[c][0]
-                            # _last_time = self._data_time_ch_for_rl[c][-1]
+                            # print('_max, _max_index, _min, _min_index', _max, _max_index, _min, _min_index)
 
                             if _max_index < _min_index:
-                                # self._data_rl[c].append(str(_first_time))
                                 self._data_rl[c].append(str(_max))
-                                # self._data_rl[c].append(str(_last_time))
                                 self._data_rl[c].append(str(_min))
                             else:
-                                # self._data_rl[c].append(str(_first_time))
                                 self._data_rl[c].append(str(_min))
-                                # self._data_rl[c].append(str(_last_time))
                                 self._data_rl[c].append(str(_max))
 
                         self._data_value_ch_for_rl[c].clear()
@@ -1292,9 +1294,6 @@ class RecordingFileWriter:
                     # mini picture
                     if len(self._data_value_ch[c]) >= 10:
                         self._data_mini_ch[c]['10']['mean'].append( int(mean(self._data_value_ch[c][0:9])) )
-                        # self._data_mini_ch[c]['10']['random'].append( str(self._data_value_ch[c][random.randint(0,9)]) )
-                        # _bar = self.get_bar(self._data_value_ch[c], None)
-                        # self._data_mini_ch[c]['10']['bar'].extend(_bar)
                         self._data_value_ch[c].clear()
                         if int(len(self._data_mini_ch[c]['10']['mean']) / 10) - self._data_mini_ch[c]['10']['dec'] > 0:
                             self._data_mini_ch[c]['100']['mean'].append( int(mean(self._data_mini_ch[c]['10']['mean'][-10:])) )
@@ -1316,6 +1315,13 @@ class RecordingFileWriter:
                     self._data_db[c].append(str(int(t)))
                     self._data_db[c].append(str(v))
                     self._time_now = int(t)
+        # print('self._data_rl',self._data_rl)
+        # print('self._data_db', self._data_db)
+        # print('self._data_value_ch', self._data_value_ch)
+        # print('self._data_value_ch_for_rl', self._data_value_ch_for_rl)
+        # print('self._data_time_ch_for_rl', self._data_time_ch_for_rl)
+        # print('self._data_time_ch', self._data_time_ch)
+        # print('self._id_db_save', self._id_db_save)
         return
     
     # @calculate_time(1)
@@ -1363,6 +1369,7 @@ class RecordingFileWriter:
                         self._data_db[ch] = []
                         self._time[ch] = 0
                         self._time_real_time[ch] = 0
+                        self._data_all[ch] = []
                 if len(self._data_mini_ch) == 0:
                     for ch in self._channel_list:
                         self._data_mini_ch[ch] = {}
@@ -1376,6 +1383,37 @@ class RecordingFileWriter:
                             }
                 
                 self.get_data_iter(d, mqtt_thread)
+            
+            print('data', len(self._data_all[0]))
+
+            test = []
+
+            # for ch in self._data_all:
+            #     if len(self._data_all[ch]) <= 1000:
+            #         print('math.ceil(len(self._data_all[ch]) / 10) + 1', math.ceil(len(self._data_all[ch]) / 10) + 1)
+            #         test = largest_triangle_three_buckets(self._data_all[ch], math.ceil(len(self._data_all[ch]) / 10) + 1)
+            #     else:
+            #         self._data_all[ch] = largest_triangle_three_buckets(self._data_all[ch], 1000)
+            #         test = self._data_all[ch]
+                
+            #     print('test', test)
+            #     a = [' '.join(list(map(str, b))) for b in test]
+            #     print('a', a)
+            #     mes = ' '.join(a)
+            #     mqtt_thread[50+ch].on_message(mes)
+
+            for ch in self._data_all:
+                if len(self._data_all[ch]) <= 1000:
+                    self._data_all[ch] = signal.savgol_filter(self._data_all[ch], 32, 3)
+                else:
+                    self._data_all[ch] = signal.savgol_filter(self._data_all[ch], 32, 3)
+                
+                print('test', test)
+                a = [' '.join(list(map(str, b))) for b in self._data_all[ch]]
+                print('a', a)
+                mes = ' '.join(a)
+                mqtt_thread[50+ch].on_message(mes)
+            
 
             if len(self._recording_file_dict) > 0:
                 for ch in self._data_db.keys():

python/biopro/recording/lttb.py

@@ -0,0 +1,108 @@
+"""
+The MIT License (MIT)
+Copyright (c) 2015 Olivier Devoisin
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.
+"""
+
+import math
+
+class LttbException(Exception):
+    pass
+
+
+def largest_triangle_three_buckets(data, threshold):
+    """
+    Return a downsampled version of data.
+    Parameters
+    ----------
+    data: list of lists/tuples
+        data must be formated this way: [[x,y], [x,y], [x,y], ...]
+                                    or: [(x,y), (x,y), (x,y), ...]
+    threshold: int
+        threshold must be >= 2 and <= to the len of data
+    Returns
+    -------
+    data, but downsampled using threshold
+    """
+
+    # Check if data and threshold are valid
+    if not isinstance(data, list):
+        raise LttbException("data is not a list")
+    if not isinstance(threshold, int) or threshold <= 2 or threshold >= len(data):
+        raise LttbException("threshold not well defined")
+    for i in data:
+        if not isinstance(i, (list, tuple)) or len(i) != 2:
+            raise LttbException("datapoints are not lists or tuples")
+
+    # Bucket size. Leave room for start and end data points
+    every = (len(data) - 2) / (threshold - 2)
+
+    a = 0  # Initially a is the first point in the triangle
+    next_a = 0
+    max_area_point = (0, 0)
+
+    sampled = [data[0]]  # Always add the first point
+
+    for i in range(0, threshold - 2):
+        # Calculate point average for next bucket (containing c)
+        avg_x = 0
+        avg_y = 0
+        avg_range_start = int(math.floor((i + 1) * every) + 1)
+        avg_range_end = int(math.floor((i + 2) * every) + 1)
+        avg_rang_end = avg_range_end if avg_range_end < len(data) else len(data)
+
+        avg_range_length = avg_rang_end - avg_range_start
+
+        while avg_range_start < avg_rang_end:
+            avg_x += data[avg_range_start][0]
+            avg_y += data[avg_range_start][1]
+            avg_range_start += 1
+
+        avg_x /= avg_range_length
+        avg_y /= avg_range_length
+
+        # Get the range for this bucket
+        range_offs = int(math.floor((i + 0) * every) + 1)
+        range_to = int(math.floor((i + 1) * every) + 1)
+
+        # Point a
+        point_ax = data[a][0]
+        point_ay = data[a][1]
+
+        max_area = -1
+
+        while range_offs < range_to:
+            # Calculate triangle area over three buckets
+            area = math.fabs(
+                (point_ax - avg_x)
+                * (data[range_offs][1] - point_ay)
+                - (point_ax - data[range_offs][0])
+                * (avg_y - point_ay)
+            ) * 0.5
+
+            if area > max_area:
+                max_area = area
+                max_area_point = data[range_offs]
+                next_a = range_offs  # Next a is this b
+            range_offs += 1
+
+        sampled.append(max_area_point)  # Pick this point from the bucket
+        a = next_a  # This a is the next a (chosen b)
+
+    sampled.append(data[len(data) - 1])  # Always add last
+
+    return sampled

python/biopro/server/recording.py

@@ -234,6 +234,7 @@ class RecordingProcess(Process):
                     if len(self._mqtt_send_data_ch_level) == 0:
                         for ch in result.channels():
                             self._mqtt_send_data_ch_level[ch] = MqttDataMessageHandler(self._mqtt_thread, 'data_server/device_data_stream/' + str(result.device) + '/' + str(ch) ) 
+                            self._mqtt_send_data_ch_level[50+ch] = MqttDataMessageHandler(self._mqtt_thread, 'data_server/device_data_stream/' + str(result.device) + '/' + str(ch+50) ) 
                     ret.append(result)
                 else:
                     continue