Das deutsche Python-Forum

Hallo,

ich möchte auf einem Rechner direkt (und ohne GUI) die Maus-Tasten per USB-Schnittstelle auslesen.
Das klappt soweit auch ganz gut.
Allerdings habe ich das Problem, dass der Code innerhalb meines while-Blocks nur ausgeführt zu werden scheint, wenn sich Werte im Buffer befinden.
Somit wird der eingebaute Heartbeat nicht dauerhaft ausgeführt.
Wir kann man das Programm umschreiben, damit ich dauerhaft auch anderen Code ausführen kann?

Vielen Dank im voraus für eure Antworten!
Sebastian

Code: Alles auswählen

import struct
import time

path = "/dev/input/by-id/usb-_mini_keyboard-if01-event-mouse"

file = open(path, "rb")

while True:

    # Read the data from the USB interface:
    byte = file.read(16)
    (type, code, value) = struct.unpack_from('hhi', byte, offset=8)

    print("Type: " + str(type) + " Code: " + str(code) + " Value: " + str(value))

    # Detect mouse presses:
    if type == 1 and value == 1:
        if code == 272:
            print("LEFT PRESS")
        if code == 273:
            print("RIGHT PRESS")

    time.sleep(1)
    print("++++ HEARTBEAT ++++")

Mit Queues und Threads kann das Problem gelöst werden.

Die Funktion put_events macht das, was du auf Modulebene machst.
Anstatt die Daten einfach im Terminal auszugeben, werden sie in eine Queue gepackt.
Eine andere Funktion holt aus der gleichen Queue die Daten ab.

Beide Funktionen werden als Threads gestartet.
D.h. wenn die Queue leer ist, würde die Funktion consumer bei queue.get() solange blockieren, bis wieder ein Objekt verfügbar ist.

Kleinere Optimierungen: bytearray als buffer

Die heartbeat-funktion macht nichts, außer Heartbeat im Terminal auszugeben.
Selbst wenn keine Daten mehr kommen, läuft heartbeat weiter.

Code: Alles auswählen

import struct
import time
from queue import Queue
from threading import Thread


def heartbeat():
    def worker():
        while True:
            time.sleep(1)
            print("++ Heartbeat ++")

    thread = Thread(target=worker, daemon=True)
    thread.start()
    return thread


def put_events(filepath, queue):
    buffer = bytearray(16)
    with open(filepath, "rb") as file:
        while True:
            size_read = file.readinto(buffer)
            if size_read != 16:
                continue
            (type, code, value) = struct.unpack_from("hhi", buffer, offset=8)

            # um die Daten aus der laufenden Funktion zu bekommen,
            # muss man mit queues arbeiten
            queue.put((type, code, value))


def consumer(queue):
    while True:
        type, code, value = queue.get()
        print(f"Type: {type} Code: {code} Value: {value}")


if __name__ == "__main__":
    device = "/dev/input/by-id/usb-Logitech_USB_Receiver-if02-event-kbd"

    queue = Queue(maxsize=10)

    heartbeat()

    event_thread = Thread(target=put_events, args=(device, queue))
    consumer_thread = Thread(target=consumer, args=(queue,))

    event_thread.start()
    consumer_thread.start()

@Hypersim: Erst einmal allgemeine Anmerkungen zum Quelltext:

Auf Modulebene sollte nur Code stehen der Konstanten, Funktionen, und Klassen definiert. Das Hauptprogramm steht üblicherweise in einer Funktion die `main()` heisst.

Dateien sollte man wo es möglich ist mit der ``with``-Anweisung verwenden, damit die auch wieder geschlossen werden, egal warum das Programm endet.

`byte` ist ein etwas irreführender Name für 16 Bytes. Da `bytes` schon der Name eines Datentyps ist, würde sich hier als allgemeiner Name `data` anbieten. Oder man spart sich das gleich komplett dem Wert einen Namen zu geben.

`type` ist auch ein Name eines Typs/einer Funktion. Den sollte man nicht für etwas anderes verwenden. Eine Konvention wenn man einen bereits von Python vorbelegten Namen oder ein Schlüsselwort als Namen verwenden will, ist das anhängen eines einzelnen Unterstrichs.

Das zusammenstückeln von Zeichenketten und Werten mittels ``+`` und `str()` ist eher BASIC als Python. Dafür gibt es die `format()`-Methode auf Zeichenketten und f-Zeichenkettenliterale.

Wenn `code` einen Wert hat, kann es nicht gleichzeitig einen anderen haben, man braucht also nicht weiterprüfen wenn man eine Übereinstimmung gefunden hat. Das zweite ``if code …`` sollte also ein ``elif code …`` sein.

Zwischenstand:

Code: Alles auswählen

#!/usr/bin/env python3
import struct
import time

PATH = "/dev/input/by-id/usb-_mini_keyboard-if01-event-mouse"


def main():
    with open(PATH, "rb") as usb_device:
        while True:
            type_, code, value = struct.unpack_from(
                "hhi", usb_device.read(16), offset=8
            )
            print(f"Type: {type_} Code: {code} Value: {value}")
            
            if type_ == 1 and value == 1:
                if code == 272:
                    print("LEFT MOUSE BUTTON PRESS")
                elif code == 273:
                    print("RIGHT MOUSE BUTTON PRESS")

            time.sleep(1)
            print("++++ HEARTBEAT ++++")


if __name__ == "__main__":
    main()

Nebenläufige ausführung kann man beispielsweise durch das `threading`-Modul erreichen. Hier am einfachsten in dem man die Heartbeat-Ausgabe in einen eigenen Thread verlegt:

Code: Alles auswählen

#!/usr/bin/env python3
import struct
import time
from threading import Thread

PATH = "/dev/input/by-id/usb-_mini_keyboard-if01-event-mouse"


def do_heartbeat():
    while True:
        time.sleep(1)
        print("++++ HEARTBEAT ++++")


def main():
    Thread(target=do_heartbeat, daemon=True).start()

    with open(PATH, "rb") as usb_device:
        while True:
            type_, code, value = struct.unpack_from(
                "hhi", usb_device.read(16), offset=8
            )
            print(f"Type: {type_} Code: {code} Value: {value}")

            if type_ == 1 and value == 1:
                if code == 272:
                    print("LEFT MOUSE BUTTON PRESS")
                elif code == 273:
                    print("RIGHT MOUSE BUTTON PRESS")


if __name__ == "__main__":
    main()

Wow, vielen Dank für die ausführliche Antworten und die gleichzeitige Optimierung des Codes!
Da werde ich mich erstmal durchwühlen, der Code ist definitiv auf einer höheren Stufe und für mich als Anfänger nicht auf Anhieb verständlich

Gerade Klassen sind noch ein Buch mit sieben Siegeln für mich.
Ich bin aber bereit, zu lernen, und freue mich sehr über eure super Erläuterungen, vielen Dank nochmal!

Falls es noch andere Anfänger gibt, die diesen Thread sehen und über das 'if __name__...' Konstrukt stolpern:
Mir war bis jetzt nie klar, was es damit auf sich hat. Hier ist eine super Erklärung dafür:
https://www.data-science-architect.de/__name____main__/

Bin übrigens in der Zwischenzeit noch auf eine andere Lösung gestoßen:

Code: Alles auswählen

os.set_blocking(file.fileno(), False)

und dann

Code: Alles auswählen

        byte = file.read(16)
        if byte:

Funktioniert auch, aber wo wir bei Optimierungen sind, welche Lösung wäre zu bevorzugen bzw. welche Vor-/ Nachteile hätten diese?

@Hypersim: Wo/Wie gehst Du denn jetzt damit um wenn da irgendwas zwischen 1 bis 16 Bytes gelesen werden können? Das musst Du ja jetzt sicherstellen das da nicht nur 0 oder 16 Bytes bei dem `read()`-Aufruf gelesen werden sondern auch beliebiges dazwischen. Das zieht ja Code nach sich den man da jetzt extra schreiben muss.

Das Arbeiten mit Threads ist kompliziert und hat viele Stolperstellen. Das Arbeiten mit nicht-blockierenden Reads ist auch kompliziert.
Was besser ist, kommt auf den konkreten Anwendungsfall an. Für Dich mit diesem Heartbeat mit zwei völlig unabhängigen Prozessen, sind Threads besser.
Wobei ich bezweifle, dass das mit dem Heartbeat wirklich Dein Anwendungsfall ist.

@blackjack: In meinem Anwendungsfall frage ich Mausbewegungen ab, das hatte ohne Abfrage der Größe geklappt. Offensichtlich, weil die Rückgabe immer eine Mindestzahl an Daten enthält. Ich muss nur sicherstellen, dass keine leeren Daten versucht werden zu lesen.

@sirius: Den Heartbeat habe ich nur zur Vereinfachung eingebaut. Im richtigen Programm soll der GPIOs steuern.
Konkret arbeite ich mit einem Raspi 3, an dem der Sensor einer optischen Maus angeschlossen ist. Diese soll Bewegungen einer Achse eines 3D-Druckers registrieren, die wiederum eine Kamera per Raspi triggern soll.
Will damit eine Zeitraffer-Aufnahme eines 3D-Druckes machen. Durch die Konstruktion ist diese dann synchronisiert mit der Z-Achse des Druckers, was später dann ganz nett aussieht (hoffe ich

Hier zur Veranschaulichung: https://www.youtube.com/watch?v=Xoj0KeiY9OE

@Hypersim: Nein eben nicht immer wenn Du das auf Non-Blocking stellst. Das ist ja daran dann gerade das Problem an dem `read(16)` dass das eben auch weniger liefern kann und das musst Du entsprechend im Code berücksichtigen. Das musst Du bei blockierendem Aufruf nicht, weil da dann `read()` solange blockiert bis 16 Bytes zusammen gekommen sind (oder die Datei zuende ist). Und es kann durchaus sein, dass bei nicht-blockierendem Lesen das in 99% der Fälle 16 Bytes liefert, aber das ist eben nicht garantiert, und Du musst damit umgehen können wenn das nicht der Fall ist.

Das deutsche Python-Forum

While-Schleife wird wegen 'file.read' nicht mehr ausgeführt

While-Schleife wird wegen 'file.read' nicht mehr ausgeführt

Re: While-Schleife wird wegen 'file.read' nicht mehr ausgeführt

Re: While-Schleife wird wegen 'file.read' nicht mehr ausgeführt

Re: While-Schleife wird wegen 'file.read' nicht mehr ausgeführt

Re: While-Schleife wird wegen 'file.read' nicht mehr ausgeführt

Re: While-Schleife wird wegen 'file.read' nicht mehr ausgeführt

Re: While-Schleife wird wegen 'file.read' nicht mehr ausgeführt

Re: While-Schleife wird wegen 'file.read' nicht mehr ausgeführt

Re: While-Schleife wird wegen 'file.read' nicht mehr ausgeführt

Re: While-Schleife wird wegen 'file.read' nicht mehr ausgeführt