如何后台运行Python子脚本并发送CTRL_C_EVENT终止，实现重复数据采集？

核心需求 #

• 父脚本后台启动带参数的子脚本，自身可继续执行逻辑
• 向子脚本发送CTRL+C信号触发收尾流程（而非强制终止）
• 支持重复执行采集流程，每次执行前等待用户调整设备

技术选型建议 #

优先用subprocess模块：子脚本是独立Python程序，subprocess更适合管控外部进程；multiprocessing更适合代码内的进程拆分，若子脚本逻辑可整合也可选用，但subprocess适配现有子脚本改动更小。

具体实现步骤 #

1. 子脚本改造（适配父脚本管控） #

修改子脚本，新增参数跳过自身PLC检测逻辑，保留原采集和收尾流程：

# plc_collector.py
import sys
import time
from openpyxl import Workbook

def collect_and_save(filename):
    wb = Workbook()
    ws = wb.active
    try:
        # 0.02秒间隔采集标签值
        while True:
            # 替换为实际AB PLC标签采集逻辑
            tag_value = time.time() % 100
            ws.append([time.strftime("%Y-%m-%d %H:%M:%S"), tag_value])
            time.sleep(0.02)
    except KeyboardInterrupt:
        # 收尾操作：绘图、保存文件
        print(f"开始处理文件 {filename}")
        # 此处添加Excel绘图逻辑（如matplotlib插入图表）
        wb.save(filename)
        print(f"文件 {filename} 保存完成")

if __name__ == "__main__":
    if len(sys.argv) < 2:
        print("请传入保存文件名参数")
        sys.exit(1)
    
    # 跳过PLC检测：父脚本已完成检测
    skip_ping = len(sys.argv) > 2 and sys.argv[2] == "--skip-ping"
    if not skip_ping:
        # 保留原PLC ping检测逻辑（仅独立运行时触发）
        print("检测PLC连接...")
        # 原ping实现代码
    
    collect_and_save(sys.argv[1])

2. 父脚本实现（Windows环境解决CTRL_C_EVENT问题） #

Windows下默认无法直接向子进程传递CTRL+C，需通过进程组+系统API发送信号：

# controller.py
import subprocess
import time
import ctypes
import sys

# Windows系统API常量
CTRL_C_EVENT = 0
CREATE_NEW_PROCESS_GROUP = 0x00000200

def send_ctrl_c(proc):
    # 向子进程发送CTRL_C_EVENT信号
    ctypes.windll.kernel32.GenerateConsoleCtrlEvent(CTRL_C_EVENT, proc.pid)

def check_plc_connection(plc_ip):
    # 父脚本独立实现PLC连接检测
    print(f"检测PLC [{plc_ip}] 连接...")
    # 替换为实际ping/PLC连接验证逻辑
    return True

def start_collector(filename):
    # 后台静默启动子脚本，传递参数并跳过PLC检测
    return subprocess.Popen(
        [sys.executable, "plc_collector.py", filename, "--skip-ping"],
        creationflags=CREATE_NEW_PROCESS_GROUP,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        shell=False
    )

def main():
    plc_ip = "192.168.1.100"  # 替换为实际PLC地址
    
    # 第一次采集流程
    if not check_plc_connection(plc_ip):
        print("PLC连接失败，退出")
        sys.exit(1)
    input("请调整设备后按回车启动第一次采集...")
    proc = start_collector("第一次采集数据.xlsx")
    
    print("发送ON命令到PLC，启动流程")
    # 替换为实际向PLC发送ON命令的逻辑
    time.sleep(5)  # 模拟第一个定时器时长
    
    print("发送第二个命令到PLC")
    # 替换为实际发送第二个命令的逻辑
    time.sleep(3)  # 模拟第二个定时器时长
    
    print("触发子脚本收尾流程")
    send_ctrl_c(proc)
    proc.wait()
    print("第一次采集完成")

    # 第二次采集流程
    input("请调整设备后按回车启动第二次采集...")
    proc = start_collector("第二次采集数据.xlsx")
    
    print("发送ON命令到PLC，启动流程")
    time.sleep(5)
    print("发送第二个命令到PLC")
    time.sleep(3)
    
    print("触发子脚本收尾流程")
    send_ctrl_c(proc)
    proc.wait()
    print("第二次采集完成")

if __name__ == "__main__":
    main()

3. 关键细节说明 #

• 参数传递：父脚本通过subprocess.Popen的参数列表直接传递文件名和--skip-ping标志，子脚本用sys.argv接收
• Windows信号处理：设置CREATE_NEW_PROCESS_GROUP让子进程成为进程组组长，确保GenerateConsoleCtrlEvent发送的CTRL+C能被捕获
• 静默运行：通过stdout=subprocess.PIPE/stderr=subprocess.PIPE重定向输出，实现后台运行；如需查看日志可移除这两个参数
• 重复执行：通过input()等待用户确认，重复调用start_collector启动新的子进程

替代方案：用multiprocessing（需整合子脚本逻辑） #

若把子脚本采集逻辑封装为函数，可通过事件通知停止，无需处理信号：

# 整合版父脚本示例
import multiprocessing
import time
from openpyxl import Workbook

def collect_data(filename, stop_event):
    wb = Workbook()
    ws = wb.active
    while not stop_event.is_set():
        tag_value = time.time() % 100
        ws.append([time.strftime("%Y-%m-%d %H:%M:%S"), tag_value])
        time.sleep(0.02)
    # 收尾操作
    print(f"处理文件 {filename}")
    wb.save(filename)

def main():
    # 第一次采集
    input("调整设备后按回车...")
    stop_event = multiprocessing.Event()
    proc = multiprocessing.Process(target=collect_data, args=("第一次采集.xlsx", stop_event))
    proc.start()
    time.sleep(5)
    time.sleep(3)
    stop_event.set()
    proc.join()

    # 第二次采集
    input("调整设备后按回车...")
    stop_event = multiprocessing.Event()
    proc = multiprocessing.Process(target=collect_data, args=("第二次采集.xlsx", stop_event))
    proc.start()
    time.sleep(5)
    time.sleep(3)
    stop_event.set()
    proc.join()

if __name__ == "__main__":
    main()

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