如何将 python Socket.IO 与 Qt 集成

Question

import sys
import socketio
from PyQt5.QtCore import QThread
from PyQt5.QtWidgets import QApplication


class ServerConnection(QThread):

    def __init__(self):
        QThread.__init__(self)
        self.sio = socketio.AsyncClient(reconnection=True, reconnection_attempts=3,
                   reconnection_delay=5, reconnection_delay_max=5, logger=True)

    'thread run function'
    def run(self) -> None:
        self.sio.connect(url="...",socketio_path="/...", transports="websocket")
        self.sio.on('connect', self.connect, namespace=None)
        self.sio.on('socket_connected', self.socket_connected, namespace=None)
        self.sio.on('connect_error', self.connect_error, namespace=None)
        self.sio.on('/client_Unlock', self.client_unlock_ack, namespace=None)

    # @sio.on('/client_unlock')
    'custom event from server, on receiving, this socketio thread needs to inform main GUI'
    async def client_unlock_ack(self, data):
            print("")
            'from here i want to call pyqt GUI main class function'

    # @sio.event
    'connection established status'
    def connect(self):
        print('Server Connection established!')

    # @sio.on("socket_connected")
    'socket connection status check'
    async def socket_connected(self, message):
        print("Socket Connected!", message)

    # @sio.event
    def connect_error(self, data):
        print('Connection error!', data)

    # @sio.on('disconnect' or 'socket_disconnected')
    def disconnect(self):
        print('Disconnected!')

'main function call'
def main():
    app = QApplication(sys.argv)
    con = ServerConnection()
    con.start()
    sys.exit(app.exec())


if __name__ == "__main__":
    main()

'GUI class is a QMainWindow application which processes further other functionalities of my application. Socketio thread connection is triggered from GUI main thread. Basically, GUI main started->Socketio thread called from GUI thread-> once we receive socketio client-unlock event inside socketio thread class-> from client_unlock_ack, i need to trigger a signal to GUI main thread to call a member function'。请指导。

Answer 1

没有必要使用线程，因为您可以使用诸如 asyncqt 之类的库，以便 Qt 与 asyncio 一起工作，避免使用 QThreads 增加不必要的复杂性

import asyncio
import sys
from functools import cached_property

from PyQt5.QtCore import pyqtSignal, QObject, Qt
from PyQt5.QtWidgets import QApplication, QLabel, QMainWindow
from asyncqt import QEventLoop

import socketio


class Client(QObject):
    connected = pyqtSignal()
    disconnected = pyqtSignal()
    error_ocurred = pyqtSignal(object, name="errorOcurred")
    data_changed = pyqtSignal(str, name="dataChanged")

    def __init__(self, parent=None):
        super().__init__(parent)

        self.sio.on("connect", self._handle_connect, namespace=None)
        self.sio.on("connect_error", self._handle_connect_error, namespace=None)
        self.sio.on("disconnect", self._handle_disconnect, namespace=None)
        self.sio.on("/client_Unlock", self.client_unlock_ack, namespace=None)

    @cached_property
    def sio(self):
        return socketio.AsyncClient(
            reconnection=True,
            reconnection_attempts=3,
            reconnection_delay=5,
            reconnection_delay_max=5,
            logger=True,
        )

    async def start(self):
        await self.sio.connect(url="...",socketio_path="/...", transports="websocket")

    def _handle_connect(self):
        self.connected.emit()

    def _handle_disconnect(self):
        self.disconnect.emit()

    def _handle_connect_error(self, data):
        self.error_ocurred.emit(data)

    def client_unlock_ack(self, data):
        self.data_changed.emit(data)


class View(QMainWindow):
    def __init__(self, parent=None):
        super().__init__(parent)
        self.label = QLabel(alignment=Qt.AlignCenter)
        self.setCentralWidget(self.label)
        self.resize(640, 480)

    def update_data(self, message):
        self.label.setText(message)


def main():
    app = QApplication(sys.argv)
    loop = QEventLoop(app)
    asyncio.set_event_loop(loop)

    view = View()
    view.show()

    client = Client()
    client.data_changed.connect(view.update_data)

    with loop:
        asyncio.ensure_future(client.start(), loop=loop)
        loop.run_forever()


if __name__ == "__main__":
    main()

如何将 python Socket.IO 与 Qt 集成

How to integrate python Socket.IO with Qt

python

multithreading

pyqt

socket.io