Ввод данных из sys.stdin, неблокирующий

Ответ 1

Превращая блокировку, вы можете читать только символ за раз. Таким образом, нет возможности получить readline() для работы в неблокирующем контексте. Я предполагаю, что вы просто хотите прочитать нажатия клавиш для управления роботом.

Мне не повезло с использованием 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)

Лучше Windows или кросс-платформенный ответ здесь: Неблокирующий консольный вход Python

Ответ 2

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

Ответ 3

Вы можете использовать селектора для мультиплексирования ввода-вывода для ввода:

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

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)

Более полным примером (и, надеюсь, более полезным) было бы мультиплексирование данных 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()

Вы можете протестировать, используя два терминала. первый терминал:

$ 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

>>> 

Ответ 4

Могу ли я предложить nobreak? Если вы готовы использовать проклятия.

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

Ответ 5

Используйте генератор - к счастью, 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