#!/usr/bin/env python # -*-coding:UTF-8-*- import logging # import loggingCfg import time import os import sys from threading import Thread import queue import serial # to install this dependency execute: pip install pyserial import datetime # to install this dependency execute: pip install gevent import struct import serial.tools.list_ports import asyncio import gevent import re # Test Data znp_edScanReq = [0xfe, 5, 0x23, 0x1f, 0x00, 0xf8, 0xff, 0x07, 0x03] # display setting DISPLAY_FORMATE_CFG_HEX_BIT = 1 DISPLAY_FORMATE_CFG_STR_BIT = 2 DISPLAY_FORMATE_CFG = DISPLAY_FORMATE_CFG_HEX_BIT | DISPLAY_FORMATE_CFG_STR_BIT # Set output format # logging.basicConfig(format='%(asctime)s,%(msecs)d %(levelname)-8s [%(filename)s:%(lineno)d] %(message)s', # datefmt='%d-%m-%Y:%H:%M:%S') logging.basicConfig(format='%(message)s', datefmt='%d-%m-%Y:%H:%M:%S') # Set the name of the log output device, you can configure multiple different log output devices in the project # Set log printing level logging.getLogger(__name__).setLevel(logging.DEBUG) logger = logging.getLogger(__name__) def uart_refresh(): """ Serial port detection """ # Detect all existing serial ports and store the information in the list i = 0 port_list = list(serial.tools.list_ports.comports()) for port in port_list: if port: i += 1 uart_list.append(port) print(uart_list[i]) if len(uart_list) is None: return "uart_list = NULL" # Save log data to file handler = logging.FileHandler('%s.txt' % time.strftime("%Y-%m-%d %H_%M")) handler.setLevel(logging.DEBUG) logger.addHandler(handler) def sendDataToSerialPort(sp, data): # data.append(fcsCheck(data,1)) binaryData = struct.pack('B' * len(data), *data) wrt = sp.write(binaryData) # data.pop() # logger.debug(">>>>>>>>>>>>>>>>>>>>>>>>>>>>{}", decTohexStr(data)) def sendStrToSerialPort(sp, data): wrt = sp.write(data.encode("utf-8")) def decTohexStr(argv): result = '' hLen = len(argv) for i in range(hLen): if (argv[i] <= 15): result += ' 0%x' % argv[i] else: result += ' %x' % argv[i] return result def start_evt_loop(loop): asyncio.set_event_loop(loop) loop.run_forever() async def readDataFromSerial(sp): print('Read') while True: await asyncio.sleep(0.2) try: # logger.debug("+++++++++++++++") if sp.isOpen() and sp.inWaiting(): tmpLen = sp.inWaiting() recv = sp.read(tmpLen) q.put(recv, block=True, timeout=5) # recv = sp.read(1) # head = struct.unpack('B',recv)[0]; logger.debug(" 0x%x " % (head)) # logger.debug("---------------") except: logger.error('serial port is recv excp!!!!') # await asyncio.sleep(1) logger.debug('R Done after s') async def writeDataFromSerial(sp): print('Write') cnt = 0 while True: cnt = cnt + 1 print("write %d" % (cnt)) if sp.isOpen(): if not q.empty(): sendData = q.get() sendDataToSerialPort(sp, list(sendData)) # sendDataToSerialPort(sp, list(znp_edScanReq) ) await asyncio.sleep(0.5) print('W Done after s') async def create_task(event_loop): i = 0 while i < 3: # A task is generated every second and submitted to the loop in the thread with event_loop as a parameter asyncio.run_coroutine_threadsafe(production(i), event_loop) await asyncio.sleep(1) i += 1 async def production(i): while True: logger.debug("%d coroutine" % (i)) await asyncio.sleep(2) def splitUartRxStarWithWrap(uartRxStr): mRxDataStr = uartRxStr.replace(r'\r', '').replace(r'\n', 'nnnnnn') dataStrList = re.split(r'n{6,30}', mRxDataStr) # print(dataStrList) if dataStrList[-1] == '': del dataStrList[-1] return dataStrList def uartRx(sp, q): print('Read processing...') while True: try: # logger.debug("+++++++++++++++") tmpLen = sp.inWaiting() # logger.debug(tmpLen) if sp.isOpen() and tmpLen != 0: # Data bits read from serial buffer bytes type recv = sp.read(tmpLen) print("recv sp read tmplen ", recv) # cmdData = struct.unpack('B'*(tmpLen),recv) # cmdData = list(cmdData) print(type(recv)) # logger.debug("%s [RX] : %s"%(datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S.%f'), decTohexStr(cmdData))) # logger.debug("%s [RX] : %s"%(datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S.%f'), str(recv))) # tmpList = splitUartRxStarWithWrap(str(recv)) # for info in tmpList: # logger.debug("%s [RX] : %s"%(datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S.%f'), info)) # print(time.time()," --- data_recv --> ", cmdData) # # logger.debug("<<<<" + recv) q.put(recv, block=True, timeout=1) # # recv = sp.read(1) # # head = struct.unpack('B',recv)[0]; # # logger.debug(" 0x%x "%(head)) # logger.debug("---------------") except: logger.error('serial port is recv excp!!!!') # time.sleep(0.1) gevent.sleep(0.005) # logger.debug("+++++++++++++++") # await asyncio.sleep(1) # logger.debug('R Done after s') def uartTx(sp, q): print('Write processing...') cnt = 0 while True: gevent.sleep(0.8) # cnt = cnt + 1 # print("write %d"%(cnt)) if sp.isOpen(): if not q.empty(): sendData = q.get() sendDataToSerialPort(sp, list(sendData)) print('W Done after s') def bytesToHexString(bytesArr): return ''.join(['%02X ' % b for b in bytesArr]) def printUartRxData(q): print('Log Print processing...') while True: gevent.sleep(0.01) if not q.empty(): sendData = q.get() print(type(sendData)) if DISPLAY_FORMATE_CFG == DISPLAY_FORMATE_CFG_HEX_BIT: # logger.debug(sendData.hex(" ", -2)) logger.debug("%s [RX] : %s" % ( datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S.%f'), bytesToHexString(sendData))) elif DISPLAY_FORMATE_CFG == DISPLAY_FORMATE_CFG_STR_BIT: tmpList = splitUartRxStarWithWrap( (sendData.decode(encoding="utf-8", errors="ignore"))) for info in tmpList: logger.debug("%s [RX] : %s" % ( datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S.%f'), info)) else: logger.debug("%s [RX] : %s %s" % ( datetime.datetime.now().strftime( '%Y-%m-%d %H:%M:%S.%f'), bytesToHexString(sendData), sendData.decode(encoding="utf-8", errors="ignore"))) break """ Test mcu into boot """ def setDTRState(serialPort, state): serialPort.setDTR(state) def setRTSState(serialPort, state): serialPort.setRTS(state) # Work-around for adapters on Windows using the usbser.sys driver: # generate a dummy change to DTR so that the set-control-line-state # request is sent with the updated RTS state and the same DTR state # In order to bypass the driver problem under Windows system, update the RTS and DTR status at the same time serialPort.setDTR(serialPort.dtr) def enterBoot(serialPort, delay=False): # delay is a workaround for bugs with the most common auto reset # circuit and Windows, if the EN pin on the dev board does not have # enough capacitance. # If the EN pin on the development board does not have enough capacitance last_error = None # issue reset-to-bootloader: # RTS = either CH_PD/EN or nRESET (both active low = chip in reset # DTR = GPIO0 (active low = boot to flasher) # # DTR & RTS are active low signals, # ie True = pin @ 0V, False = pin @ VCC. setDTRState(serialPort, False) # IO0=HIGH # setDTRState(serialPort, True) # IO0=LOW setRTSState(serialPort, True) # EN=LOW, chip in reset time.sleep(0.1) if delay: # Some chips are more likely to trigger the esp32r0 # watchdog reset silicon bug if they're held with EN=LOW # for a longer period time.sleep(1.2) setDTRState(serialPort, True) # IO0=LOW setRTSState(serialPort, False) # EN=HIGH, chip out of reset if delay: # Sleep longer after reset. # This workaround only works on revision 0 ESP32 chips, # it exploits a silicon bug spurious watchdog reset. time.sleep(0.4) # allow watchdog reset to occur time.sleep(1) setDTRState(serialPort, False) # IO0=HIGH, done setRTSState(serialPort, False) time.sleep(1) return last_error if __name__ == "__main__": logger.debug("This is a debug log") logger.info("This is an info log") logger.critical("This is critical") logger.error("An error occurred\n") # pyinstaller -i butter.ico -F d:/desktop/seven.py uart_list = [0] uart_refresh() # Get the existing serial port information try: name = input( "Select the serial port (just enter the number of the serial port):") SERIAL_PORT_CFG = {'name': "COM" + name, 'baudrate': 115200, 'timeout': 5, 'parity': serial.PARITY_NONE, 'stopbit': serial.STOPBITS_ONE, 'flowcontrol': False} except Exception as error: print("Serial port configuration error：", error) # init serial port try: SP = serial.Serial(SERIAL_PORT_CFG.get('name'), SERIAL_PORT_CFG.get('baudrate'), timeout=SERIAL_PORT_CFG.get('timeout'), parity=SERIAL_PORT_CFG.get('parity'), stopbits=SERIAL_PORT_CFG.get('stopbit'), rtscts=SERIAL_PORT_CFG.get('flowcontrol')) except Exception as error: logger.error('open ' + SERIAL_PORT_CFG.get('name') + ' is fail!!!!') sys.exit(1) logger.info(">>>>>>>>>>>>>>>> %s is opened....." % SERIAL_PORT_CFG.get('name')) # boot into boot enterBoot(SP, delay=False) uartRxDataQueue = queue.Queue(1000) print(uartRxDataQueue) uartTxDataQueue = queue.Queue(1000) print(uartTxDataQueue) # Send the automatic baud rate calibration byte to 2652 boot # cc26xxBuadradeCommunicatonReq = [0x55, 0x55] # uartTxDataQueue.put(cc26xxBuadradeCommunicatonReq, block=True, timeout=5) startTime = int(round(time.time() * 1000)) uartTasks = [gevent.spawn(uartTx, SP, uartTxDataQueue)] uartTasks = [gevent.spawn(uartTx, SP, uartTxDataQueue), gevent.spawn(uartRx, SP, uartRxDataQueue), gevent.spawn(printUartRxData, uartRxDataQueue)] gevent.joinall(uartTasks, 5) # 3 Close all threads after seconds logger.info(">>>>>>>>>>>>>>>> end") # try: # # ser = serial.Serial() # # gevent.joinall(uartTasks, 3) # Close all threads after 3 seconds # logger.info(">>>>>>>>>>>>>>>> end") # if ser.isOpen(): # ser.close() # print("COM:", name, "Closed successfully") # if not ser.isOpen(): # print("no serial port open？") # except Exception as error: # print("COM:", name, "Failed to close") # # uart_refresh() # Get the existing serial port information # print("\n") # while True: # data_count = SP.inWaiting() # if data_count !=0 : # recv = SP.read(SP.in_waiting) # print(time.time()," --- data_recv --> ", recv) # time.sleep(0.1) # while True: # try: # # logger.debug("+++++++++++++++") # tmpLen = SP.inWaiting() # logger.debug(tmpLen) # if tmpLen !=0: # recv = SP.read(tmpLen) # # logger.debug("<<<<" + recv) # print(time.time()," --- data_recv --> ", recv) # except: # logger.error('serial port is recv excp!!!!') # time.sleep(0.1) # uartRx2(SP,uartRxDataQueue) # thread_loop = asyncio.new_event_loop() # run_loop_thread = Thread(target=start_evt_loop, args=(thread_loop,)) # run_loop_thread.start() # logger.debug("TIME: %d"%(int(round(time.time() * 1000)) - startTime)) # # main_loop = asyncio.new_event_loop() # # main_loop.run_until_complete(create_task(thread_loop)) # 主线程负责create coroutine object # asyncio.run_coroutine_threadsafe(writeDataFromSerial(SP), thread_loop) # asyncio.run_coroutine_threadsafe(readDataFromSerial(SP), thread_loop)