时间戳

首页 > 解决方案 > 使用 asyncio 从 tkinter 切换到 PyQt5

python - 使用 asyncio 从 tkinter 切换到 PyQt5

问题描述

我正在将此 Tkinter 应用程序重写为 pyqt5。我每次都使用 asyncio 刷新 Tkinter 窗口，因为我反复获取一个 url，并且每次都需要调用 self.update() 以使窗口不会冻结。

from tkinter import *
import asyncio
import aiohttp
    
class App(Tk):
    def __init__(self, loop):
        super().__init__()
        self.title("GUI Client")
        self.loop = loop
        self.protocol("WM_DELETE_WINDOW", self.close)
        self.tasks = []
        self.tasks.append(loop.create_task(self.rotator(1/60)))
        self.tasks.append(loop.create_task(self.updater()))

    async def rotator(self, interval):
        while await asyncio.sleep(interval, True):
            self.response = await get()
            print(self.response)
            
    async def updater(self):
        while await asyncio.sleep(0, True):                
            self.update()

    def close(self):
        for task in self.tasks:
            task.cancel()
        self.loop.stop()
        self.destroy()   

async def get ():
    async with aiohttp.ClientSession() as session:
        async with session.post('https://jsonplaceholder.typicode.com/posts', json={
    'title':'foo',
    'body':'bar',
    'userId':1
}, headers = {'Content-type': 'application/json; charset=UTF-8'}) as resp:
            data = await resp.json()
            return data    

loop = asyncio.get_event_loop()
app = App(loop)
loop.run_forever()
loop.close()

这是 pyqt5 gui，但它不起作用，我不知道我是否很好地使用了导入，因为我得到了

QWidget：必须在 QWidget 之前构造一个 QApplication

from PyQt5 import QtCore, QtGui, QtWidgets
import asyncio

class App(QtWidgets.QMainWindow):

    def __init__(self, loop):
        super().__init__()
        self.setGeometry(50, 50, 500, 300)
        self.setWindowTitle("GUI")
        self.setWindowIcon(QtGui.QIcon('pythonlogo.png'))
        self.show()
        self.loop = loop
        self.protocol("WM_DELETE_WINDOW", self.close)
        self.tasks = []
        self.tasks.append(loop.create_task(self.updater()))

    async def updater(self):
        while await asyncio.sleep(0, True):                
            self.update()
            
    def close(self):
        for task in self.tasks:
            task.cancel()
        self.loop.stop()
        self.destroy() 

loop = asyncio.get_event_loop()
app = App(loop)
loop.run_forever()
loop.close()

标签： pythonpython-3.xpyqt5python-asyncioaiohttp

解决方案

我添加了quamash包来结合 pyQt5 和 asyncio。

from PyQt5.QtWidgets import *
from PyQt5.QtCore import *
import quamash
import aiohttp
import asyncio

class App(QWidget):

    run = 0
    response = ''
    def __init__(self, loop):
        super().__init__()

        btn = QPushButton('Start', self)
        btn.resize(btn.sizeHint())
        btn.move(50, 50)
        btn.clicked.connect(self.start)

        self.setGeometry(300, 300, 300, 200)
        self.setWindowTitle('GUI')    
        self.show()
        self.loop = loop
        self.tasks = []
        self.tasks.append(loop.create_task(self.rotator()))
        
    async def rotator(self):
        while await asyncio.sleep(0, True):
            if (self.run == 1):
                self.response = await get()
                print(self.response)
                  
    def start (self):
        self.run = 1            
    
async def get ():
    async with aiohttp.ClientSession() as session:
        async with session.post('https://jsonplaceholder.typicode.com/posts', json={
        'title':'foo',
        'body':'bar',
        'userId':1
    }, headers = {'Content-type': 'application/json; charset=UTF-8'}) as resp:
            data = await resp.json()
            return data
      
app = QApplication(sys.argv)
loop = quamash.QEventLoop(app)
asyncio.set_event_loop(loop)

with loop:
    window = App(loop)
    window.show()
    loop.run_forever()

无需更新 pyQt5 应用程序中的窗口。

删除以下代码：

self.tasks.append(loop.create_task(self.updater()))

    async def updater(self):
        while await asyncio.sleep(0, True):                
            self.update()

推荐阅读