subprocess 命令行使用

import serial, threading
import time, sys, subprocess, os
import queueclass auto_run():# 命令的结束标志end_sign = 'END'def __init__(self):self.flag = Trueself.popen = Noneself.Retry_num = Noneself.Result_Pass = Trueself.t = Noneself.if_repeat = Trueself.recv_pop = Falseself.check_print_data = Falseself.check_deta = Falseself.result = queue.Queue()def send_cmd(self):with open("SAV526_Flash - RCY2.txt", "r", encoding="utf8") as f:send_List = f.readlines()print(send_List)max_ = len(send_List) - 1for index, cmd in enumerate(send_List):if index == 0: continuecmd = cmd.strip()cmd = cmd.replace(r"\u", "")cmd = cmd.replace(r"\n", "")print(">>%s:" % cmd)self.popen = subprocess.Popen(cmd, shell=True, stdout=subprocess.PIPE, stderr=subprocess.STDOUT)if self.check_print_data == False:self.result.put(True)threading.Thread(target=self.read_response).start()self.check_print_data = Truetime.sleep(1)self.exec_command("")def read_response(self):while self.Result_Pass:  # time outline = self.popen.stdout.readlines()if line == []:time.sleep(0.2)self.check_deta = Falsethreading.Thread(target=self.recv_timeout).start()else:for linestr in line:self.check_deta = Truetry:print(linestr.decode("utf8").strip())time.sleep(0.2)except UnicodeDecodeError as e:print(linestr.decode("gb2312").strip())time.sleep(0.2)try:self.Result_Pass = self.result.get()except:self.Result_Pass = Truedef close(self):try:self.popen.kill()self.popen.stdin.close()self.popen.stdout.close()except:passdef exec_command(self, comm):try:comm += os.linesepself.popen.stdin.write(comm.encode('utf8'))self.popen.stdin.write(('echo ' + self.end_sign + os.linesep).encode('utf8'))self.popen.stdin.flush()except:passdef recv_timeout(self):try:print("输出无数据开始计时>>>>>>>>>> ")t = 0while t <= 10:t += 1time.sleep(1)if self.check_deta == False:continueif self.check_deta == True:breakif t > 10:print("超时无信息退出>>>>>>>>>>>>>>> ")self.result.put(False)else:self.result.put(True)except:self.close()return# strip()
if __name__ == '__main__':shell = auto_run()result = shell.send_cmd()