Python cmd module - Resume prompt after asynchronous event

Question

I am maintaining an operator terminal based on cmd. The customer asked for an alerting behavior. e.g. a message shown onscreen when some asynchronous event occurs. I made a thread that periodically checks for alerts, and when it finds some, it just prints them to stdout.

This seems to work OK, but it doesn't seem very elegant, and it has a problem:

Because cmd doesn't know an alert happened, the message is followed onscreen by blank. The command prompt is not reprinted, and any user input is left pending.

Is there a better way to do asynchronous alerts during Python cmd? With the method as-is, can I interrupt cmd and get it to redraw its prompt?

I tried from my thread to poke a newline in stdin using StringIO, but this is not ideal, and I haven't gotten it work right.

Example code:

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

score 0 · Accepted Answer · edited May 23 '17 at 12:00

I ended up overriding Cmd.cmdloop with my own version, replacing the readlines() with my own readlines that use non-blocking terminal IO.

Non-Blocking terminal IO info here: Non-Blocking terminal IO

Unfortunately, this opens another can trouble in that it is messy and breaks auto-completion and command history. Fortunately, the customer was OK with having to push Enter to redo the prompt, so I don't need to worry about it anymore.

Incomplete example code showing the non-blocking terminal input approach:

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')

Python cmd module - Resume prompt after asynchronous event

1 Answers1