【原創文章】歡迎正常授權轉載(聯系作者)
【反對惡意復制粘貼，如有發現必維權】
【微信公眾號原文傳送門】

---

?這篇文章將詳細介紹利用多進程的實現—方案3(代碼獲取見文章末尾)。相比之前的稍微復雜一點，先看看demo的最終效果（視頻）。

---

1 需求分析

首先看一下UI界面，界面上各個控件的詳細信息如下表所示。

控件序號控件的類別Qobject_name功能

1	QLabel	label_imgshow	實時顯示視頻
2	QLabel	label_imgshaow_res	顯示抽幀檢測效果
3	QTextEdit	textEdit	顯示檢測的目標信息
4	QPushButton	pushButton_open	打開視頻文件
5	QLineEdit	lineEdit_cameraIndex	設置視頻流URL;支持IP攝像頭;數字為當前設備中攝像頭索引(例:本人筆記本自帶攝像頭為0)
6	QPushButton	pushButton_start	開始檢測并播放
7	QPushButton	pushButton_pause	暫停檢測及播放
8	QPushButton	pushButton_end	停止檢測及播放(全部重設)

結合上面的控件分析一下需求。方案3最初的設計需求是電腦的性能有限，無法做到實時檢測并顯示，我們希望做一個折中，不檢測視頻流中的每一幀圖像，只是從中抽取部分來檢測，檢測時不打斷畫面的實時顯示，在"后臺"中盡可能多的檢測視頻流中的圖像。之前的文章中也介紹過神經網絡的預測過程是無法在線程中實現的，因此設計了一個如下圖所示的方案，將圖像檢測神經網絡放到一個子進程中負責在“后臺”檢測目標，主進程（UI進程）負責采集圖像并實時顯示在相應的控件上。

在實際的實現過程中面臨以下幾個關鍵點。

(1) 主進程（UI進程）采集到的圖像數據如何傳遞給子進程檢測？

Python多進程之間有一些簡單的通訊方式，例如：Queue，好處是它是一個進程安全的隊列，用戶不需要關注變量的管理，但是實際用這種方式來傳遞圖片，你就會發現速度慢呀！簡直崩潰，最簡單有效的還是通過“共享內存”，速度快很多（但是我還是覺得慢，我覺得主要是數據轉來轉去導致的），但需要對內存進行管理。

(2) 子進程檢測的結果如何告知主進程并將結果顯示出來?

最簡單的就和上面一樣，再使用一塊“共享內存”來將繪制好檢測結果的圖片傳遞回主進程中，但是這顯然不是最有效率的做法，畢竟使用“共享內存”耗時也挺長的，同時圖片數據也挺大的，檢測結果其實就是幾個簡單的數，沒必要將整個圖像都傳遞回去。我的處理方式是：主進程在“抽幀”時保存一個圖像備份，子進程通過“Queue”將檢測結果（相對圖片數據小的多）傳遞回主進程，主進程收到結果后，將結果繪制在備份圖像上并顯示出來。

(3) 共享內存的管理。

使用共享內存時一定要注意這部分內存的管理，不能主進程“寫”的同時你子進程在"讀"，否則數據不就錯了嘛。嚴格點來說這里需要一個“互斥鎖”(有興趣的同學可以試試)，我實在是比較懶，不想研究，直接創建了一個狀態變量（“共享的”）來控制，主進程和子進程在讀寫“共享內存”前，通過判斷狀態變量的值來確定是否有“權利”使用該“共享內存”。

---

2 代碼詳解

(1) 構造函數

這部分需要注意的是：建立共享內存、檢測子進程、消息接收線程，代碼里面有詳細的注釋，這里不贅述。

def __init__(self, parent=None):super(MainWindow, self).__init__(parent)self.setupUi(self) ?# 圖像大小self.img_shape = (480, 720) ?# 初始化界面self.label_imgshow.setScaledContents(True) # 圖片自適應顯示self.label_imgshow_res.setScaledContents(True) # 檢測結果圖片自適應顯示self.img_none = np.ones((480, 720, 3), dtype=np.uint8)*255self.show_img(self.img_none) ?# SSD檢測初始化self.weight_path = './ssd/weights/weights_SSD300.hdf5'self.weight_path = os.path.normpath(os.path.abspath(self.weight_path))self.obj_names = ['Aeroplane', 'Bicycle', 'Bird', 'Boat', 'Bottle','Bus', 'Car', 'Cat', 'Chair', 'Cow', 'Diningtable','Dog', 'Horse', 'Motorbike', 'Person', 'Pottedplant','Sheep', 'Sofa', 'Train', 'Tvmonitor']# 需要顯示的目標list， 用于過濾self.include_class = self.obj_names# -----------檢測子進程--------------# 子進程返回結果使用self.queue = Queue()# 多進程之間的共享圖片內存，參數‘I’表示數據類型為 int # 后一個參數為內存的大小，這里Python不提供多個維度數據的共享內存# 只有數組類型滿足使用需求，因此主進程中需先將圖像數據變為數組的樣子，在子進程中再恢復self.img_share = RawArray('I', self.img_shape[0] * self.img_shape[1] * 3)# 標識當前進程的狀態，非0：保持檢測；0：停止檢測self.process_flg = RawValue('I', 1)# 當前圖像共享內存 img_share 的狀態，非0：主進程使用中；0：子進程使用中self.img_get_flg = RawValue('I', 1)# 創建檢測子進程self.detector_process = Process(target=detector_process,args=(self.img_share,self.img_shape,self.process_flg,self.img_get_flg,self.queue))self.detector_process.start() # 進程開始 # ------------------------------------------------------------- ? # -----------接收檢測結果的線程--------------# 主要考慮到Queue的get方法可能會阻塞，如果直接在計時器函數中調用# get會導致UI“假死”，卡著不動。# 雖然也可以設置阻塞時間，但是建議還是建立線程接收處理Queue中的結果# 接收檢測結果的線程self.recv_thread = Recv_res(parent=self, queue=self.queue)# 連接信號 # 這個信號用于通知UI響應顯示，接收線程中只負責接收轉發結果，后面代碼中有詳細介紹self.recv_thread.res_signal.connect(self.show_res)self.recv_thread.start() # ------------------------------------------------------------- ?# 視頻文件路徑self.camera_index = 0self.FPS = None ?# 初始化計時器self.timer = QTimer(self) # 更新計時器self.timer.timeout.connect(self.timer_update) # 超時信號連接對應的槽函數 ?# 等待加載模型self.textEdit.setText('正在加載模型，請稍后......')self.pushButton_start.setEnabled(False)self.pushButton_open.setEnabled(False)self.pushButton_pause.setEnabled(False)self.lineEdit_cameraIndex.setEnabled(False) ?# 暫停初始化為不暫停self.pause = False

(2) 檢測子進程目標函數

下面是檢測子進程的目標函數，檢測子進程開始后執行的就是這個函數，首先是初始化SSD并加載權重，之后進入幀循環檢測，子進程的消息（包括檢測結果）通過Queue傳遞返回，包括兩部分：狀態量和消息內容，設置狀態量的目的是為了下一步針對不同的消息做相應的處理。

def detector_process(img_share, img_shape, process_flg, img_get_flg, res_queue):"""SSD檢測子進程目標函數:param img_share: 待檢測圖像數據，共享內存:param img_shape: 待檢測圖像的大小 (h, w):param process_flg: 子進程狀態量 0：退出檢測進程；非0：保持檢測:param img_get_flg: 共享圖像內存 的狀態量 0：子進程占用共享內存 1：主進程占有內存:param res_queue: 返回檢測結果的通道:return:"""# 初始化SSDweight_path = './ssd/weights/weights_SSD300.hdf5'weight_path = os.path.normpath(os.path.abspath(weight_path))obj_names = ['Aeroplane', 'Bicycle', 'Bird', 'Boat', 'Bottle','Bus', 'Car', 'Cat', 'Chair', 'Cow', 'Diningtable','Dog', 'Horse', 'Motorbike', 'Person', 'Pottedplant','Sheep', 'Sofa', 'Train', 'Tvmonitor']include_class = obj_namesssd = SSD_test(weight_path=weight_path, class_nam_list=obj_names) ?# 通知UI 模型加載成功res_queue.put((3, '模型加載成功')) ?# 構建檢測循環while True:# print('process_flg:{} img_get_flg:{}'.format(process_flg.value, img_get_flg.value))# 判斷檢測器狀態，是否退出if process_flg.value == 0:print('安全退出檢測進程！')res_queue.put((0, '檢測進程已安全退出！'))break ?# 判斷共享內存當前狀態是否可以安全讀取數據if img_get_flg.value == 0:# print('開始檢測！')try:img = np.array(img_share[:], dtype=np.uint8)img_scr = np.reshape(img, (img_shape[0], img_shape[1], 3)) ?# SSD檢測preds = ssd.Predict(img_scr)# 結果過濾preds = filter(obj_names, preds, inclued_class=include_class) ?h, w = img_shape[:2]res = decode_preds(obj_names, preds, w=w, h=h) # 列表 ?# 管道返回檢測結果res_queue.put((1, res)) ?except:print('圖片檢測失敗')res_queue.put((2, '當前圖像檢測失敗!'))finally:# 釋放圖像共享內存占用，讓主進程寫入新的圖像img_get_flg.value = 1

(3) 消息接收線程類

線程創建后會先進入構造函數，啟動后執行run函數。主要的功能就是接收子進程通過Queue傳遞回來的消息，并通知UI做出相應的處理。接收到檢測子進程退出的消息后，該線程也跳出循環結束生命周期。

class Recv_res(QThread):"""檢測結果接收線程"""res_signal = pyqtSignal(list)@debug_class('Recv_res')def __init__(self, parent, queue:Queue):"""構造函數:param parent: 父實例 QObj ，Qt中父實例析構相應的子線程會安全退出，不用人工處理:param queue: 管道"""super(Recv_res, self).__init__(parent=parent)self.queue = queue ?def run(self):while True:flg, res = self.queue.get()print(flg, res)if flg == 0: # 對應檢測子進程已安全退出print('接收線程已安全退出！')self.res_signal.emit([0, res])breakelse:self.res_signal.emit([flg, res])

(4) 計時器超時槽函數

該函數主要是按時讀取視頻流中的圖像并顯示在控件上，每次判斷檢測共享內存的狀態，如果子進程釋放則將當前幀數據寫入共享內存中，之后改變狀態變量的值（釋放對共享內存的占有）。

def timer_update(self):"""計時器槽函數:return:"""if self.cap.isOpened():# 讀取圖像ret, self.img_scr = self.cap.read()# ### 視頻讀取完畢if not ret:# 計時器停止計時self.timer.stop()# 不檢測self.img_get_flg.value = 1# 對話框提示QMessageBox.information(self, '播放提示', '視頻已播放完畢！')# 釋放攝像頭if hasattr(self, 'cap'):self.cap.release()del self.cap# 釋放‘開始’按鈕self.pushButton_start.setEnabled(True)# 禁止暫停并初始化其功能self.pause = Falseself.pushButton_pause.setText('暫停')self.pushButton_pause.setEnabled(False)# 釋放視頻流選擇self.pushButton_open.setEnabled(True)self.lineEdit_cameraIndex.setEnabled(True)return ?# 圖像預處理self.img_scr = cv2.resize(self.img_scr, (self.img_shape[1], self.img_shape[0]))# 轉為RGBself.img_scr = cv2.cvtColor(self.img_scr, cv2.COLOR_BGR2RGB) ?if hasattr(self, 'detector_process'):# ### 抽幀if self.img_get_flg.value == 1:# print('開始抽幀')self.img_temp = self.img_scr.copy() # 用于顯示檢測結果self.img_share[:] = self.img_scr.reshape(-1).tolist() # 抽幀保存在中間緩存self.img_get_flg.value = 0 # 不再抽取 直到檢測完成# print('結束抽幀') ?# 顯示圖像self.show_img(self.img_scr) ?# 響應UIQApplication.processEvents()else:self.textEdit.setText('數據流未打開！！！\n請檢查')self.resst_detector()

(5) 窗口關閉事件函數

在關閉窗口之前需要關閉子進程，否則子進程會一直在后臺運行，開一次軟件創建一個，多次重復后電腦越來越卡。打開任務管理器后后發現有好多名叫“Python”的進程，這些就是創建后卻沒關閉的子進程。因此，在窗口關閉事件函數下改變檢測進程的狀態變量值，使子進程能夠正常退出。

def closeEvent(self, a0):"""關閉窗口時間函數:param a0::return:"""self.process_flg.value = 0 # 退出子進程self.detector_process.join()

其他函數就不寫了，非常簡單。

---

由于本人能力有限，歡迎批評指正。
可以加我的QQ（1152291782）交流，請注明來意。

關注下方公眾號，回復關鍵字即可獲取下載地址。

• 本文配套源代碼下載地址：

  回復“SSD界面3”獲取。

---

如果你讀后有收獲，歡迎關注我的微信公眾號

上面有更多完全免費教程，我也會不定期更新

? ? ? 打開微信掃描下方二維碼關注 ? ? ?

總結

以上是生活随笔為你收集整理的【搞事情】利用PyQt为目标检测SSD300添加界面（四）的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。