-2

我正在使用 socketserver 和 pyqt5 制作一个 udpserver，如下所示：

class UdpServer(socketserver.BaseRequestHandler):
    signal_msg=pyqtSignal(str)
    def handle(self):
        data =self.request[0]
        signal_msg.emit(data.decode('utf-8'))

然后我使用 qtbutton 开始这个类，像这样：

udpserver=socketserver.ThreadingUDPServer(('127.0.0.1',8081),UdpServer)
def udpthr(): 
    udpserver.serve_forever()
threading.Thread(target=udpthr,args=()).start()

问题是，如何在 pyqt5 中使用“signal_msg”？由于“UdpServer”没有实例，我无法访问类中的变量。

4

1 回答 1

1

The signals only make sense in the QObjects since they are used to obtain information about the type of information that you want to transmit and implement all the internal logic. Considering this then you must create a QObject with a signal that has the signature of the information you send, then pass that object as an attribute of the server so that it is used in the handle method.

import sys
import socketserver
import threading

from PyQt5 import QtCore, QtGui, QtWidgets


class Bridge(QtCore.QObject):
    dataChanged = QtCore.pyqtSignal(str)


class UDPHandler(socketserver.BaseRequestHandler):
    def handle(self):
        data = str(self.request[0], "ascii")
        self.server.bridge.dataChanged.emit(data)


class MainWindow(QtWidgets.QMainWindow):
    def __init__(self, parent=None):
        super().__init__(parent)

        self.label = QtWidgets.QLabel(alignment=QtCore.Qt.AlignCenter)
        self.label.setFont(QtGui.QFont("Arial", 20))

        self.setCentralWidget(self.label)
        self.resize(640, 480)

    @QtCore.pyqtSlot(str)
    def update_text(self, text):
        self.label.setText(text)


def main():
    app = QtWidgets.QApplication(sys.argv)

    bridge = Bridge()

    w = MainWindow()
    w.show()

    HOST, PORT = "127.0.0.1", 8081
    server = socketserver.ThreadingUDPServer((HOST, PORT), UDPHandler)
    ip, port = server.server_address
    print(ip, port)
    server.bridge = bridge

    bridge.dataChanged.connect(w.update_text)

    server_thread = threading.Thread(target=server.serve_forever)
    server_thread.daemon = True
    server_thread.start()

    ret = app.exec_()

    server.server_close()

    sys.exit(ret)


if __name__ == "__main__":
    main()
于 2021-03-04T04:45:46.503 回答