-update run script

- fixed raw data last data empty string error

ana_func.py

@@ -131,4 +131,71 @@ class dataAnalyticFunc():
                 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
 

main.py

@@ -8,7 +8,7 @@ import json,csv
 # from deviceManager import DeviceManager
 # from requests import Requests
 from mqtt import MqttThread
-from test import VTmode_run, slop_run ,CVmode_run
+from test import VTmode_run, slop_run ,CVmode_run, movstd_run
 
 class Main():
   def __init__(self, controller_id = 'dc:a6:32:0f:56:9d', mqtt_ip = '192.168.2.1') -> None:
@@ -45,9 +45,12 @@ class Main():
           dict_writer.writerow(csv_data)
       elif input_data['mode'] == 3:
         raw_data = CVmode_run()
-      else:
+      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))

new/ana_func.py

@@ -20,58 +20,94 @@ class dataAnalyticFunc():
     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):
+    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
-        # 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]]
+        # 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))
 
-    def histogram_find_peak(self, bins =100): #, plot =True):
+        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)
-        arr_list = self._raw_data['y_data'].tolist()
+        #     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)
- 
-        hist_list = self._raw_data['hist_list'].value_counts(sort=False).tolist()
+        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)
-        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)
+        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(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("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)
-        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)) }
+
+        # 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("peak_wid = "+str(peak_wid))
-        print(data_list)
-            
-        return data_list, cut_bins
+        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)
@@ -133,3 +169,71 @@ class dataAnalyticFunc():
                 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

@@ -5,6 +5,7 @@ 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]]
@@ -31,32 +32,40 @@ def slop_run(id, channel_list, win):
     return [[i for i in range(len(channel_data[0]))], cal_slop_data]
 
 
-def VTmode_run(x_data, y_data, percentage):
+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)
-    VT_data_list ,_ = VT_mode.histogram_find_peak()
+    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['ground']-data_list['ceiling'])*percentage / 1e2 + data_list['ceiling']
-    trigger_line_up = (data_list['ground']-data_list['ceiling'])*(1-percentage / 1e2) + data_list['ceiling']
+    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(100-percentage)+'% point':trigger_line_down, str(percentage)+'% point':trigger_line_up}
+    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)
-    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)]
+
+    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)
@@ -66,5 +75,11 @@ def VTmode_run(x_data, y_data, percentage):
     #     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

@@ -1,8 +1,11 @@
 import csv
 import time
 import json
-from analysis_mode import VTmode_run, slop_run ,CVmode_run
-from database_api import get_raw_id_list, get_raw_data
+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):
             """
@@ -16,7 +19,10 @@ async def async_callback(topic, payload, conn, client):
                 "data": {
                     "id": list[int],
                     "channel: list[int]
-                }
+                },
+                "others": {
+                    ...
+                },
             }
             """
             print(f"Received message on {topic}: {payload}")
@@ -24,42 +30,82 @@ async def async_callback(topic, payload, conn, client):
             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
-                for meta_id in meta['id']:
-                    # TODO write file header
-                    print(input_data)
-
-                    # create meta_data
-                    meta_data = {}
+            # 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']:
-                        meta_data[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[channel].extend(raw_data)
+                            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))
-                    x_data = meta_data[meta['channel'][0]]
-                    y_data = meta_data[meta['channel'][1]]
-                    
-                    # do analysis function
-                    result, csv_data = VTmode_run(x_data, y_data, analysis_pattern['parameter']['percentage'])
-                    data = [meta_id, x_data, y_data, *result]
-                    
-                    # 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
+            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()
 
-                    # write data
-                    dict_writer.writerow(csv_data)
             print("執行時間：%f 秒" % (time.time() - start_time))

new/database_api.py

@@ -10,5 +10,14 @@ async def get_raw_data(conn, raw_id, channel):
         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]
-        return raw_data_remove_time
+        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
+    
+        

run-data-analysis.sh

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

test.py

@@ -1,40 +1,41 @@
-import psycopg2
+# 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 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()
+# 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)
+#     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
+#     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")
@@ -99,16 +100,6 @@ def VTmode_run(id, channel_list, perc):
     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()
-
     resultList = [id , *channel_data, peak_list, triggerpoint]
 
     channel_data.append(peak_list)
@@ -116,5 +107,17 @@ def VTmode_run(id, channel_list, perc):
 
     return resultList, data_list
 
-# if __name__ == '__main__':
-#     VTmode_run()
+
+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_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_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())