如何在Python中通过ADB Shell交互控制台应用并逐行执行命令

• 操作流程：
  1. 执行adb shell获取设备Root权限
  2. 运行/oem/console --no-logging启动设备交互式控制台
  3. 等待控制台返回连接成功信息后，按回车进入命令接收状态
  4. 逐行执行以下invoke命令，每个命令执行后等待数秒：
     • invoke put.device.into.charge 1
     • invoke call.cell.hangup 1
     • invoke xyz

现有Python代码如下：

import subprocess
from subprocess import Popen

def run_adb_shell_command(command, return_process=False):
    """Run an ADB shell command."""
    cmd = command.split()
    process = Popen(cmd, stdin=subprocess.PIPE)
    if return_process:
        return process


def send_command_to_console_app(command, process):
    """SEND Commands to Console App."""
    new_line = '\n'.encode()
    # encoded_command = str(command).encode()
    s0 = command
    s1 = 'event 1'
    s2 = 'event 2'
    s3 = 'event 3'
    concat_query = "{}\n".format(s0)

    process.communicate(input=concat_query.encode('utf-8'))[0]
    process.kill()

问题：调用communicate()执行单个命令时会挂起，无法继续执行后续命令。

问题根源 #

subprocess.Popen.communicate()的设计逻辑是一次性发送所有输入，然后等待进程终止并读取全部输出。但你的场景是交互式控制台进程（启动后不会自动退出，持续等待命令输入），调用communicate()后会一直阻塞，等待进程结束，这就是挂起的核心原因。

修改方案 #

直接通过process.stdin逐行写入命令，每次写入后刷新缓冲区，配合time.sleep()等待设备处理完成，最后按需关闭进程：

1. 调整ADB命令，确保先获取Root权限再启动控制台（直接在adb shell中用su -c执行控制台命令）
2. 替换communicate()为stdin.write()+stdin.flush()，逐行发送命令
3. 每个命令执行后添加等待时间，给设备足够的处理周期
4. 最后关闭标准输入并等待进程结束（或按需终止）

修改后的代码示例：

import subprocess
from subprocess import Popen
import time

def run_adb_console():
    """启动带Root权限的交互式控制台进程"""
    # 构造命令：adb shell 下执行 su -c 启动控制台
    cmd = ["adb", "shell", "su", "-c", "/oem/console --no-logging"]
    # 同时捕获stdout/stderr可以查看控制台输出（可选）
    process = Popen(cmd, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
    # 等待控制台连接成功（可根据stdout输出判断，这里简化为固定等待）
    time.sleep(2)
    # 发送回车进入命令接收状态
    process.stdin.write("\n")
    process.stdin.flush()
    time.sleep(1)
    return process

def send_commands_to_console(process, commands, wait_time=3):
    """逐行发送命令到控制台，每个命令后等待指定时间"""
    for cmd in commands:
        process.stdin.write(f"{cmd}\n")
        process.stdin.flush()
        print(f"已发送命令：{cmd}")
        time.sleep(wait_time)
    # 所有命令执行完后，关闭stdin并等待进程结束（或按需kill）
    process.stdin.close()
    process.wait()

# 主流程
if __name__ == "__main__":
    console_process = run_adb_console()
    invoke_commands = [
        "invoke put.device.into.charge 1",
        "invoke call.cell.hangup 1",
        "invoke xyz"
    ]
    send_commands_to_console(console_process, invoke_commands)

关键说明 #

• 使用text=True让Popen以文本模式处理输入输出，避免手动编码解码
• 启动控制台后添加等待时间，确保设备完成连接并返回成功信息
• 发送回车后再等待，确保控制台进入命令接收状态
• 逐行写入命令并刷新缓冲区，确保命令被控制台即时接收
• 最后关闭stdin并调用wait()，让进程正常结束（如果控制台支持退出命令，也可以发送exit命令后再等待）

内容的提问来源于stack exchange，提问作者Shahboz