非阻塞式控制台输入?

86

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

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

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

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

我在Windows上。


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

81

对于 Windows 系统,仅适用于控制台,使用 msvcrt 模块:

import msvcrt

num = 0
done = False
while not done:
    print(num)
    num += 1

    if msvcrt.kbhit():
        print "you pressed",msvcrt.getch(),"so now i will quit"
        done = True

对于Linux,这篇文章描述了以下解决方案,需要使用termios模块:

import sys
import select
import tty
import termios

def isData():
    return select.select([sys.stdin], [], [], 0) == ([sys.stdin], [], [])

old_settings = termios.tcgetattr(sys.stdin)
try:
    tty.setcbreak(sys.stdin.fileno())

    i = 0
    while 1:
        print(i)
        i += 1

        if isData():
            c = sys.stdin.read(1)
            if c == '\x1b':         # x1b is ESC
                break

finally:
    termios.tcsetattr(sys.stdin, termios.TCSADRAIN, old_settings)

对于跨平台或需要GUI的情况,您可以使用Pygame:

import pygame
from pygame.locals import *

def display(str):
    text = font.render(str, True, (255, 255, 255), (159, 182, 205))
    textRect = text.get_rect()
    textRect.centerx = screen.get_rect().centerx
    textRect.centery = screen.get_rect().centery

    screen.blit(text, textRect)
    pygame.display.update()

pygame.init()
screen = pygame.display.set_mode( (640,480) )
pygame.display.set_caption('Python numbers')
screen.fill((159, 182, 205))

font = pygame.font.Font(None, 17)

num = 0
done = False
while not done:
    display( str(num) )
    num += 1

    pygame.event.pump()
    keys = pygame.key.get_pressed()
    if keys[K_ESCAPE]:
        done = True

1
我已经有pygame了,所以我会试一下。谢谢。 不过,还有其他人有更好的解决方案吗?我想保持它作为一个控制台应用程序。 - ImTooStupidForThis
3
如果输入是通过其他进程进行管道传输,有没有办法使它工作? - anishsane
请注意:Windows解决方案实际上并不检查stdin中是否有任何内容。它检查键盘上是否按下了某个键,因此管道输入将不会被处理。 - fakedrake
我一直在寻找一种适当的方法来中断我的线程,这个线程正在阻塞在input()上。但是使用msvcrt和getch,在这里无法得到与input()相同的箭头键历史记录行为、退格等。有没有一种简单的方法来保持控制台的行为,同时仍然能够在input()上中断线程? - AgentM
Linux解决方案似乎对我无效:“termios.error: (25, 'Inappropriate ioctl for device')”在第old_settings = termios.tcgetattr(sys.stdin)行。 有什么想法吗? - gmagno
1
值得一提的是,在PyCharm中,msvcrt的kbhit函数不能直接使用,您需要在运行配置中勾选“在输出控制台中模拟终端”选项,以便kbhit()函数能够捕获输入。 - TheTomer

50

这是我见过的最棒的解决方案1。为了防止链接失效,我在这里复制一下:

#!/usr/bin/env python
'''
A Python class implementing KBHIT, the standard keyboard-interrupt poller.
Works transparently on Windows and Posix (Linux, Mac OS X).  Doesn't work
with IDLE.

This program is free software: you can redistribute it and/or modify
it under the terms of the GNU Lesser General Public License as 
published by the Free Software Foundation, either version 3 of the 
License, or (at your option) any later version.

This program is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
GNU General Public License for more details.

'''

import os

# Windows
if os.name == 'nt':
    import msvcrt

# Posix (Linux, OS X)
else:
    import sys
    import termios
    import atexit
    from select import select


class KBHit:

    def __init__(self):
        '''Creates a KBHit object that you can call to do various keyboard things.
        '''

        if os.name == 'nt':
            pass

        else:

            # Save the terminal settings
            self.fd = sys.stdin.fileno()
            self.new_term = termios.tcgetattr(self.fd)
            self.old_term = termios.tcgetattr(self.fd)

            # New terminal setting unbuffered
            self.new_term[3] = (self.new_term[3] & ~termios.ICANON & ~termios.ECHO)
            termios.tcsetattr(self.fd, termios.TCSAFLUSH, self.new_term)

            # Support normal-terminal reset at exit
            atexit.register(self.set_normal_term)


    def set_normal_term(self):
        ''' Resets to normal terminal.  On Windows this is a no-op.
        '''

        if os.name == 'nt':
            pass

        else:
            termios.tcsetattr(self.fd, termios.TCSAFLUSH, self.old_term)


    def getch(self):
        ''' Returns a keyboard character after kbhit() has been called.
            Should not be called in the same program as getarrow().
        '''

        s = ''

        if os.name == 'nt':
            return msvcrt.getch().decode('utf-8')

        else:
            return sys.stdin.read(1)


    def getarrow(self):
        ''' Returns an arrow-key code after kbhit() has been called. Codes are
        0 : up
        1 : right
        2 : down
        3 : left
        Should not be called in the same program as getch().
        '''

        if os.name == 'nt':
            msvcrt.getch() # skip 0xE0
            c = msvcrt.getch()
            vals = [72, 77, 80, 75]

        else:
            c = sys.stdin.read(3)[2]
            vals = [65, 67, 66, 68]

        return vals.index(ord(c.decode('utf-8')))


    def kbhit(self):
        ''' Returns True if keyboard character was hit, False otherwise.
        '''
        if os.name == 'nt':
            return msvcrt.kbhit()

        else:
            dr,dw,de = select([sys.stdin], [], [], 0)
            return dr != []


# Test    
if __name__ == "__main__":

    kb = KBHit()

    print('Hit any key, or ESC to exit')

    while True:

        if kb.kbhit():
            c = kb.getch()
            if ord(c) == 27: # ESC
                break
            print(c)

    kb.set_normal_term()

1 这个软件是由Simon D. Levy开发的,它是他编写并发布在一系列软件之中,使用了Gnu Lesser General Public License


32

我最喜欢的获得非阻塞输入的方法是在线程中使用Python的input()函数:

import threading

class KeyboardThread(threading.Thread):

    def __init__(self, input_cbk = None, name='keyboard-input-thread'):
        self.input_cbk = input_cbk
        super(KeyboardThread, self).__init__(name=name)
        self.start()

    def run(self):
        while True:
            self.input_cbk(input()) #waits to get input + Return

showcounter = 0 #something to demonstrate the change

def my_callback(inp):
    #evaluate the keyboard input
    print('You Entered:', inp, ' Counter is at:', showcounter)

#start the Keyboard thread
kthread = KeyboardThread(my_callback)

while True:
    #the normal program executes without blocking. here just counting up
    showcounter += 1

操作系统无关,仅使用内部库,支持多字符输入。


5
这种方法正好符合我的需求。请查看下面的答案,其中使用了闭包更加简洁地实现了相同的功能。 - JDG
1
是的,我觉得这应该是正确的答案,它最简单和轻量级。毕竟,对于这么简单的需求,我不需要一个复杂的多线程解决方案! - Jamie Nicholl-Shelley
太好了!但是如何在第一个字符时就开始工作(而不是等待回车)? - user105939
我有点困惑。run()方法什么时候被调用? - Arjuna Deva
1
@ArjunaDeva 这假设你有一个调用函数的__main__。这只是一个可以放置在你的架构中任何位置的函数。 - Jamie Nicholl-Shelley
1
啊,我明白了,run() 是继承类的方法,“代表线程活动”(根据文档)。 - Arjuna Deva

27

这里有一个解决方案,可以在 Linux 和 Windows 下运行,并使用单独的线程:

import sys
import threading
import time
import Queue

def add_input(input_queue):
    while True:
        input_queue.put(sys.stdin.read(1))

def foobar():
    input_queue = Queue.Queue()

    input_thread = threading.Thread(target=add_input, args=(input_queue,))
    input_thread.daemon = True
    input_thread.start()

    last_update = time.time()
    while True:

        if time.time()-last_update>0.5:
            sys.stdout.write(".")
            last_update = time.time()

        if not input_queue.empty():
            print "\ninput:", input_queue.get()

foobar()

似乎这是唯一的解决方案,既可以在Windows命令行控制台上工作,也可以在Eclipse中工作! - michael_s
1
sys.stdout.write(".") 之后需要加上 sys.stdout.flush() - Gene S
那么Mac呢? - ColorCodin
@ColorCodin 也应该可以工作,因为它基于Unix。 - Kev1n91

14
在Linux上,这里是对mizipzor代码的重构,这样在多个地方使用此代码时会更加容易。
import sys
import select
import tty
import termios

class NonBlockingConsole(object):

    def __enter__(self):
        self.old_settings = termios.tcgetattr(sys.stdin)
        tty.setcbreak(sys.stdin.fileno())
        return self

    def __exit__(self, type, value, traceback):
        termios.tcsetattr(sys.stdin, termios.TCSADRAIN, self.old_settings)


    def get_data(self):
        if select.select([sys.stdin], [], [], 0) == ([sys.stdin], [], []):
            return sys.stdin.read(1)
        return False

以下是使用方法:该代码将打印一个计数器,直到您按下ESC键为止。

with NonBlockingConsole() as nbc:
    i = 0
    while 1:
        print i
        i += 1
        if nbc.get_data() == '\x1b':  # x1b is ESC
            break

1
在使用GNU / Linux时,您仍然需要在输入字符后按Enter键,但是之后它可以正常工作。至少它是非阻塞的,并且大多数情况下返回普通字符(除了像Escape或Backspace这样的特殊键之外的按键码)。谢谢! - Luc

13

我认为curses库可以提供帮助。

import curses
import datetime

stdscr = curses.initscr()
curses.noecho()
stdscr.nodelay(1) # set getch() non-blocking

stdscr.addstr(0,0,"Press \"p\" to show count, \"q\" to exit...")
line = 1
try:
    while 1:
        c = stdscr.getch()
        if c == ord('p'):
            stdscr.addstr(line,0,"Some text here")
            line += 1
        elif c == ord('q'): break

        """
        Do more things
        """

finally:
    curses.endwin()

curses不具备可移植性。 - kfsone
6
curses非常便携,它比Python本身更加便携。在我看来,这是最好的答案,但这取决于你的使用情况。例如,在使用curses之前,它会清除控制台。 - Samie Bencherif
curses 接管了屏幕,所有后续的屏幕输出必须按照它自己的方式处理,print() 不再起作用。 - Yan King Yin

5

回到最初的问题...

我也在学习Python,需要看很多文档和示例,并且费尽心思...但我认为我已经找到了一种简单、简洁、短小且兼容的解决方案...只需使用输入、列表和线程。

'''
what i thought:
- input() in another thread
- that were filling a global strings list
- strings are being popped in the main thread
'''

import threading

consoleBuffer = []

def consoleInput(myBuffer):
  while True:
    myBuffer.append(input())
 
threading.Thread(target=consoleInput, args=(consoleBuffer,), daemon=True).start() # start the thread

import time # just to demonstrate non blocking parallel processing

while True:
  time.sleep(2) # avoid 100% cpu
  print(time.time()) # just to demonstrate non blocking parallel processing
  while consoleBuffer:
    print(repr(consoleBuffer.pop(0)))

这是我找到的最简单兼容的方法,需要注意的是默认情况下stdin、stdout和stderr共用同一个终端,因此如果在您输入时打印了一些内容,您的输入可能会看起来不一致,但是按下回车键后会正常接收已输入的字符串... 如果您不想/不喜欢这种行为,请尝试找到分离输入/输出区域的方法,如重定向,或者尝试其他解决方案,如curses、tkinter、pygame等。

BONUS: ctrl-c按键可以轻松处理:

try:
  # do whatever
except KeyboardInterrupt:
  print('cancelled by user') or exit() # overload

1
谢谢,这绝对是最简单的解决方案,我建议在该线程中添加 daemon=True,这样一旦程序终止,它将会优雅地关闭。 - Puupuls
你是对的,已经添加了...感谢建议。 - atesin

3

如果你只想从循环中进行一次“跳出”,你可以截取 Ctrl-C 信号。

这是跨平台的,非常简单!

import signal
import sys

def signal_handler(sig, frame):
    print('You pressed Ctrl+C!')
    sys.exit(0)

signal.signal(signal.SIGINT, signal_handler)
while True:
    # do your work here

3

我会按照Mickey Chan所说的做,但是我会使用unicurses而不是普通的curses。 Unicurses是通用的(可以在所有或至少几乎所有的操作系统上工作)。


2

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