Python cmd 模块 - 异步事件后恢复提示

Question

我正在维护一个基于 cmd 的运算符(operator)终端。客户要求提醒行为。例如当某些异步事件发生时在屏幕上显示的消息。我创建了一个定期检查警报的线程，当它发现警报时，它只是将它们打印到标准输出。

这似乎工作正常，但看起来不是很优雅，而且有问题:

因为 cmd 不知道发生了警报，所以屏幕上的消息后面是空白。不重印命令提示符，任何用户输入都处于待处理状态。

有没有更好的方法在 Python cmd 期间进行异步警报？按原样使用方法，我可以中断 cmd 并让它重新绘制提示吗？

我尝试从我的线程中使用 StringIO 在 stdin 中插入一个换行符，但这并不理想，而且我还没有让它正常工作。

示例代码:

import cmd, sys
import threading, time
import io
import sys

class MyShell(cmd.Cmd):
    intro = '*** Terminal ***\nType help or ? to list commands.\n'
    prompt = '> '
    file = None

    def alert(self):
        time.sleep(5)
        print ('\n\n*** ALERT!\n')
        sys.stdin = io.StringIO("\n")

    def do_bye(self, arg):
        'Stop recording, close the terminal, and exit:  BYE'
        print('Exiting.')
        sys.exit(0)
        return True

    def do_async(self, arg):
        'Set a five second timer to pop an alert.'
        threading.Thread(target=self.alert).start()

    def emptyline(self):
        pass

def parse(arg):
    'Convert a series of zero or more numbers to an argument tuple'
    return tuple(map(int, arg.split()))

if __name__ == '__main__':
    MyShell().cmdloop()

Answer 1

我最终用我自己的版本覆盖了 Cmd.cmdloop，用我自己的使用非阻塞终端 IO 的 readlines 替换了 readlines()。

此处为非阻塞终端 IO 信息: Non-Blocking terminal IO

不幸的是，这打开了另一个麻烦，因为它很困惑并且破坏了自动完成和命令历史记录。幸运的是，客户可以接受按 Enter 重做提示，所以我不必再担心了。

显示非阻塞终端输入方法的不完整示例代码:

import cmd, sys
import threading, time
import io

import os

if os.name=='nt':
    import msvcrt
    def getAnyKey():
        if msvcrt.kbhit():
            return msvcrt.getch()
        return None
else:
    import sys
    import select
    import tty
    import termios
    import atexit        
    def isData():
        return select.select([sys.stdin], [], [], 0) == ([sys.stdin], [], [])    
    old_settings = termios.tcgetattr(sys.stdin)    
    def restoreSettings():
        global old_settings
        termios.tcsetattr(sys.stdin, termios.TCSADRAIN, old_settings)        
    atexit.register(restoreSettings)            
    def getAnyKey():
        try:
            if isData():
                return sys.stdin.read(1)
            return None
        except:
            pass
        return None

class MyShell(cmd.Cmd):
    prompt = '> '
    file = None
    realstdin = sys.stdin
    mocking=False
    breakReadLine=False
    def alert(self):
        time.sleep(5)
        print ('\n\n*** ALERT!\n')
        self.breakReadLine=True

    # ----- basic commands -----

    def do_bye(self, arg):
        'Stop recording, close the terminal, and exit:  BYE'
        print('Exiting.')
        sys.exit(0)
        return True

    def do_async(self, arg):
        'Set a five second timer to pop an alert.'
        threading.Thread(target=self.alert).start()

    def emptyline(self):
        pass

    def myReadLine(self):
        sys.stdout.flush()
        self.breakReadLine=False
        line=''
        while not self.breakReadLine:
            c=getAnyKey()          
            if not c is None:
                c=c.decode("utf-8")              
                if c=='\x08' and len(line):
                    line=line[0:-1]
                elif c in ['\r','\n']:
                    print('\n')
                    return line
                else:
                    line+=c
                print(c,end='')
                sys.stdout.flush()


    def mycmdloop(self, intro=None):
        """Repeatedly issue a prompt, accept input, parse an initial prefix
        off the received input, and dispatch to action methods, passing them
        the remainder of the line as argument.

        """
        self.preloop()
        if self.use_rawinput and self.completekey:
            try:
                import readline
                self.old_completer = readline.get_completer()
                readline.set_completer(self.complete)
                readline.parse_and_bind(self.completekey+": complete")
            except ImportError:
                pass
        try:
            if intro is not None:
                self.intro = intro
            if self.intro:
                self.stdout.write(str(self.intro)+"\n")
            stop = None
            while not stop:
                if self.cmdqueue:
                    line = self.cmdqueue.pop(0)
                else:
                    if self.use_rawinput:
                        try:
                            print(self.prompt,end='')
                            line = self.myReadLine()#input(self.prompt)
                        except EOFError:
                            line = 'EOF'
                    else:
                        self.stdout.write(self.prompt)
                        self.stdout.flush()
                        line = self.myReadLine()#self.stdin.readline()
                if not line is None:
                    line = line.rstrip('\r\n')
                    line = self.precmd(line)
                    stop = self.onecmd(line)
                    stop = self.postcmd(stop, line)
            self.postloop()
        finally:
            if self.use_rawinput and self.completekey:
                try:
                    import readline
                    readline.set_completer(self.old_completer)
                except ImportError:
                    pass

    def cmdloop_with_keyboard_interrupt(self, intro):
        doQuit = False
        while doQuit != True:
            try:
                if intro!='':
                    cintro=intro
                    intro=''                
                    self.mycmdloop(cintro)
                else:
                    self.intro=''                
                    self.mycmdloop()
                doQuit = True
            except KeyboardInterrupt:
                sys.stdout.write('\n')

def parse(arg):
    'Convert a series of zero or more numbers to an argument tuple'
    return tuple(map(int, arg.split()))

if __name__ == '__main__':
    #MyShell().cmdloop()
    MyShell().cmdloop_with_keyboard_interrupt('*** Terminal ***\nType help or ? to list commands.\n')