최신 코드4_printdebug

AWX_collector.py

@@ -7,10 +7,13 @@ import minimalmodbus
 import serial.rs485
 import subprocess
 import json
+import yaml
 import queue
-import argparse
 import paho.mqtt.client as mqtt
-
+import numpy as np
+from multiprocessing import shared_memory
+import argparse
+import random
 
 '''
 전달해야할 데이터 들
@@ -29,7 +32,6 @@ ER1: 1초동안 판별, 판별할 동안 counting 할 예정 2V 이하라면 CNT
 
 ER2: OFF 횟수
 
-
 '''
 
 parser = argparse.ArgumentParser()
@@ -37,20 +39,91 @@ parser.add_argument('-config',help='')
 args = parser.parse_args()
 
 ROOT_PATH = f"/usr/local/sdt/app/{args.config}"
+DEVICE_PATH = f"/etc/sdt/device-control"
+
+
 
 ###############################################################################
 #                         JSON FILE READ                                      #
 ###############################################################################
 
-with open(f"{args.config}/connect_info.json","r") as f:
+with open(f"{ROOT_PATH}/connect_info.json","r") as f:
+#with open(f"{args.config}/connect_info.json","r") as f:
     info = json.load(f)
 
+with open(f"{DEVICE_PATH}/config.yaml","r") as f:
+#with open(f"{args.config}/connect_info.json","r") as f:
+    dev_info = yaml.load(f, Loader=yaml.FullLoader)
+
+
+###############################################################################
+#                         Shared memory setting_DO                            #
+###############################################################################
+
+with open(f"{ROOT_PATH}/config.json","r") as f:
+#with open(f"{args.config}/config.json","r") as f:
+    do_dict = json.load(f)
+    for key in do_dict:
+        do_dict[key] = "1"
+
+#print("do_dict", "\n", di_dict)
+#print(do_dict["1"])
+#print(type(do_dict["1"]))
+
+a = [ do_dict[str(n+1)] for n in range(64)]
+a = np.array(a)
+
+
+###############################################################################
+#                            Shared Memory Setting_CORR                       #
+###############################################################################
+
+with open(f"{ROOT_PATH}/config_corr.json","r") as f:
+    corr_dict = json.load(f)
+    #print(corr_dict)
+    for key in corr_dict:
+        corr_dict[key]= "0"
+
+c = [corr_dict[str(n+1)] for n in range(64)]
+c = np.array(c)
+
+#print(c)
+#print(type(c[0]))
+#print(corr_dict)
+#print(corr_dict["1"])
+
+
+
+
+
+###############################################################################
+#                       Get Data from Shared Memory_DO                        #
+###############################################################################
+
+shm_DO = shared_memory.SharedMemory(create=True, size=a.nbytes, name="DO-shm")
+
+do_value = np.ndarray(a.shape, dtype=a.dtype, buffer=shm_DO.buf)
+do_value[:] = a[:]
+print("Get data from shm-DO : ", do_value)
+#print(type(do_value[0]))
+
+###############################################################################
+#                       Get Data from Shared Memory_CORR                      #
+###############################################################################
+
+shm_CORR = shared_memory.SharedMemory(create=True, size=c.nbytes, name="Correction_Test")
+
+correction_value = np.ndarray(c.shape, dtype=c.dtype, buffer=shm_CORR.buf)
+correction_value[:] = c[:]
+print("Get data from shm-CORR : ",correction_value)
+
 
 ###############################################################################
 #                         MQTT Broker Setting                                 #
 ###############################################################################
 
-MQTT_TOPIC = info['mqtt']['topic']
+#MQTT_TOPIC = info['mqtt']['topic']
+MQTT_TOPIC = f"/device-data/{dev_info['assetcode']}"
 MQTT_ID = info['mqtt']['id']
 MQTT_PW = info['mqtt']['pw']
 MQTT_HOST_IP = info['mqtt']['host_ip']
@@ -69,80 +142,57 @@ client.connect(host=MQTT_HOST_IP,port=MQTT_PORT)
 client.loop(2)  # timeout = 2초
 
 
-serial_dev1 = ""
-serial_dev2 = ""
 
-# chip0_lock = threading.Lock()
+###############################################################################
-# chip1_lock = threading.Lock()
+#                          CWT Thread-Read Serial                             #
+###############################################################################
 
-_serial_port1 = serial.Serial("/dev/ttyMAX1", 115200, parity= serial.PARITY_NONE, stopbits=serial.STOPBITS_ONE, bytesize=serial.EIGHTBITS, timeout = 1, rtscts=True)
+isThreadRun = True
-_serial_port1.rs485_mode = serial.rs485.RS485Settings()
-_serial_port2 = serial.Serial("/dev/ttyMAX0", 115200, parity= serial.PARITY_NONE, stopbits=serial.STOPBITS_ONE, bytesize=serial.EIGHTBITS, timeout = 1, rtscts=True)
-_serial_port2.rs485_mode = serial.rs485.RS485Settings()
-#_serial_port3 = serial.Serial("/dev/ttyMAX2", 115200, parity= serial.PARITY_NONE, stopbits=serial.STOPBITS_ONE, bytesize=serial.EIGHTBITS, timeout = 1, rtscts=True)
-#_serial_port3.rs485_mode = serial.rs485.RS485Settings()
-#_serial_port4 = serial.Serial("/dev/ttyMAX3", 115200, parity= serial.PARITY_NONE, stopbits=serial.STOPBITS_ONE, bytesize=serial.EIGHTBITS, timeout = 1, rtscts=True)
-#_serial_port4.rs485_mode = serial.rs485.RS485Settings()
 
-#data_queue = queue.Queue()
+def cwt_thread(evt, chip_index, port_a_ai_id , port_a_do_id, port_b_ai_id , port_b_do_id, queue_a, queue_b):
-######################################### thread 함수 작성 ################################################
 
-dict_1={}
+    print("index: ", chip_index)
+    print("a-ai: %d, a-do: %d, b-ai: %d, b-do: %d " % (port_a_ai_id, port_a_do_id, port_b_ai_id, port_b_do_id))
+    if chip_index == 0:
+        serial_dev1 = "/dev/ttyMAX1"
+        serial_dev2 = "/dev/ttyMAX0"
+    else:
+        serial_dev1 = "/dev/ttyMAX2"
+        serial_dev2 = "/dev/ttyMAX3"
 
-def read_cihp0(e,port_num1,port_num2,id1,id2,id3,id4,id5,id6):
+    print(serial_dev1)
-
+    print(serial_dev2)
-    serial_dev1 = "/dev/ttyMAX1"
+    _serial_port_A = serial.Serial(serial_dev1, 115200, parity= serial.PARITY_NONE, stopbits=serial.STOPBITS_ONE, bytesize=serial.EIGHTBITS, timeout = 1, rtscts=True)
-    serial_dev2 = "/dev/ttyMAX0"
+    _serial_port_A.rs485_mode = serial.rs485.RS485Settings()
-   # lock = chip1_lock
+    _serial_port_B = serial.Serial(serial_dev2, 115200, parity= serial.PARITY_NONE, stopbits=serial.STOPBITS_ONE, bytesize=serial.EIGHTBITS, timeout = 1, rtscts=True)
+    _serial_port_B.rs485_mode = serial.rs485.RS485Settings()
 
 
-    #_serial_port3 = serial.Serial("/dev/ttyMAX2", 115200, parity= serial.PARITY_NONE, stopbits=serial.STOPBITS_ONE, bytesize=serial.EIGHTBITS, timeout = 1, rtscts=True)
+    port_A_AI = minimalmodbus.Instrument(serial_dev1, port_a_ai_id, debug=False, close_port_after_each_call=False)  # port name, slave address (in decimal)
-    #_serial_port3.rs485_mode = serial.rs485.RS485Settings()
+    port_A_AI.serial.baudrate = 115200  
-    #_serial_port4 = serial.Serial("/dev/ttyMAX3", 115200, parity= serial.PARITY_NONE, stopbits=serial.STOPBITS_ONE, bytesize=serial.EIGHTBITS, timeout = 1, rtscts=True)
-    #_serial_port4.rs485_mode = serial.rs485.RS485Settings()
 
-    port1_AI1 = minimalmodbus.Instrument(serial_dev1, id1, debug=False, close_port_after_each_call=False)  # port name, slave address (in decimal)
+    port_A_DO = minimalmodbus.Instrument(serial_dev1, port_a_do_id, debug=False, close_port_after_each_call=False)  # port name, slave address (in decimal)
-    port1_AI1.serial.baudrate = 115200  # baudrate
+    port_A_DO.serial.baudrate = 115200  
 
-    port1_DO = minimalmodbus.Instrument(serial_dev1, id2, debug=False, close_port_after_each_call=False)  # port name, slave address (in decimal)
+    port_B_AI = minimalmodbus.Instrument(serial_dev2, port_b_ai_id, debug=False, close_port_after_each_call=False)  # port name, slave address (in decimal)
-    port1_DO.serial.baudrate = 115200  # baudrate
+    port_B_AI.serial.baudrate = 115200  
 
-    port1_AI2 = minimalmodbus.Instrument(serial_dev1, id3, debug=False, close_port_after_each_call=False)  # port name, slave address (in decimal)
+    port_B_DO = minimalmodbus.Instrument(serial_dev2, port_b_do_id, debug=False, close_port_after_each_call=False)  # port name, slave address (in decimal)
-    port1_AI2.serial.baudrate = 115200  # baudrate
+    port_B_DO.serial.baudrate = 115200  
 
-    port2_AI1 = minimalmodbus.Instrument(serial_dev2, id4, debug=False, close_port_after_each_call=False)  # port name, slave address (in decimal)
-    port2_AI1.serial.baudrate = 115200  # baudrate
 
-    port2_DO = minimalmodbus.Instrument(serial_dev2, id5, debug=False, close_port_after_each_call=False)  # port name, slave address (in decimal)
+    while isThreadRun:
-    port2_DO.serial.baudrate = 115200  # baudrate
+        evt.wait()
+        evt.clear()
 
-    port2_AI2 = minimalmodbus.Instrument(serial_dev2, id6, debug=False, close_port_after_each_call=False)  # port name, slave address (in decimal)
+        prev_a_ai = [0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0]
-    port2_AI2.serial.baudrate = 115200  # baudrate
+        prev_b_ai = [0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0]
 
-    global read_cnt_ch0
+        ai_array = [0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0]
-    read_cnt_ch0 = 0
+
-    global suc_cnt_ai
+        prev_a_ai = ai_array
-    global suc_cnt_do
+        prev_b_ai = ai_array
-    global suc_cnt_ai2
-    global suc_cnt_do2
-    suc_cnt_ai = 0
-    suc_cnt_do = 0
-    suc_cnt_ai2 = 0
-    suc_cnt_do2 = 0
-    global lists 
-    lists = []
-    while True:
-        e.wait()
-        #print("read thread get even")
-        e.clear()
-        read_cnt_ch0 = read_cnt_ch0 +1
                 
-        #value list on -> do control data
-        #추후에 do power on 할 리스트는 어디선가 읽어서 와야 해요
-        #do1 do2 나눠야 해요
-        value_list_ON = [1,1,1,1,0,0,0,0,0,0,0,0,0,0,0,0]
-        # lock.acquire()
-        timenow = time.time()
         
         '''
                 | AI1 | AI2 | DO1 | DO2 |
@@ -159,184 +209,446 @@ def read_cihp0(e,port_num1,port_num2,id1,id2,id3,id4,id5,id6):
           ->  이전 데이터가 있다면, 이전 데이터를 가지고 더미 데이터로 사용하는 방법 => 안하기로 했었음
         '''
         
-        try:
+        timenow = int(time.time()*1000)
-            ai1=port1_AI1.read_registers(0x32, 32)
+        a_isSuccess = True
-            suc_cnt_ai = suc_cnt_ai +1
+        b_isSuccess = True
-        except:
+
-#            pass
+        # PORT A
-            None
+        port_a_do_value = []
-        #    print("Port3 AI1 Fail")
+        for i in list(do_value[:16]):
+            port_a_do_value.append(int(i))
+        
+        #print("port a do: ", port_a_do_value)
+        #print("do type: {}".format(type(port_a_do_value[0])))
         
         try:
-            #print("try")
+            a_ai_data=port_A_AI.read_registers(0x32, 32)
-            port1_DO.write_bits(0, value_list_ON)
+            prev_a_ai = a_ai_data
-            suc_cnt_do = suc_cnt_do +1
+            print("a_ai ;", a_ai_data)
-            #print("do1")
+            # suc_cnt_ai = suc_cnt_ai +1
+                       
+        except:
+            a_isSuccess = False
+            print("a_ai fail")
+
+        #if isSuccess:
+        #    prev_a_ai = ai_array
+
+        try:
+            #port_A_DO.write_bits(0, port_a_do_value)
+            a_do_value = port_A_DO.write_bits(0, port_a_do_value)
+            print("a_do ;", a_do_value)
+            # suc_cnt_do = suc_cnt_do +1
+            
+        except:
+            print(port_A_DO)
+            print("a_do fail")
+            None
+
+        # PORT B
+        port_b_do_value = []
+        for i in list(do_value[16:32]):
+            port_b_do_value.append(int(i))
+        #print("port b do: ", port_b_do_value)
+        try:
+            b_ai_data=port_B_AI.read_registers(0x32, 32)
+            prev_b_ai = b_ai_data
+            print("b_ai ;", b_ai_data)
+            # suc_cnt_ai = suc_cnt_ai +1
+    
+        except:
+            b_isSuccess = False
+            print("b_ai fail")
+
+        #if isSuccess:
+        #    prev_b_ai = ai_array
+
+        try:
+            #port_B_DO.write_bits(0, port_b_do_value)
+            b_do_value = port_B_DO.write_bits(0, port_b_do_value)
+            print("b_do ;", b_do_value)
+            # suc_cnt_do = suc_cnt_do +1
+            
         except:
             None
-        #    print("Port3 DO Fail")
+            print("b_do fail")
-#        try:
+            
-#            ai2 = port3_AI2.read_registers(0x32, 32)
+        if a_isSuccess:
-#        except:
+            queue_a.put([timenow,a_ai_data, port_a_do_value])
-#            print("Port3 AI2 Fail")
+        else:
-        #print(ai1)
+            queue_a.put([0,prev_a_ai, port_a_do_value])
-        #print(ai2)
+
+        if b_isSuccess:
+            queue_b.put([timenow,b_ai_data, port_b_do_value])
+        else:
+            queue_b.put([0,prev_b_ai, port_b_do_value])
 
 
 
 
-        try:
-            global ai2
-            ai2=port2_AI1.read_registers(0x32, 32)
-            suc_cnt_ai2 = suc_cnt_ai2 +1
-        except:
-            pass
-        #    print("Port4 AI1 Fail")
 
-        try:
+###############################################################################
-            do2 = port2_DO.write_bits(0, value_list_ON)
+#                         Data process thread                                 #
-            suc_cnt_do2 = suc_cnt_do2 +1
+###############################################################################
-        except:
-            pass
-        #    print("Port4 DO Fail, ")
 
-#        try:
-#            ai2 = port4_AI2.read_registers(0x32, 32)
-#        except:
-#            print("Port4 AI2 Fail")
-        #print(ai1)
-        #print(ai2)
 
-        # lock.release()
+def data_process_thread(data_queue, ch, chamber):
+    error_cnt_1 = [0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0]
+    error_cnt_2 = [0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0]
+    data_list_for_send_1=[]
+    data_list_for_send_2=[]
+    queue_count = 0
+    queue_select_count = 0
 
+    count = 0
+    below_20 = 0
+    element_counts = {}
+    current_cycle_elements = 0
+    cycle_counts = 0 
+    
+    while isThreadRun:
+        # 0.1s 마다 발생할 것으로 예상
+        timestamp, ai, do = data_queue.get()
+
+        #==================================================
+        do_temp = []
+        for d in do:
+            #print("dddddddddddd: {}".format(d))
+            do_temp.append(int(d))
+        do = do_temp.copy()
+        #==================================================
+
+
+        #print("timestamp: {timestamp}, \n ai: {ai}, \n do: {do}" .format(timestamp=timestamp, ai=ai, do=do))
+        
+ 
+
+                        
+###############################################################################
+#              Distribute AI data, Calculate value with correction            #
+###############################################################################
+
+        #ai data -> current, voltage 전환
+        queue_count = queue_count + 1
+        current = []
+        voltage = []
+        ai_index = 0
+        correction_value_array = []
+        
+        if data_queue == ch1_queue:                  
+            correction_value_array = np.array(correction_value[:16])
+            #key_to_change = str(index+1)
+        elif data_queue == ch2_queue:                
+            correction_value_array = np.array(correction_value[16:32])
+            #key_to_change = str(index+17)
+        elif data_queue == ch3_queue:            
+            correction_value_array = np.array(correction_value[32:48])
+            #key_to_change = str(index+33)
+        elif data_queue == ch4_queue:            
+            correction_value_array = np.array(correction_value[48:64])
+            #key_to_change = str(index+49)
+           
+        for i in ai:
+            #짝수 current (확인 필요)
+            ai_index = ai_index +1
+            #print("ai list :", ai)
+            if (ai_index%2) == 0:
+                current.append(int(i))
+                #print("current :", current)
+            else:
+                voltage.append(int(i))
+                #print("voltage :", voltage)
+
+        
+        calculated_voltage = voltage + np.array(correction_value_array)
+        calculated_voltage_list = calculated_voltage.tolist()
+        #print("be calculated: ", calculated_voltage)
+        
+        #print("do list : ", do)
+        #print("calculated_voltage list :", calculated_voltage_list)
+        #mult_voltage = [do[i] * calculated_voltage_list[i] for i in range(len(do))]
+        #print("multed value :", mult_voltage)
+        #print(type(mult_voltage))
+        
+###############################################################################
+#                                 HJ ORG CODE                                 #
+###############################################################################
         '''
-        리스트를 생성, 10개를 딱 잘라서 어떻게 해야 할ㅈ지 모르겠음
+        #전압값으로 error count check
-
+        voltage_cnt = 0
+        for i in calculated_voltage_list:
+            if i <= 20000:
+                #count up
+                error_cnt_1[voltage_cnt] = error_cnt_1[voltage_cnt] +1
+                print("error cnt: ",error_cnt_1)
+                if error_cnt_1[voltage_cnt] == 10:
+                    error_cnt_1[voltage_cnt] = 0
+                    error_cnt_2[voltage_cnt] = error_cnt_2[voltage_cnt] + 1
+                    #shared memory do control modify
+                    #ch, tube number->off
+            else:
+                # count reset
+                error_cnt_1[voltage_cnt] = 0
     
+            voltage_cnt = voltage_cnt + 1
+            print("voltage count:",voltage_cnt)
         '''
       
-        ##데이터 생성 할 때
-        # ai1, ai2
-        # -> ai1 [32] -> [curr, volt, curr, volt ...]
 
                 
-        data_for_list = {
+###############################################################################
-            "slot" : 1,
+#                                 YJ TEST CODE                                #
-            "timestamp": int(timenow*1000),
+###############################################################################
-            "data":{
+        '''
-                "current": ai1,
+        ERR1 = 0
-                "volatge": ai2,
+        #count = 0
-                "do1": value_list_ON,
+        #cycle_counts = 0
-                "err1": value_list_ON,
+
-                "err2": value_list_ON
+        for i, value in enumerate(calculated_voltage_list):
-                #err1, 2
+            if value <= 5:
+               element_counts = [x + 1 if y == 1 else x for x, y in zip(element_counts, do)]
+               #element_counts[i] = element_counts.get(i,0) + 1
+               print(element_counts)
+               
+               #mult_element_count = [element_count[i] * do[i] for i in range(len(do))]
+               #이건 아닌거 같다 카운트는 계속 올라갈 거고 10일 때 DO가 켜지면 또 에러 발생임
+        cycle_counts += 1
+        
+        for index, count in element_counts.items():
+            if cycle_counts % 10 == 0:
+                if count == 10:
+                    #print(f"Error elements: {index}")
+                    error_cnt_1[index] = error_cnt_1[index] +1 
+                    #error_cnt_1 = [error_cnt_1[index] +1 * do[index] for index in range(len(error_cnt_1))]
+                    mult_error_cnt_1 = [error_cnt_1[i] * do[i] for i in range(len(do))]
+                    #mult_error_cnt_1 = [x + 1 * y for x,y in zip(error_cnt_1, do)]
+                    
+                    print("check error_cnt_1 : ", error_cnt_1)
+                    print("check multed value to do index : ", mult_error_cnt_1)
+                    
+                    global do_dict
+                                        
+                    #key_to_change = str(index+1)
+                    if data_queue == ch1_queue:                  
+                        key_to_change = str(index+1)
+                    elif data_queue == ch2_queue:                
+                        key_to_change = str(index+17)
+                    elif data_queue == ch3_queue:            
+                        key_to_change = str(index+33)
+                    elif data_queue == ch4_queue:            
+                         key_to_change = str(index+49)
+             
+                    if key_to_change in do_dict:
+                       do_dict[key_to_change] = "0"  
+                       with open(f"{ROOT_PATH}/config.json","w") as f:
+                            json.dump(do_dict, f)
+                            
+                            #print("check do off from config.json: ",do_dict.values())           
+                                      
+                    element_counts[index] = 0
+                    #error_cnt_1 = 0
+                    #mult_error_cnt_1 = 0
+                else:
+                    element_counts[index] = 0 
+        '''
+        ERR1 = 0
+        #count = 0
+        #cycle_counts = 0
+        #element_counts = [0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0]
+        
+        for i, value in enumerate(calculated_voltage_list):
+            if value <= 20000 and do[i] == 1:
+               #print(f"Updating count for index {i}")
+
+               #element_counts = [x + 1 if y == 1 else x for x, y in zip(element_counts, do)]
+               
+               #element_counts[i] += 1
+               element_counts[i] = element_counts.get(i,0) + 1
+               
+               #print("check do : " ,do)
+               #print("voltage list : ", calculated_voltage_list)
+               
+               #mult_element_count = [element_count[i] * do[i] for i in range(len(do))]
+               #이건 아닌거 같다 카운트는 계속 올라갈 거고 10일 때 DO가 켜지면 또 에러 발생임
+               
+        print("check do : " ,do)
+        print("voltage list : ", calculated_voltage_list)       
+        #print("element cosunts : ", element_counts,"\r\n")
+        cycle_counts += 1
+        
+        error_cnt_1 = [0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0]
+        for index, count in element_counts.items():
+            if cycle_counts % 10 == 0:
+                if count == 10:
+                    #print(f"Updating e_count for index {index}")
+                    #print(f"Error elements: {index}")
+                    error_cnt_1[index] += 1
+                    #error_cnt_1 = [error_cnt_1[index] +1 * do[index] for index in range(len(error_cnt_1))]
+                    
+                    #mult_error_cnt_1 = [error_cnt_1[i] * do[i] for i in range(len(do))]
+                    
+                    #mult_error_cnt_1 = [x + 1 * y for x,y in zip(error_cnt_1, do)] //zip 함수 사용법
+                    
+                    #print("check error_cnt_1 : ", error_cnt_1)
+                    #print("check multed value to do index : ", mult_error_cnt_1)
+                    
+                    global do_dict
+                                        
+                    #key_to_change = str(index+1)
+                    if data_queue == ch1_queue:                  
+                        key_to_change = str(index+1)
+                    elif data_queue == ch2_queue:                
+                        key_to_change = str(index+17)
+                    elif data_queue == ch3_queue:            
+                        key_to_change = str(index+33)
+                    elif data_queue == ch4_queue:            
+                         key_to_change = str(index+49)
+             
+                    if key_to_change in do_dict:
+                       do_dict[key_to_change] = "0"  
+                       with open(f"{ROOT_PATH}/config.json","w") as f:
+                            json.dump(do_dict, f)
+                            
+                            #print("check do off from config.json: ",do_dict.values())           
+                                      
+                    element_counts[index] = 0
+                    
+                    #mult_error_cnt_1 = 0
+                else:
+                    element_counts[index] = 0
+      
+
+        # 로깅 기능을 넣어야 겠다...?
+        # 파일 하나 열어서, 데이터를 쓴다
+        
+        
+        
+        ###############################################################################
+        #                 Make data form to json and Publish message                  #
+        ###############################################################################
+  
+        #데이터 만들어서 전달
+        if timestamp > 0:
+        #if timestamp == 0:
+           data = {
+               "timestamp" : int(timestamp),
+                "data": {
+                   "current"   : current,
+                   "voltage"   : calculated_voltage_list,
+                   "do"        : do,
+                   "er1"       : error_cnt_1
                 }
-            }
+           }
-        lists.append(data_for_list)
+           if(queue_select_count == 0):
-        ### 여기서 보냄      
+               data_list_for_send_1.append(data)
-        # 10번 시그널 받으면
+           else:
-        #     
+               data_list_for_send_2.append(data)
-        # list mqtt send
+
-        # list clear
+            
-        if read_cnt_ch0 % 10 == 0:
+        '''
-            data = {
+        ############## Error Do 데이터 잘려오는 증상 추정 원인 개선 방안 중 하나 ################ 
+        # -> string 만들어서 전달하자
+        json_obj1 = json.loads(data)
+        if(queue_select_count == 0):
+            data_list_for_send_1.append(data)
+        else:
+            data_list_for_send_2.append(data)
+            
+        '''
+
+        if queue_count == 10:
+            queue_count = 0
+            mqtt_msg = {}
+            if (queue_select_count == 0):
+                if not data_list_for_send_1:
+                  print("data empty")
+                mqtt_msg = {
                     "modelCode":"test4",
-                    "assetCode":"NQ-R04-TEST-003",
+                    "assetCode":"S0NQR0423090001",  #나중에는 config에서 셋팅되는 값, or model 값으로 변경 (asset)
                     "dataType":"DATA",
                     "data": {
-                        "chamber":1,
+                        "channel": ch,  # 채널 입력 <20>자
-                        "list": lists
+                        "chamber":chamber,    #chamber 도 입력 <20>자
-                        }
+                        "slot": ch,     #슬롯 값은 chamber와 ch 값으로 만든다
+                        "list": data_list_for_send_1
                     }
-            global dict_1
+                }
-            data_json_str = json.dumps(data, indent=0)
+                queue_select_count = 1
+                data_list_for_send_2.clear()
+            else:
+                mqtt_msg = {
+                    "modelCode":"test4",
+                    "assetCode":"S0NQR0423090001",  #나중에는 config에서 셋팅되는 값, or model 값으로 변경 (asset)
+                    "dataType":"DATA",
+                    "data": {
+                        "channel": ch,  # 채널 입력 <20>자
+                        "chamber":chamber,    #chamber 도 입력 <20>자
+                        "slot": ch,     #슬롯 값은 chamber와 ch 값으로 만든다
+                        "list": data_list_for_send_2
+                    }
+                }
+                queue_select_count = 0
+                data_list_for_send_1.clear()
+
+
+#            print(type(mqtt_msg['data']['list'][0]['do'][0]))
+
+            data_json_str = json.dumps(mqtt_msg, indent=4)
+            print(data_json_str)
             publish_json_message(client, data_json_str)
-            #print(data_json_str)
-       # time.sleep(1)
- 
-    
-   # PRINT("CHip1 read count:", read_cnt)
 
 
         
-######################################### thread 함수 작성 ################################################
 
 
-def event_trigger_thread():
+###############################################################################
-        print("inpuuuuuuuuuuuuuuuuuuuuuuuuuuuuuuuuuuuuuuuuuuuuuuuuuuuuuuuut\r\n")
+#                            Thread Event Timer                               #
-        if input("Enter to exit(): "):
+###############################################################################
-          e.set()
-          #exit()
-        return
 
-def timer_event_loop(e):
+#event는 개별 생성을 해준다 th1, th2
+
+def timer_event_loop(e1, e2):
     cnt = 0
     # global read_cnt
+
     while True:
-        # print("event set, meanning : periodic sensing")
+
-         e.set()
+         e1.set()
+         e2.set()
          time.sleep(0.1)
 
-         #cnt = cnt +1
-         #if cnt > 6000:
-            # print("chip0 read count:", read_cnt_ch0)
-             #print("\r\nchip1 read count:", read_cnt_ch0)
-             #print("\r\nai1:", suc_cnt_ai, "\r\nai2:", suc_cnt_ai2)
-             #print(ai2)
-             #break
-
-    #print(cnt)
-#    print(read_cnt)
     exit()
-#        timer = threading.Timer(3, lambda: print("test event"))
-#        timer.start()
-#        timer.join()
-# #        timer = threading.Timer(3, lambda: e.set())
 
 
-################################# 작성 중 #####################################
-# def publish_json_message(client, data):
-#     topic = "devicedata/axr/nodeq1"
-#     client.publish(topic, payload=data)
 
 
-# client = mqtt.Client("python_pub")
-# client.username_pw_set("sdt", "251327")
-# client.connect(host="13.209.39.139",port=32259)
-
-# # get_value()
-
-
-# data = json.dumps(dict_1)
-# #publish_json_message(client, data)
-
-# client.loop(2)  # timeout = 2초
-
-################################# 작성 중 ######################################
 
 if __name__ == '__main__':
-    e = threading.Event()
+    ###############################################################################
-    read_data_chip0 = threading.Thread(target=read_cihp0,args=(e,1,2,1,2,3,5,6,7))
+    #                         Queue x 4         max- infinite                     #
-    # read_data_chip1 = threading.Thread(target=read_cihp1,args=(e,2,3,9,10,11,13,14,15))
+    ###############################################################################
-    # read_data_ch3 = threading.Thread(target=read_thread,args=(e,3,9,10,11))
-    # read_data_ch4 = threading.Thread(target=read_thread,args=(e,4,13,14,15))
-    event_trigger = threading.Thread(target=event_trigger_thread)
-    event_thread = threading.Thread(target=timer_event_loop, args=(e,))
-#    timer = threading.Timer(3,lambda: e.set()) # 1 time
 
-   # read_data_ch1.start()
+    ch1_queue = queue.Queue()
-   # read_data_ch2.start()
+    ch2_queue = queue.Queue()
-   # read_data_chip1.start()
+    ch3_queue = queue.Queue()
+    ch4_queue = queue.Queue()
+
+    chip0_evt = threading.Event()
+    chip1_evt = threading.Event()
+
+    read_data_chip0 = threading.Thread(target=cwt_thread,args=(chip0_evt,   0,  1,2,    13,14,   ch1_queue, ch2_queue))  #현재 4개씩 연결되 어 있어서 테스트 용
+    read_data_chip1 = threading.Thread(target=cwt_thread,args=(chip1_evt,   1,  5,6,    7,8, ch3_queue, ch4_queue))
+
+    data_process_ch1 = threading.Thread(target=data_process_thread,args=(ch1_queue,   1,  1))
+    data_process_ch2 = threading.Thread(target=data_process_thread,args=(ch2_queue,   2,  1))
+    data_process_ch3 = threading.Thread(target=data_process_thread,args=(ch3_queue,   3,  1))
+    data_process_ch4 = threading.Thread(target=data_process_thread,args=(ch4_queue,   4,  1))
+
+    event_thread = threading.Thread(target=timer_event_loop, args=(chip0_evt,chip1_evt,))
     read_data_chip0.start()
-#    timer.start()
+    read_data_chip1.start()
-    event_trigger.start()
+    data_process_ch1.start()
+    data_process_ch2.start()
+    data_process_ch3.start()
+    data_process_ch4.start()
     event_thread.start()
-   # wait_and_timer.start()
-
-
-
-#    while True:
-#        input("Enter to trigger the event: ")
-#        exit():
-#        e.set()
-         # time.sleep(600

AWX_collector.py~

@@ -0,0 +1,507 @@
+
+import threading
+import time
+import logging
+import serial
+import minimalmodbus
+import serial.rs485
+import subprocess
+import json
+import yaml
+import queue
+import paho.mqtt.client as mqtt
+import numpy as np
+from multiprocessing import shared_memory
+import argparse
+import random
+
+'''
+전달해야할 데이터 들
+mes code | 상태 | 전압 | 전류 | 4pin (DO) | ER1 | ER2
+tube 번호| 상태 | volt | curr | do        | cnt | cnt
+
+tube 번호: 1 - 128 , chamber / slot
+ - slot(n) - tube(n)
+ - 1-1 ~ 8-16 : 128개다
+
+상태: 정상이냐, 초기화, 불량
+
+ER1: 1초동안 판별, 판별할 동안 counting 할 예정 2V 이하라면 CNT UP
+ if) er: 10이 되었음 (1초 유지) -> 0으로 변환, ER2 cnt up
+ if) 2v 이상인 경우가 발생했다 -> 0으로 변환
+
+ER2: OFF 횟수
+
+'''
+
+parser = argparse.ArgumentParser()
+parser.add_argument('-config',help='')
+args = parser.parse_args()
+
+ROOT_PATH = f"/usr/local/sdt/app/{args.config}"
+DEVICE_PATH = f"/etc/sdt/device-control"
+
+###############################################################################
+#                         JSON FILE READ                                      #
+###############################################################################
+
+with open(f"{ROOT_PATH}/connect_info.json","r") as f:
+#with open(f"{args.config}/connect_info.json","r") as f:
+    info = json.load(f)
+
+with open(f"{DEVICE_PATH}/config.yaml","r") as f:
+#with open(f"{args.config}/connect_info.json","r") as f:
+    dev_info = yaml.load(f, Loader=yaml.FullLoader)
+
+
+###############################################################################
+#                         Shared memory setting                               #
+###############################################################################
+
+
+
+with open(f"{ROOT_PATH}/config.json","r") as f:
+#with open(f"{args.config}/config.json","r") as f:
+    data = json.load(f)
+
+print(data)
+print(data["1"])
+
+a = [ data[str(n+1)] for n in range(64)]
+a = np.array(a)
+
+
+###############################################################################
+#                       Get Data from Shared Memory                           #
+###############################################################################
+
+shm = shared_memory.SharedMemory(create=True, size=a.nbytes, name="DO-shm")
+b = np.ndarray(a.shape, dtype=a.dtype, buffer=shm.buf)
+b[:] = a[:]
+
+
+###############################################################################
+#                         MQTT Broker Setting                                 #
+###############################################################################
+
+#MQTT_TOPIC = info['mqtt']['topic']
+MQTT_TOPIC = f"/device-data/{dev_info['assetcode']}"
+MQTT_ID = info['mqtt']['id']
+MQTT_PW = info['mqtt']['pw']
+MQTT_HOST_IP = info['mqtt']['host_ip']
+MQTT_PORT = info['mqtt']['port']
+
+
+def publish_json_message(client, data):
+    topic = MQTT_TOPIC
+    client.publish(topic, payload=data)
+
+
+client = mqtt.Client("python_pub")
+client.username_pw_set(MQTT_ID, MQTT_PW)
+client.connect(host=MQTT_HOST_IP,port=MQTT_PORT)
+
+client.loop(2)  # timeout = 2초
+
+
+
+
+###############################################################################
+#                         Serial setting                                 #
+###############################################################################
+
+_serial_port3 = serial.Serial("/dev/ttyMAX2", 115200, parity= serial.PARITY_NONE, stopbits=serial.STOPBITS_ONE, bytesize=serial.EIGHTBITS, timeout = 1, rtscts=True)
+_serial_port3.rs485_mode = serial.rs485.RS485Settings()
+_serial_port4 = serial.Serial("/dev/ttyMAX3", 115200, parity= serial.PARITY_NONE, stopbits=serial.STOPBITS_ONE, bytesize=serial.EIGHTBITS, timeout = 1, rtscts=True)
+_serial_port4.rs485_mode = serial.rs485.RS485Settings()
+
+
+######################################### thread 함수 작성 ################################################
+
+
+
+###############################################################################
+#                         Serial thread                                  #
+###############################################################################
+isThreadRun = True
+
+def cwt_thread(evt, chip_index, port_a_ai_id , port_a_do_id, port_b_ai_id , port_b_do_id, queue_a, queue_b):
+
+    print("index: ", chip_index)
+    print("a-ai: %d, a-do: %d, b-ai: %d, b-do: %d " % (port_a_ai_id, port_a_do_id, port_b_ai_id, port_b_do_id))
+    if chip_index == 0:
+        serial_dev1 = "/dev/ttyMAX1"
+        serial_dev2 = "/dev/ttyMAX0"
+    else:
+        serial_dev1 = "/dev/ttyMAX2"
+        serial_dev2 = "/dev/ttyMAX3"
+
+    print(serial_dev1)
+    print(serial_dev2)
+    _serial_port_A = serial.Serial(serial_dev1, 115200, parity= serial.PARITY_NONE, stopbits=serial.STOPBITS_ONE, bytesize=serial.EIGHTBITS, timeout = 1, rtscts=True)
+    _serial_port_A.rs485_mode = serial.rs485.RS485Settings()
+    _serial_port_B = serial.Serial(serial_dev2, 115200, parity= serial.PARITY_NONE, stopbits=serial.STOPBITS_ONE, bytesize=serial.EIGHTBITS, timeout = 1, rtscts=True)
+    _serial_port_B.rs485_mode = serial.rs485.RS485Settings()
+
+
+    port_A_AI = minimalmodbus.Instrument(serial_dev1, port_a_ai_id, debug=False, close_port_after_each_call=False)  # port name, slave address (in decimal)
+    port_A_AI.serial.baudrate = 115200  # baudrate
+
+    port_A_DO = minimalmodbus.Instrument(serial_dev1, port_a_do_id, debug=False, close_port_after_each_call=False)  # port name, slave address (in decimal)
+    port_A_DO.serial.baudrate = 115200  # baudrate
+
+    port_B_AI = minimalmodbus.Instrument(serial_dev2, port_b_ai_id, debug=False, close_port_after_each_call=False)  # port name, slave address (in decimal)
+    port_B_AI.serial.baudrate = 115200  # baudrate
+
+    port_B_DO = minimalmodbus.Instrument(serial_dev2, port_b_do_id, debug=False, close_port_after_each_call=False)  # port name, slave address (in decimal)
+    port_B_DO.serial.baudrate = 115200  # baudrate
+
+
+    while isThreadRun:
+        evt.wait()
+        #print("read thread get even")
+        evt.clear()
+
+        
+        #prev_a_ai = [0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0]
+        #prev_b_ai = [0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0]
+
+        ai_array = [0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0]
+        random_index = random.randint(0, len(ai_array) -1)
+        ai_array = [random.randint(1,50000) for _ in range(32)]
+
+        prev_a_ai = ai_array
+        prev_b_ai = ai_array
+        #value list on -> do control data
+        #추<><ECB694>에 do power on 할 리스트는 어디선가 읽어서 와야 해요
+        #do1 do2 나눠야 해요
+
+
+
+        '''
+                | AI1 | AI2 | DO1 | DO2 |
+        Success |  0  |  0  |  x  |  x  |
+        Success |  0  |  1  |  x  |  x  |
+        Success |  1  |  0  |  x  |  x  |
+        Success |  1  |  1  |  x  |  x  | => 유효한 데이터
+
+
+        최소 한개 이상 실패 했을 때
+         -> retry 해볼 것이냐
+
+         -> 데이터를 어떻게 할 것이냐
+          ->  이전 데이터가 있다면, 이전 데이터를 가지고 더미 데이터로 사용하는 방법 => 안하기로 했었음
+        '''
+        timenow = int(time.time()*1000)
+        isSuccess = True
+       
+        print(f"[test]: ", ai_array)
+        # PORT A
+        port_a_do_value = list(b[:16])
+        try:
+            #a_ai_data=port_A_AI.read_registers(0x32, 32)
+            a_ai_data=ai_array
+            # suc_cnt_ai = suc_cnt_ai +1
+        except:
+            isSuccess = False
+
+        if isSuccess:
+            priv_a_ai = ai_array
+
+        try:
+            port_A_DO.write_bits(0, port_a_do_value)
+            # suc_cnt_do = suc_cnt_do +1
+        except:
+            None
+
+        # PORT B
+        port_b_do_value = list(b[17:32])
+        try:
+            #b_ai_data=port_B_AI.read_registers(0x32, 32)
+            b_ai_data = ai_array
+            #prev_b_ai = ai_array
+            # suc_cnt_ai = suc_cnt_ai +1
+        except:
+            isSuccess = False
+
+        if isSuccess:
+            prev_b_ai = ai_array
+
+        try:
+            port_B_DO.write_bits(0, port_b_do_value)
+            # suc_cnt_do = suc_cnt_do +1
+        except:
+            None
+
+        if isSuccess:
+            queue_a.put([timenow,ai_array, port_a_do_value])
+            queue_b.put([timenow,ai_array, port_b_do_value])
+        else:
+            queue_a.put([0,prev_a_ai, port_a_do_value])
+            queue_b.put([0,prev_a_ai, port_b_do_value])
+
+
+
+
+
+
+        # lock.release()
+
+        '''
+        리스트를 생성, 10개를 딱 잘라서 어떻게 해야 할ㅈ지 모르겠음
+
+
+        '''
+        ##데이터 생성 할 때
+        # ai1, ai2
+        # -> ai1 [32] -> [curr, volt, curr, volt ...]
+
+        '''
+        data_for_list = {
+            "slot" : 1,
+            "timestamp": int(timenow*1000),
+            "data":{
+                "current": ai1,
+                "volatge": ai2,
+                "do1": value_list_ON,
+                "err1": value_list_ON,
+                "err2": value_list_ON
+                #err1, 2
+                }
+            }
+        lists.append(data_for_list)
+        ### 여기서 보냄
+        # 10번 시그널 <20>으면
+        #
+        # list mqtt send
+        # list clear
+        if read_cnt_ch0 % 10 == 0:
+            data = {
+                    "modelCode":"test4",
+                    "assetCode":"NQ-R04-TEST-003",
+                    "dataType":"DATA",
+                    "data": {
+                        "chamber":1,
+                        "list": lists
+                        }
+                    }
+
+            data_json_str = json.dumps(data, indent=0)
+            publish_json_message(client, data_json_str)
+            #print(data_json_str)
+       # time.sleep(1)
+        '''
+
+###############################################################################
+#                           dummy data thread                                 #
+###############################################################################
+
+
+###############################################################################
+#                         data process thread                                 #
+###############################################################################
+
+
+def data_process_thread(data_queue, ch, chamber):
+    error_cnt_1 = [0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0]
+    error_cnt_2 = [0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0]
+    data_list_for_send_1=[]
+    data_list_for_send_2=[]
+    queue_count = 0
+    queue_select_count = 0
+    while isThreadRun:
+        # 0.1s 마다 발생할 것으로 예상
+        timestamp, ai, do = data_queue.get()
+
+        #==================================================
+        do_temp = []
+        for d in do:
+            do_temp.append(int(d))
+        do = do_temp.copy()
+        #==================================================
+
+
+        #print("timestamp: {timestamp}, \n ai: {ai}, \n do: {do}" .format(timestamp=timestamp, ai=ai, do=do))
+
+        #ai data -> current, voltage 전환
+        queue_count = queue_count + 1
+        current = []
+        voltage = []
+        ai_index = 0
+        for i in ai:
+            #짝수 current (확인 필요)
+            ai_index = ai_index +1
+            if (ai_index%2) == 0:
+                current.append(int(i))
+            else:
+                voltage.append(int(i))
+
+
+        #전압값으로 error count check
+        voltage_cnt = 0
+        for i in voltage:
+            if i < 2:
+                #count up
+                error_cnt_1[voltage_cnt] = error_cnt_1[voltage_cnt] +1
+                if error_cnt_1[voltage_cnt] == 10:
+                    error_cnt_1[voltage_cnt] = 0
+                    error_cnt_2[voltage_cnt] = error_cnt_2[voltage_cnt] + 1
+                    #shared memory do control modify
+                    #ch, tube number->off
+            else:
+                # count reset
+                error_cnt_1[voltage_cnt] = 0
+
+
+
+            voltage_cnt = voltage_cnt + 1
+
+        # 로깅 기능을 넣어야 겠다...?
+        # 파일 하나 열어서, 데이터를 쓴다
+
+        # 데이터 만들어서 전달
+        if timestamp > 0:
+            data = {
+                "timestamp" : int(timestamp),
+                "data": {
+                    "current"   : current,
+                    "voltage"   : voltage,
+                    "do"        : do,
+                    "er1"       : error_cnt_1,
+                    "er2"       : error_cnt_2
+                }
+            }
+            if(queue_select_count == 0):
+                data_list_for_send_1.append(data)
+            else:
+                data_list_for_send_2.append(data)
+
+
+        if queue_count == 10:
+            queue_count = 0
+            mqtt_msg = {}
+            if (queue_select_count == 0):
+                mqtt_msg = {
+                    "modelCode":"test4",
+                    "assetCode":"NQ-R04-TEST-003",  #나중에는 config에서 셋팅되는 값, or model 값으로 변경 (asset)
+                    "dataType":"DATA",
+                    "data": {
+                        "channel": ch,  # 채널 입력 <20>자
+                        "chamber":chamber,    #chamber 도 입력 <20>자
+                        "slot": ch,     #슬롯 값은 chamber와 ch 값으로 만든다
+                        "list": data_list_for_send_1
+                    }
+                }
+                queue_select_count = 1
+                data_list_for_send_2.clear()
+            else:
+                mqtt_msg = {
+                    "modelCode":"test4",
+                    "assetCode":"NQ-R04-TEST-003",  #나중에는 config에서 셋팅되는 값, or model 값으로 변경 (asset)
+                    "dataType":"DATA",
+                    "data": {
+                        "channel": ch,  # 채널 입력 <20>자
+                        "chamber":chamber,    #chamber 도 입력 <20>자
+                        "slot": ch,     #슬롯 값은 chamber와 ch 값으로 만든다
+                        "list": data_list_for_send_2
+                    }
+                }
+                queue_select_count = 0
+                data_list_for_send_1.clear()
+
+
+#            print(type(mqtt_msg['data']['list'][0]['do'][0]))
+            data_json_str = json.dumps(mqtt_msg, indent=4)
+#            print(data_json_str)
+#            print(data_json_str)
+            publish_json_message(client, data_json_str)
+
+        ##erro cnt check routine
+
+
+
+
+
+
+######################################### thread 함수 작성 ################################################
+
+
+#event는 개별 생성을 해준다 th1, th2
+
+
+def timer_event_loop(e1, e2):
+    cnt = 0
+    # global read_cnt
+
+    while True:
+        # print("event set, meanning : periodic sensing")
+         e1.set()
+         e2.set()
+         time.sleep(0.1)
+
+         #cnt = cnt +1
+         #if cnt > 6000:
+            # print("chip0 read count:", read_cnt_ch0)
+             #print("\r\nchip1 read count:", read_cnt_ch0)
+             #print("\r\nai1:", suc_cnt_ai, "\r\nai2:", suc_cnt_ai2)
+             #print(ai2)
+             #break
+
+    #print(cnt)
+#    print(read_cnt)
+    exit()
+#        timer = threading.Timer(3, lambda: print("test event"))
+#        timer.start()
+#        timer.join()
+# #        timer = threading.Timer(3, lambda: e.set())
+
+
+################################# 작성 중 #####################################
+# def publish_json_message(client, data):
+#     topic = "devicedata/axr/nodeq1"
+#     client.publish(topic, payload=data)
+
+
+# client = mqtt.Client("python_pub")
+# client.username_pw_set("sdt", "251327")
+# client.connect(host="13.209.39.139",port=32259)
+
+# # get_value()
+
+
+# data = json.dumps(dict_1)
+# #publish_json_message(client, data)
+
+# client.loop(2)  # timeout = 2초
+
+################################# 작성 중 ######################################
+
+if __name__ == '__main__':
+    ###############################################################################
+    #                         Queue x 4         max- infinite                          #
+    ###############################################################################
+
+    ch1_queue = queue.Queue()
+    ch2_queue = queue.Queue()
+    ch3_queue = queue.Queue()
+    ch4_queue = queue.Queue()
+
+    chip0_evt = threading.Event()
+    chip1_evt = threading.Event()
+
+    read_data_chip0 = threading.Thread(target=cwt_thread,args=(chip0_evt,   0,  1,2,    5,6,   ch1_queue, ch2_queue))  #현재 4 개씩 연결되 어 있어서 테스트 용
+    read_data_chip1 = threading.Thread(target=cwt_thread,args=(chip1_evt,   1,  9,10,    13,14, ch3_queue, ch4_queue))
+
+    data_process_ch1 = threading.Thread(target=data_process_thread,args=(ch1_queue,   1,  1))
+    data_process_ch2 = threading.Thread(target=data_process_thread,args=(ch2_queue,   2,  1))
+    data_process_ch3 = threading.Thread(target=data_process_thread,args=(ch3_queue,   3,  1))
+    data_process_ch4 = threading.Thread(target=data_process_thread,args=(ch4_queue,   4,  1))
+
+    event_thread = threading.Thread(target=timer_event_loop, args=(chip0_evt,chip1_evt,))
+    read_data_chip0.start()
+    read_data_chip1.start()
+    data_process_ch1.start()
+    data_process_ch2.start()
+    data_process_ch3.start()
+    data_process_ch4.start()
+    event_thread.start()

AWX_collector_1.py

@@ -0,0 +1,493 @@
+
+import threading
+import time
+import logging
+import serial
+import minimalmodbus
+import serial.rs485
+import subprocess
+import json
+import yaml
+import queue
+import paho.mqtt.client as mqtt
+import numpy as np
+from multiprocessing import shared_memory
+import argparse
+
+
+'''
+전달해야할 데이터 들
+mes code | 상태 | 전압 | 전류 | 4pin (DO) | ER1 | ER2
+tube 번호| 상태 | volt | curr | do        | cnt | cnt
+
+tube 번호: 1 - 128 , chamber / slot
+ - slot(n) - tube(n)
+ - 1-1 ~ 8-16 : 128개다
+
+상태: 정상이냐, 초기화, 불량
+
+ER1: 1초동안 판별, 판별할 동안 counting 할 예정 2V 이하라면 CNT UP 
+ if) er: 10이 되었음 (1초 유지) -> 0으로 변환, ER2 cnt up 
+ if) 2v 이상인 경우가 발생했다 -> 0으로 변환
+
+ER2: OFF 횟수
+
+'''
+
+parser = argparse.ArgumentParser()
+parser.add_argument('-config',help='')
+args = parser.parse_args()
+
+ROOT_PATH = f"/usr/local/sdt/app/{args.config}"
+DEVICE_PATH = f"/etc/sdt/device-control"
+
+###############################################################################
+#                         JSON FILE READ                                      #
+###############################################################################
+
+with open(f"{ROOT_PATH}/connect_info.json","r") as f:
+#with open(f"{args.config}/connect_info.json","r") as f:
+    info = json.load(f)
+
+with open(f"{DEVICE_PATH}/config.yaml","r") as f:
+#with open(f"{args.config}/connect_info.json","r") as f:
+    dev_info = yaml.load(f, Loader=yaml.FullLoader)
+
+
+###############################################################################
+#                         Shared memory setting                               #
+###############################################################################
+
+
+
+with open(f"{ROOT_PATH}/config.json","r") as f:
+#with open(f"{args.config}/config.json","r") as f:
+    data = json.load(f)
+    
+print(data)
+print(data["1"])
+
+a = [ data[str(n+1)] for n in range(64)]
+a = np.array(a)
+
+
+###############################################################################
+#                       Get Data from Shared Memory                           #
+###############################################################################
+
+shm = shared_memory.SharedMemory(create=True, size=a.nbytes, name="DO-shm")
+b = np.ndarray(a.shape, dtype=a.dtype, buffer=shm.buf)
+b[:] = a[:]  
+
+
+###############################################################################
+#                         MQTT Broker Setting                                 #
+###############################################################################
+
+#MQTT_TOPIC = info['mqtt']['topic']
+MQTT_TOPIC = f"/device-data/{dev_info['assetcode']}"
+MQTT_ID = info['mqtt']['id']
+MQTT_PW = info['mqtt']['pw']
+MQTT_HOST_IP = info['mqtt']['host_ip']
+MQTT_PORT = info['mqtt']['port']
+
+
+def publish_json_message(client, data):
+    topic = MQTT_TOPIC
+    client.publish(topic, payload=data)
+
+
+client = mqtt.Client("python_pub")
+client.username_pw_set(MQTT_ID, MQTT_PW)
+client.connect(host=MQTT_HOST_IP,port=MQTT_PORT)
+
+client.loop(2)  # timeout = 2초
+
+
+
+
+###############################################################################
+#                         Serial setting                                 #
+###############################################################################
+
+_serial_port3 = serial.Serial("/dev/ttyMAX2", 115200, parity= serial.PARITY_NONE, stopbits=serial.STOPBITS_ONE, bytesize=serial.EIGHTBITS, timeout = 1, rtscts=True)
+_serial_port3.rs485_mode = serial.rs485.RS485Settings()
+_serial_port4 = serial.Serial("/dev/ttyMAX3", 115200, parity= serial.PARITY_NONE, stopbits=serial.STOPBITS_ONE, bytesize=serial.EIGHTBITS, timeout = 1, rtscts=True)
+_serial_port4.rs485_mode = serial.rs485.RS485Settings()
+
+
+######################################### thread 함수 작성 ################################################
+
+
+
+###############################################################################
+#                         Serial thread                                  #
+###############################################################################
+isThreadRun = True
+
+def cwt_thread(evt, chip_index, port_a_ai_id , port_a_do_id, port_b_ai_id , port_b_do_id, queue_a, queue_b):
+
+    print("index: ", chip_index)
+    print("a-ai: %d, a-do: %d, b-ai: %d, b-do: %d " % (port_a_ai_id, port_a_do_id, port_b_ai_id, port_b_do_id))
+    if chip_index == 0:
+        serial_dev1 = "/dev/ttyMAX1"
+        serial_dev2 = "/dev/ttyMAX0"
+    else:
+        serial_dev1 = "/dev/ttyMAX2"
+        serial_dev2 = "/dev/ttyMAX3"
+
+    print(serial_dev1)
+    print(serial_dev2)
+    _serial_port_A = serial.Serial(serial_dev1, 115200, parity= serial.PARITY_NONE, stopbits=serial.STOPBITS_ONE, bytesize=serial.EIGHTBITS, timeout = 1, rtscts=True)
+    _serial_port_A.rs485_mode = serial.rs485.RS485Settings()
+    _serial_port_B = serial.Serial(serial_dev2, 115200, parity= serial.PARITY_NONE, stopbits=serial.STOPBITS_ONE, bytesize=serial.EIGHTBITS, timeout = 1, rtscts=True)
+    _serial_port_B.rs485_mode = serial.rs485.RS485Settings()
+
+
+    port_A_AI = minimalmodbus.Instrument(serial_dev1, port_a_ai_id, debug=False, close_port_after_each_call=False)  # port name, slave address (in decimal)
+    port_A_AI.serial.baudrate = 115200  # baudrate
+
+    port_A_DO = minimalmodbus.Instrument(serial_dev1, port_a_do_id, debug=False, close_port_after_each_call=False)  # port name, slave address (in decimal)
+    port_A_DO.serial.baudrate = 115200  # baudrate
+
+    port_B_AI = minimalmodbus.Instrument(serial_dev2, port_b_ai_id, debug=False, close_port_after_each_call=False)  # port name, slave address (in decimal)
+    port_B_AI.serial.baudrate = 115200  # baudrate
+
+    port_B_DO = minimalmodbus.Instrument(serial_dev2, port_b_do_id, debug=False, close_port_after_each_call=False)  # port name, slave address (in decimal)
+    port_B_DO.serial.baudrate = 115200  # baudrate
+
+
+    while isThreadRun:
+        evt.wait()
+        #print("read thread get even")
+        evt.clear()
+
+        prev_a_ai = [0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0]
+        prev_b_ai = [0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0]
+        
+        #value list on -> do control data
+        #추후에 do power on 할 리스트는 어디선가 읽어서 와야 해요
+        #do1 do2 나눠야 해요
+        
+        
+
+        '''
+                | AI1 | AI2 | DO1 | DO2 |
+        Success |  0  |  0  |  x  |  x  |
+        Success |  0  |  1  |  x  |  x  |
+        Success |  1  |  0  |  x  |  x  |
+        Success |  1  |  1  |  x  |  x  | => 유효한 데이터
+
+        
+        최소 한개 이상 실패 했을 때
+         -> retry 해볼 것이냐
+
+         -> 데이터를 어떻게 할 것이냐
+          ->  이전 데이터가 있다면, 이전 데이터를 가지고 더미 데이터로 사용하는 방법 => 안하기로 했었음      
+        '''
+        timenow = int(time.time()*1000)
+        isSuccess = True
+
+        # PORT A
+        port_a_do_value = list(b[:16])
+        try:
+            a_ai_data=port_A_AI.read_registers(0x32, 32)
+            # suc_cnt_ai = suc_cnt_ai +1
+        except:
+            isSuccess = False
+
+        if isSuccess:
+            prev_a_ai = a_ai_data
+
+        try:
+            port_A_DO.write_bits(0, port_a_do_value)
+            # suc_cnt_do = suc_cnt_do +1
+        except:
+            None
+        
+        # PORT B
+        port_b_do_value = list(b[17:32])
+        try:
+            b_ai_data=port_B_AI.read_registers(0x32, 32)
+            prev_b_ai = b_ai_data
+            # suc_cnt_ai = suc_cnt_ai +1
+        except:
+            isSuccess = False
+
+        if isSuccess:
+            prev_b_ai = b_ai_data
+    
+        try:
+            port_B_DO.write_bits(0, port_b_do_value)
+            # suc_cnt_do = suc_cnt_do +1
+        except:
+            None
+
+        if isSuccess:
+            queue_a.put([timenow,prev_a_ai, port_a_do_value])
+            queue_b.put([timenow,prev_b_ai, port_b_do_value])
+        else:
+            queue_a.put([0,prev_a_ai, port_a_do_value])
+            queue_b.put([0,prev_b_ai, port_b_do_value])
+            
+
+
+        
+
+
+        # lock.release()
+
+        '''
+        리스트를 생성, 10개를 딱 잘라서 어떻게 해야 할ㅈ지 모르겠음
+
+
+        '''
+        ##데이터 생성 할 때
+        # ai1, ai2
+        # -> ai1 [32] -> [curr, volt, curr, volt ...]
+        
+        '''
+        data_for_list = {
+            "slot" : 1,
+            "timestamp": int(timenow*1000),
+            "data":{
+                "current": ai1,
+                "volatge": ai2,
+                "do1": value_list_ON,
+                "err1": value_list_ON,
+                "err2": value_list_ON
+                #err1, 2
+                }
+            }
+        lists.append(data_for_list)
+        ### 여기서 보냄      
+        # 10번 시그널 받으면
+        #     
+        # list mqtt send
+        # list clear
+        if read_cnt_ch0 % 10 == 0:
+            data = {
+                    "modelCode":"test4",
+                    "assetCode":"NQ-R04-TEST-003",
+                    "dataType":"DATA",
+                    "data": {
+                        "chamber":1,
+                        "list": lists
+                        }
+                    }
+
+            data_json_str = json.dumps(data, indent=0)
+            publish_json_message(client, data_json_str)
+            #print(data_json_str)
+       # time.sleep(1)
+        '''
+
+
+###############################################################################
+#                         data process thread                                 #
+###############################################################################
+
+
+def data_process_thread(data_queue, ch, chamber):
+    error_cnt_1 = [0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0]
+    error_cnt_2 = [0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0]
+    data_list_for_send_1=[]
+    data_list_for_send_2=[]
+    queue_count = 0
+    queue_select_count = 0
+    while isThreadRun:
+        # 0.1s 마다 발생할 것으로 예상
+        timestamp, ai, do = data_queue.get()
+
+        #==================================================
+        do_temp = []
+        for d in do:
+            do_temp.append(int(d))
+        do = do_temp.copy()        
+        #==================================================
+
+
+        #print("timestamp: {timestamp}, \n ai: {ai}, \n do: {do}" .format(timestamp=timestamp, ai=ai, do=do))
+
+        #ai data -> current, voltage 전환
+        queue_count = queue_count + 1
+        current = []
+        voltage = []
+        ai_index = 0
+        for i in ai:
+            #짝수 current (확인 필요)
+            ai_index = ai_index +1
+            if (ai_index%2) == 0:
+                current.append(int(i))
+            else:
+                voltage.append(int(i))
+
+        
+        #전압값으로 error count check
+        voltage_cnt = 0
+        for i in voltage:
+            if i < 2:
+                #count up
+                error_cnt_1[voltage_cnt] = error_cnt_1[voltage_cnt] +1
+                if error_cnt_1[voltage_cnt] == 10:
+                    error_cnt_1[voltage_cnt] = 0
+                    error_cnt_2[voltage_cnt] = error_cnt_2[voltage_cnt] + 1
+                    #shared memory do control modify
+                    #ch, tube number->off
+            else:
+                # count reset
+                error_cnt_1[voltage_cnt] = 0
+
+
+
+            voltage_cnt = voltage_cnt + 1
+
+        # 로깅 기능을 넣어야 겠다...?
+        # 파일 하나 열어서, 데이터를 쓴다
+
+        # 데이터 만들어서 전달
+        if timestamp > 0:
+            data = {
+                "timestamp" : int(timestamp),
+                "data": {
+                    "current"   : current,
+                    "voltage"   : voltage,
+                    "do"        : do,
+                    "er1"       : error_cnt_1,
+                    "er2"       : error_cnt_2
+                }
+            }
+            if(queue_select_count == 0):
+                data_list_for_send_1.append(data)
+            else:
+                data_list_for_send_2.append(data)
+
+        
+        if queue_count == 10:
+            queue_count = 0
+            mqtt_msg = {}
+            if (queue_select_count == 0):
+                mqtt_msg = {
+                    "modelCode":"test4",
+                    "assetCode":"NQ-R04-TEST-003",  #나중에는 config에서 셋팅되는 값, or model 값으로 변경 (asset)
+                    "dataType":"DATA",
+                    "data": {
+                        "channel": ch,  # 채널 입력 받자
+                        "chamber":chamber,    #chamber 도 입력 받자
+                        "slot": ch,     #슬롯 값은 chamber와 ch 값으로 만든다
+                        "list": data_list_for_send_1
+                    }
+                }
+                queue_select_count = 1
+                data_list_for_send_2.clear()
+            else:
+                mqtt_msg = {
+                    "modelCode":"test4",
+                    "assetCode":"NQ-R04-TEST-003",  #나중에는 config에서 셋팅되는 값, or model 값으로 변경 (asset)
+                    "dataType":"DATA",
+                    "data": {
+                        "channel": ch,  # 채널 입력 받자
+                        "chamber":chamber,    #chamber 도 입력 받자
+                        "slot": ch,     #슬롯 값은 chamber와 ch 값으로 만든다
+                        "list": data_list_for_send_2
+                    }
+                }
+                queue_select_count = 0
+                data_list_for_send_1.clear()
+
+            
+#            print(type(mqtt_msg['data']['list'][0]['do'][0]))
+            data_json_str = json.dumps(mqtt_msg, indent=4)
+#            print(data_json_str)
+#            print(data_json_str)
+            publish_json_message(client, data_json_str)
+
+        ##erro cnt check routine
+
+
+
+
+
+
+######################################### thread 함수 작성 ################################################
+
+
+#event는 개별 생성을 해준다 th1, th2
+
+
+def timer_event_loop(e1, e2):
+    cnt = 0
+    # global read_cnt
+
+    while True:
+        # print("event set, meanning : periodic sensing")
+         e1.set()
+         e2.set()
+         time.sleep(0.1)
+
+         #cnt = cnt +1
+         #if cnt > 6000:
+            # print("chip0 read count:", read_cnt_ch0)
+             #print("\r\nchip1 read count:", read_cnt_ch0)
+             #print("\r\nai1:", suc_cnt_ai, "\r\nai2:", suc_cnt_ai2)
+             #print(ai2)
+             #break
+
+    #print(cnt)
+#    print(read_cnt)
+    exit()
+#        timer = threading.Timer(3, lambda: print("test event"))
+#        timer.start()
+#        timer.join()
+# #        timer = threading.Timer(3, lambda: e.set())
+
+
+################################# 작성 중 #####################################
+# def publish_json_message(client, data):
+#     topic = "devicedata/axr/nodeq1"
+#     client.publish(topic, payload=data)
+
+
+# client = mqtt.Client("python_pub")
+# client.username_pw_set("sdt", "251327")
+# client.connect(host="13.209.39.139",port=32259)
+
+# # get_value()
+
+
+# data = json.dumps(dict_1)
+# #publish_json_message(client, data)
+
+# client.loop(2)  # timeout = 2초
+
+################################# 작성 중 ######################################
+
+if __name__ == '__main__':
+    ###############################################################################
+    #                         Queue x 4         max- infinite                          #
+    ###############################################################################
+
+    ch1_queue = queue.Queue()
+    ch2_queue = queue.Queue()
+    ch3_queue = queue.Queue()
+    ch4_queue = queue.Queue()
+
+    chip0_evt = threading.Event()
+    chip1_evt = threading.Event()
+
+    read_data_chip0 = threading.Thread(target=cwt_thread,args=(chip0_evt,   0,  1,2,    5,6,   ch1_queue, ch2_queue))  #현재 4개씩 연결되어 있어서 테스트 용
+    read_data_chip1 = threading.Thread(target=cwt_thread,args=(chip1_evt,   1,  9,10,    13,14, ch3_queue, ch4_queue))
+
+    data_process_ch1 = threading.Thread(target=data_process_thread,args=(ch1_queue,   1,  1))
+    data_process_ch2 = threading.Thread(target=data_process_thread,args=(ch2_queue,   2,  1))
+    data_process_ch3 = threading.Thread(target=data_process_thread,args=(ch3_queue,   3,  1))
+    data_process_ch4 = threading.Thread(target=data_process_thread,args=(ch4_queue,   4,  1))
+
+    event_thread = threading.Thread(target=timer_event_loop, args=(chip0_evt,chip1_evt,))
+    read_data_chip0.start()
+    read_data_chip1.start()
+    data_process_ch1.start()
+    data_process_ch2.start()
+    data_process_ch3.start()
+    data_process_ch4.start()
+    event_thread.start()

AWX_collector_copy.py

@@ -0,0 +1,493 @@
+
+import threading
+import time
+import logging
+import serial
+import minimalmodbus
+import serial.rs485
+import subprocess
+import json
+import yaml
+import queue
+import paho.mqtt.client as mqtt
+import numpy as np
+from multiprocessing import shared_memory
+import argparse
+
+
+'''
+전달해야할 데이터 들
+mes code | 상태 | 전압 | 전류 | 4pin (DO) | ER1 | ER2
+tube 번호| 상태 | volt | curr | do        | cnt | cnt
+
+tube 번호: 1 - 128 , chamber / slot
+ - slot(n) - tube(n)
+ - 1-1 ~ 8-16 : 128개다
+
+상태: 정상이냐, 초기화, 불량
+
+ER1: 1초동안 판별, 판별할 동안 counting 할 예정 2V 이하라면 CNT UP 
+ if) er: 10이 되었음 (1초 유지) -> 0으로 변환, ER2 cnt up 
+ if) 2v 이상인 경우가 발생했다 -> 0으로 변환
+
+ER2: OFF 횟수
+
+'''
+
+parser = argparse.ArgumentParser()
+parser.add_argument('-config',help='')
+args = parser.parse_args()
+
+ROOT_PATH = f"/usr/local/sdt/app/{args.config}"
+DEVICE_PATH = f"/etc/sdt/device-control"
+
+###############################################################################
+#                         JSON FILE READ                                      #
+###############################################################################
+
+with open(f"{ROOT_PATH}/connect_info.json","r") as f:
+#with open(f"{args.config}/connect_info.json","r") as f:
+    info = json.load(f)
+
+with open(f"{DEVICE_PATH}/config.yaml","r") as f:
+#with open(f"{args.config}/connect_info.json","r") as f:
+    dev_info = yaml.load(f, Loader=yaml.FullLoader)
+
+
+###############################################################################
+#                         Shared memory setting                               #
+###############################################################################
+
+
+
+with open(f"{ROOT_PATH}/config.json","r") as f:
+#with open(f"{args.config}/config.json","r") as f:
+    data = json.load(f)
+    
+print(data)
+print(data["1"])
+
+a = [ data[str(n+1)] for n in range(64)]
+a = np.array(a)
+
+
+###############################################################################
+#                       Get Data from Shared Memory                           #
+###############################################################################
+
+shm = shared_memory.SharedMemory(create=True, size=a.nbytes, name="DO-shm")
+b = np.ndarray(a.shape, dtype=a.dtype, buffer=shm.buf)
+b[:] = a[:]  
+
+
+###############################################################################
+#                         MQTT Broker Setting                                 #
+###############################################################################
+
+#MQTT_TOPIC = info['mqtt']['topic']
+MQTT_TOPIC = f"/device-data/{dev_info['assetcode']}"
+MQTT_ID = info['mqtt']['id']
+MQTT_PW = info['mqtt']['pw']
+MQTT_HOST_IP = info['mqtt']['host_ip']
+MQTT_PORT = info['mqtt']['port']
+
+
+def publish_json_message(client, data):
+    topic = MQTT_TOPIC
+    client.publish(topic, payload=data)
+
+
+client = mqtt.Client("python_pub")
+client.username_pw_set(MQTT_ID, MQTT_PW)
+client.connect(host=MQTT_HOST_IP,port=MQTT_PORT)
+
+client.loop(2)  # timeout = 2초
+
+
+
+
+###############################################################################
+#                         Serial setting                                 #
+###############################################################################
+
+_serial_port3 = serial.Serial("/dev/ttyMAX2", 115200, parity= serial.PARITY_NONE, stopbits=serial.STOPBITS_ONE, bytesize=serial.EIGHTBITS, timeout = 1, rtscts=True)
+_serial_port3.rs485_mode = serial.rs485.RS485Settings()
+_serial_port4 = serial.Serial("/dev/ttyMAX3", 115200, parity= serial.PARITY_NONE, stopbits=serial.STOPBITS_ONE, bytesize=serial.EIGHTBITS, timeout = 1, rtscts=True)
+_serial_port4.rs485_mode = serial.rs485.RS485Settings()
+
+
+######################################### thread 함수 작성 ################################################
+
+
+
+###############################################################################
+#                         Serial thread                                  #
+###############################################################################
+isThreadRun = True
+
+def cwt_thread(evt, chip_index, port_a_ai_id , port_a_do_id, port_b_ai_id , port_b_do_id, queue_a, queue_b):
+
+    print("index: ", chip_index)
+    print("a-ai: %d, a-do: %d, b-ai: %d, b-do: %d " % (port_a_ai_id, port_a_do_id, port_b_ai_id, port_b_do_id))
+    if chip_index == 0:
+        serial_dev1 = "/dev/ttyMAX1"
+        serial_dev2 = "/dev/ttyMAX0"
+    else:
+        serial_dev1 = "/dev/ttyMAX2"
+        serial_dev2 = "/dev/ttyMAX3"
+
+    print(serial_dev1)
+    print(serial_dev2)
+    _serial_port_A = serial.Serial(serial_dev1, 115200, parity= serial.PARITY_NONE, stopbits=serial.STOPBITS_ONE, bytesize=serial.EIGHTBITS, timeout = 1, rtscts=True)
+    _serial_port_A.rs485_mode = serial.rs485.RS485Settings()
+    _serial_port_B = serial.Serial(serial_dev2, 115200, parity= serial.PARITY_NONE, stopbits=serial.STOPBITS_ONE, bytesize=serial.EIGHTBITS, timeout = 1, rtscts=True)
+    _serial_port_B.rs485_mode = serial.rs485.RS485Settings()
+
+
+    port_A_AI = minimalmodbus.Instrument(serial_dev1, port_a_ai_id, debug=False, close_port_after_each_call=False)  # port name, slave address (in decimal)
+    port_A_AI.serial.baudrate = 115200  # baudrate
+
+    port_A_DO = minimalmodbus.Instrument(serial_dev1, port_a_do_id, debug=False, close_port_after_each_call=False)  # port name, slave address (in decimal)
+    port_A_DO.serial.baudrate = 115200  # baudrate
+
+    port_B_AI = minimalmodbus.Instrument(serial_dev2, port_b_ai_id, debug=False, close_port_after_each_call=False)  # port name, slave address (in decimal)
+    port_B_AI.serial.baudrate = 115200  # baudrate
+
+    port_B_DO = minimalmodbus.Instrument(serial_dev2, port_b_do_id, debug=False, close_port_after_each_call=False)  # port name, slave address (in decimal)
+    port_B_DO.serial.baudrate = 115200  # baudrate
+
+
+    while isThreadRun:
+        evt.wait()
+        #print("read thread get even")
+        evt.clear()
+
+        prev_a_ai = [0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0]
+        prev_b_ai = [0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0]
+        
+        #value list on -> do control data
+        #추후에 do power on 할 리스트는 어디선가 읽어서 와야 해요
+        #do1 do2 나눠야 해요
+        
+        
+
+        '''
+                | AI1 | AI2 | DO1 | DO2 |
+        Success |  0  |  0  |  x  |  x  |
+        Success |  0  |  1  |  x  |  x  |
+        Success |  1  |  0  |  x  |  x  |
+        Success |  1  |  1  |  x  |  x  | => 유효한 데이터
+
+        
+        최소 한개 이상 실패 했을 때
+         -> retry 해볼 것이냐
+
+         -> 데이터를 어떻게 할 것이냐
+          ->  이전 데이터가 있다면, 이전 데이터를 가지고 더미 데이터로 사용하는 방법 => 안하기로 했었음      
+        '''
+        timenow = int(time.time()*1000)
+        isSuccess = True
+
+        # PORT A
+        port_a_do_value = list(b[:16])
+        try:
+            a_ai_data=port_A_AI.read_registers(0x32, 32)
+            # suc_cnt_ai = suc_cnt_ai +1
+        except:
+            isSuccess = False
+
+        if isSuccess:
+            prev_a_ai = a_ai_data
+
+        try:
+            port_A_DO.write_bits(0, port_a_do_value)
+            # suc_cnt_do = suc_cnt_do +1
+        except:
+            None
+        
+        # PORT B
+        port_b_do_value = list(b[17:32])
+        try:
+            b_ai_data=port_B_AI.read_registers(0x32, 32)
+            prev_b_ai = b_ai_data
+            # suc_cnt_ai = suc_cnt_ai +1
+        except:
+            isSuccess = False
+
+        if isSuccess:
+            prev_b_ai = b_ai_data
+    
+        try:
+            port_B_DO.write_bits(0, port_b_do_value)
+            # suc_cnt_do = suc_cnt_do +1
+        except:
+            None
+
+        if isSuccess:
+            queue_a.put([timenow,prev_a_ai, port_a_do_value])
+            queue_b.put([timenow,prev_b_ai, port_b_do_value])
+        else:
+            queue_a.put([0,prev_a_ai, port_a_do_value])
+            queue_b.put([0,prev_b_ai, port_b_do_value])
+            
+
+
+        
+
+
+        # lock.release()
+
+        '''
+        리스트를 생성, 10개를 딱 잘라서 어떻게 해야 할ㅈ지 모르겠음
+
+
+        '''
+        ##데이터 생성 할 때
+        # ai1, ai2
+        # -> ai1 [32] -> [curr, volt, curr, volt ...]
+        
+        '''
+        data_for_list = {
+            "slot" : 1,
+            "timestamp": int(timenow*1000),
+            "data":{
+                "current": ai1,
+                "volatge": ai2,
+                "do1": value_list_ON,
+                "err1": value_list_ON,
+                "err2": value_list_ON
+                #err1, 2
+                }
+            }
+        lists.append(data_for_list)
+        ### 여기서 보냄      
+        # 10번 시그널 받으면
+        #     
+        # list mqtt send
+        # list clear
+        if read_cnt_ch0 % 10 == 0:
+            data = {
+                    "modelCode":"test4",
+                    "assetCode":"NQ-R04-TEST-003",
+                    "dataType":"DATA",
+                    "data": {
+                        "chamber":1,
+                        "list": lists
+                        }
+                    }
+
+            data_json_str = json.dumps(data, indent=0)
+            publish_json_message(client, data_json_str)
+            #print(data_json_str)
+       # time.sleep(1)
+        '''
+
+
+###############################################################################
+#                         data process thread                                 #
+###############################################################################
+
+
+def data_process_thread(data_queue, ch, chamber):
+    error_cnt_1 = [0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0]
+    error_cnt_2 = [0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0]
+    data_list_for_send_1=[]
+    data_list_for_send_2=[]
+    queue_count = 0
+    queue_select_count = 0
+    while isThreadRun:
+        # 0.1s 마다 발생할 것으로 예상
+        timestamp, ai, do = data_queue.get()
+
+        #==================================================
+        do_temp = []
+        for d in do:
+            do_temp.append(int(d))
+        do = do_temp.copy()        
+        #==================================================
+
+
+        #print("timestamp: {timestamp}, \n ai: {ai}, \n do: {do}" .format(timestamp=timestamp, ai=ai, do=do))
+
+        #ai data -> current, voltage 전환
+        queue_count = queue_count + 1
+        current = []
+        voltage = []
+        ai_index = 0
+        for i in ai:
+            #짝수 current (확인 필요)
+            ai_index = ai_index +1
+            if (ai_index%2) == 0:
+                current.append(int(i))
+            else:
+                voltage.append(int(i))
+
+        
+        #전압값으로 error count check
+        voltage_cnt = 0
+        for i in voltage:
+            if i < 2:
+                #count up
+                error_cnt_1[voltage_cnt] = error_cnt_1[voltage_cnt] +1
+                if error_cnt_1[voltage_cnt] == 10:
+                    error_cnt_1[voltage_cnt] = 0
+                    error_cnt_2[voltage_cnt] = error_cnt_2[voltage_cnt] + 1
+                    #shared memory do control modify
+                    #ch, tube number->off
+            else:
+                # count reset
+                error_cnt_1[voltage_cnt] = 0
+
+
+
+            voltage_cnt = voltage_cnt + 1
+
+        # 로깅 기능을 넣어야 겠다...?
+        # 파일 하나 열어서, 데이터를 쓴다
+
+        # 데이터 만들어서 전달
+        if timestamp > 0:
+            data = {
+                "timestamp" : int(timestamp),
+                "data": {
+                    "current"   : current,
+                    "voltage"   : voltage,
+                    "do"        : do,
+                    "er1"       : error_cnt_1,
+                    "er2"       : error_cnt_2
+                }
+            }
+            if(queue_select_count == 0):
+                data_list_for_send_1.append(data)
+            else:
+                data_list_for_send_2.append(data)
+
+        
+        if queue_count == 10:
+            queue_count = 0
+            mqtt_msg = {}
+            if (queue_select_count == 0):
+                mqtt_msg = {
+                    "modelCode":"test4",
+                    "assetCode":"NQ-R04-TEST-003",  #나중에는 config에서 셋팅되는 값, or model 값으로 변경 (asset)
+                    "dataType":"DATA",
+                    "data": {
+                        "channel": ch,  # 채널 입력 받자
+                        "chamber":chamber,    #chamber 도 입력 받자
+                        "slot": ch,     #슬롯 값은 chamber와 ch 값으로 만든다
+                        "list": data_list_for_send_1
+                    }
+                }
+                queue_select_count = 1
+                data_list_for_send_2.clear()
+            else:
+                mqtt_msg = {
+                    "modelCode":"test4",
+                    "assetCode":"NQ-R04-TEST-003",  #나중에는 config에서 셋팅되는 값, or model 값으로 변경 (asset)
+                    "dataType":"DATA",
+                    "data": {
+                        "channel": ch,  # 채널 입력 받자
+                        "chamber":chamber,    #chamber 도 입력 받자
+                        "slot": ch,     #슬롯 값은 chamber와 ch 값으로 만든다
+                        "list": data_list_for_send_2
+                    }
+                }
+                queue_select_count = 0
+                data_list_for_send_1.clear()
+
+            
+#            print(type(mqtt_msg['data']['list'][0]['do'][0]))
+            data_json_str = json.dumps(mqtt_msg, indent=4)
+#            print(data_json_str)
+#            print(data_json_str)
+            publish_json_message(client, data_json_str)
+
+        ##erro cnt check routine
+
+
+
+
+
+
+######################################### thread 함수 작성 ################################################
+
+
+#event는 개별 생성을 해준다 th1, th2
+
+
+def timer_event_loop(e1, e2):
+    cnt = 0
+    # global read_cnt
+
+    while True:
+        # print("event set, meanning : periodic sensing")
+         e1.set()
+         e2.set()
+         time.sleep(0.1)
+
+         #cnt = cnt +1
+         #if cnt > 6000:
+            # print("chip0 read count:", read_cnt_ch0)
+             #print("\r\nchip1 read count:", read_cnt_ch0)
+             #print("\r\nai1:", suc_cnt_ai, "\r\nai2:", suc_cnt_ai2)
+             #print(ai2)
+             #break
+
+    #print(cnt)
+#    print(read_cnt)
+    exit()
+#        timer = threading.Timer(3, lambda: print("test event"))
+#        timer.start()
+#        timer.join()
+# #        timer = threading.Timer(3, lambda: e.set())
+
+
+################################# 작성 중 #####################################
+# def publish_json_message(client, data):
+#     topic = "devicedata/axr/nodeq1"
+#     client.publish(topic, payload=data)
+
+
+# client = mqtt.Client("python_pub")
+# client.username_pw_set("sdt", "251327")
+# client.connect(host="13.209.39.139",port=32259)
+
+# # get_value()
+
+
+# data = json.dumps(dict_1)
+# #publish_json_message(client, data)
+
+# client.loop(2)  # timeout = 2초
+
+################################# 작성 중 ######################################
+
+if __name__ == '__main__':
+    ###############################################################################
+    #                         Queue x 4         max- infinite                          #
+    ###############################################################################
+
+    ch1_queue = queue.Queue()
+    ch2_queue = queue.Queue()
+    ch3_queue = queue.Queue()
+    ch4_queue = queue.Queue()
+
+    chip0_evt = threading.Event()
+    chip1_evt = threading.Event()
+
+    read_data_chip0 = threading.Thread(target=cwt_thread,args=(chip0_evt,   0,  1,2,    5,6,   ch1_queue, ch2_queue))  #현재 4개씩 연결되어 있어서 테스트 용
+    read_data_chip1 = threading.Thread(target=cwt_thread,args=(chip1_evt,   1,  9,10,    13,14, ch3_queue, ch4_queue))
+
+    data_process_ch1 = threading.Thread(target=data_process_thread,args=(ch1_queue,   1,  1))
+    data_process_ch2 = threading.Thread(target=data_process_thread,args=(ch2_queue,   2,  1))
+    data_process_ch3 = threading.Thread(target=data_process_thread,args=(ch3_queue,   3,  1))
+    data_process_ch4 = threading.Thread(target=data_process_thread,args=(ch4_queue,   4,  1))
+
+    event_thread = threading.Thread(target=timer_event_loop, args=(chip0_evt,chip1_evt,))
+    read_data_chip0.start()
+    read_data_chip1.start()
+    data_process_ch1.start()
+    data_process_ch2.start()
+    data_process_ch3.start()
+    data_process_ch4.start()
+    event_thread.start()

AWX_shm_updater.py

@@ -0,0 +1,23 @@
+import numpy as np
+from multiprocessing import shared_memory
+import time
+import json
+import argparse
+
+
+existing_shm = shared_memory.SharedMemory(name="DO-shm")
+
+c = np.ndarray((64,), dtype=np.int32, buffer=existing_shm.buf)
+
+parser = argparse.ArgumentParser()
+parser.add_argument('-config',help='')
+args = parser.parse_args()
+
+ROOT_PATH = f"/usr/local/sdt/app/{args.config}"
+
+while True:
+    with open(f"{ROOT_PATH}/config.json", 'r') as file:
+        jsonData = json.load(file)
+    for n in range(64):
+        c[n]=jsonData[str(n+1)]
+    time.sleep(2)

AWX_shm_updater.py~

@@ -0,0 +1,18 @@
+import numpy as np
+from multiprocessing import shared_memory
+import time
+import json
+
+
+
+existing_shm = shared_memory.SharedMemory(name="DO-shm")
+
+c = np.ndarray((64,), dtype=np.int32, buffer=existing_shm.buf)
+
+file_path = "./config.json"
+while True:
+    with open(file_path, 'r') as file:
+        jsonData = json.load(file)
+    for n in range(64):
+        c[n]=jsonData[str(n+1)]
+    time.sleep(2)

AWX_shm_updater.service

@@ -0,0 +1,10 @@
+[Unit]
+Description=awx_shm_updater
+
+[Service]
+ExecStart=/usr/bin/python3 /usr/local/sdt/app/appName/AWX_shm_updater.py -config appName
+Restart=always
+RestartSec=10
+
+[Install]
+WantedBy=multi-user.target

AWX_shm_updater.service~

@@ -0,0 +1,10 @@
+[Unit]
+Description=awx_collector
+
+[Service]
+ExecStart=/usr/bin/python3 /usr/local/sdt/app/appName/AWX_collector.py -config appName
+Restart=always
+RestartSec=10
+
+[Install]
+WantedBy=multi-user.target

config.json

@@ -1,9 +1,9 @@
 {
-    "1": 0,
+    "1": 1,
-    "2": 0,
+    "2": 1,
     "3": 1,
-    "4": 0,
+    "4": 1,
-    "5": 1,
+    "5": 0,
     "6": 0,
     "7": 0,
     "8": 0,
@@ -30,5 +30,37 @@
     "29": 0,
     "30": 0,
     "31": 0,
-    "32": 0
+    "32": 0,
+    "33": 0,
+    "34": 0,
+    "35": 0,
+    "36": 1,
+    "37": 0,
+    "38": 0,
+    "39": 0,
+    "40": 0,
+    "41": 0,
+    "42": 0,
+    "43": 0,
+    "44": 0,
+    "45": 0,
+    "46": 1,
+    "47": 0,
+    "48": 0,
+    "49": 0,
+    "50": 0,
+    "51": 0,
+    "52": 0,
+    "53": 0,
+    "54": 0,
+    "55": 0,
+    "56": 1,
+    "57": 0,
+    "58": 0,
+    "59": 0,
+    "60": 0,
+    "61": 0,
+    "62": 0,
+    "63": 0,
+    "64": 0
 }

config.json~

@@ -0,0 +1,66 @@
+{
+    "1": 1,
+    "2": 1,
+    "3": 1,
+    "4": 1,
+    "5": 1,
+    "6": 1,
+    "7": 1,
+    "8": 1,
+    "9": 1,
+    "10": 0,
+    "11": 0,
+    "12": 0,
+    "13": 0,
+    "14": 0,
+    "15": 0,
+    "16": 1,
+    "17": 0,
+    "18": 0,
+    "19": 0,
+    "20": 0,
+    "21": 0,
+    "22": 0,
+    "23": 0,
+    "24": 0,
+    "25": 0,
+    "26": 0,
+    "27": 0,
+    "28": 0,
+    "29": 0,
+    "30": 0,
+    "31": 0,
+    "32": 0,
+    "33": 0,
+    "34": 0,
+    "35": 0,
+    "36": 1,
+    "37": 0,
+    "38": 0,
+    "39": 0,
+    "40": 0,
+    "41": 0,
+    "42": 0,
+    "43": 0,
+    "44": 0,
+    "45": 0,
+    "46": 1,
+    "47": 0,
+    "48": 0,
+    "49": 0,
+    "50": 0,
+    "51": 0,
+    "52": 0,
+    "53": 0,
+    "54": 0,
+    "55": 0,
+    "56": 1,
+    "57": 0,
+    "58": 0,
+    "59": 0,
+    "60": 0,
+    "61": 0,
+    "62": 0,
+    "63": 0,
+    "64": 0
+}

config_corr.json

@@ -0,0 +1,67 @@
+{
+    "1": 1,
+    "2": 2,
+    "3": 3,
+    "4": 4,
+    "5": 5,
+    "6": 6,
+    "7": 7,
+    "8": 8,
+    "9": 9,
+    "10": 5,
+    "11": 5,
+    "12": 5,
+    "13": 5,
+    "14": 5,
+    "15": 15,
+    "16": 16,
+    "17": 17,
+    "18": 5,
+    "19": 5,
+    "20": 5,
+    "21": 5,
+    "22": 5,
+    "23": 5,
+    "24": 5,
+    "25": 5,
+    "26": 5,
+    "27": 5,
+    "28": 5,
+    "29": 5,
+    "30": 5,
+    "31": 31,
+    "32": 32,
+    "33": 33,
+    "34": 5,
+    "35": 5,
+    "36": 5,
+    "37": 5,
+    "38": 5,
+    "39": 5,
+    "40": 0,
+    "41": 0,
+    "42": 0,
+    "43": 0,
+    "44": 0,
+    "45": 0,
+    "46": 5,
+    "47": 47,
+    "48": 48,
+    "49": 49,
+    "50": 0,
+    "51": 0,
+    "52": 0,
+    "53": 0,
+    "54": 0,
+    "55": 0,
+    "56": 5,
+    "57": 0,
+    "58": 0,
+    "59": 0,
+    "60": 0,
+    "61": 0,
+    "62": 0,
+    "63": 0,
+    "64": 64
+}
+

install.sh

@@ -3,5 +3,11 @@
 pip install -r requirements.txt
 sudo sed -i "s/appName/$1/g" /usr/local/sdt/app/$1/AWX_collector.service
 sudo cp /usr/local/sdt/app/$1/AWX_collector.service /etc/systemd/system/$1.service
+
+sudo sed -i "s/appName/$1/g" /usr/local/sdt/app/$1/AWX_shm_updater.service
+sudo cp /usr/local/sdt/app/$1/AWX_shm_updater.service /etc/systemd/system/AWX_shm_updater.service
 sudo systemctl start $1
 sudo systemctl enable $1
+
+sudo systemctl start AWX_shm_updater
+sudo systemctl enable AWX_shm_updater

install.sh~

@@ -1,6 +1,9 @@
 #!/bin/bash
 
+pip install -r requirements.txt
 sudo sed -i "s/appName/$1/g" /usr/local/sdt/app/$1/AWX_collector.service
 sudo cp /usr/local/sdt/app/$1/AWX_collector.service /etc/systemd/system/$1.service
+
+sudo cp /usr/local/sdt/app/$1/AWX_shm_updater.service /etc/systemd/system/AWX_shm_updater.service
 sudo systemctl start $1
 sudo systemctl enable $1

shm_updater.py~

@@ -0,0 +1,18 @@
+import numpy as np
+from multiprocessing import shared_memory
+import time
+import json
+
+
+
+existing_shm = shared_memory.SharedMemory(name="do_shm2")
+
+c = np.ndarray((32,), dtype=np.int32, buffer=existing_shm.buf)
+
+file_path = "./config.json"
+while True:
+    with open(file_path, 'r') as file:
+        jsonData = json.load(file)
+    for n in range(32):
+        c[n]=jsonData[str(n+1)]
+    time.sleep(2)