如何在Systembox类外部获取运行输出并写入pandas DataFrame
问题原因
你修改后的代码无输出是三个核心错误导致的:
- return逻辑无效:原
_AWS10000方法中调用self.print_result(self.result)时,你仅将结果return但没有任何外部代码接收这个返回值,返回值直接被丢弃 - 调用逻辑错误:你写在全局作用域的
aaa.print_result(result)里的result是未定义变量,完全不指向类内部的self.result属性 - 执行时机错误:你把实例化、打印逻辑写在
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
相关产品推荐
相关产品推荐

