python 从sys.stdin获取输入,非阻塞

yiytaume  于 2023-10-14  发布在  Python
关注(0)|答案(8)|浏览(142)

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

import sys

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

问题是,输入是通过流进入的,使用上面的方法,会阻止我打印任何东西,直到流关闭。我能做些什么来使这个工作吗?

iklwldmw

iklwldmw1#

现在你可以使用set_blocking了。这可以在Linux上工作,并且可以通过改变fd的创建方式来适应其他系统。

class CooperativeInterrupt:

    def __init__(self):
        self.fd = open('/dev/stdin', 'rb')
        os.set_blocking(self.fd.fileno(), False)

    def __del__(self):
        if self.fd is not None:
            self.fd.close()

    def has_interrupt(self):
        data = self.fd.read()
        if data is None:
            return False
        if len(data) == 0:
            raise KeyboardInterrupt()  # Stdin closed (e.g. CTRL+D).
        return True

coi = CooperativeInterrupt()
for i in range(10000000000000):
    if coi.has_interrupt():
        print(i)
    time.sleep(0.1)
qyuhtwio

qyuhtwio2#

使用生成器-谢天谢地sys.stdin已经是一个生成器了!
生成器使您能够处理无限流。当你调用它时,它总是返回下一个元素。为了生成一个生成器,你需要yield关键字。

for line in sys.stdin:
    print line

    if a_certain_situation_happens:
        break

不要忘记在循环中放置一个break语句,如果某个特定的情况发生。
您可以在以下网站找到有关发电机的更多信息:

cbjzeqam

cbjzeqam3#

通过禁用阻止,您一次只能读取一个字符。因此,没有办法让readline()在非阻塞上下文中工作。我假设你只是想通过按键来控制机器人。
我在Linux上使用select.select()没有运气,并创建了一种调整termios设置的方法。所以,这是Linux特定的,但对我来说很有效:

import atexit, termios
import sys, os
import time

old_settings=None

def init_any_key():
   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_any_key():
   global old_settings
   if old_settings:
      termios.tcsetattr(sys.stdin, termios.TCSADRAIN, old_settings)

def any_key():
   ch_set = []
   ch = os.read(sys.stdin.fileno(), 1)
   while ch is not None and len(ch) > 0:
      ch_set.append( ord(ch[0]) )
      ch = os.read(sys.stdin.fileno(), 1)
   return ch_set

init_any_key()
while True:
   key = any_key()
   if key is not None:
      print(key)
   else:
      time.sleep(0.1)

一个更好的Windows或跨平台的答案在这里:Non-blocking console input?

tvmytwxo

tvmytwxo4#

您可以使用选择器来处理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)

一个更完整的例子(希望更有用)是将来自用户的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

>>>
xuo3flqw

xuo3flqw5#

#-----------------------------------------------------------------------
# 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 get_char(block = True):
    if block or select.select([sys.stdin], [], [], 0) == ([sys.stdin], [], []):
        return sys.stdin.read(1)
    raise error('NoChar')
s5a0g9ez

s5a0g9ez6#

这是一个posix解决方案,类似于the answer by swdev
正如他们所说,你必须使用termios.VMINtermios.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()
  • 另一种是使用VMIN和VTIME
4dc9hkyq

4dc9hkyq7#

我可以推荐nobreak吗?如果你愿意使用诅咒。
https://docs.python.org/3/library/curses.html#curses.window.nodelay

v1l68za4

v1l68za48#

您应该能够读取流,

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模块将它置于raw模式。

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

相关问题