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 ('nn*** 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()
Advertisement
Answer
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 ('nn*** 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('rn')
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')