Python 从 sys.stdin 获取输入，非阻塞
声明：本页面是StackOverFlow热门问题的中英对照翻译，遵循CC BY-SA 4.0协议，如果您需要使用它，必须同样遵循CC BY-SA许可，注明原文地址和作者信息，同时你必须将它归于原作者（不是我）：StackOverFlow
原文地址: http://stackoverflow.com/questions/21791621/
Warning: these are provided under cc-by-sa 4.0 license. You are free to use/share it, But you must attribute it to the original authors (not me):
StackOverFlow
Taking input from sys.stdin, non-blocking
提问by Arthelais
I'm working on a bot for a competition that receives its input through sys.stdinand uses Python's print()for output. I have the following:
我正在开发一个用于竞赛的机器人，它通过接收输入sys.stdin并使用 Pythonprint()进行输出。我有以下几点：
import sys
def main():
while True:
line = sys.stdin.readline()
parts = line.split()
if len(parts) > 0:
# do stuff
The problem is that the input comes in through a stream and using the above, blocks me from printing anything back until the stream is closed. What can I do to make this work?
问题是输入通过流进入并使用上述内容阻止我打印任何内容，直到流关闭。我能做些什么来完成这项工作？
采纳答案by swdev
By turning blocking off you can only read a character at a time. So, there is no way to get readline()to work in a non-blocking context. I assume you just want to read key presses to control the robot.
通过关闭阻止，您一次只能读取一个字符。因此，无法readline()在非阻塞上下文中工作。我假设您只想阅读按键来控制机器人。
I have had no luck using select.select()on Linux and created a way with tweaking termiossettings. So, this is Linux specific but works for me:
我select.select()在 Linux 上使用运气不佳，并创建了一种调整termios设置的方法。所以，这是特定于 Linux 的，但对我有用：
old_settings=None
def init_anykey():
global old_settings
old_settings = termios.tcgetattr(sys.stdin)
new_settings = termios.tcgetattr(sys.stdin)
new_settings[3] = new_settings[3] & ~(termios.ECHO | termios.ICANON) # lflags
new_settings[6][termios.VMIN] = 0 # cc
new_settings[6][termios.VTIME] = 0 # cc
termios.tcsetattr(sys.stdin, termios.TCSADRAIN, new_settings)
@atexit.register
def term_anykey():
global old_settings
if old_settings:
termios.tcsetattr(sys.stdin, termios.TCSADRAIN, old_settings)
def anykey():
ch_set = []
ch = os.read(sys.stdin.fileno(), 1)
while ch != None and len(ch) > 0:
ch_set.append( ord(ch[0]) )
ch = os.read(sys.stdin.fileno(), 1)
return ch_set;
init_anykey()
while True:
key = anykey()
if key != None:
print key
else:
time.sleep(0.1)
A better Windows or cross-platform answer is here: Python nonblocking console input
更好的 Windows 或跨平台答案在这里：Python 非阻塞控制台输入
回答by Jon
Use a generator - thankfully sys.stdinis already a generator!
使用发电机 - 幸好sys.stdin已经是发电机！
A generator enables you to work on an infinite stream. Always when you call it, it returns the next element. In order to build a generator you need the yieldkeyword.
生成器使您能够处理无限流。当你调用它时，它总是返回下一个元素。为了构建生成器，您需要yield关键字。
for line in sys.stdin:
print line
if a_certain_situation_happens:
break
Do not forget to place a breakstatement into the loop if a certain, wished situation happens.
break如果发生某种希望的情况，请不要忘记将语句放入循环中。
You can find more information about generators on:
您可以在以下位置找到有关生成器的更多信息：
回答by Michael Nagy
#-----------------------------------------------------------------------
# Get a character from the keyboard. If Block is True wait for input,
# else return any available character or throw an exception if none is
# available. Ctrl+C isn't handled and continues to generate the usual
# SIGINT signal, but special keys like the arrows return the expected
# escape sequences.
#
# This requires:
#
# import sys, select
#
# This was tested using python 2.7 on Mac OS X. It will work on any
# Linux system, but will likely fail on Windows due to select/stdin
# limitations.
#-----------------------------------------------------------------------
def GetChar(Block=True):
if Block or select.select([sys.stdin], [], [], 0) == ([sys.stdin], [], []):
return sys.stdin.read(1)
raise error('NoChar')
回答by Zeh
You can use selectors for handle I/O multiplexing:
您可以使用选择器来处理 I/O 多路复用：
https://docs.python.org/3/library/selectors.html
https://docs.python.org/3/library/selectors.html
Try this out:
试试这个：
#! /usr/bin/python3
import sys
import fcntl
import os
import selectors
# set sys.stdin non-blocking
orig_fl = fcntl.fcntl(sys.stdin, fcntl.F_GETFL)
fcntl.fcntl(sys.stdin, fcntl.F_SETFL, orig_fl | os.O_NONBLOCK)
# function to be called when enter is pressed
def got_keyboard_data(stdin):
print('Keyboard input: {}'.format(stdin.read()))
# register event
m_selector = selectors.DefaultSelector()
m_selector.register(sys.stdin, selectors.EVENT_READ, got_keyboard_data)
while True:
sys.stdout.write('Type something and hit enter: ')
sys.stdout.flush()
for k, mask in m_selector.select():
callback = k.data
callback(k.fileobj)
The above code will hold on the line
上面的代码将保持在线
for k, mask in m_selector.select():
until a registered event occurs, returning a selector_key instance (k) and a mask of monitored events.
直到一个注册的事件发生，返回一个 selector_key 实例 (k) 和一个被监控事件的掩码。
In the above example we registered only one event (Enter key press):
在上面的例子中，我们只注册了一个事件（回车键按下）：
m_selector.register(sys.stdin, selectors.EVENT_READ, got_keyboard_data)
The selector key instance is defined as follows:
选择器键实例定义如下：
abstractmethod register(fileobj, events, data=None)
Therefore, the register method sets k.data as our callback function got_keyboard_data, and calls it when the Enterkey is pressed :
因此， register 方法将 k.data 设置为我们的回调函数got_keyboard_data，并在按下Enter键时调用它：
callback = k.data
callback(k.fileobj)
A more complete example (and hopefully more useful) would be to multiplex stdin data from user with incomming connections from network:
一个更完整的示例（希望更有用）是将来自用户的 stdin 数据与来自网络的传入连接复用：
import selectors
import socket
import sys
import os
import fcntl
m_selector = selectors.DefaultSelector()
# set sys.stdin non-blocking
def set_input_nonblocking():
orig_fl = fcntl.fcntl(sys.stdin, fcntl.F_GETFL)
fcntl.fcntl(sys.stdin, fcntl.F_SETFL, orig_fl | os.O_NONBLOCK)
def create_socket(port, max_conn):
server_addr = ('localhost', port)
server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
server.setblocking(False)
server.bind(server_addr)
server.listen(max_conn)
return server
def read(conn, mask):
global GO_ON
client_address = conn.getpeername()
data = conn.recv(1024)
print('Got {} from {}'.format(data, client_address))
if not data:
GO_ON = False
def accept(sock, mask):
new_conn, addr = sock.accept()
new_conn.setblocking(False)
print('Accepting connection from {}'.format(addr))
m_selector.register(new_conn, selectors.EVENT_READ, read)
def quit():
global GO_ON
print('Exiting...')
GO_ON = False
def from_keyboard(arg1, arg2):
line = arg1.read()
if line == 'quit\n':
quit()
else:
print('User input: {}'.format(line))
GO_ON = True
set_input_nonblocking()
# listen to port 10000, at most 10 connections
server = create_socket(10000, 10)
m_selector.register(server, selectors.EVENT_READ, accept)
m_selector.register(sys.stdin, selectors.EVENT_READ, from_keyboard)
while GO_ON:
sys.stdout.write('>>> ')
sys.stdout.flush()
for k, mask in m_selector.select():
callback = k.data
callback(k.fileobj, mask)
# unregister events
m_selector.unregister(sys.stdin)
# close connection
server.shutdown()
server.close()
# close select
m_selector.close()
You can test using two terminals. first terminal:
您可以使用两个终端进行测试。第一个终端：
$ python3 test.py
>>> bla
open another terminal and run:
打开另一个终端并运行：
$ nc localhost 10000
hey!
back to the first
回到第一个
>>> qwerqwer
Result (seen on the main terminal):
结果（在主终端上看到）：
$ python3 test.py
>>> bla
User input: bla
>>> Accepting connection from ('127.0.0.1', 39598)
>>> Got b'hey!\n' from ('127.0.0.1', 39598)
>>> qwerqwer
User input: qwerqwer
>>>
回答by Samy Bencherif
Might I suggest nobreak? If'n you are willing to use curses.
我可以建议nobreak吗？如果你愿意使用诅咒。
https://docs.python.org/3/library/curses.html#curses.window.nodelay
https://docs.python.org/3/library/curses.html#curses.window.nodelay
