简介基于PythonWebRTCOpenCV的智慧安防项目面向需要远程监测婴幼儿睡眠状态、防止摔床的开发者与课程设计者。系统通过后台服务将摄像头画面实时推送到手机或平板利用OpenCV对视频流进行人体存在与爬动检测一旦识别到宝宝起床便立即触发本地报警音频实现低成本的家庭防护方案。资源共10个文件包括5个Python脚本后台服务、视频跟踪、模型训练、RTC通信等、前端监控页面HTML与JS、训练好的SVM模型pkl、提示音mp3以及项目说明文档压缩包仅855KB结构紧凑便于二次开发。已有476人学习下载适合作为物联网、计算机视觉或课程设计的参考样例。通过项目说明可快速理解整体架构掌握OpenCV检测与WebRTC实时通信结合的关键技术为构建低延迟远程监护系统提供可复用的代码基础。1. 从弹窗到可靠事件这套 Python 婴儿看护系统要先解决两个问题“基于 pythonwebRTCopencv 来实现通信和检测小宝睡觉是否起床”这个标题剖开来看其实是两件事的缝合一是用 OpenCV 把摄像头画面变成检测结果二是用 WebRTC 把画面和事件推到手机或另一个房间的电脑上。单独做任何一件都容易难的是让检测不误报、让通信不掉链以及让两者各自不拖垮对方。比如 OpenCV 每秒钟处理 30 帧画面时Python 的 GIL 会抢占 WebRTC 编码线程的 CPU 时间导致视频延迟从 300ms 飙到 2 秒——这类问题如果不先从架构上想清楚后面填坑会填到怀疑人生。我给的方案是OpenCV 负责采集和推理WebRTC 只做传输两者之间用队列解耦同时用状态机把“小宝起床”从连续视频帧里提炼成离散事件。这样既能让 Python 代码质量可控也方便在将来替换检测算法或接入更多传感器。本文会给出一个可以直接跑起来的 demo而不是停留在概念层面。适合有 Python 基础、想学 WebRTC 与 OpenCV 组合用法、或者正打算做家庭安防/老人看护类项目的工程师阅读。2. 依赖选型与最小环境Python 版本、OpenCV 安装和摄像头验证动手写代码之前先把依赖选型和环境踩坑说明白否则后面每个 import 都可能卡住。本系统的核心依赖有三个opencv-python提供摄像头采集和图像处理aiortc是 Python 生态里最成熟的 WebRTC 库aiohttp做信令服务。三者配合 Python 3.9 以上的版本可以稳定运行。2.1 为什么选 aiortc 而不是其他 WebRTC 方案Python 里的 WebRTC 库很少aiortc几乎是唯一的选择。它是 asyncio 原生的与 Python 的异步编程模型吻合媒体流处理走回调机制不需要自己开多线程。与之对比的是pywebRTC它封装浏览器能力但依赖较新且对 Linux 支持一般还有通过 FFmpeg 推流到 Janus 网关的方案适合多人互动场景但家庭安防场景只需要一对一查看没必要引网关增加复杂度。aiortc的内部结构也简单RTCPeerConnection管理和维护 PeerConnection 状态MediaStreamTrack承载音视频帧RTCSessionDescription负责在两端间交换 SDP。对于这个项目我只需要从 OpenCV 读到的帧封装成一个自定义VideoStreamTrack子类aiortc 会处理后续的编码与传输。数据通道用RTCDataChannel可以走事件文本通知不做视频编码。2.2 版本选型和安装命令# 建议使用 Python 3.9 或更高版本避免老版本 asyncio 兼容问题 python3 -m venv ~/baby_guard_env source ~/baby_guard_env/bin/activate # 使用清华镜像加速opencv-python 安装包体积较大网速不好会超时 pip install -i https://pypi.tuna.tsinghua.edu.cn/simple opencv-python4.8.1.78 pip install -i https://pypi.tuna.tsinghua.edu.cn/simple aiortc1.5.0 pip install -i https://pypi.tuna.tsinghua.edu.cn/simple aiohttp3.9.1 pip install -i https://pypi.tuna.tsinghua.edu.cn/simple numpy1.24.4 # 验证安装 python3 -c import cv2; print(cv2.__version__) python3 -c import aiortc; print(aiortc.__version__)注意aiortc在安装时会编译 av 库依赖如果提示缺少libavformat等需要先安装系统级依赖。Ubuntu 用apt install libavformat-dev libavcodec-dev libavdevice-dev libavutil-dev libswscale-devCentOS 用dnf install ffmpeg-devel。macOS 用户需要先确保有 Homebrew 的 ffmpeg。Windows 下安装相对顺畅但摄像头索引号可能与 Linux 不同需要按系统区分。2.3 读取摄像头并确认画面可用import cv2 import numpy as np # 注意0 表示第一个摄像头设备树莓派 CSI 摄像头需要先配置驱动 cap cv2.VideoCapture(0, cv2.CAP_DSHOW) if os.name nt else cv2.VideoCapture(0) if not cap.isOpened(): raise RuntimeError(无法打开摄像头检查设备号和驱动) # 设置分辨率与帧率过高的分辨率会让 CPU 资源全部耗尽在编码上 cap.set(cv2.CAP_PROP_FRAME_WIDTH, 640) cap.set(cv2.CAP_PROP_FRAME_HEIGHT, 480) cap.set(cv2.CAP_PROP_FPS, 15) while True: ret, frame cap.read() if not ret: print(读取帧失败可能摄像头被占用) break # 不直接显示把画面保存成 JPEG 是后续 WebRTC 编码的标准输入格式 _, jpeg cv2.imencode(.jpg, frame, [cv2.IMWRITE_JPEG_QUALITY, 80]) frame_bytes jpeg.tobytes() # 这里只做显示验证实际项目中 frame_bytes 会被送入 WebRTC 轨道 cv2.imshow(preview, frame) if cv2.waitKey(1) 0xFF ord(q): break cap.release() cv2.destroyAllWindows()这段代码的关键是设置了 640x480 分辨率和 15 帧的采集频率。为什么不定到 1080p30因为这个系统的主要场景是婴儿床人物距离镜头 1-2 米640 宽度像素足够检测人体但帧数直接影响 OpenCV 的检测频率和 WebRTC 编码负载。1080p30 会让树莓派 4B 的 CPU 使用率冲到 80% 以上检测响应还会变慢。如果你想用 1280x720那就把检测降到每秒 5 帧否则意义不大。提示CAP_DSHOW只在 Windows 下需要Linux 和 macOS 不需要。如果出现画面颠倒或镜像用cv2.flip(frame, -1)调整方向。摄像头打不开时先排除权限问题Linux 下把用户加入video组。验证完摄像头之后下一步要把画面转换成数字特征让程序能判断小宝是否在睡觉、是否起床。这是整个系统的智慧核心。3. 用 OpenCV 实现起床检测背景减除、移动侦测和状态机起床检测在计算机视觉里属于动作检测的简化场景目标是判断目标是否从“躺着”变成了“坐起或站立”。不需要识别具体是谁只要捕捉到显著的人体位置和姿态变化。这里我用背景减除加轮廓分析来实现复杂度低、实时性高树莓派也能跑得动。3.1 背景减除还是目标检测为什么二选一目标检测需要模型权重文件和推理框架YOLOv8 或 MobileNet SSD 的精度固然高但树莓派 CPU 推理一帧要 100-300 毫秒配合摄像头 15 帧每秒就卡顿了。背景减除基于像素级差分单帧处理在 10 毫秒级别完全够用。它的问题在于光照变化会产生大量噪点所以需要做好形态学处理和轮廓面积过滤。OpenCV 自带的createBackgroundSubtractorMOG2是首选方法它基于高斯混合模型能把缓慢的光照变化当成背景学习掉。对比帧差法MOG2 在宝宝连续翻身时会保留更多背景残影检测更稳定。核心参数history控制背景更新速度值越小背景适应越快但容易把静止的宝宝也学进背景里。建议设成 500相当于 500 帧以上才把静止画面完全当作背景。varThreshold控制判定前景的灵敏度建议试试 16 到 25 之间灵敏度越高越能捕捉微小动作但也越容易误报。3.2 起床检测的完整代码从像素级变动到事件级判断import cv2 import numpy as np import time from collections import deque class WakingDetector: def __init__(self, history500, var_threshold20): self.bg_subtractor cv2.createBackgroundSubtractorMOG2( historyhistory, varThresholdvar_threshold, detectShadowsTrue ) self.kernel cv2.getStructuringElement(cv2.MORPH_ELLIPSE, (5, 5)) self.state sleeping self.last_awake_time 0 # 用 deque 保存最近 10 秒的状态快照防止单帧抖动造成误判 self.state_history deque(maxlen10) self.contour_areas deque(maxlen150) def process_frame(self, frame): # 先缩放到统一尺寸降低计算量 small cv2.resize(frame, (320, 240), interpolationcv2.INTER_AREA) # 高斯模糊去掉传感器噪点尤其夜间长曝光画面噪点明显 blur cv2.GaussianBlur(small, (5, 5), 0) fg_mask self.bg_subtractor.apply(blur) # 去阴影MOG2 检测到的阴影区域是灰度值 127 _, fg_mask cv2.threshold(fg_mask, 200, 255, cv2.THRESH_BINARY) # 形态学开运算去掉孤立像素点再闭运算合并断裂区域 fg_mask cv2.morphologyEx(fg_mask, cv2.MORPH_OPEN, self.kernel) fg_mask cv2.morphologyEx(fg_mask, cv2.MORPH_CLOSE, self.kernel) # 找出轮廓面积过滤是关键太小的区域是噪点太大的可能是光线整面变化 contours, _ cv2.findContours(fg_mask, cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_SIMPLE) max_area 0 max_box None for cnt in contours: area cv2.contourArea(cnt) if area max_area: max_area area x, y, w, h cv2.boundingRect(cnt) max_box (x, y, w, h) # 归一化面积相对画面总尺寸适应不同分辨率 norm_area max_area / (frame.shape[0] * frame.shape[1]) self.contour_areas.append(norm_area) # 只在画面有显著变化时更新时间戳 current_time time.time() if norm_area 0.03: self.last_awake_time current_time # 判断逻辑最近 10 帧的平均有效面积超过阈值且有持续肢体运动的迹象 recent_areas [a for a in list(self.contour_areas)[-15:] if a 0.03] activity_ratio len(recent_areas) / 15 if activity_ratio 0.6 and (current_time - self.last_awake_time) 3: # 如果持续出现大面积前景说明宝宝从被窝坐起或翻滚 new_state awake else: new_state sleeping self.state_history.append(new_state awake) # 连续 5 帧中检测到 4 帧以上才算切换状态防止被一次突然的动作如翻身误触发 if sum(self.state_history) 4: self.state awake elif sum(self.state_history) 1: self.state sleeping return self.state, max_box, fg_mask # 单独演示这个检测器 cap cv2.VideoCapture(0) detector WakingDetector(history500, var_threshold20) sleep_start time.time() while True: ret, frame cap.read() if not ret: break state, box, mask detector.process_frame(frame) if state awake: cv2.putText(frame, WAKE UP!, (50, 50), cv2.FONT_HERSHEY_SIMPLEX, 1.2, (0, 0, 255), 3) if box: x, y, w, h box # 放大到原图分辨率对应位置 cv2.rectangle(frame, (x*2, y*2), ((xw)*2, (yh)*2), (0, 0, 255), 2) else: cv2.putText(frame, sleeping zzz..., (50, 50), cv2.FONT_HERSHEY_SIMPLEX, 1.2, (0, 255, 0), 3) cv2.imshow(detection, frame) if cv2.waitKey(1) 0xFF ord(q): break cap.release() cv2.destroyAllWindows()这段检测流程分四层背景建模、形态学清洗、轮廓提取、状态机判定。前三层是 OpenCV 标准的视觉处理管道第四层是工程中最容易被忽视的部分。为什么需要状态机和历史队列因为单帧的norm_area 0.03不能说明宝宝醒了——他可能只是翻身、踢被子、或者因为盖被子动作挡住了摄像头。通过state_history记录最近 10 帧的状态分布并要求至少 4 帧判定为活动才能把短暂的动作和持续的清醒区分开。参数调整可以从varThreshold入手如果检测区域频繁闪烁调大到 30如果检测不到微弱的动作调小到 15。norm_area的 0.03 表示前景占到画面面积的 3%适合婴儿床场景如果摄像头视角更广就调小到 0.02。这套检测方法对光线变化的鲁棒性有限摄像头位置要避免正对窗户或强光源。3.3 夜间模式不可见光和红外摄像头的适配夜间看护是刚需但普通摄像头在黑暗中什么也看不见。常见做法是换成带红外 LED 的摄像头或者使用树莓派官方 NoIR 摄像头模块。红外画面是单通道灰度图OpenCV 处理时直接按灰度图读即可不需要转换。但要注意红外画面里宝宝皮肤会过曝发白背景减除的阈值需要相应调大 30% 左右。适配夜间模式的代码只需要在采集端做一个判断# 假设 frame 是灰度图模式下读取的画面 if len(frame.shape) 2: frame cv2.cvtColor(frame, cv2.COLOR_GRAY2BGR) # 后续走同样的流程红外场景下的误报主要来自镜头前的蚊虫和飞尘它们的面积小但高速移动会在图像里产生连续噪声。建议在检测前加一帧判断当一个轮廓的中心点在两帧之间移动了超过 50 像素就当作快速移动物体过滤掉。代码里只需维护前一帧轮廓位置移动距离超过阈值就跳过该轮廓。检测做完了接下来要把画面和状态推送到用户的手机上。这一步是 WebRTC 的主场。4. aiortc 实现端到端视频传输信令交换、视频轨道和数据通道WebRTC 本身是为浏览器设计的但 aiortc 把它带回了 Python 的世界。核心流程不变双方要交换 SDP 描述和 ICE candidate 信息然后通过 P2P 或中继传输音视频数据。这里最大的工程决策是信令服务的形态。4.1 信令方案选型WebSocket 还是 HTTP 轮询WebRTC 建立连接的第一步是交换 SDP 和 ICE candidate这个过程必须有信令通道。本地同一局域网内的场景可以用一个轻量级 aiohttp 服务器持久保存 WebSocket 连接只在交换信令时负责转发消息。外网访问场景可以部署在一个带公网 IP 的服务器上或用内网穿透工具但这就引入中继服务器延迟和带宽都会成问题。我建议的方案是先用局域网 WebSocket 跑通因为所有信令交互在 20 秒内完成WebSocket 的持久连接虽然省心但轮询实现更简单。不过摄像头端的 IP 是静态的用 WebSocket 建立信令后还有一个好处是后续可以直接通过同一个通道发送“起床”事件通知——所以最终选了 WebSocket 而不是 HTTP。4.2 把 OpenCV 帧包装成 aiortc 的 VideoStreamTrackimport asyncio import cv2 import numpy as np from fractions import Fraction from aiortc import VideoStreamTrack, RTCPeerConnection, RTCSessionDescription from aiortc.contrib.media import MediaStreamTrack, MediaPlayer class OpenCVVideoTrack(VideoStreamTrack): def __init__(self, detector): super().__init__() self.detector detector self.cap cv2.VideoCapture(0) self.cap.set(cv2.CAP_PROP_FRAME_WIDTH, 640) self.cap.set(cv2.CAP_PROP_FRAME_HEIGHT, 480) self.cap.set(cv2.CAP_PROP_FPS, 15) self._last_frame None self._state sleeping async def recv(self): # aiortc 用异步方式请求下一帧这个循环必须快不能阻塞事件循环 if self.cap is None: raise Exception(摄像头未打开) # 避免让 aiortc 等太久超时就直接重发上一帧 for _ in range(2): ret, frame self.cap.read() if ret: break else: # 读取失败后复用旧帧保持视频流不中断 frame self._last_frame.copy() if self._last_frame is not None else np.zeros((480, 640, 3), dtypenp.uint8) self._last_frame frame self._state, _, _ self.detector.process_frame(frame) # OpenCV 默认 BGR转成 RGB 后送入编码器 rgb cv2.cvtColor(frame, cv2.COLOR_BGR2RGB) # 生成 VideoFrame 对象required 参数指定时间戳和格式 from av import VideoFrame video_frame VideoFrame.from_ndarray(rgb, formatrgb24) # aiortc 需要时间戳来控制帧率 video_frame.pts asyncio.get_event_loop().time() * 1000 video_frame.time_base Fraction(1, 1000) return video_frame关键点在于recv()方法是异步的aiortc 会用它内部的时钟循环反复调用。OpenCV 的cap.read()是阻塞的默认会一直等待新帧但摄像头输入不稳定时可能导致recv()超时。干净的做法是把相机读取放到单独线程通过队列送帧到recv()但那样代码就复杂了。上面的方案用for循环尝试两次读取失败回退旧帧简单且可靠——视频偶尔卡顿不值得引发整体异常。注意VideoFrame.from_ndarray输入是 RGB 格式。如果直接用 BGR 的 frame看到的画面颜色会全部反相。这个问题排查起来很隐蔽。VideoFrame 的时间戳单位可以灵活设置我这里统一换算成毫秒便于处理。time_base表示时间的分母设成 1/1000 意味着 pts 值的单位是毫秒。4.3 WebSocket 信令服务器让手机端能查看画面import json import asyncio from aiohttp import web, WSMsgType from aiortc import RTCPeerConnection, RTCSessionDescription # 全局的 PeerConnection 对象 pcs set() video_track None detector WakingDetector() async def index(request): # 简单返回一个网页包含前端 JavaScript 代码 return web.Response(text !DOCTYPE html html headtitle小宝安防/title/head body video idvideo autoplay playsinline stylewidth: 100%;/video div idstatus状态: 等待连接/div script const pc new RTCPeerConnection(); pc.ontrack (event) { document.getElementById(video).srcObject event.streams[0]; }; // 用页面轮询方式让用户手动触发一次 offer 创建会更稳定 const offer_button document.createElement(button); offer_button.textContent 开始查看; offer_button.onclick startView; document.body.appendChild(offer_button); async function startView() { const offer await pc.createOffer(); await pc.setLocalDescription(offer); ws.send(JSON.stringify({type: offer, sdp: offer})); } const ws new WebSocket(ws:// location.host /ws); ws.onmessage async (event) { const data JSON.parse(event.data); if (data.type answer) { await pc.setRemoteDescription(data); } else if (data.type ice) { await pc.addIceCandidate(data.candidate); } else if (data.type event) { document.getElementById(status).textContent 状态: data.message; } }; /script /body /html , content_typetext/html) async def websocket_handler(request): ws web.WebSocketResponse() await ws.prepare(request) pc RTCPeerConnection() pcs.add(pc) pc.on(track) def on_track(track): print(视频轨道创建, track.kind) if track.kind video: # 给 WebRTC 对方添加视频轨道 pc.addTrack(video_track) pc.on(datachannel) def on_datachannel(channel): print(数据通道打开) # 可以在数据通道上发事件也可以直接用 WebSocket 发事件 channel.send(连接建立) # 数据通道用于事件通知WebSocket 用于信令 pc.on(connectionstatechange) async def on_connectionstatechange(): state pc.connectionState print(连接状态:, state) if state failed or state closed: pcs.discard(pc) elif state connected: # 连接成功后立即发送当前状态 await ws.send_str(json.dumps({type: event, message: str(detector.state)})) async for msg in ws: if msg.type WSMsgType.TEXT: data json.loads(msg.data) if data[type] offer: # 远端发起 offer就设置远程描述并回 answer await pc.setRemoteDescription(RTCSessionDescription(sdpdata[sdp], typeoffer)) answer await pc.createAnswer() await pc.setLocalDescription(answer) await ws.send_str(json.dumps({type: answer, sdp: pc.localDescription.sdp, state: answer})) elif data[type] ice: await pc.addIceCandidate({candidate: data[candidate], sdpMid: data.get(sdpMid), sdpMLineIndex: data.get(sdpMLineIndex)}) elif msg.type WSMsgType.CLOSE: await pc.close() break return ws # 启动 Web 服务 async def start_server(): app web.Application() app.router.add_get(/, index) app.router.add_get(/ws, websocket_handler) runner web.AppRunner(app) await runner.setup() site web.TCPSite(runner, 0.0.0.0, 8080) await site.start() print(服务器运行在 http://0.0.0.0:8080) await asyncio.Future() # 永久运行 if __name__ __main__: # 创建全局视频轨道 video_track OpenCVVideoTrack(detector) asyncio.run(start_server())前端页面在访问 8080 端口时会拿到一段 JavaScript点击按钮创建 WebRTC offer 发给 Python 后端后端处理后把 answer 返回视频流从 Python 端推送到浏览器端。这个方案在同一个局域网内可以实现 500 毫秒以内的延迟比 RTMP 推流方案快得多。4.4 RTCDataChannel 还是 WebSocket 发送事件通知视频流走 WebRTC 媒体通道起床事件怎么推送两个选择走RTCDataChannel直接点到点发送或者复用 WebSocket 信令通道。后者的优势是信令通道在连接建立后一直存在事件发送只需要一行代码ws.send_str()缺点是 WebSocket 断开会失去事件推送能力。前者的好处是事件与媒体流走同一对 PeerConnection 的连接连接状态天然同步而且事件不经过服务器。两者各有优劣但我的建议是事件通知用 WebSocket因为它和信令服务耦合最简单你不需要额外维护数据通道的生命周期。数据通道适合音视频之外的实时流数据比如心率传感器的波形推送。事件推送的具体实现# 在检测器的主循环里当状态变化时通过 WebSocket 推送 async def event_loop(ws, detector): prev_state None while True: state, _, _ detector.process_frame(video_track._last_frame) if state ! prev_state: message 小宝醒啦 if state awake else 小宝已入睡 await ws.send_str(json.dumps({type: event, message: message})) prev_state state await asyncio.sleep(1) # 每秒检查一次把事件检测和媒体发送解耦用叠加在 WebSocket 通道上的异步循环去轮询检测器状态这样状态变化即使发生在两帧读取之间也不会漏报占位。detector.process_frame已经取出过一帧画面所以这个循环不消耗额外摄像头读取只需要复检最近一帧的状态变化。走到这一步一个能看画面、能报警的最小系统已经成型。接下来的问题是如何让它在真实环境中稳定运行——树莓派的散热、夜间光线变化、网络波动这才是决定家里人是否愿意用的关键。5. 系统联调与排错延迟、误报率和持续运行的三个检查点很多项目第一次演示成功了但放在床头跑一晚上就崩。这里把三个高频故障的排查方式写清楚视频延迟偏高、连通性失败、内存泄漏导致的掉线。每类问题我给你标准检查顺序。5.1 视频延迟超过 1 秒时的排查路径延迟高有四个来源采集延迟、编码延迟、传输延迟、解码延迟。先用一个测试区分哪个环节# 在服务端测试编码前帧率 ffprobe -v error -select_streams v:0 -show_entries streamr_frame_rate -of defaultnoprint_wrappers1 http://127.0.0.1:8080 # 在客户端看实际帧间隔 # Chrome 控制台执行观察帧到达之间的间隔 videoEl.getVideoPlaybackQuality().totalVideoFrames如果服务端采集到编码的耗时就超过 200ms 单帧然后检查cap.set(cv2.CAP_PROP_FPS, 15)是否生效。部分摄像头驱动不支持 15 帧实际可能工作在 5 帧那延迟就会异常。编码延迟通常是 CPU 瓶颈aiortc默认使用 H.264 编码但如果 Python 环境里没安装openh264会回退到 VP8VP8 的 CPU 占用比 H.264 高出约 30%在树莓派上编码一帧 640x480 可能耗 150ms。安装方式pip install openh264安装后 aiortc 会自动探测并优先用 H.264。如果还不生效手动在RTCPeerConnection构造时指定编解码器偏好。网络传输延迟在同一局域网内通常不超过 10ms可以忽略。如果走公网考虑在 WebRTC 配置里指定 STUN 服务器帮助建立 P2P 直连避免流量经过谷歌或声网的公共中继服务器。提示手机上看到视频卡顿还有可能是浏览器的playsinline属性没加上iOS 会默认全屏播放导致交互异常。5.2 误报率压不下来问题往往在状态机而不是检测算法小区里有人走动、风吹窗帘、窗外车灯扫过都会形成大面积前景。解决方向把检测区域缩小到婴儿床的固定范围。在代码里设定 ROI感兴趣区域只对婴儿床区域内做背景减除。# 把婴儿床区域设定为全画幅的中间部分比如下方的 70% 区域 roi frame[int(frame.shape[0]*0.15):frame.shape[0], :] # 把 roi 送进 detector.process_frame 之前 state, box, mask detector.process_frame(roi)这个改动把 14 平米房间的干扰减少到 2 平米的婴儿床范围误报率能降一个数量级。代价是如果宝宝爬到床沿外就检测不到了可以通过双 ROI 解决——一个检测床内、一个检测床沿综合判断。另外问题还可能出在detectShadowsTrue上夜间红外灯照射下阴影检测会把宝宝头部的暗区识别为前景。如果阴影造成误报直接改为detectShadowsFalse。5.3 长期运行掉线、画面冻结的对策看门狗和内存控制树莓派长时间运行 Python 进程摄像头驱动偶尔会无响应。处理方法是在主循环加 try/catch摄像头断开时自动重连开一条低优先级线程每 5 分钟检查一次摄像头句柄的存活状态def camera_watchdog(): while True: time.sleep(300) global cap if cap is None or not cap.isOpened(): print(摄像头断开正在重连...) cap.release() cap cv2.VideoCapture(0) if not cap.isOpened(): print(重连失败10 秒后再试) time.sleep(10) continue print(重连成功)内存泄漏主要来自VideoFrame对象未正确释放。检测getPeformanceEntry时留意内存增长曲线如果每 1000 帧增加超过 20MB说明提交流程里有对象一直持有。最常见的原因是VideoFrame.pts没被 aiortc 正确消费导致 to_ndarray 拿了引用正确处理是每次 recv 关联的视频帧在下一次调用前被自动覆盖。另外OpenCV 的VideoCapture本身也会缓存帧长时间运行后内部缓冲会变大可以在每 1000 帧调用一次cap.grab()丢弃缓存。有一个隐藏的坑aiortc 在视频轨道暂停发送时会持续缓冲帧内存随之上涨。处理方法是重写VideoStreamTrack的stop()方法把内部缓存的帧全部清空。简单的做法是在recv()里限制队列长度不满足就await asyncio.sleep(0.001)让事件循环喘息。排错完稳定运行再把最小系统往实用的方向推一层把检测结果接入常规的推送链路让它真正能叫醒你。6. 事件通知接入微信 / 邮件和本地录像的三个落地技巧光在手机上守着看画面不够。小宝起床了你不可能一直盯着屏幕。把检测状态变成推送管道里的一个触发源整个智慧安防才闭环。6.1 通过 Server酱或自建 Webhook 发微信通知家庭项目最常见的做法是用 Server酱sct.ftqq.com注册后拿到一个 SendKey用 HTTP POST 就能给微信推送消息。服务端只需要把状态变化事件映射成请求即可import requests def send_wechat_notification(title, content, send_key): url fhttps://sctapi.ftqq.com/{send_key}.send payload {title: title, desp: content} resp requests.post(url, datapayload, timeout5) if resp.status_code ! 200: print(f推送失败: {resp.text}) # 在异步事件循环里调用需要放到 executor 里避免阻塞事件循环 loop asyncio.get_running_loop() await loop.run_in_executor(None, send_wechat_notification, 小宝起床了, f检测时间: {time.strftime(%H:%M:%S)}, SEND_KEY)注意requests.post这种同步 I/O 不能直接写在 aiortc 异步回调里因为它会卡住事件循环导致视频流和信令消息堆积。run_in_executor是标准解法。没有 Server酱账号的话Telegram Bot API 也是常见替代但国内使用不稳定。邮件通知则适合做古早但永久可用的保底通道用法是 SMTP 加 TLS 发件。无论用哪个通道push 失败的状态要留在内存队列里间隔 5 分钟重试一次最多三次避免半夜推送管道抖动导致漏报。6.2 事件触发录像节省存储空间和检索时间24 小时录像到远程存储意味着每天几十 GB 数据完全没必要。改成事件驱动录像状态从sleeping变成awake时开始录像状态变回sleeping后 30 秒停止。这样一天的存储不到 1GB且检索方便。import cv2 class EventRecorder: def __init__(self): self.video_writer None self.recording_path None self.last_state None def update(self, state, frame, fps15): if state awake and self.video_writer is None: # 生成基于时间戳的文件名 self.recording_path f/home/pi/baby_videos/{time.strftime(%Y%m%d_%H%M%S)}.avi fourcc cv2.VideoWriter_fourcc(*XVID) self.video_writer cv2.VideoWriter(self.recording_path, fourcc, fps, (640, 480)) print(f开始录像: {self.recording_path}) elif state sleeping and self.video_writer is not None: # 停止录像并释放资源 self.video_writer.release() self.video_writer None print(停止录像) if self.video_writer is not None: small cv2.resize(frame, (640, 480)) self.video_writer.write(small)录制的文件按事件时间戳命名之后用ffprobe快速定位某个时间点或者用 OpenCV 做缩略图切片就能在手机上快速看回放。6.3 离床检测的补充方案用传感器给视觉兜底视觉方案在完全黑暗和遮挡情况下会有盲区比如宝宝滚到被子下面完全被遮住背景减除可能把被子当成背景学掉。加一个便宜的红外距离传感器铺在床垫下检测离床压力变化当压力消失超过 30 秒且视觉状态是awake则触发更高优先级的报警。用树莓派的 GPIO 接口读取import RPi.GPIO as GPIO pressure_pin 17 GPIO.setmode(GPIO.BCM) GPIO.setup(pressure_pin, GPIO.IN, pull_up_downGPIO.PUD_UP) # 在 event loop 中读取任何时候压力为 0 且状态为 awake 就加急推送 is_on_bed GPIO.input(pressure_pin) # 高电平表示压力存在 if not is_on_bed and detector.state awake: # 优先级最高的推送或者触发声音警报 await loop.run_in_executor(None, send_wechat_notification, 紧急警报, 宝宝已离开床垫, SEND_KEY)视觉与传感器的双通道校验能让这个系统的可靠性提升到一个真正可以放心用的水准视觉负责判断动作传感器负责校验物理状态。两者结论一致时才是最可信的事件。这也是你后续扩展多房间布防、跌倒检测报警的一个架构模板——检测模块千变万化管道和事件机制是通用的。本文还有配套的精品资源点击获取