Python cmdモジュール - 非同期イベント後にプロンプ​​トを再開

Question

cmdに基づいてオペレータ端末を維持しています。顧客は警告動作を要求しました。例えばいくつかの非同期イベントが発生したときに画面に表示されるメッセージ。アラートを定期的にチェックするスレッドを作成し、見つかったときにstdoutに出力します。

は、これはOK動作しているようですが、それは非常にエレガントないないようだ、それは問題があります。警告を知らない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

私は非ブロックターミナルIOを使用し、私自身readlinesで）（readlinesを交換し、私自身のバージョンでCmd.cmdloopを上書きすることになりました。

ここでノンブロッキングターミナルIO情報： Non-Blocking terminal IO

残念ながら、これは厄介で、自動補完とコマンド履歴を壊すことで、別の缶トラブルを開きます。幸いにも、顧客はEnterを押してプロンプトをやり直さなくてもOKだったので、もうこれ以上心配する必要はありません。

非ブロッキング端末入力手法を示す不完全なコード例：

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