在 subprocess 之前,创建一个新进程的方式有很多种,如 os.system()、os.spawn* 方法等。为了统一创建进程的方式,python 社区提议使用 subprocess 模块来统一创建进程,替换 os.system 和 os.spawn*。
方法签名如下
subprocess.run(args, *, stdin=None, input=None, stdout=None, stderr=None, capture_output=False, shell=False, cwd=None, timeout=None, check=False, encoding=None, errors=None, text=None, env=None, universal_newlines=None, **other_popen_kwargs)
run 方法是创建进程的推荐方法,只有 run 方法无法满足需要的时候才考虑使用 popen 方法。run 会堵塞式地执行 args 参数提供的命令,直到命令执行结束或者超时。args 可以是字符串数组或者字符串,当 args 为字符串时,shell 参数必须为 True
我想创建一个子进程,使用 adb install 命令给 vivo/oppo/xiaomi 手机安装 app,但是 vivo/oppo/xiaomi 手机会弹出二次确认窗口,如果不进行点击操作,则无法安装 App。所以我就想创建一个子进程,超时后再接着由 UI 自动化的方式来点击确认按钮。这就要求超时后安装进程不能退出。run 方法虽然有 timeout 参数,但是子进程会在超时被 kill 掉。
以下动画演示了手动安装 app 时需要二次确认,手动点击继续安装后,可以在控制台看到 Perform Streamed Install Success 的输出。
以下动画演示了使用 run 命令,超时后安装进程被 kill 掉了,在控制台无法看到成功安装的日志输出。
对应的代码:
import subprocess
if __name__ == '__main__':
proc=subprocess.run('adb install -g -r -t app-uiautomator.apk', shell=True, timeout=10, capture_output=True)
subprocess 模块里进程的创建和管理都是 Popen 类处理的,它让开发者非常灵活地处理一些不常见的场景。
class subprocess.Popen(args, bufsize=- 1, executable=None, stdin=None, stdout=None, stderr=None, preexec_fn=None, close_fds=True, shell=False, cwd=None, env=None, universal_newlines=None, startupinfo=None, creationflags=0, restore_signals=True, start_new_session=False, pass_fds=(), *, group=None, extra_groups=None, user=None, umask=- 1, encoding=None, errors=None, text=None, pipesize=- 1, process_group=None)
Popen 类有以下几个方法:
Popen.communicate(input=None, timeout=None)
communicate()返回一个元组(stdout_data, stderr_data),如果Popen指定了text模式,stdout_data将为字符串,否则为byte
input 向标准输入发送信息
timeout 如果进行在 timeout 指定的时间之内没有结束,则抛出一个 TimeoutExpired 异常,且进程不会被杀死
抛出异常后可以继续重新调用 communicate()
* Popen.send_signal(signal) 向子进程发送信号
* Popen.terminate() 停止子进程
* Popen.poll() 检查进程是否结束,没有结束返回 None,否则返回执行状态的数值
* Popen.wait(timeout=None) 等待子进程执行结束,如果 timeout 指定的时间之后进程没有结束,则抛出 TimeoutExpired
在我的执行场景中,需要使用 communicate 方法,以下是使用 Popen 类调用 adb 命令安装 app 的动画展示,超时后,手动点击继续安装,app 最终安装成功。
import subprocess
from subprocess import TimeoutExpired
if __name__ == '__main__':
try:
proc=subprocess.Popen('adb install -g -r -t app-uiautomator.apk', shell=True, text=True, stdout=subprocess.PIPE)
stdout,errs = proc.communicate(timeout=10)
print(f'stdout1:{stdout}')
except TimeoutExpired as te:
print('timeout')
stdout,errs = proc.communicate()
print(f'stdout2:\n{stdout}')
这样的话,我的目的就达到了。