寻求PyQt5集成Python Socket程序并展示运行输出的示例

问题背景 #

已有一个可建立Socket连接的Python程序，其中RDES()函数运行正常，但将其集成到PyQt5界面时遇到阻碍，需要实现：

• 在PyQt5界面中导入该Socket程序并运行RDES()函数
• 实时展示RDES()中print语句的输出内容

解决方案要点 #

• 重定向sys.stdout：自定义输出流类，将print内容转发到PyQt文本控件
• 子线程运行RDES()：避免阻塞UI主线程导致界面卡顿
• 线程安全更新UI：通过信号槽机制在子线程中更新界面内容

完整示例代码 #

1. 原Socket程序（保存为socket_rdes.py）

# -*- coding: utf-8 -*-
import socket
import time
import sys

def RDES():
    T = 0
    while True:
        while T == 0:
            time.sleep(3)
            HOST = ''
            PORT2 = 20134
            
            s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            
            print('Socket created')
            
            try:
                l = s.bind((HOST, PORT2))
                print(l)
            except socket.error as msg:
                print(f'Bid failed: {msg}')
                sys.exit()
                
            print('Socket bind complete')
            s.listen(10)
            print('Socket now listening')
            conn, addr = s.accept()
            print(f'Connected to {addr[0]}:{str(addr[1])}')
            
            T2 = 'myturn'
            k = ''
            y = 1
            T = 1
            data = ''
            
        while T == 1:
            if T2 == 'myturn':
                data = conn.recv(2042)
                if not data:
                    print('Connection closed by peer')
                    T = 0
                    break
                line = data.decode('utf-8')
                print(f'Received data: {line}')
                # 可添加自定义业务逻辑
                conn.sendall(b'Received your message')

2. PyQt5界面程序

# -*- coding: utf-8 -*-
import sys
from PyQt5.QtWidgets import (QApplication, QMainWindow, QTextEdit, 
                             QPushButton, QVBoxLayout, QWidget)
from PyQt5.QtCore import QThread, pyqtSignal, pyqtSlot
from socket_rdes import RDES

# 自定义输出流，重定向stdout到UI信号
class Stream:
    def __init__(self, signal):
        self.signal = signal

    def write(self, text):
        self.signal.emit(text)

    def flush(self):
        pass

# 子线程类，运行RDES避免阻塞UI
class RDESThread(QThread):
    output_signal = pyqtSignal(str)

    def run(self):
        sys.stdout = Stream(self.output_signal)
        try:
            RDES()
        except Exception as e:
            self.output_signal.emit(f'Error occurred: {str(e)}')
        finally:
            sys.stdout = sys.__stdout__

# 主窗口
class MainWindow(QMainWindow):
    def __init__(self):
        super().__init__()
        self.init_ui()

    def init_ui(self):
        self.setWindowTitle('Socket RDES集成界面')
        self.setGeometry(100, 100, 800, 600)

        central_widget = QWidget()
        self.setCentralWidget(central_widget)
        layout = QVBoxLayout(central_widget)

        # 输出展示文本框
        self.output_text = QTextEdit()
        self.output_text.setReadOnly(True)
        layout.addWidget(self.output_text)

        # 启动按钮
        self.start_btn = QPushButton('启动RDES服务')
        self.start_btn.clicked.connect(self.start_rdes)
        layout.addWidget(self.start_btn)

        self.rdes_thread = None

    @pyqtSlot(str)
    def update_output(self, text):
        self.output_text.append(text)

    def start_rdes(self):
        if not (self.rdes_thread and self.rdes_thread.isRunning()):
            self.rdes_thread = RDESThread()
            self.rdes_thread.output_signal.connect(self.update_output)
            self.rdes_thread.start()
            self.start_btn.setText('服务已启动')
            self.start_btn.setEnabled(False)

if __name__ == '__main__':
    app = QApplication(sys.argv)
    window = MainWindow()
    window.show()
    sys.exit(app.exec_())

关键说明 #

• 输出重定向：通过Stream类替换sys.stdout，将print内容转为信号发送到UI线程
• 线程隔离：RDES()是无限循环逻辑，必须放在子线程运行，防止UI冻结
• 线程安全：使用pyqtSignal和pyqtSlot实现跨线程UI更新，避免直接在子线程操作控件
• 异常捕获：子线程中捕获异常并输出，便于调试排查问题

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