跨Windows/Linux平台：用Python脚本B终止脚本A的循环并安全关闭

要实现通过B.py发送指令让A.py安全终止主循环（执行capturer.close()等清理逻辑），可以采用**进程间通信（IPC）**的跨平台方案，以下是两种可靠实现方式：

方案一：文件标志法（简单无依赖） #

通过创建/删除一个标记文件作为停止信号，A.py在循环中定期检查该文件是否存在，存在则退出循环。

修改A.py代码 #

import time
import os
from image_procesing.recognizer import Recognizer
from configuration.config import ProgramData
from external_services_communication.commincation import Knx, KnxAndInflux, InfluxDb
from image_procesing.video_capturer import Capturer

# 定义停止标志文件路径，可根据实际路径调整
STOP_FLAG_FILE = "./stop_a_flag.txt"

# 启动时清理旧的标志文件
if os.path.exists(STOP_FLAG_FILE):
    os.remove(STOP_FLAG_FILE)

recognizer = Recognizer(ProgramData.coco_data)
capturer = Capturer(ProgramData.connection_string)
start_time = None
last_count = None

# 将原while True改为检查标志文件的循环
while not os.path.exists(STOP_FLAG_FILE):
    success, frame = capturer.get_status_and_frame()

    if not success:
        break

    # ... 原有业务代码（无需修改） ...
    # 示例中的start_time = time.time()保留

    # 可选：短暂休眠降低CPU占用，可根据业务调整间隔
    time.sleep(0.01)

# 执行安全清理
capturer.close()
# 清理标志文件
if os.path.exists(STOP_FLAG_FILE):
    os.remove(STOP_FLAG_FILE)
print("A.py已安全退出")

编写B.py代码 #

import os
import time

STOP_FLAG_FILE = "./stop_a_flag.txt"

def stop_script_a():
    # 创建停止标志文件
    with open(STOP_FLAG_FILE, "w") as f:
        f.write("stop")
    
    # 等待A.py处理并清理文件
    time.sleep(2)
    if not os.path.exists(STOP_FLAG_FILE):
        print("A.py已成功关闭")
    else:
        print("A.py未响应关闭指令，可考虑强制终止")

if __name__ == "__main__":
    stop_script_a()

优缺点

• ✅ 优点：实现简单，无需额外依赖，完全跨Windows/Linux
• ❌ 缺点：A.py崩溃时可能残留标志文件，需启动时清理；响应速度取决于循环检查频率

方案二：本地套接字法（可靠响应快） #

通过本地TCP套接字实现进程间通信，B.py向A.py发送停止指令，A.py实时监听并处理，无文件残留问题。

修改A.py代码 #

import time
import socket
import threading
from image_procesing.recognizer import Recognizer
from configuration.config import ProgramData
from external_services_communication.commincation import Knx, KnxAndInflux, InfluxDb
from image_procesing.video_capturer import Capturer

# 本地套接字配置（跨平台通用）
LOCAL_HOST = "127.0.0.1"
LOCAL_PORT = 65432
running = threading.Event()
running.set()

def stop_listener():
    """后台线程监听停止指令"""
    stop_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    stop_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    stop_socket.bind((LOCAL_HOST, LOCAL_PORT))
    stop_socket.listen(1)
    
    while running.is_set():
        try:
            conn, addr = stop_socket.accept()
            with conn:
                data = conn.recv(1024).decode().strip()
                if data == "STOP":
                    running.clear()
                    print("收到停止指令，准备退出")
                    break
        except Exception:
            pass
    stop_socket.close()

# 启动监听线程（后台守护线程）
listener_thread = threading.Thread(target=stop_listener, daemon=True)
listener_thread.start()

recognizer = Recognizer(ProgramData.coco_data)
capturer = Capturer(ProgramData.connection_string)
start_time = None
last_count = None

# 基于事件的循环判断
while running.is_set():
    success, frame = capturer.get_status_and_frame()

    if not success:
        break

    # ... 原有业务代码（无需修改） ...
    # 示例中的start_time = time.time()保留

# 执行安全清理
capturer.close()
print("A.py已安全退出")

编写B.py代码 #

import socket

LOCAL_HOST = "127.0.0.1"
LOCAL_PORT = 65432

def send_stop_command():
    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
        try:
            s.connect((LOCAL_HOST, LOCAL_PORT))
            s.sendall(b"STOP")
            print("停止指令已发送")
        except ConnectionRefusedError:
            print("A.py未在运行或未监听指定端口")

if __name__ == "__main__":
    send_stop_command()

优缺点

• ✅ 优点：响应及时，无文件残留，支持扩展多指令（如暂停/恢复），完全跨平台
• ❌ 缺点：需处理套接字异常，代码量略多

注意事项 #

1. 若capturer.get_status_and_frame()是阻塞式调用，建议使用方案二的线程监听方式，避免主循环无法及时响应停止指令
2. 两种方案均基于Python标准库，无需安装第三方包，完全兼容Windows和Linux系统
3. 可根据业务需求调整检查频率（文件法）或套接字端口（套接字法）

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