使用 asyncio 从 tkinter 切换到 PyQt5
Switching from tkinter to PyQt5 with asyncio
我正在将此 Tkinter 应用重写为 pyqt5。
我每次都使用 asyncio 来刷新 Tkinter window,因为我反复获取 url,并且每次都需要调用 self.update(),所以 window 不会'冻结。
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: Must construct a QApplication before a QWidget
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()
我添加了 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 应用程序中的 window。
删除了以下代码:
self.tasks.append(loop.create_task(self.updater()))
async def updater(self):
while await asyncio.sleep(0, True):
self.update()
我正在将此 Tkinter 应用重写为 pyqt5。 我每次都使用 asyncio 来刷新 Tkinter window,因为我反复获取 url,并且每次都需要调用 self.update(),所以 window 不会'冻结。
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: Must construct a QApplication before a QWidget
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()
我添加了 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 应用程序中的 window。
删除了以下代码:
self.tasks.append(loop.create_task(self.updater()))
async def updater(self):
while await asyncio.sleep(0, True):
self.update()