2023年7月9日 星期日

Python 學習筆記 : 在 Tkinter 中使用執行緒執行耗時函式的方法

上週完成維運自動化軟體後馬上開工寫第二個自動化軟體 (與 CSR 有關), 因為近來有關部門提交的數據量大幅增長, 我以前用網頁技術撰寫的舊版軟體因為須與系統在 timing 上同步, 導致執行時間過長, 因此決定改用 Python 來改寫. 正因為要執行的數據較多, 因此想在按下自動執行按鈕後用彈出視窗來顯示一個進度條與例如 '正在執行資料更新作業, 請稍後' 這樣的提示, 避免因為執行中無回應而讓操作者以為程式當掉了. 

在 Tkinter 中彈出視窗元件是 TopView, 它跟 root 視窗一樣可以放置各種元件, 這裡則只是要放一個進度條 ProgressBar 與一個用來顯示執行狀態的 Label 元件而已 (本來想將提示字串直接顯示在進度條上面, 但查詢 ProgressBar 物件的方法發現並無此功能, 所以進度提示必須放在 Label 上). 最重要的是, 因為執行資料更新作業與顯示執行進度必須同時進行, 因此進度條視窗必須利用另一個執行緒來跑, 否則兩個都在主執行緒的話, 進度條會在漫長的執行作業結束後才顯示, 這樣就失去意義了. 

以下是簡化過後的範例, 主視窗 root 上只有一個開始執行的按鈕, 按下去會觸發執行一個 start_running_task() 函式, 此函式會一個彈出視窗來顯示進度條, 然後建立一個執行緒來執行傳進來的那個無法確定多久可以執行完畢的函式, 彈出視窗會定期 (這裡設定每 100 ms) 去檢查該作業執行緒是否還存在 (工作完成執行緒就會自動結束), 如果執行緒已結束就關閉顯示進度條的彈出視窗, 範例程式碼如下 : 


測試 1 : 使用執行緒執行耗時函式 [看原始碼]

import tkinter as tk
from tkinter import ttk
from tkinter import messagebox as msgbox
import time
import threading

def start_running_task(task, msg):
    def check_thread():   # 檢查執行緒
        if thread.is_alive():   # 執行緒還沒結束 : 繼續每 100 ms 檢查一次
            progress_win.after(100, check_thread) # 每 100 ms 檢查執行緒 
        else:  # 若執行緒已結束關閉進度條視窗            
            progress_win.destroy()
            msgbox.showinfo('通知訊息', '執行作業已完成')
    # 建立 TopLevel 彈出視窗
    progress_win=tk.Toplevel(root)  
    progress_win.title('處理進度')
    progress_win.update_idletasks()  # 強制更新視窗更新, 處理所有事件
    progress_win.geometry('400x140')  # 將彈出視窗左上角拉到指定座標並設定大小
    progressbar=ttk.Progressbar(progress_win, mode='indeterminate', length=250)
    progressbar.pack(padx=3, pady=30)
    progressbar.start(5)   # 起始進度條並設定速度 (值越小跑越快)
    label=ttk.Label(progress_win)  # 顯示提示詞用
    label.config(text=msg, font=('Helvetica', 10, 'bold'))  # 設定字型大小
    label.pack(padx=3, pady=3)    
    thread=threading.Thread(target=task)  # 建立執行緒
    thread.start()    # 啟始執行緒
    progress_win.after(100, check_thread)  # 每 100 ms 檢查執行緒是否還存活    

def task():
    time.sleep(10)   # 模擬漫長的執行時間

root=tk.Tk()
root.title('執行緒測試')
root.geometry('600x400')
start_button=ttk.Button(root, text='開始執行', command=lambda: \
                        start_running_task(task, '執行中請稍候 ...'))
start_button.pack()
root.mainloop()

這裡在長時間作業函式 task() 中我使用 time.sleep() 來模擬那漫長的資料更新作業 (實際的系統是透過 Telnet 連線後向遠端主機連續丟出大量資料更新指令). 當按下按鈕時 command 參數的 lambda 會呼叫 start_running_task() 並傳入長時間作業函式 task 與要顯示在彈出視窗的提示字串,  start_running_task() 函式會建立一個執行緒來執行此作業函式 task(), 並用 after() 方法每 100 毫秒呼叫回呼函式 check_thread() 來檢查作業函式 task() 是否執行結束, 是的話就關閉彈出視窗, 否則繼續每 100 毫秒檢查一次直到 task() 結束, 執行緒終止才關掉彈出視窗, 結果如下 : 


 


其實不是每個作業函式都要丟給 start_running_task(), 只有作業時間較長 (通常使用者耐性大約 5~10 秒) 的作業才需要, 一般 3~5 秒就完成的函式進度條才閃一下就結束了, 丟給 start_running_task() 沒意思. 

如果要讓顯示進度條的彈出視窗顯示在整個螢幕的正中央, 可以加入設定彈出視窗 geometry 的程式碼, 如下面黃色背景的部分 :


測試 2 : 使用執行緒執行耗時函式 (彈出視窗置中) [看原始碼]

import tkinter as tk
from tkinter import ttk
from tkinter import messagebox as msgbox
import time
import threading

def start_running_task(task, msg):
    def check_thread():   # 檢查執行緒
        if thread.is_alive():   # 執行緒還沒結束 : 繼續每 100 ms 檢查一次
            progress_win.after(100, check_thread) # 每 100 ms 檢查執行緒 
        else:  # 若執行緒已結束關閉進度條視窗            
            progress_win.destroy()
            msgbox.showinfo('通知訊息', '執行作業已完成')
    # 建立 TopLevel 彈出視窗
    progress_win=tk.Toplevel(root)  
    progress_win.title('處理進度')
    progress_win.update_idletasks()  # 強制更新視窗更新, 處理所有事件
    # 設定彈出視窗左上角座標與大小始其置中
    width=400   # 彈出視窗寬度
    height=140  # 彈出視窗高度    
    screen_width=progress_win.winfo_screenwidth()    # 取得彈出視窗寬度
    screen_height=progress_win.winfo_screenheight()  # 取得彈出視窗高度
    x=(screen_width // 2) - (width // 2)    # 彈出視窗的左上角 x 座標
    y=(screen_height // 2) - (height // 2)  # 彈出視窗的左上角 y 座標  
    geometry=f'{width}x{height}+{x}+{y}'    # 設定視窗左上角座標與大小
    progress_win.geometry(geometry)  # 將彈出視窗左上角拉到指定座標並設定大小
    progressbar=ttk.Progressbar(progress_win, mode='indeterminate', length=250)
    progressbar.pack(padx=3, pady=30)
    progressbar.start(5)   # 起始進度條並設定速度 (值越小跑越快)
    label=ttk.Label(progress_win)  # 顯示提示詞用
    label.config(text=msg, font=('Helvetica', 10, 'bold'))  # 設定字型大小
    label.pack(padx=3, pady=3)    
    thread=threading.Thread(target=task)  # 建立執行緒
    thread.start()    # 啟始執行緒
    progress_win.after(100, check_thread)  # 每 100 ms 檢查執行緒是否還存活    

def task():
    time.sleep(10)   # 模擬漫長的執行時間

root=tk.Tk()
root.title('執行緒測試')
root.geometry('600x400')
start_button=ttk.Button(root, text='開始執行', command=lambda: \
                        start_running_task(task, '執行中請稍候 ...'))
start_button.pack()
root.mainloop()

這樣彈出視窗就會出現在全螢幕的正中央了. 我查書與爬文都沒找到滿意合用的, 以上的做法是我不斷調整 prompt 詢問 ChatGPT 好多次 (它的回答不一定有效) 摸索測試出來的可行架構, 特地記下來以便往後的專案參考. 如何善用 AI 加速軟體專案開發很值得探討. 


2023-07-10 補充 :

如果要在執行作業完成時顯示所耗費的時間, 可以在 start_running_task() 裡面計算起訖時間, 作法如下面範例所示 : 


測試 3 : 使用執行緒執行耗時函式 (顯示執行時間) [看原始碼]

import tkinter as tk
from tkinter import ttk
from tkinter import messagebox as msgbox
import time
import threading

def start_running_task(task, msg):
    def check_thread():   # 檢查執行緒
        nonlocal start_time   # 參照外部變數
        if thread.is_alive():   # 執行緒還沒結束 : 繼續每 100 ms 檢查一次
            progress_win.after(100, check_thread) # 每 100 ms 檢查執行緒 
        else:  # 若執行緒已結束關閉進度條視窗            
            progress_win.destroy()
            end_time=time.time()     # 結束時間
            elapsed=round(end_time - start_time, 2)   # 計算耗時
            msgbox.showinfo('通知訊息', f'執行作業已完成, 耗時 {elapsed} 秒')
    start_time=time.time()  # 開始時間 (計算執行時間用)
    # 建立 TopLevel 彈出視窗
    progress_win=tk.Toplevel(root)  
    progress_win.title('處理進度')
    progress_win.update_idletasks()  # 強制更新視窗更新, 處理所有事件
    progress_win.geometry('400x140')  # 將彈出視窗左上角拉到指定座標並設定大小
    progressbar=ttk.Progressbar(progress_win, mode='indeterminate', length=250)
    progressbar.pack(padx=3, pady=30)
    progressbar.start(5)   # 起始進度條並設定速度 (值越小跑越快)
    label=ttk.Label(progress_win)  # 顯示提示詞用
    label.config(text=msg, font=('Helvetica', 10, 'bold'))  # 設定字型大小
    label.pack(padx=3, pady=3)    
    thread=threading.Thread(target=task)  # 建立執行緒執行 task() 函式
    thread.start()    # 啟始執行緒
    progress_win.after(100, check_thread)  # 每 100 ms 檢查執行緒是否還存活    

def task():
    time.sleep(10)   # 模擬漫長的執行時間

root=tk.Tk()
root.title('執行緒測試')
root.geometry('600x400')
start_button=ttk.Button(root, text='開始執行', command=lambda: \
                        start_running_task(task, '執行中請稍候 ...'))
start_button.pack()
root.mainloop()

這裡因為檢查執行緒的關係使用了巢狀函式, 內部函式若要參照外部變數 start_time, 必須於內部函式中宣告為 nonlocal, 執行結果如下 : 




如果要讓 start_running_task() 可以傳入參數控制彈出視窗中 Label 顯示的訊息, 以及執行緒結束對話框內顯示的訊息, 改寫如下 : 


測試 4 : 使用執行緒執行耗時函式 (以參數控制顯示訊息) [看原始碼]

import tkinter as tk
from tkinter import ttk
from tkinter import messagebox as msgbox
import time
import threading

def start_running_task(task, msg_start='作業執行中 ...', msg_end='已執行完成'):
    def check_thread():   # 檢查執行緒
        nonlocal start_time   # 參照外部變數
        if thread.is_alive():   # 執行緒還沒結束 : 繼續每 100 ms 檢查一次
            progress_win.after(100, check_thread) # 每 100 ms 檢查執行緒 
        else:  # 若執行緒已結束關閉進度條視窗            
            progress_win.destroy()
            end_time=time.time()
            elapsed=round(end_time - start_time, 2)   # 計算耗時
            msgbox.showinfo('通知訊息', f'{msg_end}, 耗時 {elapsed} 秒')
    start_time=time.time()  # 計算執行時間用
    # 建立 TopLevel 彈出視窗
    progress_win=tk.Toplevel(root)  
    progress_win.title('處理進度')
    progress_win.update_idletasks()  # 強制更新視窗更新, 處理所有事件
    progress_win.geometry('400x140')  # 將彈出視窗左上角拉到指定座標並設定大小
    progressbar=ttk.Progressbar(progress_win, mode='indeterminate', length=250)
    progressbar.pack(padx=3, pady=30)
    progressbar.start(5)   # 起始進度條並設定速度 (值越小跑越快)
    label=ttk.Label(progress_win)  # 顯示提示詞用
    label.config(text=msg_start, font=('Helvetica', 10, 'bold'))  # 設定字型大小
    label.pack(padx=3, pady=3)    
    thread=threading.Thread(target=task)  # 建立執行緒
    thread.start()    # 啟始執行緒
    progress_win.after(100, check_thread)  # 每 100 ms 檢查執行緒是否還存活    

def task():
    time.sleep(10)   # 模擬漫長的執行時間

root=tk.Tk()
root.title('執行緒測試')
root.geometry('600x400')
start_button=ttk.Button(root, text='開始執行', command=lambda: \
                        start_running_task(task))
start_button.pack()
root.mainloop()

此例修改了 start_running_task() 的參數結構, 改為可傳入 msg_start 與 msg_end 兩個參數, 兩者皆有預設值, 如要更改可在呼叫 start_running_task() 時傳入參數, 例如 : 

start_running_task(task, '執行中請稍候 ...', '工作以執行完成')

下面是以預設值執行的結果 : 





如果執行結束時需要從 task() 函式取得資訊 (例如執行的指令失敗的個數) 顯示在結束的對話框要怎麼做? 這種情況要修改上面的程式碼, 添加一個內部中介函式例如 run_task() 來取得 task() 的傳回值, 並且將檢查執行緒的定時器函式 after() 移到 run_task() 裡面來並增加一個傳回值參數, 同時 check_thread() 要添加一個參數來取得 run_task() 傳回來的結果, 透過這個傳回值來製作完成對話框要顯示的資訊, 這樣也省了上面範例的 end_msg 參數了, 程式如下 : 


測試 5 : 使用執行緒執行耗時函式 (從作業函式取得傳回值) [看原始碼]

import tkinter as tk
from tkinter import ttk
from tkinter import messagebox as msgbox
import time
import threading

def start_running_task(task, msg='作業執行中 ...'):
    def check_thread(result):   # 檢查執行緒, result 為 task() 之傳回值
        nonlocal start_time     # 參照外部變數
        if thread.is_alive():   # 執行緒還沒結束 : 繼續每 100 ms 檢查一次
            progress_win.after(100, check_thread) # 每 100 ms 檢查執行緒 
        else:  # 若執行緒已結束關閉進度條視窗            
            progress_win.destroy()
            end_time=time.time()
            elapsed=round(end_time - start_time, 2)   # 計算耗時
            msgbox.showinfo('通知訊息', f'{result}\n耗時 {elapsed} 秒')
    def run_task():   # 中介函式
        result=task()     # 取得作業函式傳回值
        progress_win.after(100, check_thread, result)  # 每 100 ms 檢查執行緒是否還存活

    start_time=time.time()  # 計算執行時間用
    # 建立 TopLevel 彈出視窗
    progress_win=tk.Toplevel(root)  
    progress_win.title('處理進度')
    progress_win.update_idletasks()  # 強制更新視窗更新, 處理所有事件
    progress_win.geometry('400x140')  # 將彈出視窗左上角拉到指定座標並設定大小
    progressbar=ttk.Progressbar(progress_win, mode='indeterminate', length=250)
    progressbar.pack(padx=3, pady=30)
    progressbar.start(5)   # 起始進度條並設定速度 (值越小跑越快)
    label=ttk.Label(progress_win)  # 顯示提示詞用
    label.config(text=msg, font=('Helvetica', 10, 'bold'))  # 設定字型大小
    label.pack(padx=3, pady=3)    
    thread=threading.Thread(target=run_task)  # 建立執行緒執行中介函式 run_task()
    thread.start()    # 啟始執行緒

def job():
    time.sleep(5)   # 模擬漫長的執行時間
    error=2
    return '執行作業已完成, 錯誤數=' + str(error)

root=tk.Tk()
root.title('執行緒測試')
root.geometry('600x400')
start_button=ttk.Button(root, text='開始執行', command=lambda: \
                        start_running_task(job))
start_button.pack()
root.mainloop()

注意, 此例為了避免混淆, 以為內部函式 run_task() 呼叫的 task() 是直接呼叫外部函式 task(), 特地將要執行的漫長作業從 task() 改成 job(), 在按下執行按鈕時呼叫 start_running_task(job) 將 job 函式傳給 start_running_task(), 所以 run_task() 呼叫的 task() 其實就是外部的 job(), 結果如下 :





最後這個例子才是最終用在我的軟體專案中的做法. 

沒有留言 :