多线程环境下py_canoe与PySimpleGUI结合的异常问题求助

核心原因 #

py_canoe本质是封装了CANoe的COM接口，而CANoe的COM对象采用单线程公寓（STA）模型，要求所有对该对象的调用必须在创建它的线程中执行。跨线程直接操作CANoe实例会触发线程同步冲突，导致偶发的停止、卡死现象——这就是你遇到问题的根源，和PySimpleGUI的线程实现方式无关。

可行实现方案 #

方案1：主线程处理CANoe操作，拆分任务避免阻塞GUI #

把耗时的CANoe操作拆分成多个小步骤，每次执行一个小步骤后回到PySimpleGUI的事件循环，更新GUI后再继续执行下一个步骤。这样既不会阻塞GUI响应，又能保证所有CANoe操作都在主线程中进行。

示例修改思路：

# 把原execute_smoketests拆分成多个可分步执行的子函数
def smoke_test_step1(choice):
    # 执行第一个CANoe操作步骤（比如加载配置、初始化总线）
    pass

def smoke_test_step2(choice):
    # 执行第二个CANoe操作步骤（比如启动测试、采集数据）
    pass

def smoke_test_step3(choice):
    # 执行第三个CANoe操作步骤（比如停止测试、生成报告）
    pass

# 主线程GUI事件循环中分步执行
compile_start = datetime.now()
time_delta = 0
current_step = 0
smoke_tests_done = False

while True:
    event, values = Pd_conf.MainWindow.read(timeout=100)
    # 触发测试开始
    if event == "--START_SMOKE_TEST--":
        current_step = 0
        smoke_tests_done = False
        compile_start = datetime.now()
    # 利用超时事件分步执行测试
    elif event == "__TIMEOUT__" and not smoke_tests_done:
        if current_step == 0:
            smoke_test_step1(choice)
            current_step += 1
        elif current_step == 1:
            smoke_test_step2(choice)
            current_step += 1
        elif current_step == 2:
            smoke_test_step3(choice)
            smoke_tests_done = True
            Pd_conf.MainWindow.write_event_value("--PRECONDITION_CHECK_DONE--", None)
        
        # 更新时长显示
        elapsed = (datetime.now() - compile_start).seconds
        if elapsed > time_delta:
            time_delta = elapsed
            print(f"Duration : {timedelta(seconds=time_delta)}", replace=f"Duration : {timedelta(seconds=time_delta-1)}")
    # 测试完成处理
    elif event == "--PRECONDITION_CHECK_DONE--":
        print("Smoke tests completed")
        break

方案2：线程间通信，子线程仅做触发/进度计算，主线程执行CANoe操作 #

如果需要用子线程处理进度计时或业务逻辑判断，不要在子线程中直接调用py_canoe，而是通过队列或PySimpleGUI的write_event_value实现线程间通信，让主线程执行所有CANoe相关操作。

示例修改：

import threading
from queue import Queue
import time

# 全局队列用于线程间传递CANoe操作指令
canoe_cmd_queue = Queue()

def smoke_test_worker(choice, window):
    # 子线程仅负责计时和发送指令，不触碰CANoe操作
    compile_start = datetime.now()
    time_delta = 0
    # 发送启动测试的指令
    canoe_cmd_queue.put(("start_test", choice))
    
    while True:
        # 检查是否收到测试完成信号
        if not canoe_cmd_queue.empty() and canoe_cmd_queue.get() == "test_done":
            break
        # 计算时长并通知主线程更新GUI
        elapsed = (datetime.now() - compile_start).seconds
        if elapsed > time_delta:
            time_delta = elapsed
            window.write_event_value("--UPDATE_DURATION--", time_delta)
        time.sleep(1)

# 主线程GUI事件循环
while True:
    event, values = Pd_conf.MainWindow.read(timeout=100)
    # 触发测试启动
    if event == "--START_SMOKE_TEST--":
        threading.Thread(target=smoke_test_worker, args=(choice, Pd_conf.MainWindow), daemon=True).start()
    # 更新时长显示
    elif event == "--UPDATE_DURATION--":
        time_delta = values[event]
        print(f"Duration : {timedelta(seconds=time_delta)}", replace=f"Duration : {timedelta(seconds=time_delta-1)}")
    # 处理CANoe操作指令
    elif not canoe_cmd_queue.empty():
        cmd = canoe_cmd_queue.get()
        if cmd[0] == "start_test":
            # 主线程执行完整的CANoe测试操作
            execute_smoketests(cmd[1])
            # 通知子线程测试完成
            canoe_cmd_queue.put("test_done")
            Pd_conf.MainWindow.write_event_value("--PRECONDITION_CHECK_DONE--", None)
    # 测试完成处理
    elif event == "--PRECONDITION_CHECK_DONE--":
        print("Smoke tests completed")
        break

关键注意事项 #

• 所有CANoe实例的创建、调用必须在**同一个线程（通常是主线程）**中完成，绝对不能跨线程操作CANoe对象。
• 更新GUI必须通过PySimpleGUI提供的线程安全方法（如write_event_value），禁止在子线程中直接修改GUI元素。
• 如果考虑用多进程替代线程，需注意每个进程要独立启动CANoe实例，进程间无法共享CANoe对象。

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