从 sys.stdin 获取输入，非阻塞

Question

我正在为一场比赛开发一个机器人，该机器人通过sys.stdin 接收输入，并使用 Python 的print() 进行输出。我有以下内容：

import sys

def main():
    while True:
        line = sys.stdin.readline()
        parts = line.split()
        if len(parts) > 0:
            # do stuff

问题是输入通过流进入并使用上述内容阻止我打印任何内容，直到流关闭。我可以做些什么来完成这项工作？

Answer 1

您应该可以使用任一方式读取流

sys.stdin.read(1)

读取 utf-8 解码字符或：

sys.stdin.buffer.read(1)

读取原始字符。

如果我想从标准输入获取原始数据并及时对其进行处理，而不需要先读取换行符或填充内部缓冲区，我会这样做。这适用于在 tty 不可用的情况下通过 ssh 远程运行程序，请参阅：

ssh me@host '/usr/bin/python -c "import sys; print(sys.stdin.isatty())"'

要让程序在这种情况下按预期工作，还需要考虑其他一些事情。您需要在完成后刷新输出以避免缓冲延迟，并且当您根本没有刷新输出时，很容易假设程序没有读取输入。

stdout.write("my data")
stdout.flush()

但通常问题不是输入读取，而是提供输入流的终端（或程序）没有在您期望的时候将其移交，或者它可能没有在您期望的时候读取您的输出。如果您有一个 tty 开头（请参阅上面的 ssh 检查），您可以使用 tty 模块将其置于原始模式。

import sys
import termios
import tty

old = termios.tcgetattr(sys.stdin)
tty.setraw(sys.stdin)
c = None
try:
    c = sys.stdin.read(1)[0]
finally:
    termios.tcsetattr(sys.stdin, termios.TCSADRAIN, old)
print(c)

...如果使用 Mac/Linux。如果使用 Windows，您可以使用 msvcrt.getch()。

Answer 2

通过关闭屏蔽，您一次只能读取一个字符。因此，没有办法让readline() 在非阻塞上下文中工作。我假设您只想阅读按键来控制机器人。

我在 Linux 上使用 select.select() 没有运气，并创建了一种调整 termios 设置的方法。所以，这是 Linux 特有的，但对我有用：

import atexit, termios
import sys, os
import time


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)

这里有更好的 Windows 或跨平台答案：Non-blocking console input?

Answer 3

这是一个 posix 解决方案，接近于 swdev。

正如他所说，您必须使用 VMIN 和 VTIME 来捕获多个字符，而无需用户按 [enter]。尝试仅使用原始模式将是一个问题，因为像箭头这样的特殊键可能会弄乱下一次按键。

这里我们使用 tty.setcbreak() 或 tty.setraw() 作为快捷方式，但它们有short internals。

import termios
import tty
import sys
import select

def get_enter_key():
    fd = sys.stdin.fileno()
    orig_fl = termios.tcgetattr(fd)
    try:
        tty.setcbreak(fd)  # use tty.setraw() instead to catch ^C also
        mode = termios.tcgetattr(fd)
        CC = 6
        mode[CC][termios.VMIN] = 0
        mode[CC][termios.VTIME] = 0
        termios.tcsetattr(fd, termios.TCSAFLUSH, mode)
        keypress, _, _ = select.select([fd], [], [])
        if keypress:
            return sys.stdin.read(4095)
    finally:
        termios.tcsetattr(fd, termios.TCSANOW, orig_fl)

try:
    while True:
        print(get_enter_key())
except KeyboardInterrupt:
    print('exiting')
    sys.exit()

请注意，您可以在此处添加两个潜在的超时：

• 将最后一个参数添加到select.select()
• 另一个是playing with VMIN and VTIME

Answer 4

我可以建议nobreak吗？如果你愿意使用诅咒。

https://docs.python.org/3/library/curses.html#curses.window.nodelay

Answer 5

您可以使用选择器来处理 I/O 多路复用：

https://docs.python.org/3/library/selectors.html

试试这个：

#! /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)

上面的代码就行了

for k, mask in m_selector.select():

直到注册的事件发生，返回一个 selector_key 实例 (k) 和一个被监控事件的掩码。

在上面的例子中我们只注册了一个事件（Enter key press）：

m_selector.register(sys.stdin, selectors.EVENT_READ, got_keyboard_data)

选择器键实例定义如下：

abstractmethod register(fileobj, events, data=None)

因此，register方法将k.data设置为我们的回调函数got_keyboard_data，并在Enter键被按下时调用它：

    callback = k.data
    callback(k.fileobj)

一个更完整的示例（并且希望更有用）是从用户的标准输入数据与来自网络的传入连接进行多路复用：

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()

您可以使用两个终端进行测试。 第一个终端：

$ python3 test.py 
>>> bla

打开另一个终端并运行：

 $ nc localhost 10000
 hey!

回到第一个

>>> qwerqwer     

结果（在主终端上看到）：

$ 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

>>> 

Answer 6

使用生成器 - 幸好sys.stdin 已经是生成器了！

生成器使您能够处理无限流。总是当你调用它时，它会返回下一个元素。要构建生成器，您需要 yield 关键字。

for line in sys.stdin:
    print line

    if a_certain_situation_happens:
        break        

如果某种情况发生，不要忘记将break 语句放入循环中。

您可以在以下位置找到有关生成器的更多信息：

• http://www.dabeaz.com/generators/index.html
• http://linuxgazette.net/100/pramode.html

Answer 7

#-----------------------------------------------------------------------
# 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')