You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

如何在Systembox类外部获取运行输出并写入pandas DataFrame

问题原因

你修改后的代码无输出是三个核心错误导致的:

  1. return逻辑无效:原_AWS10000方法中调用self.print_result(self.result)时,你仅将结果return但没有任何外部代码接收这个返回值,返回值直接被丢弃
  2. 调用逻辑错误:你写在全局作用域的aaa.print_result(result)里的result是未定义变量,完全不指向类内部的self.result属性
  3. 执行时机错误:你把实例化、打印逻辑写在if __name__ == "__main__"外部,程序启动时会先执行这部分代码,此时Systembox还未开始请求数据,self.result是空列表,自然没有输出
解决方法

两种符合PyQt设计逻辑的实现方案,任选一种即可:

方案1:自定义信号(推荐,最适配Qt事件驱动机制)

通过Qt信号实现数据就绪时主动推送给外部处理逻辑:

第一步:修改Systembox类定义

class Systembox(QAxWidget):
    # 新增自定义信号,参数为批量数据的列表
    batch_data_ready = pyqtSignal(list)
    
    def __init__(self):
        super().__init__()
        self._create_sysembox_instance()
        self._set_signal_slots()
        self.c = 0
        self.result = []
    # 其余原有方法保持不变,仅修改_AWS10000方法
    def _AWS10000(self, rqname, trcode):
        cnn = self._get_repeat_cnt(trcode, rqname)
        for i in range(cnn):
            date = self._comm_get_data(trcode, rqname, i, "DATA")
            self.c = self.c + 1
            if self.c > 299:
                self.remained_data = False
                self.c = self.c - 300
            if len(self.result) < 300:
                self.result.append(date)
            if len(self.result) >= 300:
                # 数据凑够300条就推送信号
                self.batch_data_ready.emit(self.result)
                self.result = []
    # 可选:新增方法,所有数据跑完后把剩余不足300条的数据也推送
    def flush_remaining(self):
        if self.result:
            self.batch_data_ready.emit(self.result)
            self.result = []

第二步:外部绑定处理逻辑

if __name__ == "__main__":
    import pandas as pd
    app = QApplication(sys.argv)
    systembox = Systembox()
    
    # 自定义数据处理函数,可直接调用其他模块的处理逻辑
    def process_batch_data(data_list):
        # 示例:转DataFrame,可自行扩展后续处理
        df = pd.DataFrame(data_list, columns=["你要的列名"])
        print(df)
        # 调用其他模块示例:from data_handler import save_to_db; save_to_db(df)
    
    # 绑定信号和处理函数,数据就绪会自动触发该函数
    systembox.batch_data_ready.connect(process_batch_data)
    
    systembox.comm_connect()
    while True:
        app.processEvents() # 必须加,确保Qt事件循环正常运行、信号能触发
        while systembox.remained_data == True:
            time.sleep(TR_REQ_TIME_INTERVAL)
            app.processEvents()
    # 所有请求跑完后刷剩余数据
    systembox.flush_remaining()
    sys.stdout.close()
    sys.exit(app.exec_())

方案2:回调函数方案(不需要依赖Qt信号)

通过传入自定义回调函数实现外部处理:

第一步:修改Systembox类定义

class Systembox(QAxWidget):
    # 初始化时传入回调函数
    def __init__(self, data_callback=None):
        super().__init__()
        self.data_callback = data_callback
        self._create_sysembox_instance()
        self._set_signal_slots()
        self.c = 0
        self.result = []
    # 其余原有方法保持不变,仅修改_AWS10000方法
    def _AWS10000(self, rqname, trcode):
        cnn = self._get_repeat_cnt(trcode, rqname)
        for i in range(cnn):
            date = self._comm_get_data(trcode, rqname, i, "DATA")
            self.c = self.c + 1
            if self.c > 299:
                self.remained_data = False
                self.c = self.c - 300
            if len(self.result) < 300:
                self.result.append(date)
            if len(self.result) >= 300:
                # 调用传入的回调函数
                if self.data_callback:
                    self.data_callback(self.result)
                self.result = []

第二步:外部传入处理函数

def process_batch_data(data_list):
    import pandas as pd
    df = pd.DataFrame(data_list, columns=["列名"])
    print(df)

if __name__ == "__main__":
    app = QApplication(sys.argv)
    # 实例化时直接传入处理函数
    systembox = Systembox(data_callback=process_batch_data)
    # 其余原有逻辑保持不变
    systembox.comm_connect()
    while True:
        app.processEvents()
        while systembox.remained_data == True:
            time.sleep(TR_REQ_TIME_INTERVAL)
            app.processEvents()
    sys.stdout.close()
    sys.exit(app.exec_())

内容的提问来源于stack exchange,提问作者yodoro

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.10.06 09:15:01