Non-blocking console input?

小开

最佳答案

对于 Windows，仅使用控制台，使用 msvcrt模块:

import msvcrt


num = 0
done = False
while not done:
print(num)
num += 1


if msvcrt.kbhit():
print "you pressed",msvcrt.getch(),"so now i will quit"
done = True

对于 Linux，这个文章描述了以下解决方案，它需要 termios模块:

import sys
import select
import tty
import termios


def isData():
return select.select([sys.stdin], [], [], 0) == ([sys.stdin], [], [])


old_settings = termios.tcgetattr(sys.stdin)
try:
tty.setcbreak(sys.stdin.fileno())


i = 0
while 1:
print(i)
i += 1


if isData():
c = sys.stdin.read(1)
if c == '\x1b':         # x1b is ESC
break


finally:
termios.tcsetattr(sys.stdin, termios.TCSADRAIN, old_settings)

对于跨平台，或者如果你也想要一个 GUI，你可以使用 Pygame:

import pygame
from pygame.locals import *


def display(str):
text = font.render(str, True, (255, 255, 255), (159, 182, 205))
textRect = text.get_rect()
textRect.centerx = screen.get_rect().centerx
textRect.centery = screen.get_rect().centery


screen.blit(text, textRect)
pygame.display.update()


pygame.init()
screen = pygame.display.set_mode( (640,480) )
pygame.display.set_caption('Python numbers')
screen.fill((159, 182, 205))


font = pygame.font.Font(None, 17)


num = 0
done = False
while not done:
display( str(num) )
num += 1


pygame.event.pump()
keys = pygame.key.get_pressed()
if keys[K_ESCAPE]:
done = True

小开

在 Linux 上，这里有一个 mizipzor 代码的重构，它使这个过程变得更加容易，以防您必须在多个地方使用这个代码。

import sys
import select
import tty
import termios


class NonBlockingConsole(object):


def __enter__(self):
self.old_settings = termios.tcgetattr(sys.stdin)
tty.setcbreak(sys.stdin.fileno())
return self


def __exit__(self, type, value, traceback):
termios.tcsetattr(sys.stdin, termios.TCSADRAIN, self.old_settings)




def get_data(self):
if select.select([sys.stdin], [], [], 0) == ([sys.stdin], [], []):
return sys.stdin.read(1)
return False

下面是使用方法: 这段代码将打印一个计数器，该计数器将一直增长，直到您按下 ESC。

with NonBlockingConsole() as nbc:
i = 0
while 1:
print i
i += 1
if nbc.get_data() == '\x1b':  # x1b is ESC
break

小开

下面是一个在 linux 和 windows 下使用独立线程运行的解决方案:

import sys
import threading
import time
import Queue


def add_input(input_queue):
while True:
input_queue.put(sys.stdin.read(1))


def foobar():
input_queue = Queue.Queue()


input_thread = threading.Thread(target=add_input, args=(input_queue,))
input_thread.daemon = True
input_thread.start()


last_update = time.time()
while True:


if time.time()-last_update>0.5:
sys.stdout.write(".")
last_update = time.time()


if not input_queue.empty():
print "\ninput:", input_queue.get()


foobar()

小开

这是我见过的最棒的解决方案¹。粘贴在这里，以防链接中断:

#!/usr/bin/env python
'''
A Python class implementing KBHIT, the standard keyboard-interrupt poller.
Works transparently on Windows and Posix (Linux, Mac OS X).  Doesn't work
with IDLE.


This program is free software: you can redistribute it and/or modify
it under the terms of the GNU Lesser General Public License as
published by the Free Software Foundation, either version 3 of the
License, or (at your option) any later version.


This program is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
GNU General Public License for more details.


'''


import os


# Windows
if os.name == 'nt':
import msvcrt


# Posix (Linux, OS X)
else:
import sys
import termios
import atexit
from select import select




class KBHit:


def __init__(self):
'''Creates a KBHit object that you can call to do various keyboard things.
'''


if os.name == 'nt':
pass


else:


# Save the terminal settings
self.fd = sys.stdin.fileno()
self.new_term = termios.tcgetattr(self.fd)
self.old_term = termios.tcgetattr(self.fd)


# New terminal setting unbuffered
self.new_term[3] = (self.new_term[3] & ~termios.ICANON & ~termios.ECHO)
termios.tcsetattr(self.fd, termios.TCSAFLUSH, self.new_term)


# Support normal-terminal reset at exit
atexit.register(self.set_normal_term)




def set_normal_term(self):
''' Resets to normal terminal.  On Windows this is a no-op.
'''


if os.name == 'nt':
pass


else:
termios.tcsetattr(self.fd, termios.TCSAFLUSH, self.old_term)




def getch(self):
''' Returns a keyboard character after kbhit() has been called.
Should not be called in the same program as getarrow().
'''


s = ''


if os.name == 'nt':
return msvcrt.getch().decode('utf-8')


else:
return sys.stdin.read(1)




def getarrow(self):
''' Returns an arrow-key code after kbhit() has been called. Codes are
0 : up
1 : right
2 : down
3 : left
Should not be called in the same program as getch().
'''


if os.name == 'nt':
msvcrt.getch() # skip 0xE0
c = msvcrt.getch()
vals = [72, 77, 80, 75]


else:
c = sys.stdin.read(3)[2]
vals = [65, 67, 66, 68]


return vals.index(ord(c.decode('utf-8')))




def kbhit(self):
''' Returns True if keyboard character was hit, False otherwise.
'''
if os.name == 'nt':
return msvcrt.kbhit()


else:
dr,dw,de = select([sys.stdin], [], [], 0)
return dr != []




# Test
if __name__ == "__main__":


kb = KBHit()


print('Hit any key, or ESC to exit')


while True:


if kb.kbhit():
c = kb.getch()
if ord(c) == 27: # ESC
break
print(c)


kb.set_normal_term()

_{¹
由 Simon D. Levy制作，是他根据 GNU宽通用公共许可证编写并发行的软件的编制的一部分}

小开

With python3.3 and above you can use the asyncio module as mentioned in this answer. You will have to re factor your code though to work with asyncio. 使用 python syncio.create _ server 实例提示用户输入

小开

我觉得诅咒图书馆能帮上忙。

import curses
import datetime


stdscr = curses.initscr()
curses.noecho()
stdscr.nodelay(1) # set getch() non-blocking


stdscr.addstr(0,0,"Press \"p\" to show count, \"q\" to exit...")
line = 1
try:
while 1:
c = stdscr.getch()
if c == ord('p'):
stdscr.addstr(line,0,"Some text here")
line += 1
elif c == ord('q'): break


"""
Do more things
"""


finally:
curses.endwin()

小开

我会照米奇说的做，但我会用 unicurses代替正常的诅咒。 Unicurses是通用的(适用于所有或至少几乎所有操作系统)

小开

我最喜欢的获取非阻塞输入的方法是在线程中使用 python input () :

import threading


class KeyboardThread(threading.Thread):


def __init__(self, input_cbk = None, name='keyboard-input-thread'):
self.input_cbk = input_cbk
super(KeyboardThread, self).__init__(name=name)
self.start()


def run(self):
while True:
self.input_cbk(input()) #waits to get input + Return


showcounter = 0 #something to demonstrate the change


def my_callback(inp):
#evaluate the keyboard input
print('You Entered:', inp, ' Counter is at:', showcounter)


#start the Keyboard thread
kthread = KeyboardThread(my_callback)


while True:
#the normal program executes without blocking. here just counting up
showcounter += 1

操作系统独立，只有内部库，支持多字符输入

小开

下面是围绕上述解决方案之一的类包装器:

#!/usr/bin/env python3


import threading


import queue


class NonBlockingInput:


def __init__(self, exit_condition):
self.exit_condition = exit_condition
self.input_queue = queue.Queue()
self.input_thread = threading.Thread(target=self.read_kbd_input, args=(), daemon=True)
self.input_thread.start()


def read_kbd_input(self):
done_queueing_input = False
while not done_queueing_input:
console_input = input()
self.input_queue.put(console_input)
if console_input.strip() == self.exit_condition:
done_queueing_input = True


def input_queued(self):
return_value = False
if self.input_queue.qsize() > 0:
return_value = True
return return_value


def input_get(self):
return_value = ""
if self.input_queue.qsize() > 0:
return_value = self.input_queue.get()
return return_value


if __name__ == '__main__':


NON_BLOCK_INPUT = NonBlockingInput(exit_condition='quit')


DONE_PROCESSING = False
INPUT_STR = ""
while not DONE_PROCESSING:
if NON_BLOCK_INPUT.input_queued():
INPUT_STR = NON_BLOCK_INPUT.input_get()
if INPUT_STR.strip() == "quit":
DONE_PROCESSING = True
else:
print("{}".format(INPUT_STR))

小开

因为我发现以上答案中的一个很有用，这里有一个类似方法的例子。这段代码在接受输入时创建节拍器效果。

区别在于这段代码使用了一个闭包而不是一个类，这让我感觉更直接一些。此示例还包含一个标志，通过 my_thread.stop = True杀死线程，但不使用全局变量。我是通过(ab)使用 python 函数是对象这一事实来实现这一点的，因此可以对它们进行修补，甚至可以从它们自身进行修补。

注意: 停止线程应谨慎进行。如果您的线程拥有需要某种清理进程的数据，或者如果该线程产生了自己的线程，那么这种方法将毫不客气地终止这些进程。

# Begin metronome sound while accepting input.
# After pressing enter, turn off the metronome sound.
# Press enter again to restart the process.


import threading
import time
import winsound  # Only on Windows


beat_length = 1  # Metronome speed




def beat_thread():
beat_thread.stop = False  # Monkey-patched flag
frequency, duration = 2500, 10
def run():  # Closure
while not beat_thread.stop:  # Run until flag is True
winsound.Beep(frequency, duration)
time.sleep(beat_length - duration/1000)
threading.Thread(target=run).start()




while True:
beat_thread()
input("Input with metronome. Enter to finish.\n")
beat_thread.stop = True  # Flip monkey-patched flag
input("Metronome paused. Enter to continue.\n\n")

小开

If you just want a single "escape" from a loop, you can intercept the Ctrl-C signal.

这是跨平台的，非常简单！

import signal
import sys


def signal_handler(sig, frame):
print('You pressed Ctrl+C!')
sys.exit(0)


signal.signal(signal.SIGINT, signal_handler)
while True:
# do your work here

小开

回到最初的问题。

我也正在学习 Python，它花费了我很多文档和例子阅读和头部裂纹... 但我认为我达到了一个简单，简单，短和兼容的解决方案... 只使用输入，列表和线程

'''
what i thought:
- input() in another thread
- that were filling a global strings list
- strings are being popped in the main thread
'''


import threading


consoleBuffer = []


def consoleInput(myBuffer):
while True:
myBuffer.append(input())
 

threading.Thread(target=consoleInput, args=(consoleBuffer,), daemon=True).start() # start the thread


import time # just to demonstrate non blocking parallel processing


while True:
time.sleep(2) # avoid 100% cpu
print(time.time()) # just to demonstrate non blocking parallel processing
while consoleBuffer:
print(repr(consoleBuffer.pop(0)))

直到这是我找到的最简单和兼容的方法，注意默认的 stdin stdout 和 stderr 共享同一个终端，所以如果在你输入的时候有东西在控制台上打印出来，那么你输入的“本地回声”可能看起来不一致，但是在按下输入键之后，输入的字符串被很好地接收到了... 如果你不想/喜欢这种行为，找到一种方法来分离输入/输出区域，比如重定向，或者尝试另一种解决方案,。

额外的好处: ctrl-c按键可以很容易地处理

try:
# do whatever
except KeyboardInterrupt:
print('cancelled by user') or exit() # overload

小开

My example below does allow for non-blocking reads from stdin under both Windows (only tested under Windows 10) and Linux without requiring external dependencies or using threading. It works for copypasted text, it disables ECHO, so it could be used for e.g. some sort of custom UI and uses a loop, so it would be easy to process anything that was input into it.

考虑到上面的内容，这个示例是用于交互式 TTY 的，而不是管道输入。

#!/usr/bin/env python3
import sys


if(sys.platform == "win32"):
import msvcrt
import ctypes
from ctypes import wintypes
kernel32 = ctypes.windll.kernel32
oldStdinMode = ctypes.wintypes.DWORD()
# Windows standard handle -10 refers to stdin
kernel32.GetConsoleMode(kernel32.GetStdHandle(-10), ctypes.byref(oldStdinMode))
# Disable ECHO and line-mode
# https://learn.microsoft.com/en-us/windows/console/setconsolemode
kernel32.SetConsoleMode(kernel32.GetStdHandle(-10), 0)
else:
# POSIX uses termios
import select, termios, tty
oldStdinMode = termios.tcgetattr(sys.stdin)
_ = termios.tcgetattr(sys.stdin)
# Disable ECHO and line-mode
_[3] = _[3] & ~(termios.ECHO | termios.ICANON)
# Don't block on stdin.read()
_[6][termios.VMIN] = 0
_[6][termios.VTIME] = 0
termios.tcsetattr(sys.stdin, termios.TCSAFLUSH, _)


def readStdin():
if(sys.platform == "win32"):
return msvcrt.getwch() if(msvcrt.kbhit()) else ""
else:
return sys.stdin.read(1)


def flushStdin():
if(sys.platform == "win32"):
kernel32.FlushConsoleInputBuffer(kernel32.GetStdHandle(-10))
else:
termios.tcflush(sys.stdin, termios.TCIFLUSH)


try:
userInput = ""
print("Type something: ", end = "", flush = True)
flushStdin()
while 1:
peek = readStdin()
if(len(peek) > 0):
# Stop input on NUL, Ctrl+C, ESC, carriage return, newline, backspace, EOF, EOT
if(peek not in ["\0", "\3", "\x1b", "\r", "\n", "\b", "\x1a", "\4"]):
userInput += peek
# This is just to show the user what they typed.
# Can be skipped, if one doesn't need this.
sys.stdout.write(peek)
sys.stdout.flush()
else:
break
flushStdin()
print(f"\nuserInput length: {len(userInput)}, contents: \"{userInput}\"")
finally:
if(sys.platform == "win32"):
kernel32.SetConsoleMode(kernel32.GetStdHandle(-10), oldStdinMode)
else:
termios.tcsetattr(sys.stdin, termios.TCSAFLUSH, oldStdinMode)

小开

我正在使用 Linux 编写一个程序，它有一个更大的主循环，需要定期更新，但也需要以非阻塞方式读取字符。但是重置显示，也会丢失输入缓冲区。这是我想到的解决办法。每次屏幕更新后，它都会将终端设置为非阻塞，等待主循环通过，然后解释 stdin。之后，终端被重置为原始设置。

#!/usr/bin/python3
import sys, select, os, tty, termios, time


i = 0
l = True
oldtty = termios.tcgetattr(sys.stdin)
stdin_no = sys.stdin.fileno()


while l:
os.system('clear')
print("I'm doing stuff. Press a 'q' to stop me!")
print(i)
tty.setcbreak(stdin_no)
time.sleep(0.5)
if sys.stdin in select.select([sys.stdin], [], [], 0.0)[0]:
line = sys.stdin.read(1)
print (line, len(line))
        

if "q" in line:
l = False
else:
pass
termios.tcsetattr(stdin_no, termios.TCSADRAIN, oldtty)
i += 1

小开

Marco 的解决方案是正确的想法，但我决定将它简化到没有任何类的最小可能代码。此外，它还向您展示了如何使用队列库获取用户输入，而不仅仅是打印输入:

import time, threading, queue




def collect(que):
msg = input()
que.put(msg)


que = queue.Queue()
thread = threading.Thread(target=collect, args=[que])
thread.start()


while thread.is_alive():
time.sleep(1)
print("The main thread continues while we wait for you...")


msg = que.get()
print('You typed:', msg)

In this example, the main thread continues indefinitely (processing data or whatever), while periodically checking to see if the user has input any data in the spawned thread. When that happens it returns the user input.

我已经成功地在我自己的脚本中使用了这个想法来创建一个调试器，在这个调试器中，我可以在主循环的任何时候输入“ print variable name”，它会实时地给出值，而不会停止。