- control output system info with argument

-add comment of payload type

new/ana_func.py

@@ -0,0 +1,135 @@
+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
+

new/analysis_mode.py

@@ -0,0 +1,71 @@
+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)
+    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 / 1e2 + data_list['ceiling']
+    trigger_line_up = (data_list['ground']-data_list['ceiling'])*(1-percentage / 1e2) + data_list['ceiling']
+
+    trigger = {str(100-percentage)+'% point':trigger_line_down, str(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)]
+    # 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()
+
+    return [peak_list, triggerpoint] , data_list

new/callback_func.py

@@ -0,0 +1,71 @@
+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 concurrent.futures import ProcessPoolExecutor
+
+async def async_callback(topic, payload, conn, client):
+            """
+            payload format
+            {
+                "pattern": {
+                    "id": number,
+                    "name": string,
+                    "parameter": object,
+                },
+                "data": {
+                    "id": list[int],
+                    "channel: list[int]
+                }
+            }
+            """
+            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']
+
+            # 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
+
+                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]
+                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)
+            print("執行時間：%f 秒" % (time.time() - start_time))

new/database_api.py

@@ -0,0 +1,14 @@
+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]
+        return raw_data_remove_time

new/main.py

@@ -0,0 +1,30 @@
+import asyncio
+import psycopg2
+from mqtt_client import MqttClient
+from utils.system_info import cpu_info, ram_info
+import threading
+import sys
+
+def set_interval(func, sec):
+    def func_wrapper():
+        set_interval(func, sec)
+        func()
+    t = threading.Timer(sec, func_wrapper)
+    t.start()
+    return t
+
+def call():
+    cpu_info()
+    ram_info()
+
+async def main():
+    if len(sys.argv) > 1:
+        set_interval(call, 1)
+    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')

new/utils/system_info.py

@@ -0,0 +1,32 @@
+import psutil
+
+def get_size(bytes, suffix="B"):
+    """
+    Scale bytes to its proper format
+    e.g:
+        1253656 => '1.20MB'
+        1253656678 => '1.17GB'
+    """
+    factor = 1024
+    for unit in ["", "K", "M", "G", "T", "P"]:
+        if bytes < factor:
+            return f"{bytes:.2f}{unit}{suffix}"
+        bytes /= factor
+
+def cpu_info():
+    # let's print CPU information
+    print("="*40, "CPU Info", "="*40)
+    print("CPU Usage Per Core:")
+    for i, percentage in enumerate(psutil.cpu_percent(percpu=True, interval=1)):
+        print(f"Core {i}: {percentage}%")
+    print(f"Total CPU Usage: {psutil.cpu_percent()}%")
+
+def ram_info():
+    # Memory Information
+    print("="*40, "Memory Information", "="*40)
+    # get the memory details
+    svmem = psutil.virtual_memory()
+    print(f"Total: {get_size(svmem.total)}")
+    print(f"Available: {get_size(svmem.available)}")
+    print(f"Used: {get_size(svmem.used)}")
+    print(f"Percentage: {svmem.percent}%")

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