【已解决】Python用os.system执行命令行命令希望加上timeout超时设置

折腾：

【未解决】adb无线WiFi连接操作安卓手机时网络切换导致异常

期间，去测试命令：

adbConnectCmd="adb connect 192.168.31.84:5555"
os.system(adbConnectCmd)

发现耗时很久。。。

后来，感觉是1，2分钟后才最后返回。

adb在终端中输出：

failed to connect to '192.168.31.84:5555': Operation timed out

代码返回了：

然后是没有找到设备的。

算了，去看看：

os.system值是否有超时参数设置，比如设置为1，2秒即可。

python os.system timeout

python 3.x – Setting timeout when using os.system function – Stack Overflow

https://stackoverflow.com/questions/41094707/setting-timeout-when-using-os-system-function

说是让换subprocess.Popen

python timeout using os.system – Stack Overflow

linux – Python os.system timeout with strings – Stack Overflow

理论上可以用：

os.system('''timeout 1s bash -c "ffmpeg -i "+path+"+" | cat''')

但是，建议换用：

subprocess.call

subprocess — Subprocess management — Python 3.9.1 documentation

Set timeout for a shell command in python | Random Howtos

def timeout_command(command, timeout):
  """call shell-command and either return its output or kill it
  if it doesn't normally exit within timeout seconds and return None"""
  import subprocess, datetime, os, time, signal
  start = datetime.datetime.now()
  process = subprocess.Popen(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
  while process.poll() is None:
    time.sleep(0.1)
    now = datetime.datetime.now()
    if (now - start).seconds&gt; timeout:
      os.kill(process.pid, signal.SIGKILL)
      os.waitpid(-1, os.WNOHANG)
      return None
  return process.stdout.read()

测试：

>>> output = timeout_command(["sleep", "10"], 2)
None
>>> output = timeout_command(["sleep", "1"], 2)

不过记得之前自己又实现过自己的subprocess.Popen 或 subprocess.call

去看看

other/AndroidAutomation/libs/utils.py

def launchTerminalRunShellCommand(shellFile, isForceNewInstance=True, isUseiTerm2=False):
    curProcess = subprocess.Popen(cmdList, stdin=subprocess.PIPE, stdout=subprocess.PIPE)

src/common/DevicesMethods.py

    # def get_cmd_lines(self,cmd,text=False):
    #         p = subprocess.Popen(cmd, stdout=subprocess.PIPE)

src/common/DownloadApps.py

    def get_app_info(self, AppPath):

        cmd = 'aapt dump badging {}'.format(AppPath)
        lines = CommonUtils.get_cmd_lines(cmd)

不过已有：

utils/common_utils.py

    @staticmethod
    def get_cmd_lines(cmd, text=False):
        # 执行cmd命令，将结果保存为列表
        resultStr = ""
        resultStrList = []
        try:
            consoleOutputByte = subprocess.check_output(cmd, shell=True) # b'xxx\n'
            try:
                resultStr = consoleOutputByte.decode("utf-8")
            except UnicodeDecodeError:
                # TODO: use chardet auto detect encoding
                # consoleOutputStr = consoleOutputByte.decode("gbk")
                resultStr = consoleOutputByte.decode("gb18030")


            if not text:
                resultStrList = resultStr.splitlines()
        except Exception as err:
            print("err=%s when run cmd=%s" % (err, cmd))


        if text:
            return resultStr
        else:
            return resultStrList

或许其中也可以给check_output加上timeout？

去看看

subprocess — Subprocess management — Python 3.9.1 documentation

subprocess.check_output(args, *, stdin=None, stderr=None, shell=False, cwd=None, encoding=None, errors=None, universal_newlines=None, timeout=None, text=None, **other_popen_kwargs)¶

也是有timeout参数的。

后记：

VSCode中，Command+🖱鼠标点击，也可以看到函数定义中的timeout参数

然后此处感觉上述函数可以优化为：

调用，且加上timeout参数=1或2秒，看看：

adb connect 192.168.31.84:5555

命令的返回：

成功：包含connected

connected to 192.168.31.84:5555

出错：包含failed 或 timed out

failed to connect to ‘192.168.31.84:5555’: Operation timed out

另外，本来想把：get_cmd_lines

去加到自己的库函数：crifanLibPython中呢，发现已经有类似的了：

然后去加上timeout参数：

def getCommandOutput(consoleCommand, consoleOutputEncoding="utf-8", timeout=2):
    """get command output from terminal


    Args:
        consoleCommand (str): console/terminal command string
        consoleOutputEncoding (str): console output encoding, default is utf-8
        timeout (int): wait max timeout for run console command
    Returns:
        console output (str)
    Raises:
    """
    # print("getCommandOutput: consoleCommand=%s" % consoleCommand)
    isRunCmdOk = False
    consoleOutput = ""
    try:
        # consoleOutputByte = subprocess.check_output(consoleCommand)


        consoleOutputByte = subprocess.check_output(consoleCommand, shell=True, timeout=timeout)


        # commandPartList = consoleCommand.split(" ")
        # print("commandPartList=%s" % commandPartList)
        # consoleOutputByte = subprocess.check_output(commandPartList)
        # print("type(consoleOutputByte)=%s" % type(consoleOutputByte)) # <class 'bytes'>
        # print("consoleOutputByte=%s" % consoleOutputByte) # b'640x360\n'


        consoleOutput = consoleOutputByte.decode(consoleOutputEncoding) # '640x360\n'
        consoleOutput = consoleOutput.strip() # '640x360'
        isRunCmdOk = True
    except subprocess.CalledProcessError as callProcessErr:
        cmdErrStr = str(callProcessErr)
        print("Error %s for run command %s" % (cmdErrStr, consoleCommand))


    # print("isRunCmdOk=%s, consoleOutput=%s" % (isRunCmdOk, consoleOutput))
    return isRunCmdOk, consoleOutput

再去优化：

src/common/DevicesMethods.py

    def connectDevice(self):
        """连接/重连 Android/iOS 设备"""
        isConnected = False
        if self.isAndroid:
            adbConnectCmd = "adb connect %s" % self.device
            logging.info("Try connect Android device: %s", adbConnectCmd)
            # os.system(adbConnectCmd) # when failed, will wait too long time: ~ 1 minutes
            cmdOutputStr = CommonUtils.get_cmd_lines(adbConnectCmd, timeout=1)
            if "connected" in cmdOutputStr:
                isConnected = True
            elif "failed" in cmdOutputStr:
                isConnected = False
        elif self.isiOS:
            logging.warning("TODO: add support start iOS device")


        return isConnected

去调试看看

【总结】

utils/common_utils.py

    @staticmethod
    def get_cmd_lines(cmd, text=False, timeout=2):
        # 执行cmd命令，将结果保存为列表
        resultStr = ""
        resultStrList = []
        try:
            # consoleOutputByte = subprocess.check_output(cmd, shell=True) # b'xxx\n'
            consoleOutputByte = subprocess.check_output(cmd, shell=True, timeout=timeout) # b'xxx\n'
            try:
                resultStr = consoleOutputByte.decode("utf-8")
            except UnicodeDecodeError:
                # TODO: use chardet auto detect encoding
                # consoleOutputStr = consoleOutputByte.decode("gbk")
                resultStr = consoleOutputByte.decode("gb18030")


            if not text:
                resultStrList = resultStr.splitlines()
        except Exception as err:
            print("err=%s when run cmd=%s" % (err, cmd))


        if text:
            return resultStr
        else:
            return resultStrList

调用：

    def connectDevice(self):
        """连接/重连 Android/iOS 设备"""
        isConnected = False
        if self.isAndroid:
            adbConnectCmd = "adb connect %s" % self.device
            logging.info("Try connect Android device: %s", adbConnectCmd)
            # os.system(adbConnectCmd) # when failed, will wait too long time: ~ 1 minutes
            cmdOutputStr = CommonUtils.get_cmd_lines(adbConnectCmd, timeout=1)

然后超过1秒后，就返回，报错：

err=Command 'adb connect 192.168.31.84:5555' timed out after 1 seconds when run cmd=adb connect 192.168.31.84:5555

是我们希望的效果了。

关于get_cmd_lines的等价函数：

getCommandOutput

def getCommandOutput(consoleCommand, consoleOutputEncoding="utf-8", timeout=2):
    """get command output from terminal


    Args:
        consoleCommand (str): console/terminal command string
        consoleOutputEncoding (str): console output encoding, default is utf-8
        timeout (int): wait max timeout for run console command
    Returns:
        console output (str)
    Raises:
    """
    # print("getCommandOutput: consoleCommand=%s" % consoleCommand)
    isRunCmdOk = False
    consoleOutput = ""
    try:
        # consoleOutputByte = subprocess.check_output(consoleCommand)


        consoleOutputByte = subprocess.check_output(consoleCommand, shell=True, timeout=timeout)


        # commandPartList = consoleCommand.split(" ")
        # print("commandPartList=%s" % commandPartList)
        # consoleOutputByte = subprocess.check_output(commandPartList)
        # print("type(consoleOutputByte)=%s" % type(consoleOutputByte)) # <class 'bytes'>
        # print("consoleOutputByte=%s" % consoleOutputByte) # b'640x360\n'


        consoleOutput = consoleOutputByte.decode(consoleOutputEncoding) # '640x360\n'
        consoleOutput = consoleOutput.strip() # '640x360'
        isRunCmdOk = True
    except subprocess.CalledProcessError as callProcessErr:
        cmdErrStr = str(callProcessErr)
        print("Error %s for run command %s" % (cmdErrStr, consoleCommand))


    # print("isRunCmdOk=%s, consoleOutput=%s" % (isRunCmdOk, consoleOutput))
    return isRunCmdOk, consoleOutput

详见：

https://github.com/crifan/crifanLibPython/blob/master/crifanLib/crifanSystem.py

【后记】

去回复帖子

python 3.x – Setting timeout when using os.system function – Stack Overflow

python timeout using os.system – Stack Overflow

linux – Python os.system timeout with strings – Stack Overflow

转载请注明：在路上 » 【已解决】Python用os.system执行命令行命令希望加上timeout超时设置