python自定義時(shí)鐘類、定時(shí)任務(wù)類

import time
import threading


class _Clock:
  """
  自定義的時(shí)鐘類，用于獲取幾種不同格式的當(dāng)前時(shí)間。
   decimal : 設(shè)置time_float的精度，控制其保留幾位小數(shù)。
   time_diff : 設(shè)置該時(shí)鐘與UTC+0時(shí)區(qū)的時(shí)差。如果不設(shè)置，會(huì)自動(dòng)采用
   本地時(shí)區(qū)。
  """

  def __init__(self, name=None, decimal=3, time_diff=None):
    self.name = name
    self.decimal = decimal
    self.time_diff = time_diff
    self.time_format = "%Y/%m/%d %H:%M:%S" # 時(shí)間字符串的格式

  @property
  def time_float(self):
    """ UTC+0時(shí)區(qū)的時(shí)間戳，精度由self.decimal決定 """
    return round(time.time(), self.decimal)

  @property
  def time_int(self):
    """ UTC+0時(shí)區(qū)的時(shí)間戳，精度為秒 """
    return int(time.time())

  @property
  def time_tuple(self):
    """ 本地時(shí)區(qū)的時(shí)間元組 """
    if self.time_diff == None:
      return time.localtime(self.time_int)
    else:
      return time.gmtime(self.time_int+self.time_diff)

  @property
  def time_str(self):
    """ 本地時(shí)間的格式化字符串 """
    return time.strftime(self.time_format, self.time_tuple)

class Timer(_Clock):
  """
  自定義的計(jì)時(shí)器，像秒表一樣，可以隨時(shí)查看當(dāng)前計(jì)時(shí)、暫停計(jì)時(shí)、繼續(xù)計(jì)時(shí)。
   · 創(chuàng)建一個(gè)計(jì)時(shí)器之后，它就會(huì)開始計(jì)時(shí)。
   · 默認(rèn)使用time.time()獲取時(shí)間，精度為毫秒。
   · 可以直接調(diào)用_Clock類的方法來獲取當(dāng)前時(shí)間。
  """

  def __init__(self, *args, **kwargs):
    _Clock.__init__(self, *args, **kwargs)
    self.record = [] # 記錄每次使用的 (開始時(shí)刻,暫停時(shí)刻,計(jì)時(shí)時(shí)長(zhǎng))
    self.status = "initial"
    self.go()

  @property
  def count(self):
    """ 當(dāng)前計(jì)時(shí)值 """
    count = 0
    for line in self.record:
      if line[2] == None:
        count += self.time_float - line[0]
      else:
        count += line[2]
    return round(count, self.decimal)

  def go(self):
    """ 開始計(jì)時(shí) """
    if self.status != "timing":
      self.record.append((self.time_float, None, None))
      self.status = "timing"

  def pause(self):
    """ 暫停計(jì)時(shí) """
    # 如果該計(jì)時(shí)器在計(jì)時(shí)中，就暫停它，并計(jì)算這一段的計(jì)時(shí)時(shí)長(zhǎng)
    if self.status == "timing":
      last_line = self.record[-1]
      self.record.remove(last_line)
      current_time = self.time_float
      self.record.append(
        (last_line[0], current_time, round(current_time - last_line[0], self.decimal)))
      self.status = "paused"

class Schedule(threading.Thread):
  """
  自定義的定時(shí)任務(wù)表，添加第一個(gè)定時(shí)任務(wù)后就創(chuàng)建一個(gè)線程，開始循環(huán)檢查
  是否執(zhí)行任務(wù)表中的任務(wù)。
   · 調(diào)用stop()來終止該線程。
  """

  def __init__(self, *args, **kwargs):
    threading.Thread.__init__(self, *args, **kwargs)
    self._askToStop = False
    self._schedule = [] # 保存定時(shí)任務(wù)表
    self.status = "initial"

  def _get_time(self):
    """ 獲取當(dāng)前時(shí)間 """
    return time.time()

  def addTask(self, countDown, func, *args, **kwargs):
    """ 
    在任務(wù)表中增加一項(xiàng)定時(shí)任務(wù)：在倒計(jì)時(shí)countDown結(jié)束之后調(diào)用
    函數(shù)func，并傳入?yún)?shù)*args和**kwargs。
     · 定時(shí)任務(wù)只會(huì)被執(zhí)行一次，執(zhí)行后就會(huì)被從任務(wù)表中刪除。
     · 定時(shí)任務(wù)只會(huì)在倒計(jì)時(shí)結(jié)束之后被執(zhí)行，但無法保證無延遲。
    """
    if self.status == "initial": # 第一次添加定時(shí)任務(wù)時(shí)創(chuàng)建一個(gè)新線程
      self.status = "running"
      self.start()

    task = []
    if isinstance(countDown, (int, float)) and countDown > 0:
      task.append(self._get_time()+countDown) # 準(zhǔn)備在指定時(shí)刻執(zhí)行該任務(wù)
    else:
      raise ValueError("'countDown' must be a positive int or float.")
    if callable(func):
      task.append(func)
    else:
      raise ValueError("'func' must be callable.")
    task.append(args) # 保存元組參數(shù)
    task.append(kwargs) # 保存字典參數(shù)

    self._schedule.append(task)
    self._schedule.sort(key=lambda task: task[0]) # 將任務(wù)表按時(shí)間戳的大小排序

  def _doTask(self):
    """ 檢查任務(wù)表中各項(xiàng)任務(wù)的時(shí)間，判斷是否要執(zhí)行它。 """
    current_time = self._get_time()
    i = 0
    while i < len(self._schedule): # 遍歷任務(wù)表
      task = self._schedule[i]
      if task[0] <= current_time:
        # 如果該任務(wù)的時(shí)間不晚于當(dāng)前時(shí)間，就創(chuàng)建一個(gè)線程去執(zhí)行該任務(wù)，避免阻塞定時(shí)器線程
        t1 = CreatThread(task[1], *task[2], **task[3])
        t1.start()
        i += 1
      else:
        break # 如果該任務(wù)的時(shí)間戳大于當(dāng)前時(shí)間，就提前結(jié)束遍歷

    del self._schedule[:i] # 刪除過時(shí)的任務(wù)

  def run(self):
    """ 線程循環(huán)運(yùn)行的內(nèi)容 """
    while not self._askToStop:
      self._doTask()

    # 結(jié)束時(shí)進(jìn)行清理
    self.status == "stopped"
    return 0

  def stop(self):
    self._askToStop = True


class CreatThread(threading.Thread):
  """ 一個(gè)簡(jiǎn)單的創(chuàng)建線程的類 """

  def __init__(self, func, *args, **kwargs):
    threading.Thread.__init__(self)
    self.func = func
    self.args = args
    self.kwargs = kwargs

  def run(self):
    self.func(*self.args, **self.kwargs)