去年为一个仓库安防项目开发了一套智能监控系统,当摄像头捕捉到特定目标(如未授权人员、危险物品)时,系统会自动触发邮件报警并附上实时画面。这种结合计算机视觉和自动化工作流的技术方案,在安防、工业质检、智慧零售等领域都有广泛应用场景。本文将拆解从目标检测到邮件触发的完整实现链条,包含模型选型、接口设计、服务编排等关键环节。
YOLOv5是目前工程化落地最成熟的方案,相比Faster R-CNN等两阶段检测器,其单阶段设计在保持较高精度的同时能满足实时性要求。实测在Tesla T4显卡上,YOLOv5s版本对640x480分辨率视频能达到140FPS。对于需要更高精度的场景,可以切换到YOLOv5x版本(代价是帧率降至40FPS)。
注意:模型选择需考虑硬件条件。树莓派等边缘设备推荐使用MobileNet-SSD,其量化后模型仅20MB左右
主流有两种实现路径:
python复制import smtplib
from email.mime.multipart import MIMEMultipart
def send_email(image):
msg = MIMEMultipart()
msg['Subject'] = 'Security Alert'
server = smtplib.SMTP('smtp.example.com', 587)
server.starttls()
server.login("user@example.com", "password")
server.sendmail("user@example.com", "admin@example.com", msg.as_string())
python复制from sendgrid import SendGridAPIClient
from sendgrid.helpers.mail import Mail
message = Mail(
from_email='alerts@yourdomain.com',
to_emails='admin@yourdomain.com',
subject='Intrusion Detected',
html_content='<img src="cid:detection_result">')
message.add_attachment(image)
sg = SendGridAPIClient(os.environ.get('SENDGRID_API_KEY'))
response = sg.send(message)
采用生产者-消费者模式构建高效处理流水线:
python复制import threading
from queue import Queue
frame_queue = Queue(maxsize=30)
alert_queue = Queue()
def capture_thread():
cap = cv2.VideoCapture("rtsp://camera_feed")
while True:
ret, frame = cap.read()
if not ret: break
frame_queue.put(frame)
def detection_thread():
model = torch.hub.load('ultralytics/yolov5', 'yolov5s')
while True:
frame = frame_queue.get()
results = model(frame)
if is_alert_condition(results):
alert_queue.put((frame, results))
def alert_thread():
while True:
frame, results = alert_queue.get()
annotated_frame = draw_boxes(frame, results)
send_alert_email(annotated_frame)
根据不同场景设计触发逻辑:
python复制def is_alert_condition(results):
# 示例:检测到人员且置信度高于阈值
df = results.pandas().xyxy[0]
high_conf_persons = df[(df['name']=='person') & (df['confidence']>0.8)]
return len(high_conf_persons) > 0
当需要低延迟处理时,可采用以下优化:
bash复制python export.py --weights yolov5s.pt --include onnx
trtexec --onnx=yolov5s.onnx --saveEngine=yolov5s.trt
python复制from multiprocessing import Pool
def process_frame(frame):
model = torch.hub.load(...)
return model(frame)
with Pool(4) as p:
results = p.map(process_frame, frame_batch)
关键措施包括:
python复制from retrying import retry
@retry(stop_max_attempt_number=3, wait_exponential_multiplier=1000)
def send_email_with_retry(image):
try:
# 压缩图像至200KB以下
img = Image.fromarray(image)
img.save('temp.jpg', quality=85, optimize=True)
# 发送逻辑...
except Exception as e:
log_error(e)
raise
初期系统发送的报警邮件时间戳全是UTC,导致现场人员混淆。解决方案:
python复制from datetime import datetime
import pytz
local_tz = pytz.timezone('Asia/Shanghai')
alert_time = datetime.now(local_tz).strftime('%Y-%m-%d %H:%M:%S')
通过以下方法减少误报:
python复制def in_roi(box, roi):
x1, y1, x2, y2 = box
return (roi[0] < x1 < roi[2]) and (roi[1] < y1 < roi[3])
好的报警邮件应包含:
html复制<div>
<h3>Security Alert - Unauthorized Access</h3>
<img src="cid:detection_img" width="640">
<p>Detected at: {time} | Camera: Entrance North</p>
<a href="https://yourportal/confirm?alert_id=123">Mark as Resolved</a>
</div>
除邮件外,可扩展通知渠道:
设计分级响应策略:
将报警记录存入数据库便于后续分析:
python复制import sqlite3
def log_alert(event):
conn = sqlite3.connect('alerts.db')
c = conn.cursor()
c.execute('''INSERT INTO alerts
(timestamp, camera_id, object_class, confidence)
VALUES (?,?,?,?)''',
(event['time'], event['camera'], event['class'], event['conf']))
conn.commit()
这套系统在实际部署中需要根据具体场景调整检测阈值和报警策略。建议先用历史视频数据测试不同参数下的误报率/漏报率,找到业务可接受的平衡点。对于关键场所,可以考虑增加多摄像头协同验证逻辑来提高可靠性。