非阻塞式控制台输入?

86

我正在尝试用Python制作一个简单的IRC客户端(作为我学习这门语言的项目)。

我有一个循环来接收和解析IRC服务器发送给我的内容,但如果我使用raw_input输入一些东西,它会立即停止循环,直到我输入了一些东西(显然)。

如何在不停止循环的情况下输入内容?

(我认为我不需要发布代码,我只是想在while 1:循环不停止的情况下输入内容。)

我在Windows上。


你使用了哪个网络模块?Twisted、sockets、asyncore? - DevPlayer
3
请看这个示例:非阻塞、多线程示例:https://dev59.com/UW035IYBdhLWcg3weP3f#53344690 - Gabriel Staples
15个回答

2

因为我发现其中上面的答案有用,这里提供一个类似方法的例子。这段代码创建了一个节拍器效果,并且可以接受输入。

区别在于这段代码使用闭包而不是类,对我来说感觉更加直观一些。这个例子还通过my_thread.stop = True引入了终止线程的标志,但没有使用全局变量。我通过滥用Python函数是对象并且可以在自己内部进行猴子补丁的特性来做到这一点。

注意:停止线程应该谨慎处理。如果你的线程有需要清理的数据或者生成它自己的线程,这种方法将会不加任何仪式地杀死这些进程。

# Begin metronome sound while accepting input.
# After pressing enter, turn off the metronome sound.
# Press enter again to restart the process.

import threading
import time
import winsound  # Only on Windows

beat_length = 1  # Metronome speed


def beat_thread():
    beat_thread.stop = False  # Monkey-patched flag
    frequency, duration = 2500, 10
    def run():  # Closure
        while not beat_thread.stop:  # Run until flag is True
            winsound.Beep(frequency, duration)
            time.sleep(beat_length - duration/1000)
    threading.Thread(target=run).start()


while True:
    beat_thread()
    input("Input with metronome. Enter to finish.\n")
    beat_thread.stop = True  # Flip monkey-patched flag
    input("Metronome paused. Enter to continue.\n\n")


1

我正在使用Linux编写一个程序,它有一个更大的主循环,需要定期更新,但也需要以非阻塞方式读取字符。 但是重置显示器也会丢失输入缓冲区。 这是我想出的解决方案。 每次屏幕更新后,将终端设置为非阻塞状态,等待主循环传递,然后解释stdin。 之后,终端将被重置为原始设置。

#!/usr/bin/python3
import sys, select, os, tty, termios, time

i = 0
l = True
oldtty = termios.tcgetattr(sys.stdin)
stdin_no = sys.stdin.fileno()

while l:
    os.system('clear')
    print("I'm doing stuff. Press a 'q' to stop me!")
    print(i)
    tty.setcbreak(stdin_no)
    time.sleep(0.5)
    if sys.stdin in select.select([sys.stdin], [], [], 0.0)[0]:
        line = sys.stdin.read(1)
        print (line, len(line))
        
        if "q" in line:
            l = False
        else: 
            pass
    termios.tcsetattr(stdin_no, termios.TCSADRAIN, oldtty)
    i += 1



我认为这是最好的答案。我使用类似的代码,没有ttytermios。诀窍和整个美妙之处在于使用select.select命令进行非阻塞检查是否有可用字符进行读取。字符直接由read(1)函数读取,其参数1是必需的。即使在Micropython中,这种解决方案也可以工作。完美,可惜我没有早点发现它。 - Martin Vyskočil

1
以下是对上述解决方案之一的类包装器:
#!/usr/bin/env python3

import threading

import queue

class NonBlockingInput:

    def __init__(self, exit_condition):
        self.exit_condition = exit_condition
        self.input_queue = queue.Queue()
        self.input_thread = threading.Thread(target=self.read_kbd_input, args=(), daemon=True)
        self.input_thread.start()

    def read_kbd_input(self):
        done_queueing_input = False
        while not done_queueing_input:
            console_input = input()
            self.input_queue.put(console_input)
            if console_input.strip() == self.exit_condition:
                done_queueing_input = True

    def input_queued(self):
        return_value = False
        if self.input_queue.qsize() > 0:
            return_value = True
        return return_value

    def input_get(self):
        return_value = ""
        if self.input_queue.qsize() > 0:
            return_value = self.input_queue.get()
        return return_value

if __name__ == '__main__':

    NON_BLOCK_INPUT = NonBlockingInput(exit_condition='quit')

    DONE_PROCESSING = False
    INPUT_STR = ""
    while not DONE_PROCESSING:
        if NON_BLOCK_INPUT.input_queued():
            INPUT_STR = NON_BLOCK_INPUT.input_get()
            if INPUT_STR.strip() == "quit":
                DONE_PROCESSING = True
            else:
                print("{}".format(INPUT_STR))

1

marco提出的解决方案是正确的想法,但我决定将其简化为最小可能的代码,而不使用任何类。此外,它实际上向您展示了如何使用队列库获取用户输入,而不仅仅是打印它:

import time, threading, queue


def collect(que):
    msg = input()
    que.put(msg)

que = queue.Queue()
thread = threading.Thread(target=collect, args=[que])
thread.start()

while thread.is_alive():
    time.sleep(1)
    print("The main thread continues while we wait for you...")

msg = que.get()
print('You typed:', msg)

在这个例子中,主线程持续运行(处理数据或其他任务),同时定期检查用户是否在生成的线程中输入了任何数据。当发生这种情况时,它会返回用户输入。
我已经成功地在我的脚本中使用这个想法来创建一个调试器,在主循环的任何点上键入“print variable name”,它会实时给我提供值而不会停止。

0

下面的示例允许在Windows(仅在Windows 10下测试)和Linux下进行非阻塞读取stdin,而不需要外部依赖项或使用线程。它适用于复制粘贴的文本,禁用了ECHO,因此可以用于某种自定义UI,并使用循环,因此可以轻松处理输入到其中的任何内容。

考虑到上述内容,该示例适用于交互式TTY,而不是管道输入。

#!/usr/bin/env python3
import sys

if(sys.platform == "win32"):
    import msvcrt
    import ctypes
    from ctypes import wintypes
    kernel32 = ctypes.windll.kernel32
    oldStdinMode = ctypes.wintypes.DWORD()
    # Windows standard handle -10 refers to stdin
    kernel32.GetConsoleMode(kernel32.GetStdHandle(-10), ctypes.byref(oldStdinMode))
    # Disable ECHO and line-mode
    # https://learn.microsoft.com/en-us/windows/console/setconsolemode
    kernel32.SetConsoleMode(kernel32.GetStdHandle(-10), 0)
else:
    # POSIX uses termios
    import select, termios, tty
    oldStdinMode = termios.tcgetattr(sys.stdin)
    _ = termios.tcgetattr(sys.stdin)
    # Disable ECHO and line-mode
    _[3] = _[3] & ~(termios.ECHO | termios.ICANON)
    # Don't block on stdin.read()
    _[6][termios.VMIN] = 0
    _[6][termios.VTIME] = 0
    termios.tcsetattr(sys.stdin, termios.TCSAFLUSH, _)

def readStdin():
    if(sys.platform == "win32"):
        return msvcrt.getwch() if(msvcrt.kbhit()) else ""
    else:
        return sys.stdin.read(1)

def flushStdin():
    if(sys.platform == "win32"):
        kernel32.FlushConsoleInputBuffer(kernel32.GetStdHandle(-10))
    else:
        termios.tcflush(sys.stdin, termios.TCIFLUSH)

try:
    userInput = ""
    print("Type something: ", end = "", flush = True)
    flushStdin()
    while 1:
        peek = readStdin()
        if(len(peek) > 0):
            # Stop input on NUL, Ctrl+C, ESC, carriage return, newline, backspace, EOF, EOT
            if(peek not in ["\0", "\3", "\x1b", "\r", "\n", "\b", "\x1a", "\4"]):
                userInput += peek
                # This is just to show the user what they typed.
                # Can be skipped, if one doesn't need this.
                sys.stdout.write(peek)
                sys.stdout.flush()
            else:
                break
    flushStdin()
    print(f"\nuserInput length: {len(userInput)}, contents: \"{userInput}\"")
finally:
    if(sys.platform == "win32"):
        kernel32.SetConsoleMode(kernel32.GetStdHandle(-10), oldStdinMode)
    else:
        termios.tcsetattr(sys.stdin, termios.TCSAFLUSH, oldStdinMode)

网页内容由stack overflow 提供, 点击上面的
可以查看英文原文,
原文链接