測系統(tǒng)：從目標(biāo)檢測到報警閉環(huán)的工程實(shí)踐)
工廠生產(chǎn)安全管控和產(chǎn)線作業(yè)監(jiān)督最典型的痛點(diǎn)就是“視頻監(jiān)控只能錄不能看”。以前車間裝了幾十路攝像頭出了問題要回放幾個小時錄像靠人眼一幀幀找線索。設(shè)備和人員違規(guī)往往只能事后發(fā)現(xiàn)很難做到及時預(yù)警。最近在做車間安全監(jiān)測相關(guān)的項目時把 AI 視頻分析能力接到了實(shí)際的監(jiān)控流上實(shí)現(xiàn)了安全帽檢測、區(qū)域入侵報警、目標(biāo)跟蹤去重、報警截圖留存等一套基礎(chǔ)閉環(huán)。這篇文章就把整個系統(tǒng)的設(shè)計思路、核心算法邏輯、可運(yùn)行代碼、常見坑位一起整理出來。既不堆概念也不只貼代碼而是盡量做到“照著能跑、改改能用”。本文適合有 Python 基礎(chǔ)、想入門 AI 視頻分析的開發(fā)者也適合正在做智慧工廠、安防監(jiān)測項目的后端或算法工程師。1. 背景與核心概念1.1 傳統(tǒng)工廠視頻監(jiān)控的短板傳統(tǒng)視頻監(jiān)控系統(tǒng)主要由攝像頭、硬盤錄像機(jī)NVR、解碼器和客戶端組成。它的核心能力是“錄像”和“回放”但是缺少“理解視頻內(nèi)容”的能力。帶來的問題也很明顯監(jiān)控畫面多保安很難同時盯住十幾個屏幕。24 小時連續(xù)值班漏看率很高。事件發(fā)生后排查困難回看錄像成本大。違規(guī)行為不能實(shí)時發(fā)現(xiàn)比如進(jìn)入危險區(qū)、未戴安全帽。也就是說傳統(tǒng)監(jiān)控解決的是“能不能看到”的問題AI 視頻分析解決的是“看到了能不能自動理解”的問題。1.2 什么是工廠 AI 視頻分析智能監(jiān)測系統(tǒng)工廠 AI 視頻分析智能監(jiān)測系統(tǒng)是一套把計算機(jī)視覺模型和應(yīng)用系統(tǒng)結(jié)合起來的技術(shù)方案。攝像頭采集到的視頻幀經(jīng)過目標(biāo)檢測、目標(biāo)跟蹤、規(guī)則判定等處理后能自動轉(zhuǎn)化為結(jié)構(gòu)化的事件信息。例如檢測到人員進(jìn)入危險區(qū)域觸發(fā)“區(qū)域入侵”報警。檢測到人員未佩戴安全帽觸發(fā)“未戴安全帽”報警。檢測到煙霧或火焰觸發(fā)“火災(zāi)隱患”報警。檢測到叉車超速或逆行觸發(fā)“車輛違規(guī)”報警。整個系統(tǒng)不是簡單的“接個攝像頭跑個模型”而是一個完整的工程鏈路涉及視頻接入、編解碼、模型推理、事件管理、消息推送、數(shù)據(jù)存儲和權(quán)限控制。1.3 常見應(yīng)用場景工廠場景里比較成熟的應(yīng)用包括場景監(jiān)測內(nèi)容價值安全帽佩戴檢測人員是否佩戴安全帽降低頭部傷害風(fēng)險區(qū)域入侵檢測人員是否進(jìn)入危險區(qū)域/禁區(qū)防止誤入設(shè)備區(qū)域離崗/睡崗檢測關(guān)鍵崗位是否有人值守保證崗位責(zé)任落實(shí)煙火檢測車間、倉庫是否出現(xiàn)煙霧火焰火災(zāi)早期預(yù)警傳送帶異常檢測傳送帶跑偏、停轉(zhuǎn)、物體堆積保障產(chǎn)線連續(xù)運(yùn)行車輛違規(guī)檢測叉車、貨車是否超速或越線減少廠內(nèi)交通事故遺留物檢測場地內(nèi)是否有異常遺留物體防止物料遺落或被遺留工具絆倒這些場景的核心技術(shù)基本一致視頻抽幀 目標(biāo)檢測 目標(biāo)跟蹤 業(yè)務(wù)規(guī)則判定。差異在于模型目標(biāo)和業(yè)務(wù)邏輯不同。2. 系統(tǒng)整體設(shè)計與數(shù)據(jù)流2.1 邏輯架構(gòu)一個可上線的工廠 AI 視頻分析系統(tǒng)通常分層設(shè)計采集層攝像頭、NVR、RTSP 流、本地視頻文件等。接入層負(fù)責(zé)拉流、解碼、抽幀、碼流異常處理。算法層目標(biāo)檢測、分類、跟蹤、行為識別。業(yè)務(wù)層區(qū)域規(guī)則、報警去重、事件記錄、截圖保存。展示層實(shí)時視頻流、報警列表、統(tǒng)計分析。分層的好處是業(yè)務(wù)邏輯和算法邏輯解耦。算法模型可以單獨(dú)升級不影響報警系統(tǒng)攝像頭接入也可以獨(dú)立擴(kuò)展。2.2 核心數(shù)據(jù)流整個系統(tǒng)的數(shù)據(jù)流可以概括為視頻源推流到 NVR 或直接提供 RTSP 地址。算法服務(wù)按設(shè)定的幀間隔拉流解碼。解碼后的幀送入目標(biāo)檢測模型。檢測結(jié)果送入跟蹤器為同一目標(biāo)分配穩(wěn)定的 ID。結(jié)合 ROI 區(qū)域、安全帽類別等規(guī)則判定事件。觸發(fā)報警后寫入報警表發(fā)送到消息隊列或管理后臺。前端展示實(shí)時視頻流和報警信息。這里需要注意檢測是對單幀圖片做的但工廠場景中對象可能在連續(xù)幀里反復(fù)出現(xiàn)。如果只做檢測不做跟蹤同一人連續(xù) 10 幀都在禁區(qū)就會觸發(fā) 10 次報警。所以目標(biāo)跟蹤是工程實(shí)現(xiàn)中非常關(guān)鍵的一環(huán)。2.3 部署形態(tài)常見部署方式有三種邊緣盒子直接部署在車間現(xiàn)場接入一路或多路攝像頭成本低、易替換。工控機(jī)/GPU 服務(wù)器部署在機(jī)房統(tǒng)一拉取廠區(qū)所有攝像頭便于集中管理。云服務(wù)器適合跨廠區(qū)、多園區(qū)統(tǒng)一管理但需要考慮視頻上傳帶寬和延遲?？紤]到工廠網(wǎng)絡(luò)環(huán)境復(fù)雜攝像頭網(wǎng)段、防火墻策略、帶寬限制都可能影響系統(tǒng)。實(shí)際項目中優(yōu)先推薦邊緣部署 中心管理平臺邊緣負(fù)責(zé)實(shí)時告警中心負(fù)責(zé)數(shù)據(jù)匯聚和展示。2.4 子系統(tǒng)劃分從項目開發(fā)的角度可以拆成這些模塊視頻接入模塊管理攝像頭 RTSP 地址處理斷流重連。算法推理模塊加載模型執(zhí)行推理輸出檢測結(jié)果。規(guī)則引擎模塊判斷檢測結(jié)果是否滿足報警條件。報警管理模塊去重、持久化、發(fā)送通知。配置管理模塊管理 ROI 區(qū)域、閾值參數(shù)、攝像頭映射關(guān)系。前端展示模塊視頻流展示、報警畫面回放、統(tǒng)計大屏。如果項目規(guī)模再大一點(diǎn)還可以引入 Kafka 或 RabbitMQ 做消息解耦引入 Elasticsearch 存儲報警日志和截圖索引但初版系統(tǒng)不建議把所有組件都上齊否則光搭環(huán)境就會消耗大量時間。3. 環(huán)境準(zhǔn)備與項目結(jié)構(gòu)3.1 硬件環(huán)境開發(fā)階段可以直接使用本地電腦運(yùn)行建議配置如下CPU8 核及以上。內(nèi)存16GB 及以上。顯卡NVIDIA GPU如 RTX 3060 及以上如果沒有 GPU 也可以運(yùn)行但推理速度會明顯下降。生產(chǎn)環(huán)境部署時建議使用支持 CUDA 的 GPU 設(shè)備或使用 Intel OpenVINO / NVIDIA TensorRT 等推理加速方案。3.2 開發(fā)語言與依賴開發(fā)語言使用 Python依賴庫主要包括opencv-python視頻解碼、圖像處理、繪制標(biāo)注框。ultralyticsYOLOv8 模型加載和推理。flask提供視頻流和報警接口。flask-cors解決前端跨域訪問問題。numpy數(shù)值處理和數(shù)組的坐標(biāo)變換。由于不同版本的依賴庫 API 可能會有變化本文示例不鎖定版本安裝時建議使用較新的穩(wěn)定版本下面給出參考依賴文件。3.3 項目目錄結(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簡化版目標(biāo)跟蹤器為檢測目標(biāo)分配 ID。analyzer.py核心分析模塊負(fù)責(zé)檢測、跟蹤和規(guī)則判定。alarm_log.py報警管理負(fù)責(zé)去重和記錄。app.pyFlask 應(yīng)用負(fù)責(zé)視頻流接收、分析循環(huán)和 HTTP 接口。templates/index.html簡易監(jiān)控大屏。依賴文件內(nèi)容如下# requirements.txt opencv-python ultralytics flask flask-cors numpy安裝依賴pip install -r requirements.txt如果網(wǎng)絡(luò)環(huán)境受限官方源安裝失敗可以換成國內(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)檢測模塊處理需要注意幾點(diǎn)RTSP 地址必須能被運(yùn)行算法服務(wù)的機(jī)器訪問到。斷流后不能直接崩潰需要做重連機(jī)制。抽幀間隔會影響實(shí)時性和性能的平衡。網(wǎng)絡(luò)不好的時候cap.read()可能會阻塞建議放到獨(dú)立線程中讀取。4.2 目標(biāo)檢測模型推理目標(biāo)檢測是整條鏈路的核心本文使用 YOLOv8 做演示。YOLOv8 是 Ultralytics 推出的目標(biāo)檢測模型系列支持多種型號yolov8n是輕量級版本適合快速驗證。加載模型from ultralytics import YOLO model YOLO(weights/yolov8n.pt)如果本地weights/yolov8n.pt不存在Ultralytics 會自動嘗試下載預(yù)訓(xùn)練權(quán)重。生產(chǎn)環(huán)境建議提前下載到本地不依賴運(yùn)行時的網(wǎng)絡(luò)環(huán)境。對一幀圖片做推理results model.predict(frame, conf0.5, iou0.45, verboseFalse)conf是置信度閾值低于該閾值的檢測框會被過濾iou是 NMS 交并比閾值用來抑制重疊框。從results[0].boxes中可以拿到檢測框坐標(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 對象不要直接拿去當(dāng)列表用轉(zhuǎn)換成 Python 原生類型后再處理更穩(wěn)妥。4.3 目標(biāo)跟蹤降重YOLOv8 自帶的檢測結(jié)果沒有跨幀 ID。要避免每個目標(biāo)產(chǎn)生大量重復(fù)報警需要給目標(biāo)分配穩(wěn)定的跟蹤 ID。生產(chǎn)項目中可以用 ByteTrack、DeepSORT 等跟蹤算法。這里為了演示寫一個基于 IoU 匹配的簡化跟蹤器思路是將上一幀的跟蹤目標(biāo)與當(dāng)前幀檢測框兩兩計算 IoU。選擇 IoU 最大的檢測框作為該目標(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()]這個簡化版適合單路視頻流和運(yùn)動幅度不大的場景不適用于擁擠人群和交叉頻繁的場景。真實(shí)項目里還是建議用 ByteTrack。4.4 區(qū)域判定區(qū)域入侵是工廠場景里最常見的規(guī)則。做法是先定義一個 ROI 區(qū)域通常用歸一化坐標(biāo)表示這樣攝像頭分辨率變化時不需要修改坐標(biāo)。例如rois [ [(0.2, 0.3), (0.8, 0.3), (0.8, 0.8), (0.2, 0.8)] ]判斷一個檢測目標(biāo)是否在區(qū)域內(nèi)時可以取目標(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í)際原因的。畫面上檢測框的頂部可能因為人體站在圍欄外而判定為內(nèi)但腳部已經(jīng)進(jìn)入?yún)^(qū)域通常以腳部觸碰地面位置作為人員實(shí)際位置更合理。4.5 報警去重與記錄報警事件如果不做去重視頻流每秒 25 幀單目標(biāo)可能在 1 秒內(nèi)觸發(fā)多次報警報警表會瞬間被刷爆。一種簡單的去重策略是“同類型 同目標(biāo) ID 時間間隔”。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(報警觸發(fā):, alarm) return True return False去重周期需要按業(yè)務(wù)調(diào)10 秒一次適合大多數(shù)工廠安全報警但如果是需要秒級觸發(fā)的消防報警這個間隔就要縮短。生產(chǎn)項目中還需要把報警記錄寫入 MySQL、Elasticsearch 或 MongoDB同時保存報警前后的截圖和視頻段。這些數(shù)據(jù)既能用于管理員復(fù)核也能作為模型迭代的訓(xùn)練素材。5. 實(shí)戰(zhàn)工廠安全帽監(jiān)測與區(qū)域入侵報警5.1 需求定義本章實(shí)現(xiàn)一個最小可運(yùn)行的工廠 AI 視頻分析演示系統(tǒng)功能如下支持 RTSP 或本地視頻文件輸入。實(shí)時檢測畫面中的目標(biāo)并畫出檢測框。對目標(biāo)進(jìn)行簡易跟蹤分配 ID。判斷目標(biāo)是否進(jìn)入 ROI 區(qū)域觸發(fā)區(qū)域入侵報警。如果模型支持頭盔類別則輸出未戴安全帽報警。通過瀏覽器實(shí)時查看視頻流和報警列表。整體流程為采集線程不斷拉幀分析線程對幀做檢測、跟蹤、規(guī)則判斷Flask 提供 MJPEG 視頻流和 JSON 報警接口給前端展示。5.2 全局配置 config.pyimport os class Config: # 視頻源支持本地文件和 RTSP/RTMP 流 VIDEO_SOURCE os.getenv(VIDEO_SOURCE, test.mp4) # 模型路徑y(tǒng)olov8n.pt 會自動下載也可替換為自訓(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)同一報警類型的最小間隔單位秒 ALARM_INTERVAL int(os.getenv(ALARM_INTERVAL, 10))如果HELMET_CLASS設(shè)置為 -1表示當(dāng)前模型沒有安全帽類別只啟用區(qū)域入侵報警。這樣可以保證在只使用 COCO 預(yù)訓(xùn)練模型時也能跑通流程。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] # 縮小分辨率降低推理耗時 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這里有兩個關(guān)鍵細(xì)節(jié)第一模型推理是在縮小后的幀上做的但檢測框要映射回原圖尺寸否則畫框位置會偏移。第二SimpleTracker.update傳入的 boxes 是原圖坐標(biāo)所以 tracker 的 IoU 計算也是基于原圖坐標(biāo)的不同分辨率下不會影響匹配效果。5.5 報警管理 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中只是把耗時分析放到了同一個線程。如果一路視頻還能接受多路視頻并發(fā)時建議用生產(chǎn)者-消費(fèi)者隊列把采集、分析、展示徹底解耦。5.7 前端頁面 templates/index.html!DOCTYPE html html langzh-CN head meta charsetutf-8 title工廠AI視頻分析智能監(jiān)測系統(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)測系統(tǒng)/h1 div classcontainer div classvideo-box img src/video_feed alt實(shí)時視頻流 /div div classalarm-box h3實(shí)時報警/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(加載報警失敗, err)); }, 2000); /script /body /html5.8 運(yùn)行與驗證假設(shè)本地已經(jīng)有一份視頻文件test.mp4或者有一個可訪問的 RTSP 地址。第一步驗證模型權(quán)重可以正常加載python -c from ultralytics import YOLO; YOLO(weights/yolov8n.pt)如果本機(jī)沒有權(quán)重文件且網(wǎng)絡(luò)可以訪問官方模型倉庫會自動下載。為了穩(wěn)定最好提前下載好放到weights目錄。第二步啟動服務(wù)python app.py第三步瀏覽器訪問http://localhost:8080/預(yù)期現(xiàn)象頁面顯示實(shí)時視頻畫面視頻幀中會出現(xiàn)綠色檢測框和 ROI 黃色線段。檢測到目標(biāo)進(jìn)入 ROI 區(qū)域后右側(cè)報警列表出現(xiàn)ROI_INTRUSION記錄。如果使用包含安全帽類別的自訓(xùn)練模型則輸出NO_HELMET報警。如果想修改視頻源可以設(shè)置環(huán)境變量VIDEO_SOURCErtsp://192.168.1.100:554/stream1 python app.py注意 RTSP 如果攜帶用戶名密碼URL 格式通常是rtsp://用戶名:密碼IP:端口/路徑如果視頻源無法打開前端畫面會一直黑屏調(diào)試時要優(yōu)先確認(rèn) URL 和網(wǎng)絡(luò)連通性。6. 生產(chǎn)環(huán)境常見問題與排查問題現(xiàn)象常見原因解決思路啟動后模型加載很慢首次加載需要初始化 PyTorch 和權(quán)重文件提前將模型權(quán)重放置本地避免運(yùn)行時下載推理速度很慢沒有使用 GPU或輸入分辨率過高降低抽幀頻率、縮小推理分辨率、使用 TensorRT/OpenVINO攝像頭黑屏RTSP 地址不對或網(wǎng)絡(luò)不可達(dá)用 VLC 或 ffprobe 驗證碼流能否播放視頻卡頓明顯解碼速度低于幀率分析耗時過長開啟多線程、加大抽幀間隔漏檢率較高目標(biāo)太小、遮擋嚴(yán)重、閾值太高降低置信度閾值增加訓(xùn)練數(shù)據(jù)啟用大分辨率模型誤報很多區(qū)域坐標(biāo)設(shè)置不準(zhǔn)確或置信度閾值太低調(diào)整 ROI 坐標(biāo)提高置信度閾值同一目標(biāo)連續(xù)報警沒有做跟蹤去重接目標(biāo)跟蹤模塊增加報警間隔GPU 使用率低數(shù)據(jù)預(yù)處理與模型推理串行使用隊列做讀幀/預(yù)處理/推理流水線RTSP 斷流后服務(wù)直接卡死cap.read()長時間阻塞設(shè)置解碼超時斷流后自動重新連接內(nèi)存持續(xù)增長循環(huán)中重復(fù)創(chuàng)建 Tensor 或歷史數(shù)據(jù)未清理定期清理報警記錄減少對象引用問題排查時建議固定順序先確認(rèn)視頻源能不能看。再確認(rèn)能不能出檢測框。再確認(rèn)報警規(guī)則有沒有觸發(fā)。最后看性能和穩(wěn)定性。不要一上來就調(diào)模型很多項目跑不通都是視頻接入層的問題。7. 最佳實(shí)踐與工程建議7.1 模型選型與迭代模型選型要結(jié)合業(yè)務(wù)精度和部署顯存綜合考慮。輕量級模型推理快但小目標(biāo)檢測效果差大模型精度高但對 GPU 資源要求高。建議的原則驗證階段先用現(xiàn)成預(yù)訓(xùn)練模型跑通流程。正式項目必須用工廠現(xiàn)場數(shù)據(jù)微調(diào)模型。數(shù)據(jù)采集要覆蓋不同光線、角度、距離和人員姿態(tài)。定期用線上報警截圖補(bǔ)充訓(xùn)練集形成“標(biāo)注-訓(xùn)練-驗證-上線”的迭代閉環(huán)。用 COCO 預(yù)訓(xùn)練模型只能檢測出person、car等常見類別無法識別安全帽是否佩戴。所以安全帽檢測必須訓(xùn)練專用模型。7.2 性能優(yōu)化單路視頻在 GPU 上通常壓力不大但工廠往往有幾十路甚至上百路攝像頭性能問題成為最主要瓶頸。可以從幾個方向優(yōu)化分辨率控制不需要每幀都做全分辨率推理可以抽幀 縮放檢測后再映射回原坐標(biāo)。多路并發(fā)一個進(jìn)程處理多路視頻時用異步隊列收集各路待推理幀批量推理。推理加速GPU 環(huán)境使用 TensorRTCPU 環(huán)境使用 OpenVINO。硬件分層實(shí)時性要求高的通道用邊緣盒子非實(shí)時通道集中分析。7.3 報警去重與聯(lián)動安全報警的價值在于準(zhǔn)和穩(wěn)。同一目標(biāo)在禁區(qū)站了 10 分鐘業(yè)務(wù)上只關(guān)心“誰、什么時候進(jìn)入、是否還在”不需要每秒都收到一條消息。建議報警模型設(shè)計為產(chǎn)生初次報警后進(jìn)入“持續(xù)報警中”狀態(tài)。同一目標(biāo)離開區(qū)域后再進(jìn)入才產(chǎn)生第二次報警。重要報警支持截圖保存、現(xiàn)場聲光聯(lián)動、短信/電話通知。7.4 數(shù)據(jù)合規(guī)與權(quán)限安全工廠視頻分析系統(tǒng)涉及大量員工畫面和產(chǎn)線畫面數(shù)據(jù)隱私是必須考慮的問題。需要注意在廠區(qū)部署攝像頭的區(qū)域應(yīng)告知員工相關(guān)監(jiān)控事項。視頻數(shù)據(jù)要設(shè)置訪問權(quán)限后臺操作日志可追溯。算法分析結(jié)果和原始視頻按需留存避免無限制保存。攝像頭接入和報警記錄操作需要授權(quán)不能用管理員賬號直接運(yùn)行在無監(jiān)控的測試環(huán)境。生產(chǎn)環(huán)境變更時要先在測試環(huán)境驗證模型和配置再逐步灰度發(fā)布到生產(chǎn)通道。7.5 日志、監(jiān)控與可維護(hù)性系統(tǒng)上線后穩(wěn)定運(yùn)行比功能豐富更重要。建議在代碼中增加結(jié)構(gòu)化日志比如記錄每路攝像頭最近一次心跳時間。每一幀推理耗時。報警類型、置信度、檢測框坐標(biāo)。配置項統(tǒng)一放到配置中心模型和規(guī)則參數(shù)支持熱更新。這樣調(diào)整 ROI 區(qū)域和閾值時不需要重新編譯或重啟整個服務(wù)。8. 總結(jié)與學(xué)習(xí)路線本文完整梳理了工廠 AI 視頻分析智能監(jiān)測系統(tǒng)的設(shè)計和實(shí)現(xiàn)路徑重點(diǎn)包括相機(jī)視頻流接入與抽幀處理。YOLOv8 目標(biāo)檢測模型的調(diào)用和結(jié)果解析。基于 IoU 的簡化目標(biāo)跟蹤解決報警重復(fù)問題。ROI 區(qū)域判定和安全帽類別判斷。Flask 提供實(shí)時視頻流與報警接口。簡易監(jiān)控大屏的展示。想進(jìn)一步深入可以按下面的順序?qū)W習(xí)目標(biāo)檢測基礎(chǔ) - 目標(biāo)跟蹤ByteTrack / DeepSORT - 視頻流工程化RTSP / 多線程 / 隊列 - 模型訓(xùn)練與數(shù)據(jù)集構(gòu)建 - 服務(wù)化部署Docker / TensorRT / 彈性伸縮初學(xué)階段不建議一上手就搭建完整大平臺。可以先從“一路視頻 一個模型 一個規(guī)則 一個報警”的最小閉環(huán)練起跑通后逐步增加通道數(shù)、報警規(guī)則、消息推送和可視化大屏。如果本文對你有幫助可以收藏備用也歡迎在評論區(qū)交流你在工廠視頻分析項目中遇到的問題。 工廠AI視頻分析智能監(jiān)測系統(tǒng)：從目標(biāo)檢測到報警閉環(huán)的工程實(shí)踐
工廠生產(chǎn)安全管控和產(chǎn)線作業(yè)監(jiān)督最典型的痛點(diǎn)就是“視頻監(jiān)控只能錄不能看”。以前車間裝了幾十路攝像頭出了問題要回放幾個小時錄像靠人眼一幀幀找線索。設(shè)備和人員違規(guī)往往只能事后發(fā)現(xiàn)很難做到及時預(yù)警。最近在做車間安全監(jiān)測相關(guān)的項目時把 AI 視頻分析能力接到了實(shí)際的監(jiān)控流上實(shí)現(xiàn)了安全帽檢測、區(qū)域入侵報警、目標(biāo)跟蹤去重、報警截圖留存等一套基礎(chǔ)閉環(huán)。這篇文章就把整個系統(tǒng)的設(shè)計思路、核心算法邏輯、可運(yùn)行代碼、常見坑位一起整理出來。既不堆概念也不只貼代碼而是盡量做到“照著能跑、改改能用”。本文適合有 Python 基礎(chǔ)、想入門 AI 視頻分析的開發(fā)者也適合正在做智慧工廠、安防監(jiān)測項目的后端或算法工程師。1. 背景與核心概念1.1 傳統(tǒng)工廠視頻監(jiān)控的短板傳統(tǒng)視頻監(jiān)控系統(tǒng)主要由攝像頭、硬盤錄像機(jī)NVR、解碼器和客戶端組成。它的核心能力是“錄像”和“回放”但是缺少“理解視頻內(nèi)容”的能力。帶來的問題也很明顯監(jiān)控畫面多保安很難同時盯住十幾個屏幕。24 小時連續(xù)值班漏看率很高。事件發(fā)生后排查困難回看錄像成本大。違規(guī)行為不能實(shí)時發(fā)現(xiàn)比如進(jìn)入危險區(qū)、未戴安全帽。也就是說傳統(tǒng)監(jiān)控解決的是“能不能看到”的問題AI 視頻分析解決的是“看到了能不能自動理解”的問題。1.2 什么是工廠 AI 視頻分析智能監(jiān)測系統(tǒng)工廠 AI 視頻分析智能監(jiān)測系統(tǒng)是一套把計算機(jī)視覺模型和應(yīng)用系統(tǒng)結(jié)合起來的技術(shù)方案。攝像頭采集到的視頻幀經(jīng)過目標(biāo)檢測、目標(biāo)跟蹤、規(guī)則判定等處理后能自動轉(zhuǎn)化為結(jié)構(gòu)化的事件信息。例如檢測到人員進(jìn)入危險區(qū)域觸發(fā)“區(qū)域入侵”報警。檢測到人員未佩戴安全帽觸發(fā)“未戴安全帽”報警。檢測到煙霧或火焰觸發(fā)“火災(zāi)隱患”報警。檢測到叉車超速或逆行觸發(fā)“車輛違規(guī)”報警。整個系統(tǒng)不是簡單的“接個攝像頭跑個模型”而是一個完整的工程鏈路涉及視頻接入、編解碼、模型推理、事件管理、消息推送、數(shù)據(jù)存儲和權(quán)限控制。1.3 常見應(yīng)用場景工廠場景里比較成熟的應(yīng)用包括場景監(jiān)測內(nèi)容價值安全帽佩戴檢測人員是否佩戴安全帽降低頭部傷害風(fēng)險區(qū)域入侵檢測人員是否進(jìn)入危險區(qū)域/禁區(qū)防止誤入設(shè)備區(qū)域離崗/睡崗檢測關(guān)鍵崗位是否有人值守保證崗位責(zé)任落實(shí)煙火檢測車間、倉庫是否出現(xiàn)煙霧火焰火災(zāi)早期預(yù)警傳送帶異常檢測傳送帶跑偏、停轉(zhuǎn)、物體堆積保障產(chǎn)線連續(xù)運(yùn)行車輛違規(guī)檢測叉車、貨車是否超速或越線減少廠內(nèi)交通事故遺留物檢測場地內(nèi)是否有異常遺留物體防止物料遺落或被遺留工具絆倒這些場景的核心技術(shù)基本一致視頻抽幀 目標(biāo)檢測 目標(biāo)跟蹤 業(yè)務(wù)規(guī)則判定。差異在于模型目標(biāo)和業(yè)務(wù)邏輯不同。2. 系統(tǒng)整體設(shè)計與數(shù)據(jù)流2.1 邏輯架構(gòu)一個可上線的工廠 AI 視頻分析系統(tǒng)通常分層設(shè)計采集層攝像頭、NVR、RTSP 流、本地視頻文件等。接入層負(fù)責(zé)拉流、解碼、抽幀、碼流異常處理。算法層目標(biāo)檢測、分類、跟蹤、行為識別。業(yè)務(wù)層區(qū)域規(guī)則、報警去重、事件記錄、截圖保存。展示層實(shí)時視頻流、報警列表、統(tǒng)計分析。分層的好處是業(yè)務(wù)邏輯和算法邏輯解耦。算法模型可以單獨(dú)升級不影響報警系統(tǒng)攝像頭接入也可以獨(dú)立擴(kuò)展。2.2 核心數(shù)據(jù)流整個系統(tǒng)的數(shù)據(jù)流可以概括為視頻源推流到 NVR 或直接提供 RTSP 地址。算法服務(wù)按設(shè)定的幀間隔拉流解碼。解碼后的幀送入目標(biāo)檢測模型。檢測結(jié)果送入跟蹤器為同一目標(biāo)分配穩(wěn)定的 ID。結(jié)合 ROI 區(qū)域、安全帽類別等規(guī)則判定事件。觸發(fā)報警后寫入報警表發(fā)送到消息隊列或管理后臺。前端展示實(shí)時視頻流和報警信息。這里需要注意檢測是對單幀圖片做的但工廠場景中對象可能在連續(xù)幀里反復(fù)出現(xiàn)。如果只做檢測不做跟蹤同一人連續(xù) 10 幀都在禁區(qū)就會觸發(fā) 10 次報警。所以目標(biāo)跟蹤是工程實(shí)現(xiàn)中非常關(guān)鍵的一環(huán)。2.3 部署形態(tài)常見部署方式有三種邊緣盒子直接部署在車間現(xiàn)場接入一路或多路攝像頭成本低、易替換。工控機(jī)/GPU 服務(wù)器部署在機(jī)房統(tǒng)一拉取廠區(qū)所有攝像頭便于集中管理。云服務(wù)器適合跨廠區(qū)、多園區(qū)統(tǒng)一管理但需要考慮視頻上傳帶寬和延遲?？紤]到工廠網(wǎng)絡(luò)環(huán)境復(fù)雜攝像頭網(wǎng)段、防火墻策略、帶寬限制都可能影響系統(tǒng)。實(shí)際項目中優(yōu)先推薦邊緣部署 中心管理平臺邊緣負(fù)責(zé)實(shí)時告警中心負(fù)責(zé)數(shù)據(jù)匯聚和展示。2.4 子系統(tǒng)劃分從項目開發(fā)的角度可以拆成這些模塊視頻接入模塊管理攝像頭 RTSP 地址處理斷流重連。算法推理模塊加載模型執(zhí)行推理輸出檢測結(jié)果。規(guī)則引擎模塊判斷檢測結(jié)果是否滿足報警條件。報警管理模塊去重、持久化、發(fā)送通知。配置管理模塊管理 ROI 區(qū)域、閾值參數(shù)、攝像頭映射關(guān)系。前端展示模塊視頻流展示、報警畫面回放、統(tǒng)計大屏。如果項目規(guī)模再大一點(diǎn)還可以引入 Kafka 或 RabbitMQ 做消息解耦引入 Elasticsearch 存儲報警日志和截圖索引但初版系統(tǒng)不建議把所有組件都上齊否則光搭環(huán)境就會消耗大量時間。3. 環(huán)境準(zhǔn)備與項目結(jié)構(gòu)3.1 硬件環(huán)境開發(fā)階段可以直接使用本地電腦運(yùn)行建議配置如下CPU8 核及以上。內(nèi)存16GB 及以上。顯卡NVIDIA GPU如 RTX 3060 及以上如果沒有 GPU 也可以運(yùn)行但推理速度會明顯下降。生產(chǎn)環(huán)境部署時建議使用支持 CUDA 的 GPU 設(shè)備或使用 Intel OpenVINO / NVIDIA TensorRT 等推理加速方案。3.2 開發(fā)語言與依賴開發(fā)語言使用 Python依賴庫主要包括opencv-python視頻解碼、圖像處理、繪制標(biāo)注框。ultralyticsYOLOv8 模型加載和推理。flask提供視頻流和報警接口。flask-cors解決前端跨域訪問問題。numpy數(shù)值處理和數(shù)組的坐標(biāo)變換。由于不同版本的依賴庫 API 可能會有變化本文示例不鎖定版本安裝時建議使用較新的穩(wěn)定版本下面給出參考依賴文件。3.3 項目目錄結(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簡化版目標(biāo)跟蹤器為檢測目標(biāo)分配 ID。analyzer.py核心分析模塊負(fù)責(zé)檢測、跟蹤和規(guī)則判定。alarm_log.py報警管理負(fù)責(zé)去重和記錄。app.pyFlask 應(yīng)用負(fù)責(zé)視頻流接收、分析循環(huán)和 HTTP 接口。templates/index.html簡易監(jiān)控大屏。依賴文件內(nèi)容如下# requirements.txt opencv-python ultralytics flask flask-cors numpy安裝依賴pip install -r requirements.txt如果網(wǎng)絡(luò)環(huán)境受限官方源安裝失敗可以換成國內(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)檢測模塊處理需要注意幾點(diǎn)RTSP 地址必須能被運(yùn)行算法服務(wù)的機(jī)器訪問到。斷流后不能直接崩潰需要做重連機(jī)制。抽幀間隔會影響實(shí)時性和性能的平衡。網(wǎng)絡(luò)不好的時候cap.read()可能會阻塞建議放到獨(dú)立線程中讀取。4.2 目標(biāo)檢測模型推理目標(biāo)檢測是整條鏈路的核心本文使用 YOLOv8 做演示。YOLOv8 是 Ultralytics 推出的目標(biāo)檢測模型系列支持多種型號yolov8n是輕量級版本適合快速驗證。加載模型from ultralytics import YOLO model YOLO(weights/yolov8n.pt)如果本地weights/yolov8n.pt不存在Ultralytics 會自動嘗試下載預(yù)訓(xùn)練權(quán)重。生產(chǎn)環(huán)境建議提前下載到本地不依賴運(yùn)行時的網(wǎng)絡(luò)環(huán)境。對一幀圖片做推理results model.predict(frame, conf0.5, iou0.45, verboseFalse)conf是置信度閾值低于該閾值的檢測框會被過濾iou是 NMS 交并比閾值用來抑制重疊框。從results[0].boxes中可以拿到檢測框坐標(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 對象不要直接拿去當(dāng)列表用轉(zhuǎn)換成 Python 原生類型后再處理更穩(wěn)妥。4.3 目標(biāo)跟蹤降重YOLOv8 自帶的檢測結(jié)果沒有跨幀 ID。要避免每個目標(biāo)產(chǎn)生大量重復(fù)報警需要給目標(biāo)分配穩(wěn)定的跟蹤 ID。生產(chǎn)項目中可以用 ByteTrack、DeepSORT 等跟蹤算法。這里為了演示寫一個基于 IoU 匹配的簡化跟蹤器思路是將上一幀的跟蹤目標(biāo)與當(dāng)前幀檢測框兩兩計算 IoU。選擇 IoU 最大的檢測框作為該目標(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()]這個簡化版適合單路視頻流和運(yùn)動幅度不大的場景不適用于擁擠人群和交叉頻繁的場景。真實(shí)項目里還是建議用 ByteTrack。4.4 區(qū)域判定區(qū)域入侵是工廠場景里最常見的規(guī)則。做法是先定義一個 ROI 區(qū)域通常用歸一化坐標(biāo)表示這樣攝像頭分辨率變化時不需要修改坐標(biāo)。例如rois [ [(0.2, 0.3), (0.8, 0.3), (0.8, 0.8), (0.2, 0.8)] ]判斷一個檢測目標(biāo)是否在區(qū)域內(nèi)時可以取目標(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í)際原因的。畫面上檢測框的頂部可能因為人體站在圍欄外而判定為內(nèi)但腳部已經(jīng)進(jìn)入?yún)^(qū)域通常以腳部觸碰地面位置作為人員實(shí)際位置更合理。4.5 報警去重與記錄報警事件如果不做去重視頻流每秒 25 幀單目標(biāo)可能在 1 秒內(nèi)觸發(fā)多次報警報警表會瞬間被刷爆。一種簡單的去重策略是“同類型 同目標(biāo) ID 時間間隔”。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(報警觸發(fā):, alarm) return True return False去重周期需要按業(yè)務(wù)調(diào)10 秒一次適合大多數(shù)工廠安全報警但如果是需要秒級觸發(fā)的消防報警這個間隔就要縮短。生產(chǎn)項目中還需要把報警記錄寫入 MySQL、Elasticsearch 或 MongoDB同時保存報警前后的截圖和視頻段。這些數(shù)據(jù)既能用于管理員復(fù)核也能作為模型迭代的訓(xùn)練素材。5. 實(shí)戰(zhàn)工廠安全帽監(jiān)測與區(qū)域入侵報警5.1 需求定義本章實(shí)現(xiàn)一個最小可運(yùn)行的工廠 AI 視頻分析演示系統(tǒng)功能如下支持 RTSP 或本地視頻文件輸入。實(shí)時檢測畫面中的目標(biāo)并畫出檢測框。對目標(biāo)進(jìn)行簡易跟蹤分配 ID。判斷目標(biāo)是否進(jìn)入 ROI 區(qū)域觸發(fā)區(qū)域入侵報警。如果模型支持頭盔類別則輸出未戴安全帽報警。通過瀏覽器實(shí)時查看視頻流和報警列表。整體流程為采集線程不斷拉幀分析線程對幀做檢測、跟蹤、規(guī)則判斷Flask 提供 MJPEG 視頻流和 JSON 報警接口給前端展示。5.2 全局配置 config.pyimport os class Config: # 視頻源支持本地文件和 RTSP/RTMP 流 VIDEO_SOURCE os.getenv(VIDEO_SOURCE, test.mp4) # 模型路徑y(tǒng)olov8n.pt 會自動下載也可替換為自訓(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)同一報警類型的最小間隔單位秒 ALARM_INTERVAL int(os.getenv(ALARM_INTERVAL, 10))如果HELMET_CLASS設(shè)置為 -1表示當(dāng)前模型沒有安全帽類別只啟用區(qū)域入侵報警。這樣可以保證在只使用 COCO 預(yù)訓(xùn)練模型時也能跑通流程。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] # 縮小分辨率降低推理耗時 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這里有兩個關(guān)鍵細(xì)節(jié)第一模型推理是在縮小后的幀上做的但檢測框要映射回原圖尺寸否則畫框位置會偏移。第二SimpleTracker.update傳入的 boxes 是原圖坐標(biāo)所以 tracker 的 IoU 計算也是基于原圖坐標(biāo)的不同分辨率下不會影響匹配效果。5.5 報警管理 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中只是把耗時分析放到了同一個線程。如果一路視頻還能接受多路視頻并發(fā)時建議用生產(chǎn)者-消費(fèi)者隊列把采集、分析、展示徹底解耦。5.7 前端頁面 templates/index.html!DOCTYPE html html langzh-CN head meta charsetutf-8 title工廠AI視頻分析智能監(jiān)測系統(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)測系統(tǒng)/h1 div classcontainer div classvideo-box img src/video_feed alt實(shí)時視頻流 /div div classalarm-box h3實(shí)時報警/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(加載報警失敗, err)); }, 2000); /script /body /html5.8 運(yùn)行與驗證假設(shè)本地已經(jīng)有一份視頻文件test.mp4或者有一個可訪問的 RTSP 地址。第一步驗證模型權(quán)重可以正常加載python -c from ultralytics import YOLO; YOLO(weights/yolov8n.pt)如果本機(jī)沒有權(quán)重文件且網(wǎng)絡(luò)可以訪問官方模型倉庫會自動下載。為了穩(wěn)定最好提前下載好放到weights目錄。第二步啟動服務(wù)python app.py第三步瀏覽器訪問http://localhost:8080/預(yù)期現(xiàn)象頁面顯示實(shí)時視頻畫面視頻幀中會出現(xiàn)綠色檢測框和 ROI 黃色線段。檢測到目標(biāo)進(jìn)入 ROI 區(qū)域后右側(cè)報警列表出現(xiàn)ROI_INTRUSION記錄。如果使用包含安全帽類別的自訓(xùn)練模型則輸出NO_HELMET報警。如果想修改視頻源可以設(shè)置環(huán)境變量VIDEO_SOURCErtsp://192.168.1.100:554/stream1 python app.py注意 RTSP 如果攜帶用戶名密碼URL 格式通常是rtsp://用戶名:密碼IP:端口/路徑如果視頻源無法打開前端畫面會一直黑屏調(diào)試時要優(yōu)先確認(rèn) URL 和網(wǎng)絡(luò)連通性。6. 生產(chǎn)環(huán)境常見問題與排查問題現(xiàn)象常見原因解決思路啟動后模型加載很慢首次加載需要初始化 PyTorch 和權(quán)重文件提前將模型權(quán)重放置本地避免運(yùn)行時下載推理速度很慢沒有使用 GPU或輸入分辨率過高降低抽幀頻率、縮小推理分辨率、使用 TensorRT/OpenVINO攝像頭黑屏RTSP 地址不對或網(wǎng)絡(luò)不可達(dá)用 VLC 或 ffprobe 驗證碼流能否播放視頻卡頓明顯解碼速度低于幀率分析耗時過長開啟多線程、加大抽幀間隔漏檢率較高目標(biāo)太小、遮擋嚴(yán)重、閾值太高降低置信度閾值增加訓(xùn)練數(shù)據(jù)啟用大分辨率模型誤報很多區(qū)域坐標(biāo)設(shè)置不準(zhǔn)確或置信度閾值太低調(diào)整 ROI 坐標(biāo)提高置信度閾值同一目標(biāo)連續(xù)報警沒有做跟蹤去重接目標(biāo)跟蹤模塊增加報警間隔GPU 使用率低數(shù)據(jù)預(yù)處理與模型推理串行使用隊列做讀幀/預(yù)處理/推理流水線RTSP 斷流后服務(wù)直接卡死cap.read()長時間阻塞設(shè)置解碼超時斷流后自動重新連接內(nèi)存持續(xù)增長循環(huán)中重復(fù)創(chuàng)建 Tensor 或歷史數(shù)據(jù)未清理定期清理報警記錄減少對象引用問題排查時建議固定順序先確認(rèn)視頻源能不能看。再確認(rèn)能不能出檢測框。再確認(rèn)報警規(guī)則有沒有觸發(fā)。最后看性能和穩(wěn)定性。不要一上來就調(diào)模型很多項目跑不通都是視頻接入層的問題。7. 最佳實(shí)踐與工程建議7.1 模型選型與迭代模型選型要結(jié)合業(yè)務(wù)精度和部署顯存綜合考慮。輕量級模型推理快但小目標(biāo)檢測效果差大模型精度高但對 GPU 資源要求高。建議的原則驗證階段先用現(xiàn)成預(yù)訓(xùn)練模型跑通流程。正式項目必須用工廠現(xiàn)場數(shù)據(jù)微調(diào)模型。數(shù)據(jù)采集要覆蓋不同光線、角度、距離和人員姿態(tài)。定期用線上報警截圖補(bǔ)充訓(xùn)練集形成“標(biāo)注-訓(xùn)練-驗證-上線”的迭代閉環(huán)。用 COCO 預(yù)訓(xùn)練模型只能檢測出person、car等常見類別無法識別安全帽是否佩戴。所以安全帽檢測必須訓(xùn)練專用模型。7.2 性能優(yōu)化單路視頻在 GPU 上通常壓力不大但工廠往往有幾十路甚至上百路攝像頭性能問題成為最主要瓶頸。可以從幾個方向優(yōu)化分辨率控制不需要每幀都做全分辨率推理可以抽幀 縮放檢測后再映射回原坐標(biāo)。多路并發(fā)一個進(jìn)程處理多路視頻時用異步隊列收集各路待推理幀批量推理。推理加速GPU 環(huán)境使用 TensorRTCPU 環(huán)境使用 OpenVINO。硬件分層實(shí)時性要求高的通道用邊緣盒子非實(shí)時通道集中分析。7.3 報警去重與聯(lián)動安全報警的價值在于準(zhǔn)和穩(wěn)。同一目標(biāo)在禁區(qū)站了 10 分鐘業(yè)務(wù)上只關(guān)心“誰、什么時候進(jìn)入、是否還在”不需要每秒都收到一條消息。建議報警模型設(shè)計為產(chǎn)生初次報警后進(jìn)入“持續(xù)報警中”狀態(tài)。同一目標(biāo)離開區(qū)域后再進(jìn)入才產(chǎn)生第二次報警。重要報警支持截圖保存、現(xiàn)場聲光聯(lián)動、短信/電話通知。7.4 數(shù)據(jù)合規(guī)與權(quán)限安全工廠視頻分析系統(tǒng)涉及大量員工畫面和產(chǎn)線畫面數(shù)據(jù)隱私是必須考慮的問題。需要注意在廠區(qū)部署攝像頭的區(qū)域應(yīng)告知員工相關(guān)監(jiān)控事項。視頻數(shù)據(jù)要設(shè)置訪問權(quán)限后臺操作日志可追溯。算法分析結(jié)果和原始視頻按需留存避免無限制保存。攝像頭接入和報警記錄操作需要授權(quán)不能用管理員賬號直接運(yùn)行在無監(jiān)控的測試環(huán)境。生產(chǎn)環(huán)境變更時要先在測試環(huán)境驗證模型和配置再逐步灰度發(fā)布到生產(chǎn)通道。7.5 日志、監(jiān)控與可維護(hù)性系統(tǒng)上線后穩(wěn)定運(yùn)行比功能豐富更重要。建議在代碼中增加結(jié)構(gòu)化日志比如記錄每路攝像頭最近一次心跳時間。每一幀推理耗時。報警類型、置信度、檢測框坐標(biāo)。配置項統(tǒng)一放到配置中心模型和規(guī)則參數(shù)支持熱更新。這樣調(diào)整 ROI 區(qū)域和閾值時不需要重新編譯或重啟整個服務(wù)。8. 總結(jié)與學(xué)習(xí)路線本文完整梳理了工廠 AI 視頻分析智能監(jiān)測系統(tǒng)的設(shè)計和實(shí)現(xiàn)路徑重點(diǎn)包括相機(jī)視頻流接入與抽幀處理。YOLOv8 目標(biāo)檢測模型的調(diào)用和結(jié)果解析。基于 IoU 的簡化目標(biāo)跟蹤解決報警重復(fù)問題。ROI 區(qū)域判定和安全帽類別判斷。Flask 提供實(shí)時視頻流與報警接口。簡易監(jiān)控大屏的展示。想進(jìn)一步深入可以按下面的順序?qū)W習(xí)目標(biāo)檢測基礎(chǔ) - 目標(biāo)跟蹤ByteTrack / DeepSORT - 視頻流工程化RTSP / 多線程 / 隊列 - 模型訓(xùn)練與數(shù)據(jù)集構(gòu)建 - 服務(wù)化部署Docker / TensorRT / 彈性伸縮初學(xué)階段不建議一上手就搭建完整大平臺。可以先從“一路視頻 一個模型 一個規(guī)則 一個報警”的最小閉環(huán)練起跑通后逐步增加通道數(shù)、報警規(guī)則、消息推送和可視化大屏。如果本文對你有幫助可以收藏備用也歡迎在評論區(qū)交流你在工廠視頻分析項目中遇到的問題。