Python调用exe执行完成后自动终止进程并停止Jupyter单元格运行

原代码存在的问题 #

1. 重复调用subprocess.Popen创建进程，第一次创建的进程未被处理，会成为残留进程
2. 未向exe发送退出指令，导致exe在完成任务后停留在交互界面等待输入，communicate会一直阻塞，只能靠超时强制终止，无法准确判断任务是否完成

解决方案 #

方案1：添加退出命令（优先推荐，若exe支持） #

很多交互式exe在完成任务后，需要发送特定退出命令（如exit、quit）才会自动关闭。在命令列表末尾添加对应的退出指令，让exe执行完任务后主动退出，这样communicate就能正常等待进程结束，无需设置超时。

修正后的代码：

import subprocess

# Path to the executable
exe_path = r"C:\temp\Pipeline\Automate_LiDAR_processing\Automate_LiDAR_Processing_V2.2.exe"

commands = [
    "1\n",
    r"C:\temp\Pipeline\Automate_LiDAR_processing\Templates\Reconstruction\PWMMS_HA\HDL32EF\n",
    r"C:\temp\Pipeline\hdl32e_front_left\n",
    r"C:\temp\Pipeline\Code\20231101_Transit_US231_Trajectory_TC_revised.txt\n",
    "test_output\n",
    r"C:\temp\Pipeline\output_folder_2\n",
    "exit\n"  # 添加退出命令，根据exe实际支持的指令调整
]

# 仅创建一次进程
process = subprocess.Popen(
    exe_path,
    stdin=subprocess.PIPE,
    stdout=subprocess.PIPE,
    stderr=subprocess.PIPE,
    text=True
)

# 一次性发送所有命令（包括退出指令），并等待进程结束
output, error = process.communicate(input=''.join(commands))

# 打印输出和错误信息
print("输出内容:\n", output)
if error:
    print("错误信息:\n", error)

方案2：实时监控输出，捕获任务完成标志 #

如果不知道exe的退出命令，可以实时读取stdout，当捕获到任务完成的特定输出（比如"处理完成"、"Processing finished"等）时，立即终止进程。

示例代码：

import subprocess
import time

exe_path = r"C:\temp\Pipeline\Automate_LiDAR_processing\Automate_LiDAR_Processing_V2.2.exe"

commands = [
    "1\n",
    r"C:\temp\Pipeline\Automate_LiDAR_processing\Templates\Reconstruction\PWMMS_HA\HDL32EF\n",
    r"C:\temp\Pipeline\hdl32e_front_left\n",
    r"C:\temp\Pipeline\Code\20231101_Transit_US231_Trajectory_TC_revised.txt\n",
    "test_output\n",
    r"C:\temp\Pipeline\output_folder_2\n"
]

process = subprocess.Popen(
    exe_path,
    stdin=subprocess.PIPE,
    stdout=subprocess.PIPE,
    stderr=subprocess.PIPE,
    text=True,
    bufsize=1,  # 行缓冲，实时读取输出
    universal_newlines=True
)

# 发送所有命令
for command in commands:
    process.stdin.write(command)
    process.stdin.flush()
    time.sleep(0.1)  # 给exe留处理时间，可根据实际调整

# 实时监控输出，捕获完成标志
completed_flag = "Processing completed"  # 替换为exe实际的完成提示文本
while True:
    line = process.stdout.readline()
    if not line:
        break
    print(line.strip())
    # 检测到完成标志，终止进程
    if completed_flag in line:
        process.terminate()
        break

# 读取剩余输出和错误
remaining_output = process.stdout.read()
error_output = process.stderr.read()
if remaining_output:
    print("剩余输出:\n", remaining_output)
if error_output:
    print("错误信息:\n", error_output)

关键说明 #

• 方案1更可靠，前提是要确认exe支持的退出指令，可以手动运行exe，完成任务后输入exit/quit测试是否能正常关闭。
• 方案2需要先手动运行exe，记录任务完成时的输出文本，作为判断标志。
• 原代码中重复创建process的问题必须修复，否则会产生无法控制的残留进程。

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