python - 使用 Python，如何在后台等待按键时在前台运行循环？

Question

我知道这个(such as here)上有类似的话题但我的预期设计有点复杂。

我正在设计将在 SSH 窗口中运行的 CLI 脚本。该脚本将在 Ubuntu 14.10 服务器上托管和执行。它旨在在前台主动监控主机交换机上端口和客户端的当前状态。每 30 秒或根据用户定义，它将通过 SNMP 获取数据，然后刷新信息并将其显示到屏幕上。当它等待下一次刷新时，会有一个计时器指示它何时会再次查询设备以获取信息。

我想允许用户随时按下特定的键来更改输出 View 或编辑键变量。 (功能类似于 Unix top。)例如，按 t 将要求他们输入循环之间所需的秒数。 h、m 或 i 将切换显示/隐藏某些列。这些将不会暂停计时器或退出循环，因为更改将在下一次刷新时应用。 r 会强制立即刷新并应用更改。 q 或 Ctrl+C 将退出脚本。

主要事件如下所示:

Query loop <-----------------------------
     |                                   |
     |                              Process Data
     |                                   ^
     |                                   |
     v                               Query Data    #SNMPBULKWALK
   Timer <-------------                  ^
     | |               |                 |          
     | |        Check time remaining     |
     | |               ^                 |
     | |_______________|                 |
     |___________________________________|

对于按键中断，它会像这样:

Query loop <----------------------                      
    |                             |        ???<---Change variables
    |                        (Continue)                  ^
    V                             |                      |
  Timer <---------          !!INTERRUPT!!---------> Identify key
    | |           |               ^
    | |  Check time remaining     |
    | |           ^               |
    | |___________|               |
    |_____________________________|

我在这里有点难过。我被引导相信我可能需要实现线程——我没有这方面的经验——因为 while 循环本身似乎不能满足我们的需要。我也不确定如何将更改注入(inject)包含变量的对象(例如计时器、显示格式的标志)，因为我们的循环将不断使用它。

Answer 1

这并不复杂，也不需要任何包。

它唯一的问题是它需要在程序退出时将终端恢复正常。

即如果程序崩溃，终端将无法恢复，用户将看不到他正在输入的内容。

但是如果用户知道他在做什么，他可以强制重启 shell，一切都会恢复正常。

当然，您可以更轻松地使用此代码并使用 raw_input() 来完成这些工作。

from thread import start_new_thread as thread
from time import sleep
# Get only one character from stdin without echoing it to stdout
import termios
import fcntl
import sys
import os
fd = sys.stdin.fileno()
oldterm, oldflags = None, None

def prepareterm ():
    """Turn off echoing"""
    global oldterm, oldflags
    if oldterm!=None and oldflags!=None: return
    oldterm = termios.tcgetattr(fd)
    newattr = oldterm[:] # Copy of attributes to change
    newattr[3] = newattr[3] & ~termios.ICANON & ~termios.ECHO
    termios.tcsetattr(fd, termios.TCSANOW, newattr)
    oldflags = fcntl.fcntl(fd, fcntl.F_GETFL)
    fcntl.fcntl(fd, fcntl.F_SETFL, oldflags | os.O_NONBLOCK)

def restoreterm ():
    """Restore terminal to its previous state"""
    global oldterm, oldflags
    if oldterm==None and oldflags==None: return
    termios.tcsetattr(fd, termios.TCSAFLUSH, oldterm)
    fcntl.fcntl(fd, fcntl.F_SETFL, oldflags)
    oldterm, oldflags = None, None

def getchar():
    """Get character from stdin"""
    prepareterm()
    while 1:
        try:
            c = sys.stdin.read(1)
            break
        except IOError:
            try: sleep(0.001)
            except: restoreterm(); raise KeyboardInterrupt
    restoreterm()
    return c

def control ():
    """Waits for keypress"""
    global running, done
    while running:
        c = getchar().lower()
        print "Keypress:", c
        if c=="q": print "Quitting!"; running = 0; break
    done += 1

def work ():
    """Does the server-client part"""
    global done
    while running:
        # Do your stuff here!!!
        # Just to illustrate:
        print "I am protending to work!\nPress Q to kill me!"
        sleep(1)
    print "I am done!\nExiting . . ."
    done += 1

# Just to feel better
import atexit
atexit.register(restoreterm)

# Start the program
running = 1
done = 0
thread(work, ())
thread(control, ())
# Block the program not to close when threads detach from the main one:
while running:
    try: sleep(0.2)
    except: running = 0
# Wait for both threads to finish:
while done!=2:
    try: sleep(0.001)
    except: pass # Ignore KeyboardInterrupt
restoreterm() # Just in case!

在实践中，如果不将终端恢复到正常状态，程序应该永远无法退出，但糟糕的事情发生了。

现在，您可以通过仅使用一个线程并将您的工作循环放在主线程中来简化事情，并使用 raw_input 从用户获取命令。或者甚至更好，将您的服务器-客户端代码放在后台并在主线程中等待输入。

使用线程模块而不是原始线程也可能更安全。

如果您使用 asyncore 模块，您将让每个客户端各自运行，并且您的主线程将被 asyncore.loop() 占用。您可以覆盖它，即重写它来检查输入和其他你想做的事情，同时保持 asyncore 检查同步。此外，重负载需要覆盖其中的一些讨厌的函数，因为如果我没记错的话，它的缓冲区固定为 512 字节。否则，它可能是解决您的问题的好方法。

最后，为了清楚起见，没有回应用户输入的代码是从 getpass 模块中获取和改编的。只有一点点是我的。