Python非阻塞输入

Question

我有一个应用程序，它必须每4秒向我的API发出一个get请求以保持授权。我的问题是，它使用了大量的input()，这是阻塞线程，使请求。我试图写一个完整的模块来解决这个问题，但它仍然阻塞。我花了很长时间试图得到一个非阻塞输入。我几乎在SO上试过所有的东西。这是我为非阻塞输入编写的类。(它有一个函数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)

这个脚本"调用API"每隔4秒,处理用户输入进来。它使用pyinput，因为你是，所以它应该在任何平台上工作。

因为这里的'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}')