Python非阻滞输入

Question

我有一个应用程序，必须每4秒钟向我的API提出请求才能保持授权。我的问题是，它使用了大量input（），它阻止了线程以提出请求。我试图编写一个整个模块来对抗此模块，但仍在阻挡。我花了很长时间来尝试获得非阻滞输入。我尝试了几乎所有事情。这是我为非阻滞输入而写的课。 （它具有一个函数nbput（），函数几乎完全像Python的input（））

from pynput import keyboard
from pynput.keyboard import Key
import threading
import sys
from functools import partial

uinput = ''
lastKey = ''
modifierKey = ''


class NonBlockingInput:
    arg0 = ''

    def __init__(self):
        global listener
        listener = keyboard.Listener(on_press=self.on_press, on_release=self.on_release)
        print('1')

    def on_press(self, key):
        global lastKey
        global modifierKey
        try:
            sys.stdout.write(key.char)
            lastKey = key.char
        except AttributeError:
            if key == Key.space:
                lastKey = ' '
                sys.stdout.write(' ')
                return
            modifierKey = key

    def on_release(self, key):
        pass

    def nbinput(self, prompt):
        global uinput
        global listener
        global lastKey
        global modifierKey
        global arg0
        listener.start()
        sys.stdout.write(prompt)
        while True:
            if modifierKey == Key.enter:
                sys.stdout.write('\n')
                value_returned = partial(self.retrieved_data_func, arg0)
                break
            elif modifierKey == Key.backspace:
                spaceString = ''
                for _ in range(0, len(uinput)):
                    spaceString += ' '
                uinput = uinput[:-1]
                sys.stdout.write('\r')
                sys.stdout.write(spaceString)
                sys.stdout.write('\r')
                sys.stdout.write(uinput)
                modifierKey = ''
            else:
                uinput += lastKey
                lastKey = ''

    def retrieved_data_func(self):
        arg0 = 0
        return arg0


def nbput(prompt=''):
    global collectionThread
    nonBlockingInput = NonBlockingInput()
    collectionThread = threading.Thread(nonBlockingInput.nbinput(prompt))
    collectionThread.start()
    return NonBlockingInput.retrieved_data_func()


if __name__ == '__main__':
    print(nbput())```

Answer 1

我找到了一个解决方案：我创建auththread，然后调用main。

Answer 2

所有这些管道似乎都妨碍了不那么复杂的东西：

from time import sleep
from pynput import keyboard


def key_press(key):
    print(f'Pressed {key}')


def api_call():
    print('Making API call...')


listener = keyboard.Listener(on_press=key_press)
listener.start()


while True:
    api_call()
    sleep(4)

此脚本每4秒一次“调用API”，同时处理用户输入时。是，因此它应该在您使用的任何平台上工作。

由于这里的“ API调用”只是一个打印声明，因此无需花哨，但是您当然也可以不同步。 Sleep（4）就在那里模拟了4秒钟的阻止活动。

您回答说，您想要在课堂上全部，这在避免全球范围内有意义，但这不是您想一次（甚至一次）实例的班级。

这是一个示例，在避免一些复杂性的同时，似乎可以按照自己的意愿来做：

import sys
from time import sleep
from pynput import keyboard
import winsound
try:
    import msvcrt
except ImportError:
    msvcrt = None
    import termios


class NBI:
    text = ''
    listener = None

    @classmethod
    def start(cls):
        if cls.listener is None:
            cls.listener = keyboard.Listener(on_press=cls.key_press)
            cls.listener.start()
        else:
            raise Exception('Cannot start NBI twice.')

    @staticmethod
    def flush_input():
        if msvcrt is not None:
            while msvcrt.kbhit():
                msvcrt.getch()
        else:
            termios.tcflush(sys.stdin, termios.TCIOFLUSH)

    @classmethod
    def key_press(cls, key):
        if key == keyboard.Key.enter:
            sys.stdout.write('\n')
            cls.flush_input()
            cls.listener.stop()
            cls.listener = None
        if key == keyboard.Key.backspace:
            cls.text = cls.text[:-1]
            sys.stdout.write('\b \b')
            sys.stdout.flush()
        elif hasattr(key, 'char'):
            cls.text += key.char
            sys.stdout.write(key.char)
            sys.stdout.flush()
            cls.flush_input()


def api_call():
    winsound.Beep(440, 200)  # "API call"


NBI.start()
while NBI.listener:
    api_call()
    for _ in range(4):
        sleep(1)
        if not NBI.listener:
            break


print(f'You entered: {NBI.text}')