工廠生產(chǎn)安全管控和產(chǎn)線作業(yè)監(jiān)督最典型的痛點(diǎn)就是“視頻監(jiān)控只能錄不能看”。以前車間裝了幾十路攝像頭出了問題要回放幾個(gè)小時(shí)錄像靠人眼一幀幀找線索。設(shè)備和人員違規(guī)往往只能事后發(fā)現(xiàn)很難做到及時(shí)預(yù)警。最近在做車間安全監(jiān)測(cè)相關(guān)的項(xiàng)目時(shí)把 AI 視頻分析能力接到了實(shí)際的監(jiān)控流上實(shí)現(xiàn)了安全帽檢測(cè)、區(qū)域入侵報(bào)警、目標(biāo)跟蹤去重、報(bào)警截圖留存等一套基礎(chǔ)閉環(huán)。這篇文章就把整個(gè)系統(tǒng)的設(shè)計(jì)思路、核心算法邏輯、可運(yùn)行代碼、常見坑位一起整理出來。既不堆概念也不只貼代碼而是盡量做到“照著能跑、改改能用”。本文適合有 Python 基礎(chǔ)、想入門 AI 視頻分析的開發(fā)者也適合正在做智慧工廠、安防監(jiān)測(cè)項(xiàng)目的后端或算法工程師。1. 背景與核心概念1.1 傳統(tǒng)工廠視頻監(jiān)控的短板傳統(tǒng)視頻監(jiān)控系統(tǒng)主要由攝像頭、硬盤錄像機(jī)NVR、解碼器和客戶端組成。它的核心能力是“錄像”和“回放”但是缺少“理解視頻內(nèi)容”的能力。帶來的問題也很明顯監(jiān)控畫面多保安很難同時(shí)盯住十幾個(gè)屏幕。24 小時(shí)連續(xù)值班漏看率很高。事件發(fā)生后排查困難回看錄像成本大。違規(guī)行為不能實(shí)時(shí)發(fā)現(xiàn)比如進(jìn)入危險(xiǎn)區(qū)、未戴安全帽。也就是說傳統(tǒng)監(jiān)控解決的是“能不能看到”的問題AI 視頻分析解決的是“看到了能不能自動(dòng)理解”的問題。1.2 什么是工廠 AI 視頻分析智能監(jiān)測(cè)系統(tǒng)工廠 AI 視頻分析智能監(jiān)測(cè)系統(tǒng)是一套把計(jì)算機(jī)視覺模型和應(yīng)用系統(tǒng)結(jié)合起來的技術(shù)方案。攝像頭采集到的視頻幀經(jīng)過目標(biāo)檢測(cè)、目標(biāo)跟蹤、規(guī)則判定等處理后能自動(dòng)轉(zhuǎn)化為結(jié)構(gòu)化的事件信息。例如檢測(cè)到人員進(jìn)入危險(xiǎn)區(qū)域觸發(fā)“區(qū)域入侵”報(bào)警。檢測(cè)到人員未佩戴安全帽觸發(fā)“未戴安全帽”報(bào)警。檢測(cè)到煙霧或火焰觸發(fā)“火災(zāi)隱患”報(bào)警。檢測(cè)到叉車超速或逆行觸發(fā)“車輛違規(guī)”報(bào)警。整個(gè)系統(tǒng)不是簡(jiǎn)單的“接個(gè)攝像頭跑個(gè)模型”而是一個(gè)完整的工程鏈路涉及視頻接入、編解碼、模型推理、事件管理、消息推送、數(shù)據(jù)存儲(chǔ)和權(quán)限控制。1.3 常見應(yīng)用場(chǎng)景工廠場(chǎng)景里比較成熟的應(yīng)用包括場(chǎng)景監(jiān)測(cè)內(nèi)容價(jià)值安全帽佩戴檢測(cè)人員是否佩戴安全帽降低頭部傷害風(fēng)險(xiǎn)區(qū)域入侵檢測(cè)人員是否進(jìn)入危險(xiǎn)區(qū)域/禁區(qū)防止誤入設(shè)備區(qū)域離崗/睡崗檢測(cè)關(guān)鍵崗位是否有人值守保證崗位責(zé)任落實(shí)煙火檢測(cè)車間、倉(cāng)庫(kù)是否出現(xiàn)煙霧火焰火災(zāi)早期預(yù)警傳送帶異常檢測(cè)傳送帶跑偏、停轉(zhuǎn)、物體堆積保障產(chǎn)線連續(xù)運(yùn)行車輛違規(guī)檢測(cè)叉車、貨車是否超速或越線減少?gòu)S內(nèi)交通事故遺留物檢測(cè)場(chǎng)地內(nèi)是否有異常遺留物體防止物料遺落或被遺留工具絆倒這些場(chǎng)景的核心技術(shù)基本一致視頻抽幀 目標(biāo)檢測(cè) 目標(biāo)跟蹤 業(yè)務(wù)規(guī)則判定。差異在于模型目標(biāo)和業(yè)務(wù)邏輯不同。2. 系統(tǒng)整體設(shè)計(jì)與數(shù)據(jù)流2.1 邏輯架構(gòu)一個(gè)可上線的工廠 AI 視頻分析系統(tǒng)通常分層設(shè)計(jì)采集層攝像頭、NVR、RTSP 流、本地視頻文件等。接入層負(fù)責(zé)拉流、解碼、抽幀、碼流異常處理。算法層目標(biāo)檢測(cè)、分類、跟蹤、行為識(shí)別。業(yè)務(wù)層區(qū)域規(guī)則、報(bào)警去重、事件記錄、截圖保存。展示層實(shí)時(shí)視頻流、報(bào)警列表、統(tǒng)計(jì)分析。分層的好處是業(yè)務(wù)邏輯和算法邏輯解耦。算法模型可以單獨(dú)升級(jí)不影響報(bào)警系統(tǒng)攝像頭接入也可以獨(dú)立擴(kuò)展。2.2 核心數(shù)據(jù)流整個(gè)系統(tǒng)的數(shù)據(jù)流可以概括為視頻源推流到 NVR 或直接提供 RTSP 地址。算法服務(wù)按設(shè)定的幀間隔拉流解碼。解碼后的幀送入目標(biāo)檢測(cè)模型。檢測(cè)結(jié)果送入跟蹤器為同一目標(biāo)分配穩(wěn)定的 ID。結(jié)合 ROI 區(qū)域、安全帽類別等規(guī)則判定事件。觸發(fā)報(bào)警后寫入報(bào)警表發(fā)送到消息隊(duì)列或管理后臺(tái)。前端展示實(shí)時(shí)視頻流和報(bào)警信息。這里需要注意檢測(cè)是對(duì)單幀圖片做的但工廠場(chǎng)景中對(duì)象可能在連續(xù)幀里反復(fù)出現(xiàn)。如果只做檢測(cè)不做跟蹤同一人連續(xù) 10 幀都在禁區(qū)就會(huì)觸發(fā) 10 次報(bào)警。所以目標(biāo)跟蹤是工程實(shí)現(xiàn)中非常關(guān)鍵的一環(huán)。2.3 部署形態(tài)常見部署方式有三種邊緣盒子直接部署在車間現(xiàn)場(chǎng)接入一路或多路攝像頭成本低、易替換。工控機(jī)/GPU 服務(wù)器部署在機(jī)房統(tǒng)一拉取廠區(qū)所有攝像頭便于集中管理。云服務(wù)器適合跨廠區(qū)、多園區(qū)統(tǒng)一管理但需要考慮視頻上傳帶寬和延遲。考慮到工廠網(wǎng)絡(luò)環(huán)境復(fù)雜攝像頭網(wǎng)段、防火墻策略、帶寬限制都可能影響系統(tǒng)。實(shí)際項(xiàng)目中優(yōu)先推薦邊緣部署 中心管理平臺(tái)邊緣負(fù)責(zé)實(shí)時(shí)告警中心負(fù)責(zé)數(shù)據(jù)匯聚和展示。2.4 子系統(tǒng)劃分從項(xiàng)目開發(fā)的角度可以拆成這些模塊視頻接入模塊管理攝像頭 RTSP 地址處理斷流重連。算法推理模塊加載模型執(zhí)行推理輸出檢測(cè)結(jié)果。規(guī)則引擎模塊判斷檢測(cè)結(jié)果是否滿足報(bào)警條件。報(bào)警管理模塊去重、持久化、發(fā)送通知。配置管理模塊管理 ROI 區(qū)域、閾值參數(shù)、攝像頭映射關(guān)系。前端展示模塊視頻流展示、報(bào)警畫面回放、統(tǒng)計(jì)大屏。如果項(xiàng)目規(guī)模再大一點(diǎn)還可以引入 Kafka 或 RabbitMQ 做消息解耦引入 Elasticsearch 存儲(chǔ)報(bào)警日志和截圖索引但初版系統(tǒng)不建議把所有組件都上齊否則光搭環(huán)境就會(huì)消耗大量時(shí)間。3. 環(huán)境準(zhǔn)備與項(xiàng)目結(jié)構(gòu)3.1 硬件環(huán)境開發(fā)階段可以直接使用本地電腦運(yùn)行建議配置如下CPU8 核及以上。內(nèi)存16GB 及以上。顯卡NVIDIA GPU如 RTX 3060 及以上如果沒有 GPU 也可以運(yùn)行但推理速度會(huì)明顯下降。生產(chǎn)環(huán)境部署時(shí)建議使用支持 CUDA 的 GPU 設(shè)備或使用 Intel OpenVINO / NVIDIA TensorRT 等推理加速方案。3.2 開發(fā)語言與依賴開發(fā)語言使用 Python依賴庫(kù)主要包括opencv-python視頻解碼、圖像處理、繪制標(biāo)注框。ultralyticsYOLOv8 模型加載和推理。flask提供視頻流和報(bào)警接口。flask-cors解決前端跨域訪問問題。numpy數(shù)值處理和數(shù)組的坐標(biāo)變換。由于不同版本的依賴庫(kù) API 可能會(huì)有變化本文示例不鎖定版本安裝時(shí)建議使用較新的穩(wěn)定版本下面給出參考依賴文件。3.3 項(xiàng)目目錄結(jié)構(gòu)factory-ai-vision/ ├── requirements.txt ├── config.py ├── tracker.py ├── analyzer.py ├── alarm_log.py ├── app.py ├── templates/ │ └── index.html └── weights/ └── yolov8n.pt各文件職責(zé)如下config.py全局配置包括視頻源、模型路徑、閾值、ROI 區(qū)域等。tracker.py簡(jiǎn)化版目標(biāo)跟蹤器為檢測(cè)目標(biāo)分配 ID。analyzer.py核心分析模塊負(fù)責(zé)檢測(cè)、跟蹤和規(guī)則判定。alarm_log.py報(bào)警管理負(fù)責(zé)去重和記錄。app.pyFlask 應(yīng)用負(fù)責(zé)視頻流接收、分析循環(huán)和 HTTP 接口。templates/index.html簡(jiǎn)易監(jiān)控大屏。依賴文件內(nèi)容如下# requirements.txt opencv-python ultralytics flask flask-cors numpy安裝依賴pip install -r requirements.txt如果網(wǎng)絡(luò)環(huán)境受限官方源安裝失敗可以換成國(guó)內(nèi) pip 鏡像源具體地址按自己所在環(huán)境選擇即可。4. 核心原理與關(guān)鍵代碼4.1 視頻流接入與抽幀無論是本地文件還是 RTSP 網(wǎng)絡(luò)流OpenCV 的cv2.VideoCapture都能統(tǒng)一處理。核心代碼import cv2 cap cv2.VideoCapture(rtsp://user:pass192.168.1.100:554/stream1) while True: ret, frame cap.read() if not ret: print(讀取幀失敗嘗試重連) cap.release() cap cv2.VideoCapture(rtsp://user:pass192.168.1.100:554/stream1) continue # 示例每 2 幀分析一次降低 CPU/GPU 壓力 frame_index 1 if frame_index % 2 ! 0: continue # 這里交給目標(biāo)檢測(cè)模塊處理需要注意幾點(diǎn)RTSP 地址必須能被運(yùn)行算法服務(wù)的機(jī)器訪問到。斷流后不能直接崩潰需要做重連機(jī)制。抽幀間隔會(huì)影響實(shí)時(shí)性和性能的平衡。網(wǎng)絡(luò)不好的時(shí)候cap.read()可能會(huì)阻塞建議放到獨(dú)立線程中讀取。4.2 目標(biāo)檢測(cè)模型推理目標(biāo)檢測(cè)是整條鏈路的核心本文使用 YOLOv8 做演示。YOLOv8 是 Ultralytics 推出的目標(biāo)檢測(cè)模型系列支持多種型號(hào)yolov8n是輕量級(jí)版本適合快速驗(yàn)證。加載模型from ultralytics import YOLO model YOLO(weights/yolov8n.pt)如果本地weights/yolov8n.pt不存在Ultralytics 會(huì)自動(dòng)嘗試下載預(yù)訓(xùn)練權(quán)重。生產(chǎn)環(huán)境建議提前下載到本地不依賴運(yùn)行時(shí)的網(wǎng)絡(luò)環(huán)境。對(duì)一幀圖片做推理results model.predict(frame, conf0.5, iou0.45, verboseFalse)conf是置信度閾值低于該閾值的檢測(cè)框會(huì)被過濾iou是 NMS 交并比閾值用來抑制重疊框。從results[0].boxes中可以拿到檢測(cè)框坐標(biāo)、類別和置信度boxes results[0].boxes if boxes is not None and len(boxes) 0: for det in boxes: x1, y1, x2, y2 det.xyxy[0].tolist() cls_id int(det.cls[0].item()) conf float(det.conf[0].item()) print(x1, y1, x2, y2, cls_id, conf)這里需要注意det.xyxy和det.cls是 PyTorch 的 Tensor 對(duì)象不要直接拿去當(dāng)列表用轉(zhuǎn)換成 Python 原生類型后再處理更穩(wěn)妥。4.3 目標(biāo)跟蹤降重YOLOv8 自帶的檢測(cè)結(jié)果沒有跨幀 ID。要避免每個(gè)目標(biāo)產(chǎn)生大量重復(fù)報(bào)警需要給目標(biāo)分配穩(wěn)定的跟蹤 ID。生產(chǎn)項(xiàng)目中可以用 ByteTrack、DeepSORT 等跟蹤算法。這里為了演示寫一個(gè)基于 IoU 匹配的簡(jiǎn)化跟蹤器思路是將上一幀的跟蹤目標(biāo)與當(dāng)前幀檢測(cè)框兩兩計(jì)算 IoU。選擇 IoU 最大的檢測(cè)框作為該目標(biāo)的當(dāng)前位置。若 IoU 低于閾值則視為新目標(biāo)分配新 ID。class SimpleTracker: def __init__(self, iou_threshold0.3, max_missing5): self.tracks {} self.next_id 0 self.iou_threshold iou_threshold self.max_missing max_missing def _iou(self, box1, box2): x1 max(box1[0], box2[0]) y1 max(box1[1], box2[1]) x2 min(box1[2], box2[2]) y2 min(box1[3], box2[3]) if x2 x1 or y2 y1: return 0.0 inter_area (x2 - x1) * (y2 - y1) area1 (box1[2] - box1[0]) * (box1[3] - box1[1]) area2 (box2[2] - box2[0]) * (box2[3] - box2[1]) union_area area1 area2 - inter_area return inter_area / union_area if union_area 0 else 0.0 def update(self, boxes, labels): updated {} unmatched set(range(len(boxes))) for tid, track in self.tracks.items(): best_idx None best_iou 0.0 for idx in unmatched: iou self._iou(track[box], boxes[idx]) if iou best_iou: best_iou iou best_idx idx if best_idx is not None and best_iou self.iou_threshold: track[box] boxes[best_idx] track[missing] 0 track[labels] labels[best_idx] updated[tid] track unmatched.remove(best_idx) else: track[missing] 1 if track[missing] self.max_missing: updated[tid] track for idx in unmatched: self.next_id 1 updated[self.next_id] { box: boxes[idx], labels: labels[idx], missing: 0 } self.tracks updated return [(tid, track[box], track[labels]) for tid, track in self.tracks.items()]這個(gè)簡(jiǎn)化版適合單路視頻流和運(yùn)動(dòng)幅度不大的場(chǎng)景不適用于擁擠人群和交叉頻繁的場(chǎng)景。真實(shí)項(xiàng)目里還是建議用 ByteTrack。4.4 區(qū)域判定區(qū)域入侵是工廠場(chǎng)景里最常見的規(guī)則。做法是先定義一個(gè) ROI 區(qū)域通常用歸一化坐標(biāo)表示這樣攝像頭分辨率變化時(shí)不需要修改坐標(biāo)。例如rois [ [(0.2, 0.3), (0.8, 0.3), (0.8, 0.8), (0.2, 0.8)] ]判斷一個(gè)檢測(cè)目標(biāo)是否在區(qū)域內(nèi)時(shí)可以取目標(biāo)框的底部中點(diǎn)作為落腳點(diǎn)import cv2 def point_in_rois(x, y, rois, frame_w, frame_h): for roi in rois: polygon [(int(px * frame_w), int(py * frame_h)) for px, py in roi] result cv2.pointPolygonTest(polygon, (int(x), int(y)), False) if result 0: return True return Falsecv2.pointPolygonTest返回值為正表示點(diǎn)在多邊形內(nèi)部0 表示在多邊形邊緣負(fù)數(shù)表示外部。這里取底部中點(diǎn)是有實(shí)際原因的。畫面上檢測(cè)框的頂部可能因?yàn)槿梭w站在圍欄外而判定為內(nèi)但腳部已經(jīng)進(jìn)入?yún)^(qū)域通常以腳部觸碰地面位置作為人員實(shí)際位置更合理。4.5 報(bào)警去重與記錄報(bào)警事件如果不做去重視頻流每秒 25 幀單目標(biāo)可能在 1 秒內(nèi)觸發(fā)多次報(bào)警報(bào)警表會(huì)瞬間被刷爆。一種簡(jiǎn)單的去重策略是“同類型 同目標(biāo) ID 時(shí)間間隔”。import time class AlarmCenter: def __init__(self, interval10): self.last_alarm {} self.alarms [] self.interval interval def push(self, alarm): key (alarm[type], alarm[track_id]) now time.time() if key not in self.last_alarm or now - self.last_alarm[key] self.interval: self.last_alarm[key] now alarm[time] time.strftime(%Y-%m-%d %H:%M:%S) self.alarms.append(alarm) print(報(bào)警觸發(fā):, alarm) return True return False去重周期需要按業(yè)務(wù)調(diào)10 秒一次適合大多數(shù)工廠安全報(bào)警但如果是需要秒級(jí)觸發(fā)的消防報(bào)警這個(gè)間隔就要縮短。生產(chǎn)項(xiàng)目中還需要把報(bào)警記錄寫入 MySQL、Elasticsearch 或 MongoDB同時(shí)保存報(bào)警前后的截圖和視頻段。這些數(shù)據(jù)既能用于管理員復(fù)核也能作為模型迭代的訓(xùn)練素材。5. 實(shí)戰(zhàn)工廠安全帽監(jiān)測(cè)與區(qū)域入侵報(bào)警5.1 需求定義本章實(shí)現(xiàn)一個(gè)最小可運(yùn)行的工廠 AI 視頻分析演示系統(tǒng)功能如下支持 RTSP 或本地視頻文件輸入。實(shí)時(shí)檢測(cè)畫面中的目標(biāo)并畫出檢測(cè)框。對(duì)目標(biāo)進(jìn)行簡(jiǎn)易跟蹤分配 ID。判斷目標(biāo)是否進(jìn)入 ROI 區(qū)域觸發(fā)區(qū)域入侵報(bào)警。如果模型支持頭盔類別則輸出未戴安全帽報(bào)警。通過瀏覽器實(shí)時(shí)查看視頻流和報(bào)警列表。整體流程為采集線程不斷拉幀分析線程對(duì)幀做檢測(cè)、跟蹤、規(guī)則判斷Flask 提供 MJPEG 視頻流和 JSON 報(bào)警接口給前端展示。5.2 全局配置 config.pyimport os class Config: # 視頻源支持本地文件和 RTSP/RTMP 流 VIDEO_SOURCE os.getenv(VIDEO_SOURCE, test.mp4) # 模型路徑y(tǒng)olov8n.pt 會(huì)自動(dòng)下載也可替換為自訓(xùn)練模型 MODEL_PATH os.getenv(MODEL_PATH, weights/yolov8n.pt) # 置信度閾值和 NMS 閾值 CONF_THRESHOLD float(os.getenv(CONF_THRESHOLD, 0.5)) IOU_THRESHOLD float(os.getenv(IOU_THRESHOLD, 0.45)) # 抽幀間隔2 表示每 2 幀分析一次 FRAME_INTERVAL int(os.getenv(FRAME_INTERVAL, 2)) # COCO 數(shù)據(jù)集中 person 類別的 id 是 0 PERSON_CLASS int(os.getenv(PERSON_CLASS, 0)) # 安全帽未佩戴類別 id需按自訓(xùn)練模型的類別定義調(diào)整 HELMET_CLASS int(os.getenv(HELMET_CLASS, -1)) # ROI 區(qū)域使用歸一化坐標(biāo) ROIS [ [(0.1, 0.3), (0.9, 0.3), (0.9, 0.9), (0.1, 0.9)] ] # 同一目標(biāo)同一報(bào)警類型的最小間隔單位秒 ALARM_INTERVAL int(os.getenv(ALARM_INTERVAL, 10))如果HELMET_CLASS設(shè)置為 -1表示當(dāng)前模型沒有安全帽類別只啟用區(qū)域入侵報(bào)警。這樣可以保證在只使用 COCO 預(yù)訓(xùn)練模型時(shí)也能跑通流程。5.3 目標(biāo)跟蹤器 tracker.py按照第 4.3 節(jié)的SimpleTracker實(shí)現(xiàn)保存為tracker.py即可。5.4 分析模塊 analyzer.pyimport cv2 from ultralytics import YOLO from tracker import SimpleTracker class VideoAnalyzer: def __init__(self, config): self.config config self.model YOLO(config.MODEL_PATH) self.tracker SimpleTracker(iou_threshold0.3, max_missing5) def process(self, frame): h, w frame.shape[:2] # 縮小分辨率降低推理耗時(shí) infer_scale 0.5 infer_w int(w * infer_scale) infer_h int(h * infer_scale) small_frame cv2.resize(frame, (infer_w, infer_h)) results self.model.predict( small_frame, confself.config.CONF_THRESHOLD, iouself.config.IOU_THRESHOLD, verboseFalse ) boxes [] labels [] dets results[0].boxes if dets is not None and len(dets) 0: for det in dets: x1, y1, x2, y2 det.xyxy[0].tolist() cls_id int(det.cls[0].item()) conf float(det.conf[0].item()) # 坐標(biāo)還原到原圖分辨率 x1 x1 / infer_scale y1 y1 / infer_scale x2 x2 / infer_scale y2 y2 / infer_scale boxes.append([x1, y1, x2, y2]) labels.append((cls_id, conf)) tracked self.tracker.update(boxes, labels) alarms self._check_alarms(tracked, w, h) return tracked, alarms def _check_alarms(self, tracked, w, h): alarms [] for tid, box, labels in tracked: x1, y1, x2, y2 box class_ids [cls_id for cls_id, _ in labels] is_person self.config.PERSON_CLASS in class_ids if not is_person: continue # 目標(biāo)底部中點(diǎn) foot_x (x1 x2) / 2.0 foot_y y2 in_roi self._point_in_rois(foot_x, foot_y, w, h) if in_roi: alarms.append({ type: ROI_INTRUSION, track_id: tid, box: [round(v, 2) for v in box] }) if self.config.HELMET_CLASS 0 and self.config.HELMET_CLASS in class_ids: alarms.append({ type: NO_HELMET, track_id: tid, box: [round(v, 2) for v in box] }) return alarms def _point_in_rois(self, x, y, w, h): for roi in self.config.ROIS: polygon [(int(px * w), int(py * h)) for px, py in roi] if cv2.pointPolygonTest(polygon, (int(x), int(y)), False) 0: return True return False這里有兩個(gè)關(guān)鍵細(xì)節(jié)第一模型推理是在縮小后的幀上做的但檢測(cè)框要映射回原圖尺寸否則畫框位置會(huì)偏移。第二SimpleTracker.update傳入的 boxes 是原圖坐標(biāo)所以 tracker 的 IoU 計(jì)算也是基于原圖坐標(biāo)的不同分辨率下不會(huì)影響匹配效果。5.5 報(bào)警管理 alarm_log.py直接使用第 4.5 節(jié)的AlarmCenter保存為alarm_log.py。5.6 Flask 視頻服務(wù) app.pyimport threading import cv2 import numpy as np from flask import Flask, Response, render_template, jsonify from analyzer import VideoAnalyzer from alarm_log import AlarmCenter from config import Config app Flask(__name__) config Config() analyzer VideoAnalyzer(config) alarm_center AlarmCenter(intervalconfig.ALARM_INTERVAL) current_frame None frame_lock threading.Lock() running False def draw_annotations(frame, tracked): for roi in config.ROIS: pts [(int(p[0] * frame.shape[1]), int(p[1] * frame.shape[0])) for p in roi] cv2.polylines(frame, [np.array(pts)], True, (0, 255, 255), 2) for tid, box, labels in tracked: x1, y1, x2, y2 [int(v) for v in box] class_names {0: person, 1: no_helmet} label_text ,.join( class_names.get(cls_id, str(cls_id)) for cls_id, _ in labels ) cv2.rectangle(frame, (x1, y1), (x2, y2), (0, 255, 0), 2) cv2.putText( frame, f{tid}:{label_text}, (x1, max(0, y1 - 5)), cv2.FONT_HERSHEY_SIMPLEX, 0.6, (0, 255, 0), 2, ) return frame def capture_loop(): global current_frame, running cap cv2.VideoCapture(config.VIDEO_SOURCE) if not cap.isOpened(): print(無法打開視頻源:, config.VIDEO_SOURCE) running False return running True frame_index 0 while running: ret, frame cap.read() if not ret: print(讀取視頻幀失敗重新連接中...) cap.release() cap cv2.VideoCapture(config.VIDEO_SOURCE) continue frame_index 1 if config.FRAME_INTERVAL 1 and frame_index % config.FRAME_INTERVAL ! 0: continue tracked, alarms analyzer.process(frame) for alarm in alarms: alarm_center.push(alarm) frame draw_annotations(frame, tracked) with frame_lock: current_frame frame cap.release() running False def generate_mjpeg(): while True: with frame_lock: frame current_frame if frame is None: continue ret, jpeg cv2.imencode(.jpg, frame) if not ret: continue frame_bytes jpeg.tobytes() yield ( b--frame\r\n bContent-Type: image/jpeg\r\n\r\n frame_bytes b\r\n ) app.route(/) def index(): return render_template(index.html) app.route(/video_feed) def video_feed(): return Response( generate_mjpeg(), mimetypemultipart/x-mixed-replace; boundaryframe ) app.route(/alerts) def alerts(): return jsonify(alarm_center.alarms[-50:][::-1]) if __name__ __main__: threading.Thread(targetcapture_loop, daemonTrue).start() app.run(host0.0.0.0, port8080, debugFalse)這段代碼的核心結(jié)構(gòu)是“采集線程 分析線程 HTTP 展示線程”。采集線程和解碼是同步的實(shí)際上解碼仍然發(fā)生在capture_loop中只是把耗時(shí)分析放到了同一個(gè)線程。如果一路視頻還能接受多路視頻并發(fā)時(shí)建議用生產(chǎn)者-消費(fèi)者隊(duì)列把采集、分析、展示徹底解耦。5.7 前端頁面 templates/index.html!DOCTYPE html html langzh-CN head meta charsetutf-8 title工廠AI視頻分析智能監(jiān)測(cè)系統(tǒng)/title style body { font-family: Arial, sans-serif; background: #f5f6fa; margin: 0; padding: 20px; } h1 { font-size: 20px; } .container { display: flex; gap: 20px; align-items: flex-start; } .video-box { flex: 1; background: #000; border-radius: 8px; overflow: hidden; } .video-box img { width: 100%; display: block; } .alarm-box { width: 360px; background: #fff; border-radius: 8px; padding: 16px; box-shadow: 0 2px 8px rgba(0, 0, 0, 0.08); } .alarm-item { border-bottom: 1px solid #eee; padding: 8px 0; font-size: 14px; } .alarm-time { color: #888; margin-right: 8px; } /style /head body h1工廠AI視頻分析智能監(jiān)測(cè)系統(tǒng)/h1 div classcontainer div classvideo-box img src/video_feed alt實(shí)時(shí)視頻流 /div div classalarm-box h3實(shí)時(shí)報(bào)警/h3 div idalarms/div /div /div script setInterval(() { fetch(/alerts) .then((res) res.json()) .then((data) { const box document.getElementById(alarms); box.innerHTML data .map( (item) div classalarm-itemspan classalarm-time${item.time}/span${item.type} · 目標(biāo)ID${item.track_id}/div ) .join(); }) .catch((err) console.error(加載報(bào)警失敗, err)); }, 2000); /script /body /html5.8 運(yùn)行與驗(yàn)證假設(shè)本地已經(jīng)有一份視頻文件test.mp4或者有一個(gè)可訪問的 RTSP 地址。第一步驗(yàn)證模型權(quán)重可以正常加載python -c from ultralytics import YOLO; YOLO(weights/yolov8n.pt)如果本機(jī)沒有權(quán)重文件且網(wǎng)絡(luò)可以訪問官方模型倉(cāng)庫(kù)會(huì)自動(dòng)下載。為了穩(wěn)定最好提前下載好放到weights目錄。第二步啟動(dòng)服務(wù)python app.py第三步瀏覽器訪問http://localhost:8080/預(yù)期現(xiàn)象頁面顯示實(shí)時(shí)視頻畫面視頻幀中會(huì)出現(xiàn)綠色檢測(cè)框和 ROI 黃色線段。檢測(cè)到目標(biāo)進(jìn)入 ROI 區(qū)域后右側(cè)報(bào)警列表出現(xiàn)ROI_INTRUSION記錄。如果使用包含安全帽類別的自訓(xùn)練模型則輸出NO_HELMET報(bào)警。如果想修改視頻源可以設(shè)置環(huán)境變量VIDEO_SOURCErtsp://192.168.1.100:554/stream1 python app.py注意 RTSP 如果攜帶用戶名密碼URL 格式通常是rtsp://用戶名:密碼IP:端口/路徑如果視頻源無法打開前端畫面會(huì)一直黑屏調(diào)試時(shí)要優(yōu)先確認(rèn) URL 和網(wǎng)絡(luò)連通性。6. 生產(chǎn)環(huán)境常見問題與排查問題現(xiàn)象常見原因解決思路啟動(dòng)后模型加載很慢首次加載需要初始化 PyTorch 和權(quán)重文件提前將模型權(quán)重放置本地避免運(yùn)行時(shí)下載推理速度很慢沒有使用 GPU或輸入分辨率過高降低抽幀頻率、縮小推理分辨率、使用 TensorRT/OpenVINO攝像頭黑屏RTSP 地址不對(duì)或網(wǎng)絡(luò)不可達(dá)用 VLC 或 ffprobe 驗(yàn)證碼流能否播放視頻卡頓明顯解碼速度低于幀率分析耗時(shí)過長(zhǎng)開啟多線程、加大抽幀間隔漏檢率較高目標(biāo)太小、遮擋嚴(yán)重、閾值太高降低置信度閾值增加訓(xùn)練數(shù)據(jù)啟用大分辨率模型誤報(bào)很多區(qū)域坐標(biāo)設(shè)置不準(zhǔn)確或置信度閾值太低調(diào)整 ROI 坐標(biāo)提高置信度閾值同一目標(biāo)連續(xù)報(bào)警沒有做跟蹤去重接目標(biāo)跟蹤模塊增加報(bào)警間隔GPU 使用率低數(shù)據(jù)預(yù)處理與模型推理串行使用隊(duì)列做讀幀/預(yù)處理/推理流水線RTSP 斷流后服務(wù)直接卡死cap.read()長(zhǎng)時(shí)間阻塞設(shè)置解碼超時(shí)斷流后自動(dòng)重新連接內(nèi)存持續(xù)增長(zhǎng)循環(huán)中重復(fù)創(chuàng)建 Tensor 或歷史數(shù)據(jù)未清理定期清理報(bào)警記錄減少對(duì)象引用問題排查時(shí)建議固定順序先確認(rèn)視頻源能不能看。再確認(rèn)能不能出檢測(cè)框。再確認(rèn)報(bào)警規(guī)則有沒有觸發(fā)。最后看性能和穩(wěn)定性。不要一上來就調(diào)模型很多項(xiàng)目跑不通都是視頻接入層的問題。7. 最佳實(shí)踐與工程建議7.1 模型選型與迭代模型選型要結(jié)合業(yè)務(wù)精度和部署顯存綜合考慮。輕量級(jí)模型推理快但小目標(biāo)檢測(cè)效果差大模型精度高但對(duì) GPU 資源要求高。建議的原則驗(yàn)證階段先用現(xiàn)成預(yù)訓(xùn)練模型跑通流程。正式項(xiàng)目必須用工廠現(xiàn)場(chǎng)數(shù)據(jù)微調(diào)模型。數(shù)據(jù)采集要覆蓋不同光線、角度、距離和人員姿態(tài)。定期用線上報(bào)警截圖補(bǔ)充訓(xùn)練集形成“標(biāo)注-訓(xùn)練-驗(yàn)證-上線”的迭代閉環(huán)。用 COCO 預(yù)訓(xùn)練模型只能檢測(cè)出person、car等常見類別無法識(shí)別安全帽是否佩戴。所以安全帽檢測(cè)必須訓(xùn)練專用模型。7.2 性能優(yōu)化單路視頻在 GPU 上通常壓力不大但工廠往往有幾十路甚至上百路攝像頭性能問題成為最主要瓶頸。可以從幾個(gè)方向優(yōu)化分辨率控制不需要每幀都做全分辨率推理可以抽幀 縮放檢測(cè)后再映射回原坐標(biāo)。多路并發(fā)一個(gè)進(jìn)程處理多路視頻時(shí)用異步隊(duì)列收集各路待推理幀批量推理。推理加速GPU 環(huán)境使用 TensorRTCPU 環(huán)境使用 OpenVINO。硬件分層實(shí)時(shí)性要求高的通道用邊緣盒子非實(shí)時(shí)通道集中分析。7.3 報(bào)警去重與聯(lián)動(dòng)安全報(bào)警的價(jià)值在于準(zhǔn)和穩(wěn)。同一目標(biāo)在禁區(qū)站了 10 分鐘業(yè)務(wù)上只關(guān)心“誰、什么時(shí)候進(jìn)入、是否還在”不需要每秒都收到一條消息。建議報(bào)警模型設(shè)計(jì)為產(chǎn)生初次報(bào)警后進(jìn)入“持續(xù)報(bào)警中”狀態(tài)。同一目標(biāo)離開區(qū)域后再進(jìn)入才產(chǎn)生第二次報(bào)警。重要報(bào)警支持截圖保存、現(xiàn)場(chǎng)聲光聯(lián)動(dòng)、短信/電話通知。7.4 數(shù)據(jù)合規(guī)與權(quán)限安全工廠視頻分析系統(tǒng)涉及大量員工畫面和產(chǎn)線畫面數(shù)據(jù)隱私是必須考慮的問題。需要注意在廠區(qū)部署攝像頭的區(qū)域應(yīng)告知員工相關(guān)監(jiān)控事項(xiàng)。視頻數(shù)據(jù)要設(shè)置訪問權(quán)限后臺(tái)操作日志可追溯。算法分析結(jié)果和原始視頻按需留存避免無限制保存。攝像頭接入和報(bào)警記錄操作需要授權(quán)不能用管理員賬號(hào)直接運(yùn)行在無監(jiān)控的測(cè)試環(huán)境。生產(chǎn)環(huán)境變更時(shí)要先在測(cè)試環(huán)境驗(yàn)證模型和配置再逐步灰度發(fā)布到生產(chǎn)通道。7.5 日志、監(jiān)控與可維護(hù)性系統(tǒng)上線后穩(wěn)定運(yùn)行比功能豐富更重要。建議在代碼中增加結(jié)構(gòu)化日志比如記錄每路攝像頭最近一次心跳時(shí)間。每一幀推理耗時(shí)。報(bào)警類型、置信度、檢測(cè)框坐標(biāo)。配置項(xiàng)統(tǒng)一放到配置中心模型和規(guī)則參數(shù)支持熱更新。這樣調(diào)整 ROI 區(qū)域和閾值時(shí)不需要重新編譯或重啟整個(gè)服務(wù)。8. 總結(jié)與學(xué)習(xí)路線本文完整梳理了工廠 AI 視頻分析智能監(jiān)測(cè)系統(tǒng)的設(shè)計(jì)和實(shí)現(xiàn)路徑重點(diǎn)包括相機(jī)視頻流接入與抽幀處理。YOLOv8 目標(biāo)檢測(cè)模型的調(diào)用和結(jié)果解析?；?IoU 的簡(jiǎn)化目標(biāo)跟蹤解決報(bào)警重復(fù)問題。ROI 區(qū)域判定和安全帽類別判斷。Flask 提供實(shí)時(shí)視頻流與報(bào)警接口。簡(jiǎn)易監(jiān)控大屏的展示。想進(jìn)一步深入可以按下面的順序?qū)W習(xí)目標(biāo)檢測(cè)基礎(chǔ) - 目標(biāo)跟蹤ByteTrack / DeepSORT - 視頻流工程化RTSP / 多線程 / 隊(duì)列 - 模型訓(xùn)練與數(shù)據(jù)集構(gòu)建 - 服務(wù)化部署Docker / TensorRT / 彈性伸縮初學(xué)階段不建議一上手就搭建完整大平臺(tái)。可以先從“一路視頻 一個(gè)模型 一個(gè)規(guī)則 一個(gè)報(bào)警”的最小閉環(huán)練起跑通后逐步增加通道數(shù)、報(bào)警規(guī)則、消息推送和可視化大屏。如果本文對(duì)你有幫助可以收藏備用也歡迎在評(píng)論區(qū)交流你在工廠視頻分析項(xiàng)目中遇到的問題。