AWX_collector/AWX_collector.py

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

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-shm2")

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_Test2")

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 = 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초



###############################################################################
#                          CWT Thread-Read Serial                             #
###############################################################################

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  

    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  

    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  

    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  


    while isThreadRun:
        evt.wait()
        evt.clear()

        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]
        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]

        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]

        prev_a_ai = ai_array
        prev_b_ai = ai_array
                
        
        '''
                | 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)
        a_isSuccess = True
        b_isSuccess = True

        # PORT A
        port_a_do_value = []
        for i in list(do_value[:16]):
            port_a_do_value.append(int(i))
        port_a_do_value = list(do_value[:16])
        #print("port a do: ", port_a_do_value)
        #print("do type: {}".format(type(port_a_do_value[0])))
        
        try:
            a_ai_data=port_A_AI.read_registers(0x32, 32)
            prev_a_ai = a_ai_data
            # suc_cnt_ai = suc_cnt_ai +1
                       
        except:
            a_isSuccess = False

        #if isSuccess:
        #    prev_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 = []
        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
            # suc_cnt_ai = suc_cnt_ai +1
    
        except:
            b_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 a_isSuccess:
            queue_a.put([timenow,ai_array, port_a_do_value])
        else:
            queue_a.put([0,ai_array, port_a_do_value])

        if b_isSuccess:
            queue_b.put([timenow,ai_array, port_b_do_value])
        else:
            queue_b.put([0,ai_array, port_b_do_value])





###############################################################################
#                         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

    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
            if (ai_index%2) == 0:
                current.append(int(i))
            else:
                voltage.append(int(i))
                #print(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                                 #
###############################################################################
        '''
        #전압값으로 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)
        '''
      

                
###############################################################################
#                                 YJ TEST CODE                                #
###############################################################################
        '''
        ERR1 = 0
        #count = 0
        #cycle_counts = 0

        for i, value in enumerate(calculated_voltage_list):
            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
                }
           }
           if(queue_select_count == 0):
               data_list_for_send_1.append(data)
           else:
               data_list_for_send_2.append(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":"S0NQR0423090001",  #나중에는 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":"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)


        


###############################################################################
#                            Thread Event Timer                               #
###############################################################################

#event는 개별 생성을 해준다 th1, th2

def timer_event_loop(e1, e2):
    cnt = 0
    # global read_cnt

    while True:

         e1.set()
         e2.set()
         time.sleep(0.1)

    exit()





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