首页 > 解决方案 > QAxWidget 实例中的 QEventLoop 失败

问题描述

我正在使用 Windows OCX api 开发基于终端的 PyQt 应用程序。而且我在 QAxWidget 实例中实现(调用 exec)QEventLoop 时遇到了麻烦。

如果我在主 QApplication 主循环中调用 exec() ,一切都很好。但是,在主循环中实例化的 QAxWidget 实例中调用 exec() ,它不能按预期工作。

当 exec() 调用(代码中的_on_event_connect)时,错误消息显示为

"QEventLoop:exec: instance 0x18479d8 has already called exec()"

在调用 exit() 的那一刻,QEventLoop 似乎没有运行。并且应用程序在此之后挂起。

下面是我的代码的简化示例,它会在登录过程完成后立即调用 QAxWidget 方法“dynamicCall”。它由 pyQt 信号/槽机制组成,从 API 服务器获取事件。请理解您无法执行此代码,因为它需要特定的 API 安装和注册用户 ID。但我希望你能看到问题所在。

主文件

import sys
from PyQt5.QtWidgets import QApplication

from api import OpenApi

class Main():
    def __init__(self):

    self.api = OpenApi()
    self.api.comm_connect()
    # self.api.req_basic_stock_info("035420") -> if I call here, works well

if __name__ == "__main__":
    app = QApplication(sys.argv)
    main = Main()
    sys.exit(app.exec_())

api.py

import sys

from PyQt5.QtWidgets import *
from PyQt5.QAxContainer import *
from PyQt5.QtCore import *

class OpenApi(QAxWidget):
    def __init__(self):
        super().__init__()

        self.setControl("KHOPENAPI.KHOpenAPICtrl.1")
        self.OnEventConnect.connect(self._on_event_connect)
        self.OnReceiveTrData.connect(self._on_receive_tr_data)
        self._qloop = QEventLoop()

    def loop(self):
        if not self._qloop.isRunning():
            logger.info(f"-->exec")
            self._qloop.exec()
        else:
            logger.info("exec skipped")

    def unloop(self):
        if self._qloop.isRunning():
            logger.info("-->exit")
            self._qloop.exit()
        else:
            logger.info(f"exit skipped")

    def comm_connect(self):
        self.dynamicCall("CommConnect()")
        self.loop()

    def comm_rq_data(self, rqname, trcode, next, screen_no):
        self.dynamicCall("CommRqData(QString, QString, int, QString)", rqname, trcode, next, screen_no)
        self.loop()

    def _on_event_connect(self, err_code):
        if err_code == 0:
            logger.info("logged in")
        else:
            logger.info("log-in failed")
        self.unloop()

        # If I call this function here, QEventLoop fails
        
        self.req_basic_stock_info("000660")    

    def _on_receive_tr_data(self, screen_no, rqname, trcode, record_name, next,
                        unused1, unused2, unused3, unused4):
        logger.info(f"OnReceivTrData: {rqname}")
        self.unloop()

    def req_basic_stock_info(self, code):
        # I want to call this function inside instance on certain condition.
        self.set_input_value("item", code)
        scrno = "2000"
        self.comm_rq_data("ITEM_INFO_REQ", "opt10080", "0", scrno)

输出如下。

12:42:53,54 test INFO -->exec

12:42:58,961 test INFO logged in

12:42:58,962 test INFO -->exit

12:42:58,977 test INFO -->exec

QEventLoop::exec: instance 0x35c34a0 has already called exec()

12:42:59,33 test INFO OnReceivTrData:ITEM_INFO_REQ

12:42:59,35 test INFO exit skipped

正如 eyllanesc 告诉我的那样,我更改了与分配/取消分配 QEventLoop 相关的函数,如下所示。它有效。

def loop(self):
    self._qloop = QEventLoop()
    self._qloop.exec()

def unloop(self):
    try:
        self._qloop.exit()
    except AttributeError as e:
        logger.error(e)

更新

恐怕这个问题还没有解决。在 _on_event_connect 事件上调用 req_basic_stock_info 函数时成功,但我在另一个事件函数中调用它,它再次挂起而没有任何错误消息。调用机制没有区别。

标签: pythonpyqtqapplicationqeventloopqaxwidget

解决方案


调用该exit()函数并不意味着可以重用 QEventLoop,因此会发出错误消息。在这种情况下,最好使用新的 QEventLoop:

def unloop(self):
    if self._qloop.isRunning():
        logger.info("-->exit")
        self._qloop.exit()
        self._qloop = QEventLoop()
    else:
        logger.info(f"exit skipped")

推荐阅读