大佬教程收集整理的这篇文章主要介绍了如何将 python Socket.IO 与 Qt 集成，大佬教程大佬觉得挺不错的，现在分享给大家，也给大家做个参考。
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 类是一个 QMainWindow 应用程序，它处理我的应用程序的其他功能。 Socketio 线程连接是从 GUI 主线程触发的。基本上，GUI主启动->从GUI线程调用的Socketio线程->一旦我们在socketio线程类中接收到socketio客户端解锁事件->从clIEnt_unlock_ack，我需要触发一个信号到GUI主线程来调用一个成员函数'。请指导。
没有必要使用线程，因为您可以使用诸如 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 集成全部内容，希望文章能够帮你解决如何将 python Socket.IO 与 Qt 集成所遇到的程序开发问题。
如果觉得大佬教程网站内容还不错，欢迎将大佬教程推荐给程序员好友。
本图文内容来源于网友网络收集整理提供，作为学习参考使用，版权属于原作者。
如您有任何意见或建议可联系处理。小编QQ：384754419，请注明来意。