【Python】我是如何使計算時間提速25.6倍的

我是如何使計算時間提速25.6倍的

個人原始文檔：https://www.yuque.com/lart/blog/aemqfz

在顯著性目標檢測任務中有個重要的評價指標, E-measure, 須要使用在閉區間 [0, 255] 內連續變化的閾值對模型預測的灰度圖二值化. 直接的書寫方式就是使用 for 循環, 將對應的閾值送入指標得分計算函數中, 讓其計算分割後的預測結果和真值mask之間的統計類似度.
在顯著性目標檢測中, 另外一個指標, F-measure, 一樣涉及到連續變化的閾值二值化處理, 可是該指標計算僅須要precision和recall, 這兩項實際上僅須要正陽性(TP)和假陽性(FP)元素數量, 以及總的正(T)樣本元素數量. T可使用 np.count_nonzero(gt) 來計算, 而前兩項則能夠直接利用累計直方圖的策略一次性獲得全部的256個TP、FP數量對, 分別對應不一樣的閾值. 這樣就能夠很是方便且快速的計算出來這一系列的指標結果. 這其實是對於F-measure計算的一種很是有效的加速策略.
可是不一樣的是, E-measure的計算方式(須要減去對應二值圖的均值後進行計算)致使按照上面的這種針對變化閾值加速計算的策略並不容易變通, 至少我目前沒有這樣使用. 可是最後我找到了一種更加(相較於原始的 for 策略)高效的計算方式, 這裏簡單作一下思考和實驗重現的記錄.

選擇使用更合適的函數

雖然運算主要基於 numpy 的各類函數, 可是針對同一個目的不一樣的函數實現方式也是有明顯的速度差別的, 這裏簡單彙總下:

統計非零元素數量首選 np.count_nonzero(array)

我想到的針對二值圖的幾種不一樣的實現:

import time
import numpy as np

# 快速統計numpy數組的非零值建議使用np.count_nonzero，一個簡單的小實驗
def cal_nonzero(size):
    a = np.random.randn(size, size)
    a = a > 0
    start = time.time()
    print(np.count_nonzero(a), time.time() - start)
    start = time.time()
    print(np.sum(a), time.time() - start)
    start = time.time()
    print(len(np.nonzero(a)[0]), time.time() - start)
    start = time.time()
    print(len(np.where(a)), time.time() - start)

if __name__ == '__main__':
    cal_nonzero(1000)
    # 499950 6.723403930664062e-05
    # 499950 0.0006949901580810547
    # 499950 0.007088184356689453

能夠看到, 最合適的是 np.count_nonzero(array) 了.

更快的交集計算方式

import time
import numpy as np

# 快速統計numpy數組的非零值建議使用np.count_nonzero，一個簡單的小實驗
def cal_andnot(size):
    a = np.random.randn(size, size)
    b = np.random.randn(size, size)
    a = a > 0
    b = b < 0
    start = time.time()
    a_and_b_mul = a * b
    _a_and__b_mul = (1 - a) * (1 - b)
    print(time.time() - start)
    start = time.time()
    a_and_b_and = a & b
    _a_and__b_and = ~a & ~b
    print(time.time() - start)

if __name__ == '__main__':
    cal_andnot(1000)
    # 0.0036919116973876953
    # 0.0005502700805664062

可見, 對於bool數組, numpy的位運算是要更快更有效的. 並且bool數組能夠直接用來索引矩陣即 array[bool_array] , 很是方便.

邏輯的改進

通過儘量的挑選更加快速的計算函數以後, 目前速度受限的最大問題就是這個 for 循環中的256次矩陣運算了. 也就是這部分代碼:

...
    def step(self, pred: np.ndarray, gt: np.ndarray):
        pred, gt = _prepare_data(pred=pred, gt=gt)
        self.all_fg = np.all(gt)
        self.all_bg = np.all(~gt)
        self.gt_size = gt.shape[0] * gt.shape[1]

        if self.changeable_ems is not None:
            changeable_ems = self.cal_changeable_em(pred, gt)
            self.changeable_ems.append(changeable_ems)
        adaptive_em = self.cal_adaptive_em(pred, gt)
        self.adaptive_ems.append(adaptive_em)

    def cal_adaptive_em(self, pred: np.ndarray, gt: np.ndarray) -> float:
        adaptive_threshold = _get_adaptive_threshold(pred, max_value=1)
        adaptive_em = self.cal_em_with_threshold(pred, gt, threshold=adaptive_threshold)
        return adaptive_em

    def cal_changeable_em(self, pred: np.ndarray, gt: np.ndarray) -> list:
        changeable_ems = [self.cal_em_with_threshold(pred, gt, threshold=th) for th in np.linspace(0, 1, 256)]
        return changeable_ems

    def cal_em_with_threshold(self, pred: np.ndarray, gt: np.ndarray, threshold: float) -> float:
        binarized_pred = pred >= threshold
        if self.all_bg:
            enhanced_matrix = 1 - binarized_pred
        elif self.all_fg:
            enhanced_matrix = binarized_pred
        else:
            enhanced_matrix = self.cal_enhanced_matrix(binarized_pred, gt)
        em = enhanced_matrix.sum() / (gt.shape[0] * gt.shape[1] - 1 + _EPS)
        return em

    def cal_enhanced_matrix(self, pred: np.ndarray, gt: np.ndarray) -> np.ndarray:
        demeaned_pred = pred - pred.mean()
        demeaned_gt = gt - gt.mean()
        align_matrix = 2 * (demeaned_gt * demeaned_pred) / (demeaned_gt ** 2 + demeaned_pred ** 2 + _EPS)
        enhanced_matrix = (align_matrix + 1) ** 2 / 4
        return enhanced_matrix
    ...

能夠看到, 這裏對於每個閾值都要計算一遍一樣的流程, 若是每次的計算都比較耗時的話, 那麼整體時間也就很難減下來. 因此須要探究如何下降這裏的 cal_enhanced_matrix 的耗時.

前面的嘗試都是在代碼函數選擇層面的改進, 可是對於這裏, 這樣的思路已經很難產生明顯的效果了. 那麼咱們就應該轉變思路了, 應該從計算流程自己上思考. 能夠按照下面這一系列思考來引出最終的一種比較好的策略.

• 這裏計算爲何會那麼慢?
  • 由於涉及到了大量的矩陣元素級的運算, 例如元素級減法、加法、乘法、平方、除法.
• 大量的元素級運算是否能夠優化?
  • 必須能夠:<
• 如何優化元素級運算?
  • 尋找規律性、重複性的計算, 將其合併、消減, 能夠聯想numpy的稀疏矩陣的思想.
• 規律性、重複性的計算在哪裏?
  • 去均值其實是對每一個元素減去了相同的一個值, 若是被減數能夠優化, 那麼這一步就能夠被優化
  • 元素乘法和平方涉及到兩部分, demeaned_gt和demeaned_pred, 若是這兩個能夠被優化, 那麼這些運算就均可以被優化
  • 這些元素運算的連鎖關係致使了只要咱們優化了最初的pred和gt, 那麼整個流程就均可以被優化
• 如何優化pred和gt的表示?
  • 這裏須要從兩者自己的屬性上入手
• 兩者最大的特色是什麼?
  • 都是二值數組, 只有0和1
• 那如何優化?
  • 實際上就借鑑了稀疏矩陣的思想, 既然存在大量的重複性, 那麼咱們就將數值與位置解耦, 優化表示方式
• 如何解耦?
  • 以gt爲例, 能夠表示爲0和1兩種數據, 其中0對應背景, 1對應前景, 0的數量表示背景面積, 1的數量表示前景面積
• 那如何使用該思想重構前面的計算呢?

到最後一個問題, 實際上核心策略已經出現, 就是"解耦", 將數值與位置解耦. 這裏須要具體分析下, 咱們直接將 pred 和 gt 拆分紅數值和數量, 是能夠比較好的處理 demeaned_* 項的表示的, 也就是:

# demeaned_pred = pred - pred.mean()
# demeaned_gt = gt - gt.mean()
pred_fg_numel = np.count_nonzero(binarized_pred)
pred_bg_numel = self.gt_size - pred_fg_numel
gt_fg_numel = np.count_nonzero(gt)
gt_bg_numel = self.gt_size - gt_fg_numel

mean_pred_value = pred_fg_numel / self.gt_size
mean_gt_value = gt_fg_numel / self.gt_size

demeaned_pred_fg_value = 1 - mean_pred_value
demeaned_pred_bg_value = 0 - mean_pred_value
demeaned_gt_fg_value = 1 - mean_gt_value
demeaned_gt_bg_value = 0 - mean_gt_value

接下來須要進一步優化後面的乘法和加法了, 由於這裏同時涉及到了同一位置的 pred 和 gt 的值, 這就須要注意了, 由於兩者前景與背景對應關係並不明確, 這就得分狀況考慮了. 整體而言, 包含四種狀況, 就是:

1. pred: fg; gt: fg
2. pred: fg; gt: bg
3. pred: bg; gt: fg
4. pred: bg; gt: bg

而這些區域其實是對前面初步解耦區域的進一步細化, 因此咱們從新整理思路, 能夠將整個流程構造以下:

fg_fg_numel = np.count_nonzero(binarized_pred & gt)
fg_bg_numel = np.count_nonzero(binarized_pred & ~gt)

# bg_fg_numel = np.count_nonzero(~binarized_pred & gt)
bg_fg_numel = self.gt_fg_numel - fg_fg_numel
# bg_bg_numel = np.count_nonzero(~binarized_pred & ~gt)
bg_bg_numel = self.gt_size - (fg_fg_numel + fg_bg_numel + bg_fg_numel)

parts_numel = [fg_fg_numel, fg_bg_numel, bg_fg_numel, bg_bg_numel]

mean_pred_value = (fg_fg_numel + fg_bg_numel) / self.gt_size
mean_gt_value = self.gt_fg_numel / self.gt_size

demeaned_pred_fg_value = 1 - mean_pred_value
demeaned_pred_bg_value = 0 - mean_pred_value
demeaned_gt_fg_value = 1 - mean_gt_value
demeaned_gt_bg_value = 0 - mean_gt_value

combinations = [(demeaned_pred_fg_value, demeaned_gt_fg_value), (demeaned_pred_fg_value, demeaned_gt_bg_value),
                (demeaned_pred_bg_value, demeaned_gt_fg_value), (demeaned_pred_bg_value, demeaned_gt_bg_value)]

這裏忽略掉了一些沒必要要的計算, 能直接使用現有量就使用現有的量.

針對前面的這些解耦, 後面就能夠比較簡單的書寫了:

results_parts = []
for part_numel, combination in zip(parts_numel, combinations):
    # align_matrix = 2 * (demeaned_gt * demeaned_pred) / (demeaned_gt ** 2 + demeaned_pred ** 2 + _EPS)
    align_matrix_value = 2 * (combination[0] * combination[1]) / \
                            (combination[0] ** 2 + combination[1] ** 2 + _EPS)
    # enhanced_matrix = (align_matrix + 1) ** 2 / 4
    enhanced_matrix_value = (align_matrix_value + 1) ** 2 / 4
    results_parts.append(enhanced_matrix_value * part_numel)

# enhanced_matrix = enhanced_matrix.sum()
enhanced_matrix = sum(results_parts)

因爲不一樣區域元素結果一致, 而區域的面積也已知, 因此最終 cal_em_with_threshold 中的 enhanced_matrix.sum() 其實更適合放到 cal_enhanced_matrix 中, 能夠一便計算出來.

爲了儘量重用現有變量, 咱們其實反過來能夠優化 cal_em_with_threshold :

binarized_pred = pred >= threshold
if self.all_bg:
    enhanced_matrix = 1 - binarized_pred
elif self.all_fg:
    enhanced_matrix = binarized_pred
else:
    enhanced_matrix = self.cal_enhanced_matrix(binarized_pred, gt)
em = enhanced_matrix.sum() / (gt.shape[0] * gt.shape[1] - 1 + _EPS)

這裏的 self.all_bg 和 self.all_fg 實際上可使用 self.gt_fg_numel 和 self.gt_size 表示, 也就是隻需計算一次 np.count_nonzero(array) 就能夠了. 另外在 cal_em_with_threshold 中 if 的前兩個分支中, 須要將 sum 整合到各個分支內部(else分支已經被整合到了 cal_enhanced_matrix 方法中), (1-binarized_pred).sum() 和 binarized_pred.sum() 實際上就是表示背景像素數量和前景像素數量. 因此能夠藉助於更快的 np.count_nonzero(array) , 從而改爲以下形式:

binarized_pred = pred >= threshold

if self.gt_fg_numel == 0:
    binarized_pred_bg_numel = np.count_nonzero(~binarized_pred)
    enhanced_matrix_sum = binarized_pred_bg_numel
elif self.gt_fg_numel == self.gt_size:
    binarized_pred_fg_numel = np.count_nonzero(binarized_pred)
    enhanced_matrix_sum = binarized_pred_fg_numel
else:
    enhanced_matrix_sum = self.cal_enhanced_matrix(binarized_pred, gt)
em = enhanced_matrix_sum / (self.gt_size - 1 + _EPS)

效率對比

使用本地的845張灰度預測圖和二值mask真值數據進行測試比較, 整體時間對好比下:

• 'base': 503.5014679431915s
• 'best': 19.27734637260437s

雖然具體時間可能還受硬件限制, 可是相對快慢仍是比較明顯的. 變爲原來的19/504~=4%, 快了504/19~=26.5倍.

測試代碼可見個人 github : https://github.com/lartpang/CodeForArticle/tree/main/sod_metrics