- control output system info with argument

new/analysis_mode.py

@@ -31,7 +31,8 @@ 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)

new/callback_func.py

@@ -3,6 +3,7 @@ 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):
             """
@@ -25,41 +26,46 @@ async def async_callback(topic, payload, conn, client):
             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
-                for meta_id in meta['id']:
-                    # TODO write file header
-                    print(input_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
 
-                    # create meta_data
-                    meta_data = {}
-                    for channel in meta['channel']:
-                        meta_data[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)
-                    # 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]
-                    
+                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)
+                    # # 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 header
+                    # if not write_header_done:
+                    #     dict_writer.writeheader()
+                    #     write_header_done = True
 
-                    # write data
-                    dict_writer.writerow(csv_data)
+                    # # write data
+                    # dict_writer.writerow(csv_data)
             print("執行時間：%f 秒" % (time.time() - start_time))

new/main.py

@@ -1,8 +1,25 @@
 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)

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