############################################################################### import argparse import threading import time import serial import copy import binascii ############################################################################### parser = argparse.ArgumentParser(description='USB2SerialMux control helper tool.') parser.add_argument("-b", "--getBaudrate", default=False, help="Get the baudrate for the USART connection.", action='store_true') parser.add_argument("-s", "--setBaudrate", type=int, help="Something like 9600 or 115200.") parser.add_argument("-m", "--getMuxLine", default=False, help="Get the current multiplexer control line state.", action='store_true') parser.add_argument("-l", "--setMuxLine", type=int, help="Something like 0 to 7.") parser.add_argument("-r", "--resetToBtldr", default=False, help="Reset the device to the LUFA bootloader.", action='store_true') parser.add_argument("-d", "--device", type=str, help="The control device like /dev/ttyACM0 or COM3.") ############################################################################### MSG_SOD1 = 0x3c MSG_SOD2 = 0x3e MSG_EOD1 = 0x0d MSG_EOD2 = 0x0a MSG_TYPE_ANSWER_OK = 0x01 MSG_TYPE_ANSWER_NOK = 0x02 MSG_TYPE_BAUDRATE = 0x03 MSG_TYPE_MUXLINE = 0x04 MSG_TYPE_ANSWER_OK_DATA_TO_RECV = 0 MSG_TYPE_ANSWER_NOK_DATA_TO_RECV = 0 MSG_TYPE_BAUDRATE_DATA_TO_RECV = 4 MSG_TYPE_MUXLINE_DATA_TO_RECV = 1 CC_CMD_SET_BAUDRATE = 0x0A CC_CMD_GET_BAUDRATE = 0x14 CC_CMD_SET_MUX_LINE = 0x1E CC_CMD_GET_MUX_LINE = 0x28 CC_CMD_START_BTLDR = 0x32 ############################################################################### TIMEOUT_CNT_MAX = 50 MAIN_LOOP_DELAY_S = 0.05 THREAD_LOOP_DELAY_S = 0.01 ############################################################################### MUX_MIN_VAL = 0 MUX_MAX_VAL = 7 ############################################################################### ser = None device = "/dev/ttyACM0" ############################################################################### baudrate = 0xffffffff muxLine = 0xff ############################################################################### CC_STATE_WAIT_SOD1 = 0x01 CC_STATE_WAIT_SOD2 = 0x02 CC_STATE_READ_DATA = 0x03 CC_STATE_WAIT_EOD1 = 0x04 CC_STATE_WAIT_EOD2 = 0x05 CC_STATE_GET_TYPE = 0x06 cc_state = CC_STATE_WAIT_SOD1 cc_state_list = [ CC_STATE_WAIT_SOD1, CC_STATE_WAIT_SOD2, CC_STATE_READ_DATA, CC_STATE_WAIT_EOD1, CC_STATE_WAIT_EOD2 ] cc_state_fn = {} msg_type_list = [ MSG_TYPE_ANSWER_OK, MSG_TYPE_ANSWER_NOK, MSG_TYPE_BAUDRATE, MSG_TYPE_MUXLINE, ] msg_type_data_to_read = { MSG_TYPE_ANSWER_OK : MSG_TYPE_ANSWER_OK_DATA_TO_RECV, MSG_TYPE_ANSWER_NOK : MSG_TYPE_ANSWER_NOK_DATA_TO_RECV, MSG_TYPE_BAUDRATE : MSG_TYPE_BAUDRATE_DATA_TO_RECV, MSG_TYPE_MUXLINE : MSG_TYPE_MUXLINE_DATA_TO_RECV, } msg_type = 0 cc_data_read = 0 cc_data_buffer = [] # yes a separate counter to manage the order of the received messages cc_message_cnt = 0 cc_received_messages = [] ############################################################################### thread_obj = None thread_lock = None thread_started = False thread_stop = False ############################################################################### def cc_init(): global cc_state_fn global cc_state cc_state = CC_STATE_WAIT_SOD1 cc_state_fn = { CC_STATE_WAIT_SOD1 : cc_state_fn_wait_for_sod1, CC_STATE_WAIT_SOD2 : cc_state_fn_wait_for_sod2, CC_STATE_WAIT_EOD1 : cc_state_fn_wait_for_eod1, CC_STATE_WAIT_EOD2 : cc_state_fn_wait_for_eod2, CC_STATE_GET_TYPE : cc_state_fn_get_type, CC_STATE_READ_DATA : cc_state_fn_read_data, } ########## function to call by the thread def cc_dataReceiverThread(): global ser global cc_state global cc_state_fn global thread_started global thread_stop thread_started = True while thread_stop == False: # 1. read byte from serial port into incoming incoming = [] bytesToRead = ser.inWaiting() if bytesToRead > 0: incoming = list(ser.read(64)) # 2. process the received data for c in incoming: c = int(binascii.hexlify(c), 16) # call the cc_state specific function to process the currently received byte cc_state_fn[cc_state](c) if cc_state not in cc_state_list: cc_state = CC_STATE_WAIT_SOD1 time.sleep(THREAD_LOOP_DELAY_S) thread_started = False ########## def cc_startReceiverThread(): global thread_obj global thread_lock global thread_stop if thread_started == False: thread_lock = threading.Lock() thread_obj = threading.Thread(target=cc_dataReceiverThread) thread_obj.start() thread_stop = False ########## def cc_stopReceiverThread(): global thread_obj global thread_started global thread_stop if thread_started == True: thread_stop = True thread_obj.join() # wait for the thread to finish ##### CC_STATE_WAIT_SOD1 def cc_state_fn_wait_for_sod1(c): global cc_data_read global msg_type global cc_data_buffer global cc_state cc_data_read = 0 msg_type = 0 cc_data_buffer = [] if c == MSG_SOD1: cc_state = CC_STATE_WAIT_SOD2 else: cc_state = CC_STATE_WAIT_SOD1 ##### CC_STATE_WAIT_SOD2 def cc_state_fn_wait_for_sod2(c): global cc_state if c == MSG_SOD2: cc_state = CC_STATE_GET_TYPE else: cc_state = CC_STATE_WAIT_SOD1 ##### CC_STATE_GET_TYPE def cc_state_fn_get_type(c): global msg_type global cc_state if c in msg_type_list: msg_type = c if msg_type_data_to_read[msg_type] > 0: cc_state = CC_STATE_READ_DATA else: cc_state = CC_STATE_WAIT_EOD1 else: cc_state = CC_STATE_WAIT_SOD1 ##### CC_STATE_READ_DATA def cc_state_fn_read_data(c): global cc_data_buffer global cc_data_read global cc_state if cc_data_read <= msg_type_data_to_read[msg_type] - 1: cc_data_buffer.append(c) cc_data_read = cc_data_read + 1 if cc_data_read == msg_type_data_to_read[msg_type]: cc_state = CC_STATE_WAIT_EOD1 ##### CC_STATE_WAIT_EOD1 def cc_state_fn_wait_for_eod1(c): global cc_state if c == MSG_EOD1: cc_state = CC_STATE_WAIT_EOD2 else: cc_state = CC_STATE_WAIT_SOD1 ##### CC_STATE_WAIT_EOD2 def cc_state_fn_wait_for_eod2(c): global thread_lock global cc_message_cnt global cc_state if c == MSG_EOD2: is_message_read = False thread_lock.acquire() cc_received_messages.append([ cc_message_cnt, msg_type, is_message_read, copy.deepcopy(cc_data_buffer) ]) thread_lock.release() cc_message_cnt = cc_message_cnt + 1 cc_state = CC_STATE_WAIT_SOD1 ############################################################################### def send_getBaudrate(): print "send: get baudrate (0x%02x)" % (CC_CMD_GET_BAUDRATE) sendSerialData([CC_CMD_GET_BAUDRATE]) def send_setBaudrate(b): print "send: set baudrate (0x%02x)" % (CC_CMD_SET_BAUDRATE) sendSerialData([ CC_CMD_SET_BAUDRATE, (b & 0xff000000) >> 24, (b & 0xff0000) >> 16, (b & 0xff00) >> 8, (b & 0xff) ]) ############################################################################### def send_getMuxLine(): print "send: get mux line (0x%02x)" % (CC_CMD_GET_MUX_LINE) sendSerialData([CC_CMD_GET_MUX_LINE]) def send_setMuxLine(l): print "send: set mux line (0x%02x)" % (CC_CMD_SET_MUX_LINE) sendSerialData([CC_CMD_SET_MUX_LINE, l]) ############################################################################### def send_resetToBtldr(): print "send: reset to bootloader message (0x%02x)" % (CC_CMD_START_BTLDR) sendSerialData([CC_CMD_START_BTLDR]) ############################################################################### ##### def openSerialDevice(d): global ser # Why 115200? Because the host defines the baudrate for the USB serial connection. try: if "com" in d.lower(): d = "\\\\.\\" + d ser = serial.Serial(d, 115200, timeout=0) except: print "ERROR: Can't open the serial device " + d exit(1) ##### def closeSerialDevice(): global ser ser.close() ##### def sendSerialData(data): global ser ser.write(bytearray([ MSG_SOD1, MSG_SOD2 ])) ser.write(bytearray(data)) ser.write(bytearray([ MSG_EOD1, MSG_EOD2 ])) ############################################################################### if __name__ == "__main__": cc_init() # parse the commandline arguments args = parser.parse_args() dataSend = 0 timeout = 0 # 1. open serial device or abort if args.device != None: device = args.device openSerialDevice(device) # 2. start thread to poll cc_dataReceiverThread() cc_startReceiverThread() # 3. get and process the commandline arguments/parameter if args.resetToBtldr == True: send_resetToBtldr() else: if args.setBaudrate != None: baudrate = args.setBaudrate send_setBaudrate(baudrate) dataSend = dataSend + 1 if args.setMuxLine != None: muxLine = args.setMuxLine # keep the mux line in range if muxLine < MUX_MIN_VAL: muxLine = MUX_MIN_VAL if muxLine > MUX_MAX_VAL: muxLine = MUX_MAX_VAL send_setMuxLine(muxLine) dataSend = dataSend + 1 if args.getBaudrate == True: send_getBaudrate() dataSend = dataSend + 1 if args.getMuxLine == True: send_getMuxLine() dataSend = dataSend + 1 # 4. start main loop while dataSend > 0 and timeout < TIMEOUT_CNT_MAX: thread_lock.acquire() tmp_messages = copy.deepcopy(cc_received_messages) thread_lock.release() # 4.1 test for the response(s) for e in tmp_messages: if e[2] == False: # test for unread message # process it and set the data to read if e[1] == MSG_TYPE_ANSWER_OK: print "recv: OK" elif e[1] == MSG_TYPE_ANSWER_NOK: print "recv: NOT OK" elif e[1] == MSG_TYPE_BAUDRATE: baudrate = e[3][0] << 24 baudrate += e[3][1] << 16 baudrate += e[3][2] << 8 baudrate += e[3][3] print "recv: baudrate = %d" % (baudrate) elif e[1] == MSG_TYPE_MUXLINE: muxLine = e[3][0] print "recv: muxLine = %d" % (muxLine) else: print "err: unknown type 0x%02x" % (e[1]) break thread_lock.acquire() cc_received_messages[e[0]][2] = True thread_lock.release() timeout = 0 # reset the timeout # reduce the number of messages to receive dataSend = dataSend - 1 # manage the timeout behaviour time.sleep(MAIN_LOOP_DELAY_S) timeout = timeout + 1 if timeout >= TIMEOUT_CNT_MAX: print "Timeout happened" # 5. stop data processing thread cc_stopReceiverThread() # 6. close serial device closeSerialDevice() exit(0)