使用使用 AsyncioScheduler 的 PyQt 应用程序，这让我很适应。该应用程序正在使用使用异步函数的 Kasa api 控制 Kasa SmartStrip。下面列出的代码有一个假的 Kasa 实现，但结果是相同的错误。任何帮助将不胜感激。
import sys
import time
import asyncio
from datetime import datetime
from PyQt5.QtWidgets import QDialog, QApplication, QVBoxLayout, QPushButton
from asyncqt import QEventLoop, asyncSlot
from apscheduler.schedulers.asyncio import AsyncIOScheduler
# Fake Kasa SmartStrip class.
# turn_on and turn_off are async functions, so use asyncio.sleep
# to implement an async function
class SmartStrip:
def __init__(self, address):
self.address = address
async def update(self):
await asyncio.sleep(1)
return ('Tick! The time is: %s' % datetime.now())
async def turn_on(self, child):
await asyncio.sleep(1)
async def turn_off(self, child):
await asyncio.sleep(1)
# Allow control (turn_on/turn_off) the individual sockets (upper/lower)
# in the wall plug
# sched_on/sched_off allows to set an on/off time.
class MyPlug(SmartStrip):
def __init__(self, ip_address):
super().__init__(ip_address)
async def initialize(self, child):
coroutine = self.update()
result = await coroutine
print(result)
async def turn_on(self, child):
coroutine = self.initialize(child)
result = await coroutine
async def turn_off(self, child):
coroutine = self.initialize(child)
result = await coroutine
async def sched_on(self, sched):
sched.add_job(MyPlug.turn_on, 'interval', seconds=3, args=[self, 0])
async def sched_off(self, sched):
sched.add_job(MyPlug.turn_off, 'interval', seconds=7, args=[self, 0])
# A QDialog which contains buttons to (turn on/ turn off) the (upper or
# lower) sockets. Also allows scheduling to turn on/off.
class Form(QDialog):
loop = None
scheduler = None
def __init__(self, loop=None):
super(Form, self).__init__()
self.loop = asyncio.get_event_loop()
self.scheduler = AsyncIOScheduler()
scheduler = self.scheduler
layout = QVBoxLayout()
self.setLayout(layout)
self.uon_name = "Upper On"
self.uoff_name = "Upper Off"
self.lon_name = "Lower On"
self.loff_name = "Lower Off"
self.sched_name = "Schedule"
self.quit_name = "Quit"
# Upper On
self.up_on = QPushButton("Upper On")
self.up_on.clicked.connect(lambda:self.whichbtn(self.up_on))
# Upper Off
self.up_off = QPushButton("Upper Off")
self.up_off.clicked.connect(lambda:self.whichbtn(self.up_off))
# Lower On
self.lo_on = QPushButton("Lower On")
self.lo_on.clicked.connect(lambda:self.whichbtn(self.lo_on))
# Lower Off
self.lo_off = QPushButton("Lower Off")
self.lo_off.clicked.connect(lambda:self.whichbtn(self.lo_off))
# Schedule
self.sched = QPushButton("Schedule")
self.sched.clicked.connect(lambda:self.whichbtn(self.sched))
#Quit
self.quit = QPushButton(self.quit_name)
self.quit.clicked.connect(lambda:self.quit_app())
layout.addWidget(self.up_on)
layout.addWidget(self.up_off)
layout.addWidget(self.lo_on)
layout.addWidget(self.lo_off)
layout.addWidget(self.sched)
layout.addWidget(self.quit)
self.setWindowTitle("Outlet Demo")
self.mp = MyPlug("10.0.0.31")
@asyncSlot()
async def whichbtn(self, btn):
if btn.text() == "Upper On":
await self.mp.turn_on(0)
if btn.text() =="Upper Off" :
await self.mp.turn_off(0)
if btn.text() == "Lower On":
await self.mp.turn_on(1)
if btn.text() == "Lower Off":
await self.mp.turn_off(1)
if btn.text() == "Schedule":
await self.mp.sched_on(self.scheduler)
self.scheduler.start()
print(self.scheduler)
print(self.scheduler.get_jobs())
def quit_app(self):
print("quit_app")
self.loop.stop()
print("Loop running ", self.loop.is_running())
self.loop.close()
print("Loop closed ", self.loop.is_closed())
self.close()
def main():
app = QApplication(sys.argv)
loop = QEventLoop(app)
asyncio.set_event_loop(loop)
ex = Form(loop)
ex.show()
with loop:
loop.run_forever()
if __name__ == '__main__':
main()
当我运行代码时，它会弹出一个带有上/下、开/关按钮的 Qt 窗口。按钮工作正常，按下时给我：
勾选！时间是：2023-02-20 20:26:14.390022
问题是“日程安排”按钮。我希望每 3 秒收到相同的输出，但得到以下信息：
TypeError: startTimer(self, int, timerType: Qt.TimerType = Qt.CoarseTimer): argument 1 has unexpected type 'float'
简而言之，我最初的问题是关于从 AsyncioScheduler 获取错误，特别是：
TypeError: startTimer(self, int, timerType: Qt.TimerType = Qt.CoarseTimer): argument 1 has unexpected type 'float'
我在 Qt 应用程序中使用调度程序，但没有取得太大成功。我终于让它工作了，修改后的代码如下...
import sys
import asyncio
import qasync
from datetime import datetime
from asyncqt import asyncSlot
from apscheduler.schedulers.asyncio import AsyncIOScheduler
from PyQt5.QtWidgets import QDialog, QApplication, QVBoxLayout, QPushButton
# Fake Kasa SmartStrip class.
# turn_on and turn_off are async functions, so use asyncio.sleep
# to implement
class SmartStrip:
def __init__(self, address):
self.address = address
async def update(self):
await asyncio.sleep(1)
print(f"update called at {datetime.now()}")
async def turn_on(self, child):
await asyncio.sleep(1)
print(f"turn on child", child, "called at", datetime.now())
async def turn_off(self, child):
await asyncio.sleep(1)
print(f"turn off child", child, "called at", datetime.now())
# Allow control (turn_on/turn_off) the individual sockets (upper/lower)
# in the wall plug
class MyPlug(SmartStrip):
def __init__(self, ip_address):
super().__init__(ip_address)
async def turn_on_upper(self):
await self.turn_on(0)
async def turn_on_lower(self):
await self.turn_on(1)
async def turn_off_upper(self):
await self.turn_off(0)
async def turn_off_lower(self):
await self.turn_off(1)
# The Qt goodies which contain buttons to (turn on/off)
# the (upper or lower) sockets. Also allows scheduling.
class Form(QDialog):
def __init__(self, loop=None):
super(Form, self).__init__()
self.loop = asyncio.get_event_loop()
asyncio.set_event_loop(self.loop)
self.scheduler = AsyncIOScheduler({'event_loop': self.loop}, timezone="America/New_York")
layout = QVBoxLayout()
self.setLayout(layout)
# Initialize
self.initialize = QPushButton("Initialize")
self.initialize.clicked.connect(self.init_device)
# Upper On
self.up_on = QPushButton("Upper On")
self.up_on.clicked.connect(self.turn_on_upper)
# Upper Off
self.up_off = QPushButton("Upper Off")
self.up_off.clicked.connect(self.turn_off_upper)
# Lower On
self.lo_on = QPushButton("Lower On")
self.lo_on.clicked.connect(self.turn_on_lower)
# Lower Off
self.lo_off = QPushButton("Lower Off")
self.lo_off.clicked.connect(self.turn_off_lower)
# Schedule
self.sched = QPushButton("Schedule")
self.sched.clicked.connect(self.schedule_mp)
# Quit
self.quit = QPushButton("Quit")
self.quit.clicked.connect(self.quit_app)
layout.addWidget(self.initialize)
layout.addWidget(self.up_on)
layout.addWidget(self.up_off)
layout.addWidget(self.lo_on)
layout.addWidget(self.lo_off)
layout.addWidget(self.sched)
layout.addWidget(self.quit)
self.setWindowTitle("Outlet Demo")
self.mp = MyPlug("10.0.0.31")
# The following functions with the asyncSlot decorator
# are called when a Qt button is clicked. Qt requires
# the asyncSlot decorator
@asyncSlot()
async def init_device(self):
print("Initialize device ************")
await self.mp.update()
print("******************************")
@asyncSlot()
async def turn_on_upper(self):
print("turn_on_upper ****************")
await self.mp.turn_on(0)
print("******************************")
@asyncSlot()
async def turn_on_lower(self):
print("turn_on_lower ****************")
await self.mp.turn_on(1)
print("******************************")
@asyncSlot()
async def turn_off_upper(self):
print("turn_off_upper ***************")
await self.mp.turn_off(0)
print("******************************")
@asyncSlot()
async def turn_off_lower(self):
print("turn_off_lower ***************")
await self.mp.turn_off(1)
print("******************************")
# The following functions are submitted as jobs to
# the scheduler. Same as the above functions but
# without the asyncSlot decorator.
async def sched_turn_on_upper(self):
await self.turn_on_upper()
async def sched_turn_on_lower(self):
await self.turn_on_lower()
async def sched_turn_off_upper(self):
await self.turn_off_upper()
async def sched_turn_off_lower(self):
await self.turn_off_lower()
def schedule_mp(self):
# Turn on every 5 seconds
self.scheduler.add_job(self.sched_turn_on_upper, 'interval', seconds=5)
# Turn off every 7 seconds
self.scheduler.add_job(self.sched_turn_off_upper, 'interval', seconds=7)
print("*****")
print("Scheduler jobs:\n", self.scheduler.print_jobs())
print("*****")
if not self.scheduler.running:
self.scheduler.start()
def quit_app(self):
if self.scheduler.running:
self.scheduler.shutdown()
self.loop.stop()
self.loop.close()
self.close()
def main():
app = QApplication(sys.argv)
# Since I'm running a Qt application, let's use a
# QEventLoop
loop = qasync.QEventLoop(app)
asyncio.set_event_loop(loop)
ex = Form(loop)
ex.show()
with loop:
loop.run_forever()
if __name__ == '__main__':
main()