更好的键盘中断检测这个线程的Spinner类

Question

好吧，我已经写了这个类，基于一些其他Spinner类，我在谷歌代码搜索谷歌搜索。

它按预期工作，但我正在寻找更好的方法来处理KeyboardInterrupt和SystemExit异常。有更好的方法吗？

这里是我的代码：

#!/usr/bin/env python 
import itertools 
import sys 
import threading 

class Spinner(threading.Thread): 
    '''Represent a random work indicator, handled in a separate thread''' 

    # Spinner glyphs 
    glyphs = ('|', '/', '-', '\\', '|', '/', '-') 
    # Output string format 
    output_format = '%-78s%-2s' 
    # Message to output while spin 
    spin_message = '' 
    # Message to output when done 
    done_message = '' 
    # Time between spins 
    spin_delay = 0.1 

    def __init__(self, *args, **kwargs): 
     '''Spinner constructor''' 
     threading.Thread.__init__(self, *args, **kwargs) 
     self.daemon = True 
     self.__started = False 
     self.__stopped = False 
     self.__glyphs = itertools.cycle(iter(self.glyphs)) 

    def __call__(self, func, *args, **kwargs): 
     '''Convenient way to run a routine with a spinner''' 
     self.init() 
     skipped = False 

     try: 
      return func(*args, **kwargs) 
     except (KeyboardInterrupt, SystemExit): 
      skipped = True 
     finally: 
      self.stop(skipped) 

    def init(self): 
     '''Shows a spinner''' 
     self.__started = True 
     self.start() 

    def run(self): 
     '''Spins the spinner while do some task''' 
     while not self.__stopped: 
      self.spin() 

    def spin(self): 
     '''Spins the spinner''' 
     if not self.__started: 
      raise NotStarted('You must call init() first before using spin()') 

     if sys.stdin.isatty(): 
      sys.stdout.write('\r') 
      sys.stdout.write(self.output_format % (self.spin_message, 
                self.__glyphs.next())) 
      sys.stdout.flush() 
      time.sleep(self.spin_delay) 

    def stop(self, skipped=None): 
     '''Stops the spinner''' 
     if not self.__started: 
      raise NotStarted('You must call init() first before using stop()') 

     self.__stopped = True 
     self.__started = False 

     if sys.stdin.isatty() and not skipped: 
      sys.stdout.write('\b%s%s\n' % ('\b' * len(self.done_message), 
              self.done_message)) 
      sys.stdout.flush() 

class NotStarted(Exception): 
    '''Spinner not started exception''' 
    pass 

if __name__ == '__main__': 
    import time 

    # Normal example 
    spinner1 = Spinner() 
    spinner1.spin_message = 'Scanning...' 
    spinner1.done_message = 'DONE' 
    spinner1.init() 
    skipped = False 

    try: 
     time.sleep(5) 
    except (KeyboardInterrupt, SystemExit): 
     skipped = True 
    finally: 
     spinner1.stop(skipped) 

    # Callable example 
    spinner2 = Spinner() 
    spinner2.spin_message = 'Scanning...' 
    spinner2.done_message = 'DONE' 
    spinner2(time.sleep, 5) 

预先感谢您。

Answer 1

您可能不需要担心赶上SystemExit，因为它会提高sys.exit()。您可能想要在程序退出之前捕获它以清理一些资源。

捕获KeyboardInterrupt的另一种方法是注册一个信号处理程序来捕获SIGINT。然而对于你的例子，使用try..except更有意义，所以你在正确的轨道上。

一些小的建议：

• 的__call__方法也许重命名为start，使之更加清楚你开始工作。
• 您可能还想通过在start方法中附加新线程而不是在构造函数中使Spinner类可重用。
• 还要考虑当用户点击当前微调工作的CTRL-C时会发生什么 - 下一个工作是否可以开始，或者应用刚刚退出？
• 您也可以将spin_message作为start的第一个参数，以将其与要运行的任务相关联。

例如，这里是某人可能会使用微调：

dbproc = MyDatabaseProc() 
spinner = Spinner() 
spinner.done_message = 'OK' 
try: 
    spinner.start("Dropping the database", dbproc.drop, "mydb") 
    spinner.start("Re-creating the database", dbproc.create, "mydb") 
    spinner.start("Inserting data into tables", dbproc.populate) 
    ... 
except (KeyboardInterrupt, SystemExit): 
    # stop the currently executing job 
    spinner.stop() 
    # do some cleanup if needed.. 
    dbproc.cleanup()