QThread richtige Verwendung

[Dennis89]

Hallo zusammen,

ich benötige mal wieder eure Hilfe, bitte.
Folgendes möchte ich erreichen, der Benutzer klickt auf einen "Loggin"-Button, dann soll ein Fenster aufgehen, in dem steht dass der Benutzer seinen RFID-Chip an den Reader halten soll und dabei wird ein Countdown runter gezählt.
Das Lesen des RFID-Readers oder das Fenster muss in ein Thread ausgelagert werden. Ich weis nicht was davon sinnvoller ist, denn wenn der Chip ausgelesen ist, soll das Fenster geschlossen werden und wenn innerhalb des angegebenen Zeitraum kein Chip ausgelesen wird, dann soll das Fenster auch geschlossen werden.
Das Fenster mit dem Countdown und anschließendem schließen, blockierend im Haupt-Thread habe ich:

Code: Alles auswählen

import sys

from PySide6.QtCore import QThread, QTimer
from PySide6.QtWidgets import QApplication, QMessageBox, QWidget
from ui_form import Ui_FitForFactory

LOGIN_TEXT = "Zur Anmeldung Chip vorhalten\nVerbleibende Zeit:"


class LoginMessage(QMessageBox):
    def __init__(self, timeout=10):
        super(LoginMessage, self).__init__(parent=None)
        self.timer = QTimer()
        self.timeout = timeout
        self.timer.timeout.connect(self.tick)
        self.timer.setInterval(1000)
        self.setText(f"{LOGIN_TEXT} {self.timeout} s")
        self.setWindowTitle("Login")
        self.setDefaultButton(self.addButton(QMessageBox.Cancel))

    def showEvent(self, event):
        self.tick()
        self.timer.start()

    def tick(self):
        self.timeout -= 1
        if self.timeout > 0:
            self.setText(f"{LOGIN_TEXT} {self.timeout} s")
        else:
            self.timer.stop()
            self.defaultButton().animateClick()

    @staticmethod
    def create_new(timeout):
        window = LoginMessage(timeout)
        window.setIcon(QMessageBox.Information)
        return window.exec_()


class Example(QWidget):
    def __init__(self, parent=None):
        super().__init__(parent)
        self.ui = Ui_FitForFactory()
        self.ui.setupUi(self)
        self.ui.login_button.clicked.connect(self.login_process)

    def login_process(self):
        self.show_login()
        print("XX")
    
    @staticmethod
    def show_login():
        LoginMessage.create_new(timeout=10)


if __name__ == "__main__":
    app = QApplication(sys.argv)
    widget = Example()
    widget.show()
    sys.exit(app.exec())

`show_login` ist eine extra Methode, weil ich dachte, dass da noch mehr rein kommt, wegen der Kommunikation mit dem anderen Thread, wenn er mal existiert.

Meine Idee war, ich lasse das Anzeigen des Fensters in einem extra Thread laufen. Wenn ich `threading.Thread()` verwende, bekomme ich die Meldung, das ich `QThread` verwenden soll. Da finde ich aber gar keinen Einstieg. Es gibt zwar viele Beispiele mit `Worker(QObject)`-Klassen, nur bekomme ich das nicht umgesetzt, also so gar nicht, das ich auch kein Versuch zeigen kann.

Daher erst mal meine Fragen:
Welchen Teil würdet ihr auslagern?
Kann ich `QThread` nicht ähnlich wie `threading.Thread(target=self.show_login).start()` verwenden?
Dann muss ich vermutlich noch eine `Queue` mit ins Spiel bringen um zu kommunizieren und gegebenenfalls den Thread vorzeitig zubeenden. Das wäre aber mein zweiter Part.

Könnt ihr mir bitte ein paar Denkanstöße geben?

Vielen Dank und Grüße
Dennis

[__blackjack__]

@Dennis89: Ich glaube die GUI *muss* im Hauptthread laufen, womit die Frage was in einen eigenen Thread ausgelagert wird ja schon geklärt ist. Bei `QThread` braucht man keine Queue denn in Qt kommuniziert man mit dem Signal/Slot-Mechanismus. Was unter der Haube dann letztendlich eine „queued connection“ ist, wenn man es über Threads hinweg macht.

[grubenfox]

Das Lesen des RFID-Readers oder das Fenster muss in ein Thread ausgelagert werden. Ich weis nicht was davon sinnvoller ist, denn wenn der Chip ausgelesen ist, soll das Fenster geschlossen werden und wenn innerhalb des angegebenen Zeitraum kein Chip ausgelesen wird, dann soll das Fenster auch geschlossen werden.
Das Fenster mit dem Countdown und anschließendem schließen, blockierend im Haupt-Thread habe ich:

In Unkenntnis darüber was die Funktion zum Lesen des RFID-reader zurück liefert und wie lange sie eigentlich läuft, tue ich mich gedanklich schwer damit überhaupt einen zusätzlichen Thread einzubauen.
Ich gehe mal davon aus das die Lese-Routine unter einer Zehntelsekunde für einmal lesen benötigt und entweder die gelesenen Daten oder None zurückliefert.
Mit ein wenig zusätzlicher Initialisierung

Code: Alles auswählen

self.start_zeitpunkt = time.monotonic() # irgendwo beim Start des Countdowns
self.nfc_data = None # dies wohl in __init__
self.timer.setInterval(100)  # dies steht ja schon in __init__, hier nur ein kleinerer Wert

an passender Stelle(n) bin ich dann bei sowas:

Code: Alles auswählen

    def tick(self):
        if self.timeout > 0 and self.nfc_data is None:
            self.nfc_data = reader.get_data() # liefert entweder Daten oder None zurück
            if time.monotonic() - self.start_zeitpunkt >= 1:
                self.start_zeitpunkt = time.monotonic()
                self.timeout -= 1
                self.setText(f"{LOGIN_TEXT} {self.timeout} s")
        else:
            self.timer.stop()
            self.defaultButton().animateClick()

Wird also 10 pro Sekunde aufgerufen, liest den RFID-reader aus und zählt bei Bedarf (jeweils nach Ablauf einer Sekunde) den Countdown runter. Solange der Reader noch keine Daten geliefert hat und der Countdown noch größer Null ist, solange wird einfach weiter gemacht.

[grubenfox]

Ach, so langsam komme ich gedanklich doch zu einem eigenen extra-Thread... Wenn die Routine, die die Daten vom Reader liefern soll, erst mal hängt bis sie irgendwann Daten hat die sie zurückgeben kann, dann braucht's natürlich einen extra Thread. Sonst hängt ja gleich das komplette Programm...

[Dennis89]

Hallo,

danke für eure Antworten.
@grubenfox Ich habe hier gelernt, dass die Logik unabhängig von der GUI geschrieben wird, daher schon kann ich deinen Vorschlag nicht umsetzen. Und der Reader braucht nicht lange zum lesen, das ist richtig, aber er kann erst lesen, wenn der Benutzer seinen Chip gefunden hat und ihn in die Nähe des Readers bringt. Also ja das ist vergleichbar mit deinem zweiten Post.

Ich schau mal ob ich raus finden kann, ob ich `QThread` relativ einfach nutzen um mit den Signalen und Slots auch zu kommunizieren. Im Notfall könnte ich noch auf `queue.Queue()` und `threading.Thread()` ausweichen, weil die Meldung, die ich oben ansprach, bezog sich nur auf QT-Elemente. Wobei mit der Gedanke mit den Signalen schon besser gefällt.

Grüße
Dennis

[Dennis89]

Hallo zusammen,

ich habe das jetzt so gelöst:

Code: Alles auswählen

import sys
from json import loads
from pathlib import Path
from time import sleep

from PySide6.QtCore import QObject, QRegularExpression, QThread, QTimer, Signal
from PySide6.QtGui import QRegularExpressionValidator
from PySide6.QtWidgets import QApplication, QMessageBox, QWidget


from ui_form import Ui_FitForFactory

ROOT = Path(__file__).parent
MEMBERS_FILE = ROOT / "members.json"
TEST_NAME = "X"
TEST_ID = "abc123"

LOGIN_TEXT = "Zur Anmeldung Chip vorhalten\nVerbleibende Zeit:"


class Reader(QObject):
    finished = Signal()

    def __init__(self):
        super(Reader, self).__init__(parent=None)
        self.id = None

    def run(self):
        sleep(3)
        self.id = TEST_ID
        self.finished.emit()


class LoginMessage(QMessageBox):
    def __init__(self, event_, timeout=10):
        super(LoginMessage, self).__init__(parent=None)
        self.timer = QTimer()
        self.event_ = event_
        self.timeout = timeout
        self.timer.timeout.connect(self.tick)
        self.timer.setInterval(1000)
        self.setText(f"{LOGIN_TEXT} {self.timeout} s")
        self.setWindowTitle("Anmeldung")
        self.setDefaultButton(self.addButton(QMessageBox.Cancel))
        self.defaultButton().clicked.connect(self.stop_event)

    def stop_event(self):
        self.event_.quit()
        self.event_.wait()

    def showEvent(self, event):
        self.tick()
        self.timer.start()

    def tick(self):
        self.timeout -= 1
        if self.timeout > 0:
            self.setText(f"{LOGIN_TEXT} {self.timeout} s")
        if not self.event_.isRunning() or self.timeout <= 0:
            self.timer.stop()
            self.defaultButton().animateClick()

    @staticmethod
    def create_new(event_, timeout):
        window = LoginMessage(event_, timeout)
        window.setIcon(QMessageBox.Information)
        return window.exec()


class Example(QWidget):
    def __init__(self, parent=None):
        super().__init__(parent)
        self.ui = Ui_Example()
        self.ui.setupUi(self)
        validator = QRegularExpressionValidator(QRegularExpression(r"^[a-zA-Z\s]*$/"))
        self.ui.user_name.setValidator(validator)
        self.thread = QThread()
        self.reader = Reader()
        self.reader.moveToThread(self.thread)
        self.thread.started.connect(self.reader.run)
        self.reader.finished.connect(self.thread.quit)
        self.user_id = None
        self.ui.login_button.clicked.connect(self.login_process)


    def login_process(self):
        self.get_chip_id()
        if self.user_id is None:
            return
        members = load_members(MEMBERS_FILE)
        if is_member(members["members"], self.user_id):
            self.ui.hello_user.setText(
                f"Wilkommen {list(map(lambda x: x.get(self.user_id, ""), members["members"]))[0]}, viel Spass!"
            )
            self.ui.login_button.setEnabled(False)
            self.ui.logout_button.setEnabled(True)

    def get_chip_id(self):
        self.thread.start()
        LoginMessage.create_new(self.thread, timeout=10)
        self.user_id = self.reader.id
        
def load_members(file):
    return loads(file.read_text(encoding="UTF-8"))

def is_member(members, id_):
    return id_ in [id_ for member in members for id_ in member]



if __name__ == "__main__":
    app = QApplication(sys.argv)
    widget = Example()
    widget.show()
    sys.exit(app.exec())

Für Kritik bin ich dankbar.

Grüße
Dennis