简介：基于Python+WebRTC+OpenCV的智慧安防项目，面向需要远程监测婴幼儿睡眠状态、防止摔床的开发者与课程设计者。系统通过后台服务将摄像头画面实时推送到手机或平板，利用OpenCV对视频流进行人体存在与爬动检测，一旦识别到宝宝起床便立即触发本地报警音频，实现低成本的家庭防护方案。资源共10个文件，包括5个Python脚本（后台服务、视频跟踪、模型训练、RTC通信等）、前端监控页面（HTML与JS）、训练好的SVM模型pkl、提示音mp3以及项目说明文档，压缩包仅855KB，结构紧凑便于二次开发。已有476人学习下载，适合作为物联网、计算机视觉或课程设计的参考样例。通过项目说明可快速理解整体架构，掌握OpenCV检测与WebRTC实时通信结合的关键技术，为构建低延迟远程监护系统提供可复用的代码基础。
1. 从弹窗到可靠事件：这套 Python 婴儿看护系统要先解决两个问题
“基于 python+webRTC+opencv 来实现通信和检测小宝睡觉是否起床”这个标题，剖开来看其实是两件事的缝合：一是用 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-python==4.8.1.78 pip install -i https://pypi.tuna.tsinghua.edu.cn/simple aiortc==1.5.0 pip install -i https://pypi.tuna.tsinghua.edu.cn/simple aiohttp==3.9.1 pip install -i https://pypi.tuna.tsinghua.edu.cn/simple numpy==1.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-dev，CentOS 用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, history=500, var_threshold=20): self.bg_subtractor = cv2.createBackgroundSubtractorMOG2( history=history, varThreshold=var_threshold, detectShadows=True ) self.kernel = cv2.getStructuringElement(cv2.MORPH_ELLIPSE, (5, 5)) self.state = "sleeping" self.last_awake_time = 0 # 用 deque 保存最近 10 秒的状态快照，防止单帧抖动造成误判 self.state_history = deque(maxlen=10) self.contour_areas = deque(maxlen=150) def process_frame(self, frame): # 先缩放到统一尺寸，降低计算量 small = cv2.resize(frame, (320, 240), interpolation=cv2.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(history=500, var_threshold=20) 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), ((x+w)*2, (y+h)*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 的 VideoStreamTrack
import 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), dtype=np.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, format="rgb24") # 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> <head><title>小宝安防</title></head> <body> <video id="video" autoplay playsinline style="width: 100%;"></video> <div id="status">状态: 等待连接</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_type='text/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(sdp=data["sdp"], type="offer")) 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 stream=r_frame_rate -of default=noprint_wrappers=1 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，会回退到 VP8，VP8 的 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 解决——一个检测床内、一个检测床沿，综合判断。
另外，问题还可能出在detectShadows=True上，夜间红外灯照射下阴影检测会把宝宝头部的暗区识别为前景。如果阴影造成误报直接改为detectShadows=False。
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 = f"https://sctapi.ftqq.com/{send_key}.send" payload = {"title": title, "desp": content} resp = requests.post(url, data=payload, timeout=5) 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, fps=15): 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_down=GPIO.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)视觉与传感器的双通道校验，能让这个系统的可靠性提升到一个真正可以放心用的水准：视觉负责判断动作，传感器负责校验物理状态。两者结论一致时才是最可信的事件。这也是你后续扩展多房间布防、跌倒检测报警的一个架构模板——检测模块千变万化，管道和事件机制是通用的。
本文还有配套的精品资源，点击获取
