-update run script

- fixed raw data last data empty string error

ana_func.py

@@ -0,0 +1,201 @@
+import random, math, time
+import matplotlib.pyplot as plt 
+import numpy as np
+from scipy import signal
+import pandas as pd
+
+class dataAnalyticFunc():
+    def __init__(self, x_data, y_data, ref_data = None) ->None:
+        if len(x_data)!=len(y_data):
+            print('Scale of x_data and y_data are different.')
+            return None
+        self._raw_data = pd.DataFrame({'x_data':x_data,'y_data':y_data})
+        if ref_data:
+            print(ref_data[:10])
+            if len(ref_data) == len(self._raw_data):
+                self._raw_data['ref_data'] = ref_data
+        # print(self._raw_data['y_data'].tolist())
+
+    def find_peak(self, value_name, key_name):
+        signal.find_peaks_cwt(list(self._raw_data[value_name]),np.arange(100,200))
+
+    def trigger_point(self, value_data, key_data, trigger, order = 1, drop = 1):
+        if value_data not in list(self._raw_data) or key_data not in list(self._raw_data):
+            print( 'There is no '+ str(value_data) + 'in datafram. ')
+            return False
+        # print(self._raw_data[value_data].head(10))
+        last_row, last_key = [],False
+        for i,row in self._raw_data.sort_values(key_data)[::order].iterrows():
+            # print(row[key_data])
+            if row[value_data]*drop <= trigger*drop:
+                print (i,type(row))
+                last_row ,last_key = row,True
+                print(last_row[key_data])
+                break
+        # print(last_row==False)
+        if last_key ==False:
+            print('No data meet the criteria.')
+            return False
+        
+        # print(last_key)
+        # return [int(self._raw_data[key_data][last_key]), int(self._raw_data[value_data][last_key])]
+        return [last_row[key_data],last_row[value_data]]
+
+    def histogram_find_peak(self, bins =100): #, plot =True):
+        # if bins == 0:
+        #     bins = 100
+            # bins = round(len(self._raw_data)/10)
+        arr_list = self._raw_data['y_data'].tolist()
+        self._raw_data['hist_list'] = pd.cut(self._raw_data['y_data'], bins)
+ 
+        hist_list = self._raw_data['hist_list'].value_counts(sort=False).tolist()
+        cut_bins = self._raw_data['hist_list'].value_counts(sort=False).keys()
+        # print(cut_bins[0])
+        # hist_list = [len([j for j in self._raw_data_y if j>=hist_range[i] and j<hist_range[i+1]]) for i in range(len(hist_range)-1)]
+        peek_temp_list,_ = signal.find_peaks(list(hist_list))
+        peek_temp_list = list(peek_temp_list)
+        if np.sign(np.diff(hist_list))[0] < 0:
+            peek_temp_list = [0] + peek_temp_list
+        if np.sign(np.diff(hist_list))[-1] > 0:
+            peek_temp_list = peek_temp_list+[len(hist_list)-1]
+        peek_list = [i for i in peek_temp_list if hist_list[i] >= len(self._raw_data)/bins*3]
+        print(peek_list)
+        results_half = signal.peak_widths(hist_list, peek_list, rel_height=0.8)
+        # print(results_half)
+        peak_range = [ pd.Interval(cut_bins[round(results_half[2][i])].left, cut_bins[round(results_half[3][i])].right) for i in range(len(results_half[2]))]
+        # print(peak_range)
+        peak_wid = [ np.mean([ j for j in arr_list if j in i]) for i in peak_range ]
+        data_list = {'ground':round(min(peak_wid)), 'ceiling':round(max(peak_wid)) }
+        # print(peek_list)
+        # print("peak_wid = "+str(peak_wid))
+        print(data_list)
+            
+        return data_list, cut_bins
+
+    def cal_slop_array(self, window = 10):
+        # print(self._raw_data)
+        print('window')
+        if window*2+1 >= len(self._raw_data):
+            print('Scale of data are smaller than window.')
+            return False
+        slop_a = [0]*window
+        for i in range(len(self._raw_data)-window*2):
+            # print(list(self._raw_data['x_data'][i:i+window*2+1]))
+            s,_ = np.polyfit(list(self._raw_data['x_data'][i:i+window*2+1]), list(self._raw_data['y_data'][i:i+window*2+1]), 1)
+            # time.sleep(3)
+            # print(s)
+            slop_a.append(s)
+        slop_a = slop_a+[0]* window
+        self._raw_data['1D'] = slop_a
+
+        print(list(self._raw_data))
+        return slop_a
+
+    def guess_func(self,value_name='x_data', window = 10):
+        win = window
+        print('guess_func')
+        X = np.array([[math.pow(i,j) for j in range(4)] for i in np.linspace(-1*win,win,2*win+1)])
+        J = np.linalg.inv(X.T.dot(X)).dot(X.T)
+        slop_list = []
+        for i in range(len(self._raw_data[value_name][win:-1*win])):
+            slop_list.append(J.dot(np.array(self._raw_data[value_name][i:i+2*win+1])).tolist()[1])
+        slop_list = [slop_list[0]]*win +slop_list+ [slop_list[-1]]*win
+        self._raw_data[value_name+'_guess'] = slop_list
+        # print(value_name+'_guess')
+        # print(self._raw_data[value_name+'_guess'][:10])
+        return slop_list
+
+    def simple_slop(self,value_name='y_data',key_name='x_data'):
+        df = self._raw_data.sort_values(key_name)
+        slop_list = []
+        for i in range(len(df)-1):
+            df[value_name].iloc[i+1]-df[value_name].iloc[i]
+
+    def integral_func(self, int_value_name, int_key_name, sort=True ):
+        print('integral_func')
+        # df = self._raw_data[self._raw_data['y_data_guess']>0].sort_values(int_key_name) if sort else self._raw_data
+        print(self._raw_data[self._raw_data['x_data_guess']>0])
+        df = self._raw_data[self._raw_data['x_data_guess'] > 0]
+        temp_value_sum = 0
+        last_key = df[int_key_name].iloc[0]
+        temp_key_num = 0
+        integral_value = 0
+        for i in range(len(df)):
+            row = df.iloc[i]
+            if last_key == row[int_key_name] :
+                temp_value_sum += row[int_value_name]
+                temp_key_num += 1
+            else:
+                integral_value += (row[int_key_name]-last_key) * temp_value_sum/temp_key_num
+                last_key = row[int_key_name]
+                temp_value_sum = row[int_value_name]
+                temp_key_num = 1
+        return integral_value
+    
+    def SMA(self, win, n ,value_key='y_data'):
+        data_df = self._raw_data[value_key]
+        rev_sum = lambda data_df: data_df[::-1].rolling(window=win).mean()
+        self._raw_data['rolling_rev_mean'] = rev_sum(self._raw_data[value_key])#[::-1].reset_index(drop=True))
+        # print(self._raw_data)
+        
+        raw_data_rev = list(self._raw_data[value_key][::-1].reset_index(drop=True))
+        rolling_mean_rev = list(self._raw_data['rolling_rev_mean'][::-1].reset_index(drop=True))
+        win1 = len(raw_data_rev)//10
+        std_arr = [np.std(raw_data_rev[:win1])]  * win1
+        mean_arr =  [np.mean(raw_data_rev[:win1])] * win1
+        sum_temp = sum(raw_data_rev[:win1])
+        # start_time1 = time.time()
+        # std_arr2 = std_arr + [np.std(raw_data_rev[0:win1+ i ]) for i in range(len(raw_data_rev)-win1)]
+        # mean_arr2 = mean_arr + [np.mean(raw_data_rev[0:win1+ i ]) for i in range(len(raw_data_rev)-win1)]
+        end_time1 = time.time()
+        for ind_i in range(len(raw_data_rev)-win1):
+            key_temp =  win1 + ind_i
+            # sum_temp += raw_data_rev[key_temp]
+            # mean_arr.append(sum_temp/key_temp)
+            # std_arr.append(math.sqrt(sum([pow(i- mean_arr[-1],2) for i in raw_data_rev[:key_temp]])/key_temp))
+            mean_arr.append(np.mean(raw_data_rev[:key_temp]))
+            std_arr.append(np.std(raw_data_rev[:key_temp]))
+            if (rolling_mean_rev[key_temp+1]> mean_arr[-1] + n * std_arr[-1]) or (rolling_mean_rev[key_temp+1]< mean_arr[-1] - n * std_arr[-1]):
+                break
+        end_time2 = time.time()
+        # print('old',end_time1-start_time1)
+        print('time',end_time2-end_time1)
+        # print(std_arr[:win+10])
+        plt.plot([df_i + std_j*n for df_i, std_j in zip(mean_arr,std_arr)])
+        plt.plot([df_i - std_j*n for df_i, std_j in zip(mean_arr,std_arr)])
+        # plt.plot([df_i + std_j*4 for df_i, std_j in zip(mean_arr2,std_arr2)])
+        # plt.plot([df_i - std_j*4 for df_i, std_j in zip(mean_arr2,std_arr2)])
+        plt.plot(raw_data_rev[:len(std_arr)])
+        plt.plot(rolling_mean_rev)
+        plt.plot(mean_arr)
+        plt.plot(std_arr)
+        # df = list(data_df)[:600:-1]
+        # std_arr = [np.std(df[0:win+ i ]) for i in range(len(df)-win)]
+        # mean_arr = [np.mean(df[0:win+ i ]) for i in range(len(df)-win)]
+        # # print([[0,win+ i ] for i in range(len(df)-win)][:10])
+        # # print(len(std_arr),len(df))
+        # # fig, ax1 = plt.subplots()
+        # # ax2 = ax1.twinx()
+        # plt.plot(df[win:],color = 'black')
+        # plt.plot(std_arr, color='b')
+        # plt.plot([df_i + std_j*2 for df_i, std_j in zip(mean_arr,std_arr)])
+        # plt.plot(mean_arr,'.')
+        # # plt.plot(self._raw_data[value_key])
+        # # plt.plot(df.rolling(win).mean())
+        # # fig.tight_layout()
+        plt.show()
+        
+        # print(df)
+        # self._raw_data['rolling_mean'] = self._raw_data[value_key].rolling(win, center=True).mean()
+        # std_arr = self._raw_data[value_key].rolling(win, center=True).std()
+        # self._raw_data['rolling_std'] = std_arr
+        # var_arr = self._raw_data[value_key].rolling(win, center=True).var()
+        # self._raw_data['rolling_var'] = var_arr
+        # print(np.std(std_arr))
+        # print('std error: ', self._raw_data[value_key].sem()/self._raw_data[value_key].mean())
+        # # 
+        return [std_arr[-1],len(self._raw_data)-key_temp]
+    
+    def cutoff(self):
+        pass
+

csv_file/.gitignore

@@ -0,0 +1,4 @@
+# Ignore everything in this directory
+*
+# Except this file
+!.gitignore

db_test.py

@@ -56,7 +56,7 @@ def histogram_slope(x_data, y_data, slope =0, bins =0, plot =True):
 def run():
     conn = psycopg2.connect(database="postgres", user="biopro", password="BioProControlBox", host="127.0.0.1", port="5432")
     cur = conn.cursor()
-    id = 14304
+    id = 166
 
     sql_str = f'select "raw_data" from "recording_data_metas" WHERE id = {id}'
     cur.execute(sql_str)
@@ -83,6 +83,9 @@ def run():
     x_data, y_data ,peek_list = histogram_slope(channel_data[0],channel_data[1],bins = 200)
     print([y_data[i] for i in peek_list])
     print(peek_list, channel_data)
+    channel_data.append(1.5e6)
+    channel_data.append(4.5e6)
     return channel_data, peek_list
+
     # print([y_data])
 

fig/.gitignore

@@ -0,0 +1,4 @@
+# Ignore everything in this directory
+*
+# Except this file
+!.gitignore

main.py

@@ -3,23 +3,58 @@
 
 import time
 import sys
-import json
+import json,csv
 
 # from deviceManager import DeviceManager
 # from requests import Requests
 from mqtt import MqttThread
-from db_test import run 
+from test import VTmode_run, slop_run ,CVmode_run, movstd_run
 
 class Main():
-  def __init__(self, controller_id = 'b8:27:eb:18:f8:cc', mqtt_ip = '192.168.2.1') -> None:
+  def __init__(self, controller_id = 'dc:a6:32:0f:56:9d', mqtt_ip = '192.168.2.1') -> None:
       # setup mqtt thread
       self._mqttThread = MqttThread(self, controller_id, mqtt_ip, 1883, 'test')
       self._mqttThread.run()
   
   def get_analysis_data(self, input:str):
-      raw_data, peek_list = run()
-      self._mqttThread.publish(raw_data)
-     
+    start_time = time.time()
+    input_data = json.loads(input)['e']
+    # print(input_data['data_name'])
+    # print([i.split('-') for i in input_data['data_id']])
+    search_id = [int(i.split('-')[1]) for i in input_data['data_id']]
+    print(search_id)
+    with open('csv_file/output.csv', 'w', newline='') as csvfile:
+      writer = csv.writer(csvfile)
+      # print([[i[0],i[1]] for i in input_data.items()])
+      writer.writerows([[i[0],i[1]] for i in input_data.items()])
+      writer.writerow("")
+      if input_data['mode'] == 1:
+        first_flag = True
+        for i in search_id:
+          print(i)
+          raw_data, csv_data = VTmode_run(i, input_data['data_channel'], input_data['data']['persentage'])
+          fieldnames = list(csv_data.keys())
+          print(fieldnames)
+          dict_writer = csv.DictWriter(csvfile, fieldnames=fieldnames)
+          if first_flag:
+            # self._mqttThread.publish(raw_data)
+            dict_writer.writeheader()
+            first_flag = False
+          print('csv_data', csv_data)
+          
+          dict_writer.writerow(csv_data)
+      elif input_data['mode'] == 3:
+        raw_data = CVmode_run()
+      elif input_data['mode'] == 2:
+        for i in search_id:
+          raw_data = slop_run(i, input_data['data_channel'], input_data['data']['window'])
+      else:
+        for i in search_id:
+           raw_data = movstd_run(i, input_data['data_channel'])
+      self._mqttThread.publish(raw_data) 
+      end_time = time.time()
+      print("執行時間：%f 秒" % (end_time - start_time))
+
 
 if __name__ == '__main__':
     # if len(sys.argv) < 3:
@@ -30,6 +65,7 @@ if __name__ == '__main__':
 
     try:
         while True:
+          time.sleep(1)
           pass
     except (KeyboardInterrupt, SystemExit):
         print("Received keyboard interrupt, quitting ...")

mqtt.py

@@ -34,7 +34,7 @@ class MqttThread():
       print('use controller ID', self.__controller_ID)
 
       self._mqtt_client = mqtt.Client(self.__controller_ID + '_' + self._client_id)
-      self._mqtt_client.connect(self._mqtt_url, self._mqtt_port)
+      self._mqtt_client.connect(self._mqtt_url, self._mqtt_port, keepalive=3600)
       self._mqtt_client.on_connect = self.on_connect
       self._mqtt_client.on_disconnect = self.on_disconnect
       self._mqtt_client.on_message = self.on_message
@@ -65,7 +65,7 @@ class MqttThread():
     QoS = 2 if wait_for_ack else 0
     topic = f'{self.__controller_ID}/data_analysis'
     message =  json.dumps(payload)
-    print('publish', topic, message)
+    print('publish', topic)#, message)
     message_info = self._mqtt_client.publish(topic, message, QoS)
 
     if wait_for_ack:

new/ana_func.py

@@ -0,0 +1,239 @@
+import random, math, time
+import matplotlib.pyplot as plt 
+import numpy as np
+from scipy import signal
+import pandas as pd
+
+class dataAnalyticFunc():
+
+    def __init__(self, x_data, y_data, ref_data = None) ->None:
+        if len(x_data)!=len(y_data):
+            print('Scale of x_data and y_data are different.')
+            return None
+        self._raw_data = pd.DataFrame({'x_data':x_data,'y_data':y_data})
+        if ref_data:
+            print(ref_data[:10])
+            if len(ref_data) == len(self._raw_data):
+                self._raw_data['ref_data'] = ref_data
+        # print(self._raw_data['y_data'].tolist())
+
+    def find_peak(self, value_name, key_name):
+        signal.find_peaks_cwt(list(self._raw_data[value_name]),np.arange(100,200))
+
+    def trigger_point(self, value_data, key_data, trigger, gte =True ,head= False):
+        
+        if value_data not in list(self._raw_data) or key_data not in list(self._raw_data):
+            print( 'There is no '+ str(value_data) + 'in datafram. ')
+            return False
+        # start_time = time.time()
+        # # print(self._raw_data[value_data].head(10))
+        # last_row, last_key = [],False
+        # for i,row in self._raw_data.sort_values(key_data)[::order].iterrows():
+        #     # print(row[key_data])
+        #     if row[value_data]*drop <= trigger*drop:
+        #         print (i,type(row))
+        #         last_row ,last_key = row,True
+        #         print(last_row[key_data])
+        #         break
+        # # print(last_row==False)
+        # if last_key ==False:
+        #     print('No data meet the criteria.')
+        #     return False
+        # print([last_row[key_data],last_row[value_data]])
+        # print("origin mode end：%f 秒" % (time.time() - start_time))
+
+        df = self._raw_data[[ key_data,value_data]][self._raw_data[value_data]*(1 if gte else -1) >= trigger*(1 if gte else -1)]
+        # print(head)
+        # print(df[key_data].min() if head else df[key_data].max())
+        df_limit_list = df[df[key_data] == (df[key_data].min() if head else df[key_data].max())]
+        # print(df_limit_list)
+        df_lim_id = df_limit_list[value_data].idxmax() if gte else df_limit_list[value_data].idxmin()
+        tri_return = df_limit_list.loc[df_lim_id].tolist()
+        print(tri_return)
+        # return [int(self._raw_data[key_data][last_key]), int(self._raw_data[value_data][last_key])]
+        return tri_return
+
+    def histogram_find_peak(self, bins =100):
+        # if bins == 0:
+        #     bins = 100
+        #     bins = round(len(self._raw_data)/10)
+        start_time = time.time()
+        # arr_list = self._raw_data['y_data'].tolist()
+        self._raw_data['hist_list'] = pd.cut(self._raw_data['y_data'], bins)
+        print("cut db end：%f 秒" % (time.time() - start_time))
+        
+        # hist_list = self._raw_data['hist_list'].value_counts(sort=False).tolist()
+        start_time = time.time()
+        hist_list = [0]+ self._raw_data['hist_list'].value_counts(sort=False).tolist() +[0]
+        cut_bins = self._raw_data['hist_list'].value_counts(sort=False).keys()
+        print("1 end：%f 秒" % (time.time() - start_time))
+        # print(cut_bins[0])
+        # hist_list = [len([j for j in self._raw_data_y if j>=hist_range[i] and j<hist_range[i+1]]) for i in range(len(hist_range)-1)]
+        start_time = time.time()
+        peek_temp_list,_ = signal.find_peaks(list(hist_list))
+        print("find peak end：%f 秒" % (time.time() - start_time))
+        start_time = time.time()
+        peek_temp_list = list(peek_temp_list)
+        peek_list = [i for i in peek_temp_list if hist_list[i] >= len(self._raw_data)/bins*3]
+        print("2 end：%f 秒" % (time.time() - start_time))
+        start_time = time.time()
+        results_half = signal.peak_widths(hist_list, peek_list, rel_height=0.8)
+        print("find peak width end：%f 秒" % (time.time() - start_time))
+        # print(len(hist_list))
+        # print('peak list:',peek_list)
+        # print('peak half range' ,results_half)
+        # print(round(results_half[2][0]),round(results_half[3][0]))
+        start_time = time.time()
+        # peak_range = [ pd.Interval(cut_bins[(round(results_half[2][i])-1)if round(results_half[2][i])>1 else 0].left, cut_bins[(round(results_half[3][i])-1) if round(results_half[3][i])-1 < bins else bins-1].right) for i in range(len(results_half[2]))]
+        peak_range_array = [ [cut_bins[(round(results_half[2][i])-1)if round(results_half[2][i])>1 else 0].left, cut_bins[(round(results_half[3][i])-1) if round(results_half[3][i])-1 < bins else bins-1].right] for i in range(len(results_half[2]))]
+        print("3 end：%f 秒" % (time.time() - start_time))
+        # print(peak_range)
+
+        # if np.sign(np.diff(hist_list))[0] < 0 and hist_list[0]>= len(self._raw_data)/bins*3:
+        #     peek_list = [0] + peek_list
+        #     peak_range = [pd.Interval(cut_bins[0].left, cut_bins[0].right)] + peak_range
+        # if np.sign(np.diff(hist_list))[-1] > 0 and hist_list[-1]>= len(self._raw_data)/bins*3:
+        #     peek_list = peek_list + [len(hist_list)-1]
+        #     peak_range = peak_range + [pd.Interval(cut_bins[len(hist_list)-1].left, cut_bins[len(hist_list)-1].right)]
+        
+        # print('peak list:',peek_list)
+        start_time = time.time()
+        # print([self._raw_data['y_data'][self._raw_data['y_data'] in i].mean for i in peak_range])
+        peak_wid = [self._raw_data['y_data'][self._raw_data['y_data'].between(i[0],i[1])].mean() for i in peak_range_array]
+        # peak_wid = [ np.mean([ j for j in arr_list if j in i]) for i in peak_range ]
+        print(peak_wid)
+        # data_list = {'ground':round(min(peak_wid)), 'ceiling':round(max(peak_wid)) }
+        # print(peek_list)
+        print("4 end：%f 秒" % (time.time() - start_time))
+        print("peak_wid", peak_wid)
+
+        return peak_wid, cut_bins
+
+    def cal_slop_array(self, window = 10):
+        # print(self._raw_data)
+        print('window')
+        if window*2+1 >= len(self._raw_data):
+            print('Scale of data are smaller than window.')
+            return False
+        slop_a = [0]*window
+        for i in range(len(self._raw_data)-window*2):
+            # print(list(self._raw_data['x_data'][i:i+window*2+1]))
+            s,_ = np.polyfit(list(self._raw_data['x_data'][i:i+window*2+1]), list(self._raw_data['y_data'][i:i+window*2+1]), 1)
+            # time.sleep(3)
+            # print(s)
+            slop_a.append(s)
+        slop_a = slop_a+[0]* window
+        self._raw_data['1D'] = slop_a
+
+        print(list(self._raw_data))
+        return slop_a
+
+    def guess_func(self,value_name='x_data', window = 10):
+        win = window
+        print('guess_func')
+        X = np.array([[math.pow(i,j) for j in range(4)] for i in np.linspace(-1*win,win,2*win+1)])
+        J = np.linalg.inv(X.T.dot(X)).dot(X.T)
+        slop_list = []
+        for i in range(len(self._raw_data[value_name][win:-1*win])):
+            slop_list.append(J.dot(np.array(self._raw_data[value_name][i:i+2*win+1])).tolist()[1])
+        slop_list = [slop_list[0]]*win +slop_list+ [slop_list[-1]]*win
+        self._raw_data[value_name+'_guess'] = slop_list
+        # print(value_name+'_guess')
+        # print(self._raw_data[value_name+'_guess'][:10])
+        return slop_list
+
+    def simple_slop(self,value_name='y_data',key_name='x_data'):
+        df = self._raw_data.sort_values(key_name)
+        slop_list = []
+        for i in range(len(df)-1):
+            df[value_name].iloc[i+1]-df[value_name].iloc[i]
+
+    def integral_func(self, int_value_name, int_key_name, sort=True ):
+        print('integral_func')
+        # df = self._raw_data[self._raw_data['y_data_guess']>0].sort_values(int_key_name) if sort else self._raw_data
+        print(self._raw_data[self._raw_data['x_data_guess']>0])
+        df = self._raw_data[self._raw_data['x_data_guess'] > 0]
+        temp_value_sum = 0
+        last_key = df[int_key_name].iloc[0]
+        temp_key_num = 0
+        integral_value = 0
+        for i in range(len(df)):
+            row = df.iloc[i]
+            if last_key == row[int_key_name] :
+                temp_value_sum += row[int_value_name]
+                temp_key_num += 1
+            else:
+                integral_value += (row[int_key_name]-last_key) * temp_value_sum/temp_key_num
+                last_key = row[int_key_name]
+                temp_value_sum = row[int_value_name]
+                temp_key_num = 1
+        return integral_value
+
+    def SMA(self, win, n ,value_key='y_data'):
+        data_df = self._raw_data[value_key]
+        rev_sum = lambda data_df: data_df[::-1].rolling(window=win).mean()
+        self._raw_data['rolling_rev_mean'] = rev_sum(self._raw_data[value_key])#[::-1].reset_index(drop=True))
+        # print(self._raw_data)
+        
+        raw_data_rev = list(self._raw_data[value_key][::-1].reset_index(drop=True))
+        rolling_mean_rev = list(self._raw_data['rolling_rev_mean'][::-1].reset_index(drop=True))
+        win1 = len(raw_data_rev)//10
+        std_arr = [np.std(raw_data_rev[:win1])]  * win1
+        mean_arr =  [np.mean(raw_data_rev[:win1])] * win1
+        sum_temp = sum(raw_data_rev[:win1])
+        # start_time1 = time.time()
+        # std_arr2 = std_arr + [np.std(raw_data_rev[0:win1+ i ]) for i in range(len(raw_data_rev)-win1)]
+        # mean_arr2 = mean_arr + [np.mean(raw_data_rev[0:win1+ i ]) for i in range(len(raw_data_rev)-win1)]
+        end_time1 = time.time()
+        for ind_i in range(len(raw_data_rev)-win1):
+            key_temp =  win1 + ind_i
+            # sum_temp += raw_data_rev[key_temp]
+            # mean_arr.append(sum_temp/key_temp)
+            # std_arr.append(math.sqrt(sum([pow(i- mean_arr[-1],2) for i in raw_data_rev[:key_temp]])/key_temp))
+            mean_arr.append(np.mean(raw_data_rev[:key_temp]))
+            std_arr.append(np.std(raw_data_rev[:key_temp]))
+            if key_temp +1 < len(rolling_mean_rev):
+                if (rolling_mean_rev[key_temp+1]> mean_arr[-1] + n * std_arr[-1]) or (rolling_mean_rev[key_temp+1]< mean_arr[-1] - n * std_arr[-1]):
+                    break
+        end_time2 = time.time()
+        # print('old',end_time1-start_time1)
+        print('time',end_time2-end_time1)
+        # print(std_arr[:win+10])
+        plt.plot([df_i + std_j*n for df_i, std_j in zip(mean_arr,std_arr)])
+        plt.plot([df_i - std_j*n for df_i, std_j in zip(mean_arr,std_arr)])
+        # plt.plot([df_i + std_j*4 for df_i, std_j in zip(mean_arr2,std_arr2)])
+        # plt.plot([df_i - std_j*4 for df_i, std_j in zip(mean_arr2,std_arr2)])
+        plt.plot(raw_data_rev[:len(std_arr)])
+        plt.plot(rolling_mean_rev)
+        plt.plot(mean_arr)
+        plt.plot(std_arr)
+        # df = list(data_df)[:600:-1]
+        # std_arr = [np.std(df[0:win+ i ]) for i in range(len(df)-win)]
+        # mean_arr = [np.mean(df[0:win+ i ]) for i in range(len(df)-win)]
+        # # print([[0,win+ i ] for i in range(len(df)-win)][:10])
+        # # print(len(std_arr),len(df))
+        # # fig, ax1 = plt.subplots()
+        # # ax2 = ax1.twinx()
+        # plt.plot(df[win:],color = 'black')
+        # plt.plot(std_arr, color='b')
+        # plt.plot([df_i + std_j*2 for df_i, std_j in zip(mean_arr,std_arr)])
+        # plt.plot(mean_arr,'.')
+        # # plt.plot(self._raw_data[value_key])
+        # # plt.plot(df.rolling(win).mean())
+        # # fig.tight_layout()
+        plt.show()
+        
+        # print(df)
+        # self._raw_data['rolling_mean'] = self._raw_data[value_key].rolling(win, center=True).mean()
+        # std_arr = self._raw_data[value_key].rolling(win, center=True).std()
+        # self._raw_data['rolling_std'] = std_arr
+        # var_arr = self._raw_data[value_key].rolling(win, center=True).var()
+        # self._raw_data['rolling_var'] = var_arr
+        # print(np.std(std_arr))
+        # print('std error: ', self._raw_data[value_key].sem()/self._raw_data[value_key].mean())
+        # # 
+        return [mean_arr[-1],len(self._raw_data)-key_temp]
+    
+    def cutoff(self):
+        pass
+

new/analysis_mode.py

@@ -0,0 +1,85 @@
+from ana_func import dataAnalyticFunc
+import matplotlib.pyplot as plt
+import csv
+import pandas as pd
+import time
+
+async def CVmode_run():
+    
+    df = read_data_csv() 
+    print(df[0][:10])
+    data = [[df[0][i] for i in range(len(df[0])) if df[2][i] == 5], [df[1][i] for i in range(len(df[1])) if df[2][i] == 5]]
+    print(data[0][:10]) #y:Current[nA]/x:Voltage[uV]
+    CVmode = dataAnalyticFunc(data[1], data[0])
+    cal_slop_data = CVmode.guess_func()
+    print(len(cal_slop_data))
+    integel = CVmode.integral_func('x_data','y_data')
+    print(integel)
+
+    # return [data[1],data[0]]
+    return [[i for i in range(len(data[0]))],cal_slop_data]
+    # return [[i for i in range(len(df[0]))],[i/50 for i in df[0]]]
+# CVmode()
+
+def slop_run(id, channel_list, win):
+    # id = 166
+    # win = 20
+    print(win)
+    channel_data = read_data(id, channel_list)
+    slop_mode = dataAnalyticFunc(channel_data[0], channel_data[1])
+    cal_slop_data = slop_mode.cal_slop_array(window=win)
+    # return [[i for i in range(len(channel_data[0]))], channel_data[1]]
+    return [[i for i in range(len(channel_data[0]))], cal_slop_data]
+
+
+def VTmode_run(args):
+    x_data, y_data, percentage = args
+    data_list = {}
+    ### create dataAnalyticFunc instance
+    # print('x', x_data, 'y', y_data)
+    VT_mode = dataAnalyticFunc(x_data, y_data)
+    peak_wid ,_ = VT_mode.histogram_find_peak()
+    if len(peak_wid) != 2:
+        print('check if data fit VT mode.')
+        return [[], [[], []]],{}
+    VT_data_list = {'ground':round(min(peak_wid)), 'ceiling':round(max(peak_wid)) }
+    # print('peak list =', data_list['ground'])
+    data_list.update(VT_data_list)
+    peak_list = list(data_list.values())
+
+    trigger_line_down = (data_list['ceiling']-data_list['ground'])*percentage / 1e2 + data_list['ground']
+    trigger_line_up = (data_list['ceiling']-data_list['ground'])*(1-percentage / 1e2) + data_list['ground']
+
+    trigger = {str(percentage)+'% point':trigger_line_down, str(100-percentage)+'% point':trigger_line_up}
+    data_list.update(trigger)
+    
+    peak_list.append(trigger_line_down)
+    peak_list.append(trigger_line_up)
+
+    start_time = time.time()
+    triggerpoint = [VT_mode.trigger_point('y_data','x_data',trigger_line_down,gte = False,head = True), VT_mode.trigger_point('y_data','x_data',trigger_line_up,gte=True,head=False)]
+    print("trigger end：%f 秒" % (time.time() - start_time))
+    # slope_tan = math.atan2(triggerpoint[0][1]-triggerpoint[1][1],triggerpoint[0][0]-triggerpoint[1][0])
+    center_point = [(triggerpoint[0][0]+triggerpoint[1][0])/2,(triggerpoint[0][1]+triggerpoint[1][0])/2]
+    slope = (triggerpoint[0][1]-triggerpoint[1][1])/(triggerpoint[0][0]-triggerpoint[1][0])
+    center_point_dict = {'center point key': center_point[0], 'center point value':center_point[1], 'slope': slope}
+    print('trigger point',triggerpoint, slope, center_point)
+    data_list.update(center_point_dict)
+
+    # Plot
+    # plt.plot(channel_data[0],channel_data[1],'o',markersize=2)
+    # print(peak_list)
+    # for i in peak_list[1:]:
+    #     plt.hlines(i,min(channel_data[0]),max(channel_data[0]),color='r')
+    # for j in triggerpoint:
+    #     plt.plot(j[0],j[1],'o',markersize=5)
+    # plt.savefig('fig/test'+ str(id) +'.png')
+    # plt.close()
+    print([peak_list, triggerpoint] , data_list)
+    return [peak_list, triggerpoint] , data_list
+
+
+def movstd_run(channel_data):
+    movs_mode = dataAnalyticFunc(channel_data[0], channel_data[1])
+    result = movs_mode.SMA(20,5)
+    return result

new/callback_func.py

@@ -0,0 +1,111 @@
+import csv
+import time
+import json
+from analysis_mode import VTmode_run, slop_run ,CVmode_run, movstd_run
+from database_api import get_raw_id_list, get_raw_data, get_raw_data_with_time
+from concurrent.futures import ProcessPoolExecutor
+from db.base import Base, Session, engine
+from db.subject_data import SubjectData
+
+async def async_callback(topic, payload, conn, client):
+            """
+            payload format
+            {
+                "pattern": {
+                    "id": number,
+                    "name": string,
+                    "parameter": object,
+                },
+                "data": {
+                    "id": list[int],
+                    "channel: list[int]
+                },
+                "others": {
+                    ...
+                },
+            }
+            """
+            print(f"Received message on {topic}: {payload}")
+            start_time = time.time()
+            input_data = json.loads(payload)
+            meta = input_data['data']
+            analysis_pattern = input_data['pattern']
+            others = input_data.get('others')
+
+            # with open('csv_file/output.csv', 'w', newline='') as csvfile:
+            #     writer = csv.writer(csvfile)
+            #     write_header_done = False
+            meta_data = {}
+            for meta_id in meta['id']:
+                meta_data[meta_id] = {}
+                # TODO write file header
+                # create meta_data
+                if 'Time' in meta['channel']:
+                    for channel in meta['channel']:
+                         if channel != 'Time':
+                            meta_data[meta_id]['Time'] = []
+                            meta_data[meta_id][channel] = []
+                            raw_id_list =  await get_raw_id_list(conn, meta_id, channel)
+                            for raw_id in raw_id_list:
+                                time_data, raw_data =  await get_raw_data_with_time(conn, raw_id, channel)
+                                meta_data[meta_id]['Time'].extend(time_data)
+                                meta_data[meta_id][channel].extend(raw_data)
+                else:
+                    for channel in meta['channel']:
+                        meta_data[meta_id][channel] = []
+                        raw_id_list =  await get_raw_id_list(conn, meta_id, channel)
+                        for raw_id in raw_id_list:
+                            raw_data =  await get_raw_data(conn, raw_id, channel)
+                            meta_data[meta_id][channel].extend(raw_data)
+                    # results = await asyncio.gather(*(get_raw_data(conn, raw_id, channel) for raw_id in raw_id_list))
+        
+
+            with ProcessPoolExecutor() as executor:
+                channelX = meta['channel'][0]
+                channelY = meta['channel'][1]
+
+                if analysis_pattern['id'] == 0:
+                    for meta_id in meta['id']:
+                        data = [meta_id, meta_data[meta_id][channelX], meta_data[meta_id][channelY]]
+                        # mqtt publish data
+                        await client.publish("data_analysis", data)
+                elif analysis_pattern['id'] == 1:
+                    for meta_id, result in zip(meta['id'], executor.map(VTmode_run,  [[meta_data[meta_id][channelX],meta_data[meta_id][channelY],analysis_pattern['parameter']['percentage']] for meta_id in meta['id']])):
+                        print('result', meta_id, result)
+                                
+                            # do analysis function
+                            # result, csv_data = VTmode_run(x_data, y_data, analysis_pattern['parameter']['percentage'])
+                        data = [meta_id, meta_data[meta_id][channelX], meta_data[meta_id][channelY], *result[0]]
+                            
+                        # mqtt publish data
+                        await client.publish("data_analysis", data)
+                        
+                        # # create dict writer
+                        # fieldnames = list(csv_data.keys())
+                        # dict_writer = csv.DictWriter(csvfile, fieldnames=fieldnames)
+
+                        # # write data header
+                        # if not write_header_done:
+                        #     dict_writer.writeheader()
+                        #     write_header_done = True
+
+                        # # write data
+                        # dict_writer.writerow(csv_data)
+                elif analysis_pattern['id'] == 4:
+                    with Session() as session:
+                        for meta_id in meta['id']:
+                            print('data' , len(meta_data[meta_id][channelX]), len(meta_data[meta_id][channelY]))
+                            result = movstd_run([meta_data[meta_id][channelX], meta_data[meta_id][channelY]])
+                            print('movstd_run', result)
+                            print(others)
+                            subject_data = SubjectData(
+                                subject_id= others["subject"]["id"],
+                                project = others['project'],
+                                meta = others['project_meta'],
+                                mode= others['device'],
+                                data= result[0]
+                            )
+                            session.add(subject_data)
+                            session.commit()
+
+            print("執行時間：%f 秒" % (time.time() - start_time))

new/database_api.py

@@ -0,0 +1,23 @@
+async def get_raw_id_list(conn, id, channel):
+    with conn.cursor() as cursor:
+        sql_str = f'select "raw_data" from "recording_data_metas" WHERE id = {id}'
+        cursor.execute(sql_str)
+        raw_id_list = cursor.fetchone()[0][str(channel)]
+        return raw_id_list
+
+async def get_raw_data(conn, raw_id, channel):
+    with conn.cursor() as cursor:
+        sql_str = f'select "data" from "{channel}_recording_data_raws" WHERE id = {raw_id}'
+        cursor.execute(sql_str)
+        raw_data = cursor.fetchone()[0].replace('"***"',' ').split(" ")
+        raw_data_remove_time = [int(raw_data[idx]) for idx in range(len(raw_data)) if idx%2 and raw_data[idx] != '']
+        return raw_data_remove_time
+    
+async def get_raw_data_with_time(conn, raw_id, channel):
+    with conn.cursor() as cursor:
+        sql_str = f'select "data" from "{channel}_recording_data_raws" WHERE id = {raw_id}'
+        cursor.execute(sql_str)
+        raw_data = cursor.fetchone()[0].replace('"***"'," ").split(" ")
+        raw_data_remove_time = [int(raw_data[idx]) for idx in range(len(raw_data)) if idx%2 and raw_data[idx] != '']
+        raw_data_time = [int(raw_data[idx]) for idx in range(len(raw_data)) if idx%2 == 0 and raw_data[idx] != '']
+        return raw_data_time, raw_data_remove_time

new/db/base.py

@@ -0,0 +1,10 @@
+from sqlalchemy import create_engine
+# from sqlalchemy.ext.declarative import declarative_base
+from sqlalchemy.orm import sessionmaker, declarative_base
+
+SQLALCHEMY_DATABASE_URL = "postgresql://biopro:BioProControlBox@127.0.0.1:5432/postgres"
+
+engine = create_engine(SQLALCHEMY_DATABASE_URL)
+Session = sessionmaker(bind=engine)
+
+Base = declarative_base()

new/db/collection.py

@@ -0,0 +1,68 @@
+from sqlalchemy import Table, Column, String, MetaData, ForeignKey, JSON
+from sqlalchemy.sql import select, func
+from sqlalchemy.types import Integer, BigInteger, String, Boolean, TIMESTAMP, Numeric
+from sqlalchemy.dialects.postgresql import JSONB
+from .base import Session
+from .base import Base
+
+class Collection(Base):
+    __tablename__ = "collections"
+
+    id = Column(Integer, primary_key=True)
+    name = Column(String(255))
+    parent = Column(JSONB)
+    controller_id = Column(Integer)
+    type = Column(String(255))
+    description = Column(String(255))
+    deleted = Column(Boolean)
+    created_at = Column(TIMESTAMP(timezone=True), server_default=func.now())
+    updated_at = Column(TIMESTAMP(timezone=True), onupdate=func.now())
+
+    @classmethod
+    def create_collection(cls, collection_name, parent):
+        with Session() as session:
+            name = cls.check_name_duplicate(collection_name, parent, 0)
+            collection = Collection(
+                name = name,
+                parent = parent,
+                type= "folder",
+            )
+            session.add(collection)
+            session.commit()
+            print('a', collection.id)
+            return collection
+
+    @classmethod
+    def check_name_duplicate(cls, collection_name, parent, n, _session = None):
+        if _session == None:
+            with Session() as session:
+                result = session.query(Collection).filter(Collection.name == cls.generate_name(collection_name, n), Collection.parent == parent).first()
+                if result is None:
+                    return cls.generate_name(collection_name, n)
+                else:
+                    new_num = n + 1
+                    # new_name =  f"{collection_name}({new_num})" 
+                    return cls.check_name_duplicate(collection_name, parent, new_num, session)
+        else:
+            result = _session.query(Collection).filter(Collection.name == cls.generate_name(collection_name, n), Collection.parent == parent).first()
+            if result is None:
+                return cls.generate_name(collection_name, n)
+            else:
+                new_num = n + 1
+                # new_name =  f"{collection_name}({new_num})" 
+                return cls.check_name_duplicate(collection_name, parent, new_num, _session)
+
+    @classmethod
+    def generate_name(cls, collection_name, n):
+        if n==0:
+            return collection_name
+        else:
+            return f"{collection_name}({n})" 
+    
+    @classmethod
+    def find_collection(cls, collection_name, parent):
+        with Session() as session:
+            result = session.query(Collection).filter(Collection.name == collection_name, Collection.parent == parent).first()
+            return result
+    
+        

new/db/device.py

@@ -0,0 +1,51 @@
+from sqlalchemy import Table, Column, String, MetaData, ForeignKey, JSON
+from sqlalchemy.sql import select, func
+from sqlalchemy.types import Integer, BigInteger, String, Boolean, TIMESTAMP, Numeric, Float, BINARY, LargeBinary
+from sqlalchemy.dialects.postgresql import JSONB
+
+from .base import Base, Session
+
+class Device(Base):
+    __tablename__ = "devices"
+
+    id = Column(Integer, primary_key=True)
+    name = Column(String(255))
+    mac_address = Column(String(255))
+    serial_number = Column(String(255))
+    configuration = Column(JSONB)
+    library = Column(String(255))
+    library_version = Column(String(255))
+    device_version = Column(String(255))
+    type = Column(String(255))
+    battery = Column(Integer)
+    temperature = Column(Float)
+    auto_connect = Column(Boolean)
+    connect_priority = Column(Integer)
+    connect_time = Column(BigInteger)
+    parameter_set = Column(JSONB)
+    running = Column(Boolean)
+    calibration = Column(LargeBinary)
+    calibration_version = Column(Integer, default=-1)
+    user_auth = Column(JSONB)
+    deleted = Column(Boolean)
+    created_at = Column(TIMESTAMP(timezone=True), server_default=func.now())
+    updated_at = Column(TIMESTAMP(timezone=True), onupdate=func.now())
+
+    @classmethod
+    def update_device(cls, device, options):
+        with Session() as session:
+            result = session.query(Device).filter(Device.mac_address == device['mac_address']).first()
+            for key, value in options.items():
+                setattr(result, key, value)
+            session.commit()
+    
+    @classmethod
+    def get_device(cls, device):
+        with Session() as session:
+            result = session.query(Device).filter(Device.mac_address == device['mac_address']).first()
+            return result
+            
+
+
+    # def __repr__(self):
+    #     return f"User(id={self.id!r}, name={self.name!r}, fullname={self.task!r})"

new/db/meta.py

@@ -0,0 +1,24 @@
+from sqlalchemy import Table, Column, String, MetaData, ForeignKey, JSON
+from sqlalchemy.sql import select, func
+from sqlalchemy.types import Integer, BigInteger, String, Boolean, TIMESTAMP, Numeric
+from sqlalchemy.dialects.postgresql import JSONB
+
+from .base import Base, Session
+
+class RecordingDataMeta(Base):
+    __tablename__ = "recording_data_metas"
+
+    id = Column(Integer, primary_key=True)
+    path = Column(String(255))
+    name = Column(String(255))
+    parent = Column(JSONB)
+    size = Column(String(255))
+    time_duration = (String(255))
+    raw_data = Column(JSONB)
+    project = Column(Integer)
+    deleted = Column(Boolean, default = False)
+    created_at = Column(TIMESTAMP(timezone=True), server_default=func.now())
+    updated_at = Column(TIMESTAMP(timezone=True), onupdate=func.now())
+
+    # def __repr__(self):
+    #     return f"User(id={self.id!r}, name={self.name!r}, fullname={self.task!r})"

new/db/project.py

@@ -0,0 +1,24 @@
+from sqlalchemy import Table, Column, String, MetaData, ForeignKey, JSON
+from sqlalchemy.sql import select, func
+from sqlalchemy.types import Integer, BigInteger, String, Boolean, TIMESTAMP, Numeric
+from sqlalchemy.dialects.postgresql import JSONB
+
+from .base import Base
+
+class Project(Base):
+    __tablename__ = "project"
+
+    id = Column(Integer, primary_key=True)
+    name = Column(String)
+    desc = Column(String)
+    task = Column(JSONB)
+    cycle = Column(JSONB)
+    device = Column(JSONB)
+    uuid = Column(String(36))
+    user_auth = Column(JSONB)
+    deleted = Column(Boolean, default = False)
+    created_at = Column(TIMESTAMP(timezone=True), server_default=func.now())
+    updated_at = Column(TIMESTAMP(timezone=True), onupdate=func.now())
+
+    # def __repr__(self):
+    #     return f"User(id={self.id!r}, name={self.name!r}, fullname={self.task!r})"

new/db/project_meta.py

@@ -0,0 +1,29 @@
+from sqlalchemy import Table, Column, String, MetaData, ForeignKey, JSON
+from sqlalchemy.sql import select, func
+from sqlalchemy.types import Integer, BigInteger, String, Boolean, TIMESTAMP, Numeric
+from sqlalchemy.dialects.postgresql import JSONB
+
+from .base import Base, Session
+
+class MetaProjectInfo(Base):
+    __tablename__ = "project_metas"
+
+    id = Column(Integer, primary_key=True)
+    project = Column(String(36))
+    cycle = Column(JSONB)
+    task = Column(JSONB)
+    serial_number = Column(Integer)
+    deleted = Column(Boolean, default = False)
+    created_at = Column(TIMESTAMP(timezone=True), server_default=func.now())
+    updated_at = Column(TIMESTAMP(timezone=True), onupdate=func.now())
+
+    @classmethod
+    def create_project_meta(cls, project):
+        with Session() as session:
+            project_meta = MetaProjectInfo(project = project['project'], cycle= project['cycle'], task=project['task'], serial_number=int(project['serial_number']))
+            session.add(project_meta)
+            session.commit()
+            return project_meta.id
+
+    # def __repr__(self):
+    #     return f"User(id={self.id!r}, name={self.name!r}, fullname={self.task!r})"

new/db/project_report.py

@@ -0,0 +1,24 @@
+from sqlalchemy import Table, Column, String, MetaData, ForeignKey, JSON
+from sqlalchemy.sql import select, func
+from sqlalchemy.types import Integer, BigInteger, String, Boolean, TIMESTAMP, Numeric
+from sqlalchemy.dialects.postgresql import JSONB
+
+from .base import Base
+
+class ProjectReport(Base):
+    __tablename__ = "project_reports"
+
+    id = Column(Integer, primary_key=True)
+    name = Column(String)
+    desc = Column(String)
+    task = Column(JSONB)
+    cycle = Column(JSONB)
+    device = Column(JSONB)
+    uuid = Column(String(36))
+    user_auth = Column(JSONB)
+    deleted = Column(Boolean, default = False)
+    created_at = Column(TIMESTAMP(timezone=True), server_default=func.now())
+    updated_at = Column(TIMESTAMP(timezone=True), onupdate=func.now())
+
+    # def __repr__(self):
+    #     return f"User(id={self.id!r}, name={self.name!r}, fullname={self.task!r})"

new/db/recording_data.py

@@ -0,0 +1,88 @@
+from sqlalchemy import Table, Column, String, MetaData, ForeignKey, JSON
+from sqlalchemy.sql import select, func
+from sqlalchemy.types import Integer, BigInteger, String, Boolean, TIMESTAMP, Numeric, Text
+from sqlalchemy.dialects.postgresql import JSONB
+from db.base import Session
+
+from .base import Base
+
+# build a model class with a specific table name
+def get_raw_model(channel):
+    tablename = str(channel) + '_recording_data_raws'  # dynamic table name
+    class_name = 'RECORDING_DATA_RAWS' # dynamic class name
+    print('get_raw_model', tablename)
+
+    for mapper in Base.registry.mappers:
+        cls = mapper.class_
+        classname = cls.__name__
+        tblname = cls.__tablename__
+        print(cls, classname, tblname)
+        if (classname == class_name):
+            if tblname == tablename:
+                return cls
+
+    Model = type(class_name, (RECORDING_DATA_RAWS,), {
+        '__tablename__': tablename
+    })
+    return Model
+
+
+class RECORDING_DATA_RAWS(Base):
+    __abstract__ = True
+
+    id = Column(Integer, primary_key=True)
+    name = Column(String(255))
+    parent = Column(Integer)
+    size = Column(String(255))
+    path = Column(JSONB)
+    uuid = Column(String(255))
+    serial_number = Column(String(255))
+    data_format = Column(String(255))
+    channel = Column(Integer)
+    start_time = Column(String(255))
+    end_time = Column(String(255))
+    data = Column(Text)
+    compressed = Column(Boolean)
+    deleted = Column(Boolean)
+    created_at = Column(TIMESTAMP(timezone=True), server_default=func.now())
+    updated_at = Column(TIMESTAMP(timezone=True), onupdate=func.now())
+
+    # @classmethod
+    # def create_subject_data(cls, subject_id, project, meta, data):
+    #     with Session() as session:
+    #         subject = Subject(
+    #             subject_id = subject_id,
+    #             project = project,
+    #             meta= meta,
+    #             data = data
+    #         )
+    #         session.add(subject)
+    #         session.commit()
+    #         return subject
+        
+    # @classmethod
+    # def check_name_duplicate(cls, collection_name, parent, n):
+    #     with Session() as session:
+    #         result = session.query(Collection).filter(Collection.name == cls.generate_name(collection_name, n), Collection.parent == parent).first()
+    #         if result is None:
+    #             return cls.generate_name(collection_name, n)
+    #         else:
+    #             new_num = n + 1
+    #             # new_name =  f"{collection_name}({new_num})" 
+    #             return cls.check_name_duplicate(collection_name, parent, new_num)
+
+    # @classmethod
+    # def generate_name(cls, collection_name, n):
+    #     if n==0:
+    #         return collection_name
+    #     else:
+    #         return f"{collection_name}({n})" 
+    
+    # @classmethod
+    # def find_data(cls, id):
+    #     with Session() as session:
+
+    #         result = session.query(RECORDING_DATA_RAWS).first()
+    #         return result
+    
+        

new/db/subject_data.py

@@ -0,0 +1,60 @@
+from sqlalchemy import Table, Column, String, MetaData, ForeignKey, JSON
+from sqlalchemy.sql import select, func
+from sqlalchemy.types import Integer, BigInteger, String, Boolean, TIMESTAMP, Numeric
+from sqlalchemy.dialects.postgresql import JSONB
+from db.base import Session
+
+from .base import Base
+
+class SubjectData(Base):
+    __tablename__ = "subject_datas"
+
+    id = Column(Integer, primary_key=True)
+    subject_id = Column(Integer)
+    mode = Column(JSONB)
+    data = Column(JSONB)
+    user_auth = Column(JSONB)
+    meta = Column(String(255))
+    project = Column(String(255))
+    deleted = Column(Boolean)
+    created_at = Column(TIMESTAMP(timezone=True), server_default=func.now())
+    updated_at = Column(TIMESTAMP(timezone=True), onupdate=func.now())
+
+    @classmethod
+    def create_subject_data(cls, subject_id, project, meta, data):
+        with Session() as session:
+            subject = SubjectData(
+                subject_id = subject_id,
+                project = project,
+                meta= meta,
+                data = data
+            )
+            session.add(subject)
+            session.commit()
+            return subject
+        
+    # @classmethod
+    # def check_name_duplicate(cls, collection_name, parent, n):
+    #     with Session() as session:
+    #         result = session.query(Collection).filter(Collection.name == cls.generate_name(collection_name, n), Collection.parent == parent).first()
+    #         if result is None:
+    #             return cls.generate_name(collection_name, n)
+    #         else:
+    #             new_num = n + 1
+    #             # new_name =  f"{collection_name}({new_num})" 
+    #             return cls.check_name_duplicate(collection_name, parent, new_num)
+
+    # @classmethod
+    # def generate_name(cls, collection_name, n):
+    #     if n==0:
+    #         return collection_name
+    #     else:
+    #         return f"{collection_name}({n})" 
+    
+    # @classmethod
+    # def find_collection(cls, collection_name, parent):
+    #     with Session() as session:
+    #         result = session.query(Collection).filter(Collection.name == collection_name, Collection.parent == parent).first()
+    #         return result
+    
+        

new/main.py

@@ -0,0 +1,13 @@
+import asyncio
+import psycopg2
+from mqtt_client import MqttClient
+
+async def main():
+    loop = asyncio.get_event_loop()
+    conn = psycopg2.connect(database="postgres", user="biopro", password="BioProControlBox", host="127.0.0.1", port="5432")
+    client = MqttClient("dc:a6:32:0f:56:9d", "192.168.2.1", 1883, loop, conn)
+    
+    while True:
+        await asyncio.sleep(1)
+
+asyncio.run(main())

new/mqtt_client.py

@@ -0,0 +1,28 @@
+import paho.mqtt.client as mqtt
+import json
+from callback_func import async_callback
+
+class MqttClient:
+    def __init__(self, mqtt_id, broker_address, broker_port, loop, conn):
+        self._loop = loop
+        self._conn = conn
+        self._mqtt_id = mqtt_id
+        
+        self._client = mqtt.Client(self._mqtt_id)
+        self._client.connect(broker_address, broker_port, 3600)
+        self._client.on_connect = self.on_connect
+        self._client.on_message = self.on_message
+        self._client.loop_start()
+
+    def on_connect(self, client, userdata, flags, rc):
+        print("Connected with result code "+str(rc))
+        client.subscribe(f"{self._mqtt_id}_data_analysis/#")
+
+    def on_message(self, client, userdata, msg):
+        self._loop.create_task(async_callback(msg.topic, msg.payload.decode(), self._conn, self))
+    
+    async def publish(self, topic, payload, qos=0, retain=False):
+        payload = json.dumps(payload)
+        _topic = f"{self._mqtt_id}/{topic}"
+        self._client.publish(_topic, payload, qos, retain)
+        print('publish success')

run-data-analysis.sh

@@ -0,0 +1,4 @@
+#!/bin/bash
+
+cd /home/pi/data-analysis
+python3 -u new/main.py

test.py

@@ -0,0 +1,123 @@
+# import psycopg2
+import ana_func
+import matplotlib.pyplot as plt
+import csv
+import pandas as pd
+import numpy as np
+
+# def search_id_by_str(keyword):
+#     conn = psycopg2.connect(database="postgres", user="biopro", password="BioProControlBox", host="127.0.0.1", port="5432")
+#     cur = conn.cursor()
+#     sql_str = f"select id from recording_data_metas WHERE name ILIKE '%"+keyword+"%'"
+#     cur.execute(sql_str)
+#     data = cur.fetchall()
+#     data = [i[0] for i in data]
+#     return data
+
+# def read_data(id,channel_list):
+#     conn = psycopg2.connect(database="postgres", user="biopro", password="BioProControlBox", host="127.0.0.1", port="5432")
+#     cur = conn.cursor()
+
+#     sql_str = f'select "raw_data" from "recording_data_metas" WHERE id = {id}'
+#     cur.execute(sql_str)
+#     data = cur.fetchall()[0][0]
+#     channel_data = []
+#     for channel in channel_list:
+#         i = str(channel)
+#         channel_data_temp = []
+#         for j in data[i]:
+#             # print(j)
+#             sql_str = f'select "data" from "{i}_recording_data_raws" WHERE id = {j}'
+#             cur.execute(sql_str)
+#             raw_data = cur.fetchall()[0][0]
+#             raw_data = raw_data.replace('"***"',' ').split(" ")
+#             channel_data_temp =channel_data_temp + [int(raw_data[k]) for k in range(len(raw_data)) if k%2]
+#             # print(len(channel_data_temp))
+#         channel_data.append(channel_data_temp)
+
+#     return channel_data
+
+def read_data_csv(file_name=0, start_row=0):
+    df = pd.read_excel("read_file/2021-8-9-15-42-46-0_CV discussion.xlsx",skiprows = 63, usecols="B,D,F,H")
+    data = [list(df['Current[nA]']),list(df['Voltage[uV]']),list(df['Unnamed: 7'])]
+    # print(df)
+    return data
+# read_data_csv()
+
+def CVmode_run():
+    df = read_data_csv() 
+    print(df[0][:10])
+    data = [[df[0][i] for i in range(len(df[0])) if df[2][i] == 5], [df[1][i] for i in range(len(df[1])) if df[2][i] == 5]]
+    print(data[0][:10]) #y:Current[nA]/x:Voltage[uV]
+    CVmode = ana_func.dataAnalyticFunc(data[1], data[0])
+    cal_slop_data = CVmode.guess_func()
+    print(len(cal_slop_data))
+    integel = CVmode.integral_func('x_data','y_data')
+    print(integel)
+
+    # return [data[1],data[0]]
+    return [[i for i in range(len(data[0]))],cal_slop_data]
+    # return [[i for i in range(len(df[0]))],[i/50 for i in df[0]]]
+# CVmode()
+
+def slop_run(id, channel_list, win):
+    # id = 166
+    # win = 20
+    print(win)
+    channel_data = read_data(id, channel_list)
+    slop_mode = ana_func.dataAnalyticFunc(channel_data[0], channel_data[1])
+    cal_slop_data = slop_mode.cal_slop_array(window=win)
+    # return [[i for i in range(len(channel_data[0]))], channel_data[1]]
+    return [[i for i in range(len(channel_data[0]))], cal_slop_data]
+
+
+def VTmode_run(id, channel_list, perc):
+    # id = 166
+    percentage = perc/100
+    data_list = {}
+    print('channel_list', channel_list)
+    print(percentage)
+    channel_data = read_data(id, channel_list)
+    VT_mode = ana_func.dataAnalyticFunc(channel_data[0], channel_data[1])
+    VT_data_list ,_ = VT_mode.histogram_find_peak()
+    # print('peak list =', data_list['ground'])
+    data_list.update(VT_data_list)
+    peak_list = list(data_list.values())
+
+    trigger_line_down = (data_list['ground']-data_list['ceiling'])*percentage + data_list['ceiling']
+    trigger_line_up = (data_list['ground']-data_list['ceiling'])*(1-percentage) + data_list['ceiling']
+
+    trigger = {str(100-perc)+'% point':trigger_line_down, str(perc)+'% point':trigger_line_up}
+    data_list.update(trigger)
+    
+    peak_list.append(trigger_line_down)
+    peak_list.append(trigger_line_up)
+    triggerpoint = [VT_mode.trigger_point('y_data','x_data',trigger_line_down, order=-1,drop=-1), VT_mode.trigger_point('y_data','x_data',trigger_line_up)]
+    # slope_tan = math.atan2(triggerpoint[0][1]-triggerpoint[1][1],triggerpoint[0][0]-triggerpoint[1][0])
+    center_point = [(triggerpoint[0][0]+triggerpoint[1][0])/2,(triggerpoint[0][1]+triggerpoint[1][0])/2]
+    slope = (triggerpoint[0][1]-triggerpoint[1][1])/(triggerpoint[0][0]-triggerpoint[1][0])
+    center_point_dict = {'center point key': center_point[0], 'center point value':center_point[1], 'slope': slope}
+    print('trigger point',triggerpoint, slope, center_point)
+    data_list.update(center_point_dict)
+    
+    resultList = [id , *channel_data, peak_list, triggerpoint]
+
+    channel_data.append(peak_list)
+    channel_data.append(triggerpoint)
+
+    return resultList, data_list
+
+
+def movstd_run(channel_data):
+# def movstd_run(id, channel_list):
+#     channel_data = read_data(id, channel_list)
+    movs_mode = ana_func.dataAnalyticFunc(channel_data[0], channel_data[1])
+    movs_mode.SMA(20,5)
+    return np.mean(channel_data[0])
+    
+
+
+if __name__ == '__main__':
+    df = pd.read_excel("Cynthia reference.xlsx",sheet_name="ELITE EIS Data set", skiprows=7)
+    data = list(df["Impedance [ohm]"])[2:][-6000:]
+    print(movstd_run([data,data]))

test/test_asyncio.py

@@ -0,0 +1,24 @@
+import asyncio
+import time
+
+async def say_after(delay, what):
+    await asyncio.sleep(delay)
+    print(what)
+
+async def main():
+    asyncio.create_task(
+        say_after(1, time.time()))
+
+    task2 = asyncio.create_task(
+        say_after(2, 'world'))
+
+    print(f"started at {time.strftime('%X')}")
+
+    # Wait until both tasks are completed (should take
+    # around 2 seconds.)
+
+    await task2
+
+    print(f"finished at {time.strftime('%X')}")
+
+asyncio.run(main())

test/test_asyncio_csv.py

@@ -0,0 +1,42 @@
+import asyncio
+import csv
+import aiofiles
+
+async def write_rows_to_csv(rows, filename):
+    async with aiofiles.open(filename, 'w', newline='') as csvfile:
+        writer = csv.writer(csvfile)
+        for row in rows:
+            await writer.writerow(row)
+
+async def main():
+    rows = [
+        ['Name', 'Age', 'Country'],
+        ['Alice', 30, 'USA'],
+        ['Bob', 35, 'Canada'],
+        ['Charlie', 40, 'UK'],
+    ]
+
+    await write_rows_to_csv(rows, 'people.csv')
+
+asyncio.run(main())
+
+# import asyncio
+# import csv
+
+# async def write_rows_to_csv(rows, filename):
+#     async with open(filename, 'w', newline='') as csvfile:
+#         writer = csv.writer(csvfile)
+#         for row in rows:
+#             await writer.writerow(row)
+
+# async def main():
+#     rows = [
+#         ['Name', 'Age', 'Country'],
+#         ['Alice', 30, 'USA'],
+#         ['Bob', 35, 'Canada'],
+#         ['Charlie', 40, 'UK'],
+#     ]
+
+#     await write_rows_to_csv(rows, 'people.csv')
+
+# asyncio.run(main())

test/test_asyncio_executor_map.py

@@ -0,0 +1,21 @@
+import concurrent.futures
+
+def my_function(args):
+    a, b, c = args
+    # Do something with the arguments
+    return a + b + c
+
+def main():
+    with concurrent.futures.ProcessPoolExecutor() as executor:
+        # Create a list of argument tuples
+        arguments = [(1, 2, 3), (1, 2, 4), (1, 2, 5), (1, 2, 6)]
+
+        # Pass the list of argument tuples to map
+        results = executor.map(my_function, arguments)
+
+        # Print the results
+        for result in results:
+            print(result)
+
+if __name__ == '__main__':
+    main()

test/test_asyncio_in_task_control.py

@@ -0,0 +1,70 @@
+import asyncio
+
+d = []
+
+# async def task1():
+#     print("Running task 1")
+#     await asyncio.sleep(2)
+#     d.append(1)
+#     print("Task 1 completed")
+
+# async def task2():
+#     print("Running task 2")
+#     await asyncio.sleep(3)
+#     d.append(2)
+#     print("Task 2 completed")
+
+# async def task3():
+#     print("Running task 3")
+#     await asyncio.sleep(3)
+#     d.append(3)
+#     print("Task 3 completed")
+
+# async def task4():
+#     print("Running task 4")
+#     await asyncio.sleep(1)
+#     print(d)
+#     d.append(4)
+#     print("Task 4 completed")
+
+# async def main():
+#     # task_list1 = [task1(), task2()]
+#     task_list1 = [asyncio.create_task(task1()), asyncio.create_task(task2())]
+#     done, pending = await asyncio.wait(task_list1, return_when=asyncio.FIRST_COMPLETED)
+#     print("First task completed, starting task_list2")
+#     print(d)
+
+#     task_list2 = [task3(), task4()]
+#     await asyncio.gather(*task_list2)
+
+
+# asyncio.run(main())
+
+
+import asyncio
+
+async def first_completed(*tasks):
+    tasks = [asyncio.create_task(task) for task in tasks]
+
+    done, pending = await asyncio.wait(tasks, return_when=asyncio.FIRST_COMPLETED)
+
+    # for task in pending:
+    #     task.cancel()
+
+    # return done.pop().result()
+
+async def task1():
+    await asyncio.sleep(1)
+    print('1')
+    # return 'task 1 result'
+
+async def task2():
+    await asyncio.sleep(2)
+    print('2')
+    # return 'task 2 result'
+
+async def main():
+    result = await first_completed(task1(), task2())
+    print(result)
+
+asyncio.run(main())

test/test_asyncio_multi_process.py

@@ -0,0 +1,81 @@
+import concurrent.futures
+import math
+import time
+import asyncio
+
+PRIMES = [
+    112272535095293,
+    112582705942171,
+    112272535095293,
+    115280095190773,
+    115797848077099,
+    1099726899285419]
+
+def is_prime(n):
+    if n < 2:
+        return False
+    if n == 2:
+        return True
+    if n % 2 == 0:
+        return False
+
+    sqrt_n = int(math.floor(math.sqrt(n)))
+    for i in range(3, sqrt_n + 1, 2):
+        if n % i == 0:
+            return False
+    return True
+
+
+async def my_async_func(arg):
+    return is_prime(arg)
+
+def sync_wrapper(arg):
+    loop = asyncio.new_event_loop()
+    asyncio.set_event_loop(loop)
+    result = loop.run_until_complete(my_async_func(arg))
+    loop.close()
+    return result
+
+
+
+# def main():
+#     start = time.time()
+#     # Use the sync_wrapper function with Executor.map
+#     with concurrent.futures.ProcessPoolExecutor() as executor:
+#         for number, prime in zip(PRIMES, executor.map(sync_wrapper, PRIMES)):
+#             print('%d is prime: %s' % (number, prime))
+#     # process
+#     # with concurrent.futures.ProcessPoolExecutor() as executor:
+#     #     for number, prime in zip(PRIMES, executor.map(is_prime, PRIMES)):
+#     #         print('%d is prime: %s' % (number, prime))
+#     # thread
+#     # with concurrent.futures.ThreadPoolExecutor() as executor:
+#     #     for number, prime in zip(PRIMES, executor.map(is_prime, PRIMES)):
+#     #         print('%d is prime: %s' % (number, prime))
+#     #single
+#     # for prime in PRIMES:
+#     #     print(prime, is_prime(prime))
+#     print('done', time.time() - start)
+
+async def main():
+    start = time.time()
+    # Use the sync_wrapper function with Executor.map
+    with concurrent.futures.ProcessPoolExecutor() as executor:
+        for number, prime in zip(PRIMES, executor.map(sync_wrapper, PRIMES)):
+            print('%d is prime: %s' % (number, prime))
+    # process
+    # with concurrent.futures.ProcessPoolExecutor() as executor:
+    #     for number, prime in zip(PRIMES, executor.map(is_prime, PRIMES)):
+    #         print('%d is prime: %s' % (number, prime))
+    # thread
+    # with concurrent.futures.ThreadPoolExecutor() as executor:
+    #     for number, prime in zip(PRIMES, executor.map(is_prime, PRIMES)):
+    #         print('%d is prime: %s' % (number, prime))
+    #single
+    # for prime in PRIMES:
+    #     print(prime, is_prime(prime))
+    print('done', time.time() - start)
+
+if __name__ == '__main__':
+    # main()
+    asyncio.run(main())