從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

通過轉動堵住你只可以一次讀取一個字符。因此，無法使readline()在非阻塞上下文中工作。我假設你只是想閱讀按鍵來控制機器人。

我在Linux上使用select.select()沒有運氣，並創建了一個調整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) 

一個更好的Windows或跨平臺的答案就在這裏：Python nonblocking console input

Answer 2

使用發電機 - 謝天謝地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 3

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

Answer 4

您可以使用手柄選擇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）和監視的事件的掩模。

在上面的例子中，我們只登記了一個事件（輸入按鍵）：

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

鍵實例定義爲選擇器如下：

abstractmethod register(fileobj, events, data=None) 

因此，寄存器方法集ķ。數據作爲我們的回調函數got_keyboard_data，並調用它時，輸入按鍵：

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 

>>>