[Python 小東西] 排程工具 Scheduling Tool

情境敘述

情境:現在要打造一座城堡,目前已經有任務清單,例如建造牆壁、打地基、安裝窗戶等等,但還沒有排序哪個任務要先完成,目前有個檔案有每個任務的 index、每個任務的名稱以及每個任務的先備條件,目前我們需要一個能夠排序任務的小工具。

基本拓樸排序

一些設定和排序概念

Task 要儲存 index、任務名、必備條件清單。
必備條件清單也可以說是先備條件清單(prerequisite),以製作麵包為例子,步驟大致上可分為:

  • 準備材料(水、麵粉、鹽、酵母)
  • 秤重
  • 攪拌
  • 發酵
  • 放進烤箱

那麼秤重這一項步驟,它的先備條件就是準備材料;發酵這一項步驟,所需要發動條件為準備材料、秤重、攪拌都完成後才可以繼續。

PoSorter 的作用是 task 的排序,設計幾項方法:

  • topo_sort 要實現拓樸排序。呼叫 Task 設定目前的任務追蹤清單。任務狀態也有分就緒的任務和處理中任務,排序好後要將任務儲存在 sorted_tasks 中。
  • verify_sort 方法目的是要驗證每個在清單出現的任務都在先備條件之後(例如攪拌一定是出現在準備材料、秤重這兩項完成後)。因為必須逐一檢查,所以會設計一個 loop。
  • load_po_file 方法要從有部份排序任務的文件載入,當中我們需要 read_task 方法去讀取文件的每一行,並且回傳一個新的 Task 或是 None
  • read_task 方法主要是讀取檔案,分析檔案結構後儲存並製造新的 Task

上述設計中題到的檔案格式如下:

 0, Install windows, [25]
 1, Grade site, []
 2, Pour foundation, [1, 9]
 3, Frame walls and roof, [2]
 4, Install insulation, [3]
 5, Hang drywall, [4, 7, 13, 14, 20]
...

最前面是 index,再來是任務名稱,最後是先備條件的 index。

到我的 github 觀看完整程式碼

演算法

其實也沒什麼太高深的演算法XD 主要是設計好要怎麼排序每個任務,都在有符合條件的狀況下安排好。例如總不可能地基沒打好就先製作窗戶吧XD 上面已經有安排 PoSorter 的幾項 method,大致會這樣進行:

  1. 將所有任務初始化。任務有先備條件,那最好也要有追蹤清單(這裡稱為 followers)。追蹤清單是要追蹤有那些任務是認這一項任務為先備條件。例如拿做麵包的例子來說,攪拌的先備任務是準備材料秤重,那麼秤重的追蹤清單就會有攪拌(以及後面的發酵、放進烤箱)。
  2. 建立一個空的 sorted_tasks 清單。
  3. 建立一個就緒清單 ready_tasks
  4. 把沒有先備條件的任務加到 ready_tasks
  5. 如果 ready_tasks 不是空的話,重複以下動作:
    1. 從清單移除任務 task
    2. 巡覽這個 taskfollowers,因為 task 已經完成,follower 中的每個任務的先備條件數目 prereq_count 可以減一(要準備的事情少一項)。如果做完以上動作後發現 prereq_count 等於 0,表示這一項任務也準備好了,可以移去 ready_tasks 清單內。
def topo_sort(self):
    for task in self.tasks:
        task.followers = []
    
    for task in self.tasks:
        for prereq in task.prereq_tasks:
            prereq.followers.append(task)
        task.prereq_count = len(task.prereq_tasks)
    
    # Move tasks with no prerequisites onto the ready list.
    ready_tasks = []
    for task in self.tasks:
        if task.prereq_count == 0:
            ready_task.append(task)
    
    # sorted task list
    self.sorted_tasks = []
    
    while len(ready_tasks) > 0:
        ready_task = ready_tasks.pop(0)
        self.sorted_tasks.append(ready_task)
        
        for follower in ready_task.followers:
            follower.prereq_count -= 1
            if follower.prereq_count == 0:
                ready_tasks.append(follower) # no prereqs, add it to the ready list

真的要說個演算法名字,類似 Kuhn’s algorithm,但有稍微改變過,不會完整地套用這個演算法。

能夠選取任一符合格式的檔案進行任務排序
設計一個視窗告知是否成功

到我的 github 觀看完整程式碼

PERT 圖表

做法說明

這個圖表可以用於規畫專案任務,是一個排定、組織的工具,全名為計劃評估與審核技術(Program Evaluation and Review Technique),和甘特圖很像但是結構不太一樣。關於 PERT 圖表可以查看這個頁面:PERT 圖表

在上個部份中,我們設計好了安排任務順序的功能,但還沒有真正進行到如何「計劃」或是「安排」進入時間表內(任務的時程最重要了!),因此在這邊我採用 PERT 圖表試試看。

以目前這個題目的情境題來說,每一項任務會牽涉到的的工人有木工、水電工、水泥工等等,若任務不衝突的話,他們其實都可以同時作業。PERT 圖除了可以安排任務外,還可以分辨哪些任務是依賴於哪個任務的完成,例如要建造窗戶,牆必須要先做好,如果牆面的工程延遲了尚未做好,則窗戶任務就勢必得延遲。

在這個部份中,我們一樣會沿用上個部份的 TaskPoSorter 功能,另外還要製作最最最重要的 PERT 圖,因此會建議先大概理解 PERT 圖的建立方式,再回來設計這個部份的功能。在做 PERT 圖表之前,也必須初始化 Task。把先前寫的 topo_sort 一部分搬過來用在 build_pert_chart 的新方法中。

建立 PERT 圖表的演算法

和先前做拓樸排序的方式很像,因此會有部份程式碼重複。在拓樸排序的時候,會直接將沒有先備條件要執行的任務(也就是已經就緒的任務)丟到 ready_tasks。而要建立 PERT 圖表的話,會選擇將這種就緒的任務放去 new_ready_tasks 清單內,當 ready_tasks 空了後,在把 new_ready_tasks 丟去 ready_tasks(這樣 new_ready_tasks 又空了),一直這樣循環下去。

以上面的做法來說,應該會發現原本線性的任務清單(做完一件事做下一件),會變成類似這裡有一綑任務,接下來有另一綑任務(同時可能會有三、四個任務進行)。像這樣的資料會存在 columns 裡面。

畫 PERT 圖表

接下來就是我覺得最痛苦的部份:描繪出 PERT 圖表,這部分寫在 draw_pert_chart 裡面,會用 Canvas 實作。

把上面的 columns 清單內的東西具現化 🙂 總而言之就是畫線啊加箭頭之類的,因此要知道每次迴圈畫完後下一個 x 座標和 y 座標要怎麼訂,不太想多說明這部份好麻煩QQ 簡而言之可以直接結果,整個實作起來有點像設計網頁前端那樣,只是用 Canvas 做出來。比較重要的事要怎麼把 columns 內的東西提取出來當作一個一個元素,並將一個一個元素畫出、連結。

下方程式碼是實作演算法的部份:

def build_pert_chart(self):
    self.prepare_tasks()
    
    # Move tasks with no prerequisites onto the ready list.
    ready_tasks = []
    for task in self.tasks:
        if task.prereq_count == 0:
            ready_tasks.append(task)

    self.sorted_tasks = [] # sorted task list
    self.columns = [] # columns list
    
    while len(ready_tasks) > 0:
        new_column = []
        self.columns.append(new_column)
        
        new_ready_tasks = []
        while len(ready_tasks) > 0:
            ready_task = ready_tasks.pop(0)
            new_column.append(ready_task)
            self.append(ready_task)

            for follower in ready_task.followers:
                follower.prereq_count -= 1
                if follower.prereq_count == 0:
                    new_ready_tasks.append(follower) # no prereqs, add it to the new ready list
        
        ready_tasks = new_ready_tasks

再來要將圖畫出來的部份,將每個就緒的 task 當成一個 cell:

def draw_pert_chart(self, canvas):
    canvas.delete('all')
    
    x_spacing = 40
    y_spacing = 10
    margin = 10
    cell_width = 30
    cell_height = 30
    
    for task in self.tasks:
        task.cell_bounds = None
    
    x = margin
    for column in self.columns:
        y = margin
        for task in column:
            task.cell_bounds = (x, y, x + cell_width, y + cell_height)
            y += cell_height + y_spacing
        x += cell_width + x_spacing
        
    for task in self.tasks:
        if task.cell_bounds == None:
            continue
        
        x1 = task.cell_bounds[0]
        y1 = (task.cell_bounds[1] + task.cell_bounds[3]) / 2
        
        for prereq in task.prereq_tasks:
            # Get the middle of the right edge of the prereq's cell.
            x2 = prereq.cell_bounds[2]
            y2 = (prereq.cell_bounds[1] + prereq.cell_bounds[3]) / 2
            canvas.create_line(x1, y1, x2, y2, arrow='first')
        
    for task in self.tasks:
        # Skip tasks that were not placed in the chart.
        if task.cell_bounds == None:
            continue

        canvas.create_rectangle(*task.cell_bounds, fill='white', outline='blue')

        # Get the center of the task's cell.
        x = (task.cell_bounds[0] + task.cell_bounds[2]) / 2
        y = (task.cell_bounds[1] + task.cell_bounds[3]) / 2
        canvas.create_text(x, y, text=task.index, fill='black', anchor='center')

到我的 github 觀看完整程式碼

關鍵路徑 Critical path method

OK,這個部份又有更有趣的玩法了。一樣從上個部份的 PERT 圖表延伸,透過關鍵路徑法(critical path method,CPM)或是關鍵路徑分析(critical path analysis,CPA)。關於這項方法可以查看一篇文章:關鍵路徑法 去理解。越做越有以前上圖論課的味道了……

CPM 有幾個必要的資料:

  1. 列出所有任務項目
  2. 每項工作的持續時間
  3. 任務依存關係(需要先完成哪項工作、完成工作後要執行哪項工作、什麼工作應同時完成)

聽起來已經在上一個部份完成了,所以我們應該可以專注在建構模型的部份。(畫起來應該幾乎一樣,資料差在持續時間)其中關鍵路徑是整個專案從開始到結束必須完成的最長任務路徑。

另外原始檔的部份也有變化,在這個方法中需要加上任務期間,所以格式改成如下:

0, 0, Start, []
 1, 3, Install windows, [26]
 2, 4, Grade site, [0]
 3, 1, Pour foundation, [2, 10]
 4, 6, Frame walls and roof, [3]
 5, 2, Install insulation, [4]
...

如果有連結到完整的程式碼,應該會發現變動較大的部份集中在 draw_pert_chart

到我的 github 觀看完整程式碼

甘特圖 Gantt Charts

甘特圖相對於上面兩個圖來說較為常見,好啦以我淺薄的知識來說。上面兩張圖雖然也是專案規劃,但比較沒有辦法很直覺看到一個任務的執行長度,而 PERT 圖表也無法直接顯示任務可能延誤的數量。

甘特圖最大的好處就是對於任務時間長度我們會比較能夠直覺的去評估,且也可以很輕易觀察到,有些任務即使延後了,其實不會對後面的任務有影響XD 對於任何任務的開始或是結束也都很清楚,會發現真的比上面兩張平易近人許多。

另外甘特圖會用到比較大的空間(每個任務都有一個時間軸),因此這部份的實作會是以 14 個任務 (實際上檔案有 16 個任務,多了開頭和結尾)

 0, 0, Start, []
 1, 1, Windows, [12]
 2, 2, Grade, [0]
 3, 1, Foundation, [2]
 4, 3, Frame, [3]
 5, 1, Insulation, [4]
 6, 4, Drywall, [5, 8]
 7, 2, Bathrooms, [6]
 8, 6, Rough in, [4]
 9, 7, Exterior, [12, 13]
10, 1, Furnace/AC, [8]
11, 1, Moat monster, [2]
12, 4, Siding, [4]
13, 3, Roof, [4]
14, 5, Interior, [6]
15, 1, Pay fines, [1, 7, 9, 11, 14]
16, 0, Finish, [10, 14, 15]

我們可以直接拿剛剛做好的那一版再繼續改。新增一個方法叫做 draw_gantt_chart,並嘗試改成甘特圖那樣XDDD 作法:

  • 設定大小、顏色等等,像是左手邊顯示任務名稱的寬度、和關鍵路徑方法一樣分成標示 critical 和非 critical 的顏色。
  • 畫格線,因為甘特圖的特性就是要看時間間隔,格線相對重要,較能比較出任務長度。
  • 如果以表格的列和行來說,橫列的是任務名稱,直行的部份是日期。

內容大同小異,比較多變化的主要是要如何畫出長條,以及設定長條的長相。超級繁雜,凡是要顯示圖啊視窗的,都好繁雜……不過因為已經經歷過前面的繪圖過程了,甘特圖這部分說不定比較容易消化,主要是細節會多很多。基本上演算法是一樣的,只要動繪圖的部份就夠了。

畫格線

之前的圖表也都有用類似的設定,但甘特圖較為複雜,稍微標示一下,其餘的概念大同小異

margin 是邊緣保留空白)

max_duration 取最後一個 task 的結束時間
xmin = margin + 文字欄位寬度(text_width
ymin = margin
xmax = xmin + 格子寬度(day_width)* 天數(max_duration
ymax = margin + 格子高度(row_height)* 列數量(tasks 數量 + 1,要多一格標示天數)

畫橫線
畫出由座標 (xmin, ymin) 起始到 (xmax, ymin) 結束的第一條線,依序畫到最後一條。
每畫一條 y 都要加上格子高度再畫下一條,作變化的是 y
(因為要容納最上面日期的關係,要再多畫兩條)

畫直線
畫出由座標 (xmin, ymin) 起始到 (xmax, ymin) 結束的第一條線,依序畫到最後一條。
每畫一條 x 都要加上格子寬度再畫下一條,作變化的是 x
(兩邊都要有線,再多畫一條)

max_duration = self.tasks[-1].end_time
xmin = margin + text_width
ymin = margin
xmax = xmin + day_width * max_duration
ymax = margin + row_height * (len(self.tasks) + 1)

y = margin
for r in range(len(self.tasks) + 2):
    canvas.create_line(xmin, y, xmax, y, fill=grid_color)
    y += row_height

x = xmin
for c in range(max_duration + 1):
    canvas.create_line(x, ymin, x, ymax, fill=grid_color)
    x += day_width

最上面的天數編號

數字本身基本上就是每一輪迴加一,數字位置的話主要是 x 座標作變化,y 座標固定在同一個高度。

x = xmin + day_width / 2
y = margin + text_y_margin
for c in range(max_duration):
    canvas.create_text(x, y, text=f'{c + 1}', fill=day_num_color, anchor='n')
    x += day_width

任務名稱的欄位

列印每個任務編號 + 任務名稱(較簡單就先不寫在這)。記得要保留最上面的天數列,所以 task 數量之外還要加一。除了印出名稱之外,還要順便畫出長條圖。這裡要注意長條是有寬度的,所以除了 x 座標要定起始和最末端位置外,y 座標也要記得設定。

另外 task 隨著是否為 critical,也有不同的著色方式,要記得顏色也要變化。

for task in self.tasks:
    # Draw the task's rectangle.
    x1 = margin + text_width + task.start_time * day_width # 長條起始位置
    x2 = x1 + task.duration * day_width # 計算每個任務區間*格子寬度並加上起始位置
    y1 = margin + (task.index + 1) * row_height + day_y_margin
    y2 = y1 + day_height
    if task.is_critical:
        canvas.create_rectangle(x1, y1, x2, y2, fill=critical_task_color, outline=critical_task_outline)
    else:
        canvas.create_rectangle(x1, y1, x2, y2, fill=non_critical_task_color, outline=non_critical_task_outline)

畫任務間的連結

for task in self.tasks:
    # Draw arrows to prereqs.
    x1 = margin + text_width + task.start_time * day_width
    task_x = x1 + prereq_hgap
    task_top = margin + (task.index + 1) * row_height + day_y_margin
    task_bottom = margin + (task.index + 2) * row_height - day_y_margin
    for prereq in task.prereq_tasks:
        prereq_y = margin + (prereq.index + 1.5) * row_height # Vertical center of prereq.
        prereq_x = margin + text_width + prereq.end_time * day_width

        # See if we should draw from the top or bottom of the task.
        if prereq.index < task.index:
            task_y = task_top
        else:
            task_y = task_bottom

        # See if this link is critical for this task.
        link_color = non_critical_link_color
        link_width = non_critical_link_width

        if prereq.end_time == task.start_time:
            link_width = critical_link_width
            if task.is_critical:
                link_color = 'red'

        canvas.create_line(task_x,   task_y,   task_x, prereq_y, width=link_width, fill=link_color, arrow='first')
        canvas.create_line(task_x, prereq_y, prereq_x, prereq_y, width=link_width, fill=link_color)

        task_x += prereq_hgap

到我的 github 觀看完整程式碼

參考資料


後記

在這個 project 學到很多,連 PERT 圖表都是第一次聽到XDD 但我想知道這些流程除了練習這個 project 之外,對於我日後安排計劃也會有一點幫助。

另外視覺化在規劃上還真的是我的罩門,即使已經省略很多部份了,還是有太多細節弄到想哭QQ

讓我知道你在想什麼!