这份备忘是结合你从零踩坑到成功跑通的全过程整理的涵盖了原理、概念、完整复现步骤、踩坑记录以及调参指南。一、 基本原理与架构整个链路就像一个“全自动保安系统”摄像头/视频-Python脚本(大脑)-MQTT(对讲机)-Home Assistant(控制台)。Python脚本保安的大脑读取视频画面对比“没人时的背景”背景减除算法 MOG2。发现画面有差异有东西在动就画上绿框算出运动面积占比。MQTT对讲机把算出的结果带框的画面、状态、面积打包通过不同的“频道”主题 Topic广播出去。Home Assistant控制台订阅这些频道把画面展示在网页上把状态接入自动化。数据流向与 MQTT 主题映射aiot/camera1/image(二进制JPEG) - HA 的camera.camera1(显示画面)aiot/camera1/motion(ON/OFF) - HA 的binary_sensor.camera1motion(触发自动化)aiot/camera1/motion_area(数字%) - HA 的sensor.camera1motionarea(仪表盘展示)aiot/camera1/event(JSON) - 备用以后接 ThingsBoard 或存档。二、 核心概念venv虚拟环境独立沙盒避免 Python 依赖包冲突导致 HA 崩溃。位置在~/aiot-camera/.venv。每次操作前必须source .venv/bin/activate。camera vs binary_sensorcamera是给“人眼”看的图像流binary_sensor是给“机器”用的只判断 ON/OFF用于自动化。MQTT 发布/订阅脚本只管“发布”消息HA 只管“订阅”消息。只要有相同的 Topic 和 Broker 地址两边就能对接。systemdLinux 后台服务管家。把脚本交给他就能实现开机自启、崩溃自动重启不必一直开着终端窗口。三、 完整复现步骤1. 构建独立环境与依赖sudoaptinstall-ypython3-venv mosquitto-clientsmkdir-p~/aiot-cameracd~/aiot-camera python3-mvenv .venvsource.venv/bin/activate pipinstallopencv-python-headless paho-mqtt numpy(把camera_motion.py、make_demo_video.py上传到该目录)2. 测试脚本与 MQTT 链路先不碰 HA终端1生成视频并跑脚本source~/aiot-camera/.venv/bin/activatecd~/aiot-camera python make_demo_video.py python camera_motion.py demo_video.mp4终端2监听 MQTTmosquitto_sub-h127.0.0.1-taiot/camera1/#-v验收标准看到motion ON、motion OFF和motion_area在跳图片流显示乱码正常二进制。3. 修改 HA 配置 必须改对路径路径警告必须是/home/dragon/.homeassistant/带点的隐藏目录不要改到/homeassistant/无点的影子目录修改configuration.yaml在mqtt:下方、与sensor:平级的位置追加camera和binary_sensor块camera:-name:Camera1unique_id:cam1_imagetopic:aiot/camera1/imageavailability_topic:aiot/camera1/availabilitybinary_sensor:-name:Camera1Motionunique_id:cam1_motionstate_topic:aiot/camera1/motionpayload_on:ONpayload_off:OFFdevice_class:motionavailability_topic:aiot/camera1/availability⚠️ 追加到原有sensor:列表末尾不要新建第二个sensor:-unique_id:cam1_motion_areaname:Camera1MotionAreastate_topic:aiot/camera1/motion_areaunit_of_measurement:%availability_topic:aiot/camera1/availability(可选)添加自动化在文件已有automation:列表末尾追加-alias:摄像头检测到运动triggers:-trigger:stateentity_id:binary_sensor.camera1motionto:onactions:-action:persistent_notification.createdata:title:视觉告警message:摄像头1检测到运动修改ui-lovelace.yaml在views:下新增一页注意缩进对齐平级-title:摄像头path:cameraicon:mdi:cctvcards:-type:picture-entityentity:camera.camera1show_state:false-type:entitiestitle:运动检测entities:-entity:binary_sensor.camera1motion-entity:sensor.camera1motionarea(避坑HA 实体 ID 强制全小写千万不能写camera1motion之类的大小写混用)4. 重启与验收开发者工具 - YAML - 检查配置 - 重启 HA -CtrlF5强制刷新网页 - 进“摄像头”标签页。保持 Python 脚本运行看到带绿框的画面和传感器跳动即为成功。5. 后台常驻交付使用创建服务文件sudo nano /etc/systemd/system/aiot-camera.service[Unit] DescriptionAIoT camera motion detector Afternetwork.target mosquitto.service [Service] Userdragon WorkingDirectory/home/dragon/aiot-camera ExecStart/home/dragon/aiot-camera/.venv/bin/python camera_motion.py 0 Restartalways RestartSec5 [Install] WantedBymulti-user.target生效并启动sudosystemctl daemon-reloadsudosystemctlenable--nowaiot-camera journalctl-uaiot-camera-f# 查看运行日志四、 踩坑记录文件路径的“双胞胎”陷阱HA 真正读取的是/home/dragon/.homeassistant/隐藏目录。你曾经改错到homeassistant/config/导致怎么改都没反应。以后改配置一定要用grep验证或使用绝对路径。YAML 格式错乱致死YAML 对缩进极其严格。多一个空格、用中文冒号、错误缩进如把binary_sensor写进了sensor列表都会导致 HA 配置检查失败整个新配置全部罢工。集成冲突与 Token 失效小米集成invalid refresh token报错会直接卡住 HA 启动导致 MQTT 配置不加载。遇到莫名卡顿去设置 - 设备与服务里禁用/删除报错的集成。HA 重启无效的错觉如果是configuration.yaml的改动必须重启 HA 服务设置 - 系统 - 重启。仅仅刷新网页或重载 Lovelace 是不会生效的。USB 摄像头连接“独木桥”罗技 BRIO 这类高清摄像头必须把 VMware 的 USB 控制器设为USB 3.1否则带宽不够设备无法识别。连接前先彻底关机虚拟机调整设置后再开机。插上摄像头后在 VMware 菜单栏“虚拟机 - 可移动设备”中手动点击 Logitech BRIO才能穿透进虚拟机。穿透后Windows 本机腾讯会议等将无法使用该摄像头。权限导致的“无头案”执行过sudo usermod -aG video dragon后必须重启虚拟机或注销重登用户组权限才会生效。否则脚本会报Permission denied。五、 调参修改指南灵活调整如果后续更换视频源或需要优化检测效果修改camera_motion.py顶部配置更换视频源假视频python camera_motion.py demo_video.mp4USB 摄像头python camera_motion.py 0修改 systemd 服务里的ExecStart也要改成0手机IP摄像头python camera_motion.py http://手机IP:8080/video网络摄像头(RTSP)python camera_motion.py rtsp://用户:密码IP:554/xxx灵敏度调节没人也乱触发 - 调大MIN_AREA_PCT如 1.0或调大NOISE_AREA_PCT如 0.5。人走过没触发 - 调小MIN_AREA_PCT如 0.2。状态来回闪烁 - 调大HOLD_SECONDS如 5 秒。性能优化虚拟机卡顿 - 调大IMG_INTERVAL如 2.0或调小WIDTH如 480和JPEG_QUALITY如 50。未来升级YOLO只需要替换camera_motion.py中“背景减除”那段算法为 YOLO 模型输出人数person_count。MQTT 主题和 HA 配置文件完全不用动。附录#!/usr/bin/env python3 camera_motion.py —— 摄像头运动检测 → MQTT → Home Assistant 用法: python camera_motion.py # 使用下面 SOURCE 的默认值 python camera_motion.py demo_video.mp4 # 视频文件(循环播放,用来模拟摄像头) python camera_motion.py 0 # USB 摄像头 /dev/video0 python camera_motion.py http://手机IP:8080/video # 手机 IP Webcam / 网络摄像头 python camera_motion.py rtsp://用户:密码摄像头IP:554/xxx 发布的 MQTT 主题 (BASE 默认 aiot/camera1): {BASE}/image JPEG 画面(带检测框) - HA 的 mqtt camera {BASE}/motion ON / OFF (retain) - HA 的 mqtt binary_sensor {BASE}/motion_area 运动面积占画面的百分比 - HA 的 mqtt sensor {BASE}/event JSON 事件(开始/结束) - 以后转发给 ThingsBoard / 存档 {BASE}/availability online / offline (遗嘱消息) importjsonimportsysimporttimeimportcv2importpaho.mqtt.clientasmqtt# 配置(按需修改) SOURCEdemo_video.mp4# 视频源,命令行参数优先MQTT_HOST127.0.0.1# MQTT broker 地址(和 HA 里配置的一致)MQTT_PORT1883MQTT_USERNone# broker 有账号密码就填,如 userMQTT_PASSNoneBASEaiot/camera1WIDTH640# 处理前把画面缩放到这个宽度(越小越省 CPU)MIN_AREA_PCT0.5# 运动面积占画面 % 超过它,才算有运动NOISE_AREA_PCT0.2# 单个色块小于画面的这个 %,当作噪点忽略HOLD_SECONDS3# 运动消失后,保持 ON 的时间(防抖,避免开关闪烁)WARMUP_FRAMES30# 开头几帧用来学习背景,不判断IMG_INTERVAL1.0# 每隔多少秒发布一张画面/面积(太快会拖慢 HA)JPEG_QUALITY70# defmake_client():try:# paho-mqtt 2.xclientmqtt.Client(mqtt.CallbackAPIVersion.VERSION2)exceptAttributeError:# paho-mqtt 1.xclientmqtt.Client()ifMQTT_USER:client.username_pw_set(MQTT_USER,MQTT_PASS)client.will_set(f{BASE}/availability,offline,retainTrue)client.connect(MQTT_HOST,MQTT_PORT,60)client.loop_start()returnclientdefopen_source(src):capcv2.VideoCapture(src)ifnotcap.isOpened():raiseRuntimeError(f打不开视频源:{src})returncapdefparse_source():srcsys.argv[1]iflen(sys.argv)1elseSOURCEifisinstance(src,str)andsrc.isdigit():srcint(src)# 0 - USB 摄像头编号returnsrcdefmain():srcparse_source()is_fileisinstance(src,str)andnotsrc.lower().startswith((rtsp://,http://,https://))clientmake_client()client.publish(f{BASE}/availability,online,retainTrue)client.publish(f{BASE}/motion,OFF,retainTrue)capopen_source(src)fpscap.get(cv2.CAP_PROP_FPS)or25bgcv2.createBackgroundSubtractorMOG2(history200,varThreshold32,detectShadowsFalse)kernelcv2.getStructuringElement(cv2.MORPH_ELLIPSE,(5,5))motion_onFalselast_motion_ts0.0last_img_ts0.0frame_count0print(f开始检测,视频源{src},MQTT{MQTT_HOST}:{MQTT_PORT},主题前缀{BASE})try:whileTrue:ok,framecap.read()ifnotok:ifis_file:# 视频文件播完,回到开头循环cap.set(cv2.CAP_PROP_POS_FRAMES,0)continueprint(读取画面失败,2 秒后重连...)cap.release()time.sleep(2)try:capopen_source(src)exceptRuntimeErrorase:print(e)continueh,wframe.shape[:2]framecv2.resize(frame,(WIDTH,int(h*WIDTH/w)))frame_areaframe.shape[0]*frame.shape[1]frame_count1learningframe_countWARMUP_FRAMES# 开头几帧只学习背景# 1) 背景减除:得到前景(运动)掩膜,再去噪、膨胀maskbg.apply(frame)maskcv2.morphologyEx(mask,cv2.MORPH_OPEN,kernel)maskcv2.dilate(mask,kernel,iterations2)# 2) 找轮廓,过滤小噪点,累计运动面积,画框total0.0contours,_cv2.findContours(mask,cv2.RETR_EXTERNAL,cv2.CHAIN_APPROX_SIMPLE)forcin([]iflearningelsecontours):areacv2.contourArea(c)ifareaframe_area*NOISE_AREA_PCT/100:continuetotalarea x,y,bw,bhcv2.boundingRect(c)cv2.rectangle(frame,(x,y),(xbw,ybh),(0,255,0),2)pctround(total*100/frame_area,2)nowtime.time()# 3) 状态机:开始 - 保持 - 结束(带防抖)ifnotlearning:ifpctMIN_AREA_PCT:last_motion_tsnowifnotmotion_on:motion_onTrueclient.publish(f{BASE}/motion,ON,retainTrue)client.publish(f{BASE}/event,json.dumps({event:motion_start,area_pct:pct,ts:int(now)}))print(f[{time.strftime(%H:%M:%S)}] 运动开始 area{pct}%)elifmotion_onandnow-last_motion_tsHOLD_SECONDS:motion_onFalseclient.publish(f{BASE}/motion,OFF,retainTrue)client.publish(f{BASE}/event,json.dumps({event:motion_end,ts:int(now)}))print(f[{time.strftime(%H:%M:%S)}] 运动结束)# 4) 定时发布画面和面积(画面上的文字用英文,OpenCV 不支持中文)ifnow-last_img_tsIMG_INTERVAL:labelMOTIONifmotion_onelseIDLEcolor(0,0,255)ifmotion_onelse(0,200,0)cv2.putText(frame,f{label}{pct}%,(10,28),cv2.FONT_HERSHEY_SIMPLEX,0.8,color,2)ok2,bufcv2.imencode(.jpg,frame,[cv2.IMWRITE_JPEG_QUALITY,JPEG_QUALITY])ifok2:client.publish(f{BASE}/image,buf.tobytes())client.publish(f{BASE}/motion_area,str(pct))last_img_tsnowifis_file:# 视频文件按原帧率播放,模拟真实摄像头time.sleep(1.0/fps)exceptKeyboardInterrupt:print(收到 CtrlC,退出)finally:client.publish(f{BASE}/availability,offline,retainTrue)client.loop_stop()client.disconnect()cap.release()if__name____main__:main()#!/usr/bin/env python3 make_demo_video.py —— 生成一段演示视频 demo_video.mp4(没有摄像头/没有素材时使用) 内容(20 秒,15 帧/秒,640x480): 0-5 秒 空场景 5-12 秒 一个人从左走到右 12-16 秒 空场景 16-20 秒 另一个人从右走到左 画面带轻微噪声,用来模拟真实摄像头。 importcv2importnumpyasnp W,H,FPS,SECONDS640,480,15,20rngnp.random.default_rng(0)# 背景:上下渐变的灰色,再画几条地面线bgnp.tile(np.linspace(90,140,H,dtypenp.uint8)[:,None],(1,W))bgcv2.cvtColor(bg,cv2.COLOR_GRAY2BGR)foryin(300,360,420):cv2.line(bg,(0,y),(W,y),(110,110,110),1)outcv2.VideoWriter(demo_video.mp4,cv2.VideoWriter_fourcc(*mp4v),FPS,(W,H))foriinrange(FPS*SECONDS):ti/FPS framebg.copy()if5t12:# 从左到右xint(-60(t-5)/7*(W120))elif16t20:# 从右到左xint(W60-(t-16)/4*(W120))else:xNoneifxisnotNone:cv2.rectangle(frame,(x,230),(x60,400),(40,60,200),-1)# 身体cv2.circle(frame,(x30,210),22,(60,120,220),-1)# 头noiserng.normal(0,3,frame.shape)# 轻微噪声framenp.clip(frame.astype(np.float32)noise,0,255).astype(np.uint8)out.write(frame)out.release()print(已生成 demo_video.mp4)HA 自动化的原理一句话就是如果…就…专业说法叫ECA 模型触发器Trigger→ 条件Condition可选→ 动作Action。概述HA 后台一直在监听所有实体的状态变化。每当某个实体的值变了温度从 25 跳到 36运动传感器从 OFF 变 ONHA 就把这次变化拿去跟你写的每一条自动化规则做比对触发器这次变化符不符合我要等的那个条件比如温度变化这个事件发生了。条件可选用来加限制现在还满足别的要求吗比如现在是不是白天。不满足就不往下走但不会重新等触发器只是这次不执行。动作都满足就去执行——开空调、发通知、关插座。它就是当Trigger某件事发生了并且Condition当前状态也符合就去做Action某个动作。关键点动作之后会形成闭环动作执行后会改变某个设备的状态比如插座被关掉这个状态改变本身又是一次新的变化可能触发别的自动化。这就是你之前理解的状态变化→事件→动作→新状态→新事件的循环是对的。举个项目里的例子触发器光照传感器数值 50 条件 现在是晚上 动作 打开房间灯总结光照一变暗如果又是晚上就开灯。白天光照也可能变暗比如阴天、遮挡但条件不满足就不会误开灯——条件的作用就是防止触发器误判。