'How to create timer without freezing all processes?
Want to describe my task firstly. My aim is to create Discord bot that sends a particular message exported from .txt in chat with a time interval of N seconds therefore I'm currently trying to use libraries such as discord, pyqt6 and asyncio.
I've tried to use time library but the problem is that time.sleep() delays all processes so the GUI becomes unusable and I cannot implement other constructions due to my lack of experience (I swear I tried my best). Hence I tried using asyncio and I found it suitable because it will not delay GUI.
My code includes a class with different methods and finally executes in main() but I cannot comprehend how am I supposed to execute method async def func() whereas every other functions does not demand async execution. Here is my code. I will be super grateful for every advise or connected literature because I've been browsing the internet for a couple days now and I still found barely anything to comprehend. Thank you.
class DiscordBot(QtWidgets.QMainWindow, widget.Ui_Form):
def __init__(self):
super().__init__()
self.setupUi(self)
self.start_button.clicked.connect(self.printer)
def __getitem__(self, item):
file = open('phrases.txt')
a = file.read().split('\n')
for item in range(len(a)):
yield a[item]
file.close()
async def printer(self):
self.listWidget.clear()
i = 0
a = DiscordBot.__getitem__(self, item=i)
for i in a:
self.listWidget.addItem(DiscordBot.__getitem__(self, item=i))
await asyncio.sleep(5)
def main():
app = QtWidgets.QApplication(sys.argv)
window = DiscordBot()
window.show()
app.exec()
asyncio.run(window.printer())
if __name__ == '__main__':
main()
Solution 1:[1]
You're not completely understanding magic methods use cases and python classes as well, I suppose. You don't have to call methods on class, self
is much more readable. __getitem__
is a method called when you access class using brackets: self[index]
or class_instance[index]
.
Now to threading. PyQt provides QThread
implementation for this purposes, you can create new thread, move worker to it and start the thread. Worker is any QObject
you like.
So you can rewrite the code like this:
from PyQt5.QtWidgets import (QMainWindow, QApplication, QWidget,
QPushButton, QVBoxLayout, QListWidget)
from PyQt5.QtCore import QObject, QThread, pyqtSignal
import time
import sys
class Printer(QObject):
''' Worker for long-running task
This worker will run in separate thread and perform required action
'''
sent = pyqtSignal(str)
@property # omit parentheses, allow easier caching (if you want)
def phrases(self):
''' Data to send '''
# context manager will close file for you
with open('/tmp/phrases.txt') as file:
# This way you can skip reading whole file into memory
# (it's important if file is large)
for line in file: # file is already iterable!
yield line.strip('\n')
def start_printer(self):
''' Write text '''
for phrase in self.phrases: # iterate over your generator
self.sent.emit(phrase)
time.sleep(5)
class DiscordBot(QMainWindow):
def __init__(self):
super().__init__()
self.setupUi()
self.start_button.clicked.connect(self.run_task)
def setupUi(self):
# I don't have your layout, so wrote something simple to reproduce.
# delete it when you're ready.
self.resize(300, 150)
self.centralWidget = QWidget()
self.setCentralWidget(self.centralWidget)
self.start_button = QPushButton("Start", self)
self.listWidget = QListWidget(self)
layout = QVBoxLayout()
layout.addWidget(self.start_button)
layout.addWidget(self.listWidget)
self.centralWidget.setLayout(layout)
def run_task(self):
''' Start your thread '''
self.thread = QThread() # new thread
self.worker = Printer() # initialize worker
self.worker.moveToThread(self.thread) # run in separate thread
self.thread.started.connect(self.worker.start_printer) # entry point
self.thread.started.connect(self.listWidget.clear)
self.worker.sent.connect(self.listWidget.addItem)
self.thread.start() # actually start the job
def main():
app = QApplication(sys.argv)
window = DiscordBot()
window.show()
app.exec()
# no need to do smth else, app will handle everything for you
if __name__ == '__main__':
main()
If you want more details and deeper knowledge (signals handlng, for instance, to communicate with thread and report progress), here's a good article.
Thanks to @musicamante: fixed thread-safety. We really should use signals to affect widgets instead of operating directly using reference, as user might do something with UI at the same time.
Solution 2:[2]
You can send blocking functions to other threads via;
from threading import Thread
Thread(target=myfunc).start()
or asyncio's executors;
import asyncio
from concurrent.futures import ProcessPoolExecutor
executor = ProcessPoolExecutor(5)
loop = asyncio.get_event_loop()
loop.run_in_executor(executor, myfunc)
but keep in mind both methods can not share data as they run on different threads.
Sources
This article follows the attribution requirements of Stack Overflow and is licensed under CC BY-SA 3.0.
Source: Stack Overflow
Solution | Source |
---|---|
Solution 1 | |
Solution 2 | CherrieP |