diff --git a/AWX_collector.py b/AWX_collector.py index 8b80530..780a50d 100644 --- a/AWX_collector.py +++ b/AWX_collector.py @@ -41,6 +41,8 @@ args = parser.parse_args() ROOT_PATH = f"/usr/local/sdt/app/{args.config}" DEVICE_PATH = f"/etc/sdt/device-control" + + ############################################################################### # JSON FILE READ # ############################################################################### @@ -55,29 +57,62 @@ with open(f"{DEVICE_PATH}/config.yaml","r") as f: ############################################################################### -# Shared memory setting # +# Shared memory setting_DO # ############################################################################### - - with open(f"{ROOT_PATH}/config.json","r") as f: #with open(f"{args.config}/config.json","r") as f: - data = json.load(f) + do_dict = json.load(f) + for key in do_dict: + do_dict[key] = "1" -print(data) -print(data["1"]) +print(do_dict) +#print(do_dict["1"]) +#print(type(do_dict["1"])) -a = [ data[str(n+1)] for n in range(64)] +a = [ do_dict[str(n+1)] for n in range(64)] a = np.array(a) ############################################################################### -# Get Data from Shared Memory # +# Shared Memory Setting_CORR # ############################################################################### -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[:] +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) ############################################################################### @@ -105,24 +140,10 @@ client.loop(2) # timeout = 2초 - ############################################################################### -# Serial setting # +# CWT Thread-Read Serial # ############################################################################### -_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): @@ -145,39 +166,31 @@ def cwt_thread(evt, chip_index, port_a_ai_id , port_a_do_id, port_b_ai_id , port 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_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 # baudrate + 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 # baudrate + 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 # baudrate + port_B_DO.serial.baudrate = 115200 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] + 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] - 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 - #추��에 do power on 할 리스트는 어디선가 읽어서 와야 해요 - #do1 do2 나눠야 해요 - - - + + ''' | AI1 | AI2 | DO1 | DO2 | Success | 0 | 0 | x | x | @@ -192,112 +205,76 @@ def cwt_thread(evt, chip_index, port_a_ai_id , port_a_do_id, port_b_ai_id , port -> 데이터를 어떻게 할 것이냐 -> 이전 데이터가 있다면, 이전 데이터를 가지고 더미 데이터로 사용하는 방법 => 안하기로 했었음 ''' + 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) - a_ai_data=ai_array - # suc_cnt_ai = suc_cnt_ai +1 - except: - isSuccess = False + a_isSuccess = True + b_isSuccess = True - if isSuccess: - priv_a_ai = ai_array + # PORT A + port_a_do_value = [] + for i in list(b[: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 = list(b[17:32]) + port_b_do_value = [] + for i in list(b[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) - b_ai_data = ai_array - #prev_b_ai = ai_array + 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 + b_isSuccess = False - if isSuccess: - prev_b_ai = ai_array + #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: + + 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_a.put([0,prev_a_ai, port_a_do_value]) - queue_b.put([0,prev_a_ai, port_b_do_value]) + queue_b.put([0,ai_array, 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) - ''' - -############################################################################### -# dummy data thread # -############################################################################### - - ############################################################################### -# data process thread # +# Data process thread # ############################################################################### @@ -308,6 +285,13 @@ def data_process_thread(data_queue, ch, chamber): 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() @@ -315,18 +299,41 @@ def data_process_thread(data_queue, ch, chamber): #================================================== 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 @@ -334,14 +341,30 @@ def data_process_thread(data_queue, ch, chamber): 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 voltage: - if i < 2: + 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 @@ -350,43 +373,186 @@ def data_process_thread(data_queue, ch, chamber): 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: - data = { - "timestamp" : int(timestamp), + #if timestamp == 0: + data = { + "timestamp" : int(timestamp), "data": { - "current" : current, - "voltage" : voltage, - "do" : do, - "er1" : error_cnt_1, - "er2" : error_cnt_2 + "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) + } + 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":"NQ-R04-TEST-003", #나중에는 config에서 셋팅되는 값, or model 값으로 변경 (asset) "dataType":"DATA", "data": { - "channel": ch, # 채널 입력 �자 - "chamber":chamber, #chamber 도 입력 �자 + "channel": ch, # 채널 입력 �자 + "chamber":chamber, #chamber 도 입력 �자 "slot": ch, #슬롯 값은 chamber와 ch 값으로 만든다 "list": data_list_for_send_1 } @@ -399,8 +565,8 @@ def data_process_thread(data_queue, ch, chamber): "assetCode":"NQ-R04-TEST-003", #나중에는 config에서 셋팅되는 값, or model 값으로 변경 (asset) "dataType":"DATA", "data": { - "channel": ch, # 채널 입력 �자 - "chamber":chamber, #chamber 도 입력 �자 + "channel": ch, # 채널 입력 �자 + "chamber":chamber, #chamber 도 입력 �자 "slot": ch, #슬롯 값은 chamber와 ch 값으로 만든다 "list": data_list_for_send_2 } @@ -410,74 +576,40 @@ def data_process_thread(data_queue, ch, chamber): # 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) + print(data_json_str) publish_json_message(client, data_json_str) - ##erro cnt check routine + + - - - - -######################################### thread 함수 작성 ################################################ - +############################################################################### +# Thread Event Timer # +############################################################################### #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) + time.sleep(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 # + # Queue x 4 max- infinite # ############################################################################### ch1_queue = queue.Queue() @@ -488,7 +620,7 @@ if __name__ == '__main__': 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_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)) @@ -503,4 +635,4 @@ if __name__ == '__main__': data_process_ch2.start() data_process_ch3.start() data_process_ch4.start() - event_thread.start() + event_thread.start() \ No newline at end of file