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