mirror of
https://gitee.com/xiaohuolufeihua/bizhang_-obav.git
synced 2026-05-21 01:12:11 +00:00
Tools/HIL: relax command retry for now
This commit is contained in:
@@ -19,7 +19,7 @@ def monitor_firmware_upload(port, baudrate):
|
||||
timeout_newline = time.time()
|
||||
|
||||
while finished == 0:
|
||||
serial_line = ser.readline().decode("utf-8", errors='ignore')
|
||||
serial_line = ser.readline().decode("ascii", errors='ignore')
|
||||
if (len(serial_line) > 0):
|
||||
print(serial_line.replace('\n', ''))
|
||||
|
||||
@@ -40,7 +40,7 @@ def monitor_firmware_upload(port, baudrate):
|
||||
# newline every 10 seconds if still running
|
||||
if time.time() - timeout_newline > 10:
|
||||
timeout_newline = time.time()
|
||||
ser.write('\n'.encode("utf-8"))
|
||||
ser.write('\n'.encode("ascii"))
|
||||
ser.flush()
|
||||
|
||||
ser.close()
|
||||
|
||||
@@ -17,47 +17,60 @@ def do_nsh_cmd(port, baudrate, cmd):
|
||||
timeout_start = time.time()
|
||||
timeout = 10 # 10 seconds
|
||||
|
||||
# clear
|
||||
ser.write("\n".encode("ascii"))
|
||||
ser.flush()
|
||||
ser.readline()
|
||||
|
||||
success_cmd = "cmd succeeded!"
|
||||
|
||||
while True:
|
||||
serial_cmd = '{0}; echo "{1}"\n'.format(cmd, success_cmd)
|
||||
ser.write(serial_cmd.encode("utf-8"))
|
||||
ser.flush()
|
||||
serial_cmd = '{0}; echo "{1}"\n'.format(cmd, success_cmd)
|
||||
ser.write(serial_cmd.encode("ascii"))
|
||||
ser.flush()
|
||||
ser.readline()
|
||||
|
||||
serial_line = ser.readline().decode("utf-8", errors='ignore')
|
||||
# TODO: require successful command echo
|
||||
# while True:
|
||||
# serial_cmd = '{0}; echo "{1}"\n'.format(cmd, success_cmd)
|
||||
# ser.write(serial_cmd.encode("ascii"))
|
||||
# ser.flush()
|
||||
|
||||
if cmd in serial_line:
|
||||
break
|
||||
else:
|
||||
print(serial_line.replace('\n', ''))
|
||||
# serial_line = ser.readline().decode("ascii", errors='ignore')
|
||||
|
||||
if time.time() > timeout_start + timeout:
|
||||
print("Error, timeout")
|
||||
sys.exit(-1)
|
||||
break
|
||||
# if cmd in serial_line:
|
||||
# break
|
||||
# else:
|
||||
# print(serial_line, end='')
|
||||
|
||||
time.sleep(1)
|
||||
# if time.time() > timeout_start + timeout:
|
||||
# print("Error, timeout")
|
||||
# sys.exit(-1)
|
||||
# break
|
||||
|
||||
# time.sleep(1)
|
||||
|
||||
|
||||
timeout_start = time.time()
|
||||
timeout = 30 # 30 seconds
|
||||
|
||||
while True:
|
||||
serial_line = ser.readline().decode("utf-8", errors='ignore')
|
||||
serial_line = ser.readline().decode("ascii", errors='ignore')
|
||||
|
||||
if success_cmd in serial_line:
|
||||
break
|
||||
else:
|
||||
if len(serial_line) > 0:
|
||||
print(serial_line.replace('\n', ''))
|
||||
print(serial_line, end='')
|
||||
|
||||
if "nsh>" in serial_line:
|
||||
sys.exit(-1) # error, command didn't complete successfully
|
||||
#sys.exit(-1) # error, command didn't complete successfully
|
||||
break # TODO: return error on failure
|
||||
elif "NuttShell (NSH)" in serial_line:
|
||||
sys.exit(-1) # error, command didn't complete successfully
|
||||
#sys.exit(-1) # error, command didn't complete successfully
|
||||
break # TODO: return error on failure
|
||||
|
||||
if len(serial_line) <= 0:
|
||||
ser.write("\n".encode("utf-8"))
|
||||
ser.write("\n".encode("ascii"))
|
||||
ser.flush()
|
||||
|
||||
if time.time() > timeout_start + timeout:
|
||||
|
||||
@@ -24,23 +24,35 @@ def do_test(port, baudrate, test_name):
|
||||
print('|======================================================================')
|
||||
timeout_start = time.time()
|
||||
timeout = 10 # 10 seconds
|
||||
while True:
|
||||
serial_cmd = '{0}\n'.format(cmd)
|
||||
ser.write(serial_cmd.encode("utf-8"))
|
||||
ser.flush()
|
||||
|
||||
serial_line = ser.readline().decode("utf-8", errors='ignore')
|
||||
# clear
|
||||
ser.write("\n".encode("ascii"))
|
||||
ser.flush()
|
||||
ser.readline()
|
||||
|
||||
if cmd in serial_line:
|
||||
break
|
||||
else:
|
||||
print(serial_line.replace('\n', ''))
|
||||
serial_cmd = '{0}\n'.format(cmd)
|
||||
ser.write(serial_cmd.encode("ascii"))
|
||||
ser.flush()
|
||||
ser.readline()
|
||||
|
||||
if time.time() > timeout_start + timeout:
|
||||
print("Error, unable to write cmd")
|
||||
return False
|
||||
# TODO: retry command
|
||||
# while True:
|
||||
# serial_cmd = '{0}\n'.format(cmd)
|
||||
# ser.write(serial_cmd.encode("ascii"))
|
||||
# ser.flush()
|
||||
|
||||
time.sleep(1)
|
||||
# serial_line = ser.readline().decode("ascii", errors='ignore')
|
||||
|
||||
# if cmd in serial_line:
|
||||
# break
|
||||
# else:
|
||||
# print(serial_line.replace('\n', ''))
|
||||
|
||||
# if time.time() > timeout_start + timeout:
|
||||
# print("Error, unable to write cmd")
|
||||
# return False
|
||||
|
||||
# time.sleep(1)
|
||||
|
||||
|
||||
# print results, wait for final result (PASSED or FAILED)
|
||||
@@ -49,9 +61,9 @@ def do_test(port, baudrate, test_name):
|
||||
timeout_newline = timeout_start
|
||||
|
||||
while True:
|
||||
serial_line = ser.readline().decode("utf-8", errors='ignore')
|
||||
serial_line = ser.readline().decode("ascii", errors='ignore')
|
||||
if (len(serial_line) > 0):
|
||||
print(serial_line.replace('\n', ''))
|
||||
print(serial_line, end='')
|
||||
|
||||
if test_name + " PASSED" in serial_line:
|
||||
success = True
|
||||
@@ -68,7 +80,7 @@ def do_test(port, baudrate, test_name):
|
||||
|
||||
# newline every 10 seconds if still running
|
||||
if time.time() - timeout_newline > 10:
|
||||
ser.write("\n".encode("utf-8"))
|
||||
ser.write("\n".encode("ascii"))
|
||||
timeout_newline = time.time()
|
||||
|
||||
ser.close()
|
||||
|
||||
Reference in New Issue
Block a user