采集层隔离
每个摄像头独立重试 3 次,退避 2 / 5 / 10 秒;一个失败不影响另一个。取流服务先做健康检查,不在线就直接跳过,不去耗那 30 秒超时。
Home Vision Pipeline
家里两个室内摄像头加一个门口猫眼,接一个视觉模型,把连续画面压成一条可查询的状态序列: 几点睡的、醒了多久、身边有没有人、什么时候被抱出门。这篇不讲「我做了个什么」,只讲几个真正决定成败的取舍—— 两级过滤怎么把模型调用量压下去、为什么输出格式不用 JSON、状态机为什么必须忽略 unknown、 以及门口那路为什么非得是边沿触发。
摄像头本身早就解决了「看得见」。它没解决的是:我在公司,想知道的是 几点睡的、这一觉睡了多久、醒着的时候身边有没有人, 而它能给我的只有一条连续的像素流。中间缺的是一层把像素变成离散状态 + 时间戳的东西。
最直接的做法是每分钟把画面丢给视觉模型让它描述。这条路一算成本就断了—— 按后面第三节的计价模型,每分钟一次、一天按 15 小时算,光模型调用就是 900 次。 所以成本不是这个系统的优化项,是它的第一约束,整个架构是围着这条约束长出来的。
入口只有一个 scheduler.py,crontab 每分钟调一次。采集每分钟都跑,分析十分钟跑一次:
if hour < RUN_HOUR_START or hour >= RUN_HOUR_END:
return # 夜里不跑,省钱也省噪音
run_capture() # 每分钟:抓帧 + 存盘
if minute % 10 == 0:
run_analyze() # 每 10 分钟:判断 + 可能调用模型
用 crontab 而不是常驻进程,是因为这台机器还跑别的东西,而且进程崩了不需要我去管——
下一分钟系统会再拉起来一次。代价是所有跨周期的状态都得落盘,这就是
tracker_state.json 的由来。
一处遗留:
采集层其实已经算出了「这一帧和上一帧比变没变」,并写进了返回值和
last_any_change;但分析层没读它,而是自己重新算了一遍。两套判据服务的目标不同
(见下一节),不冲突,但确实是重复计算。调度层拿到 run_capture() 的返回值后也没有使用。
核心思路很朴素:调用模型之前,先用一个便宜几个数量级的方法判断值不值得调用。 这里用的是最土的帧差。
curr = Image.open(io.BytesIO(img_bytes)).convert("L").resize((160, 120))
prev = Image.open(last_path).convert("L").resize((160, 120))
diff = ImageChops.difference(curr, prev)
pixels = list(diff.convert("L").tobytes())
return sum(pixels) / len(pixels) # 0-255 的平均绝对差
两个细节值得说:先转灰度再缩到 160×120,不是为了快(虽然确实快), 而是缩放本身就是一次低通滤波,能把传感器噪点、压缩块效应、以及夜视模式下的雪花 自然抹掉;留下来的只有「大块区域变了」这种我真正关心的信号。阈值 8.0 就是在这个尺度上定的。
分析层做判断时,比的是这一批里最早那张和最晚那张:
for files in captures.values(): # 每个摄像头
first = load_gray(files[0])
last = load_gray(files[-1])
max_diff = max(max_diff, mean_abs_diff(first, last))
因为我要回答的问题是「这十分钟结束时,情况和开始时还一样吗」,而不是「中间有没有抖动」。 比相邻帧会被呼吸起伏、窗帘飘动、光照渐变这些持续的小变化不停触发;比首尾帧只对净变化敏感。
这个选择有盲区: 如果在窗口内离开又回到几乎相同的位置和姿势,首尾两帧会非常接近,这一轮就会被判定为「没变化」而跳过。 换句话说,帧差闸门过滤的是净位移,天然看不见往返。这是拿漏检换成本, 兜底靠的是下面那条强制分析。
significant_change = batch_diff > DIFF_THRESHOLD
force_check = minutes_since_last_call >= 30
if not significant_change and not force_check:
skip()
没有这条,系统会有一个很难发现的失效模式:画面静止和画面卡死,在帧差看来是一模一样的。 摄像头挂掉后持续输出同一帧,帧差恒等于 0,系统会永远跳过、永远安静,而且日志上看起来一切正常。 强制分析把「沉默」的上限钉死在 30 分钟,它既是漏检的兜底,也是故障的探测器。
代码里直接写了估算(按输入 $1.25/M、输出 $10/M token 计):
input_tokens = num_images * 258 + 800 # 每图 258,prompt 约 800
cost = (input_tokens * 1.25 + 50 * 10) / 1_000_000
# 10 张图: 3380 input → 约 $0.0047 / 次
真正决定账单的不是单次价格,而是跳过率。而跳过率由场景决定:婴儿一天里大部分时间在睡觉, 画面近乎静止——这个系统之所以便宜,是因为它监控的对象恰好大部分时间不动。 换成监控一条马路,同一套架构会立刻退化成每次都调用。
输出格式不是 JSON,是一行用竖线分隔的文本:
房间 | 活动描述 | 陪伴情况 | 环境光线
卧室 | 一直在婴儿床里睡觉 | 无人 | 关灯、夜视
客厅→卧室 | 前 5 分钟客厅玩耍,后被抱回卧室睡觉 | 妈妈 | 明亮
三个理由,按重要性排:
split("|") 不足 4 段就整条判为 unknown,不去猜、不去修。宁可少一条记录,也不要一条错的记录进状态机。最早的版本经常把大人、或者来串门的小孩认成目标。解决办法不是换更大的模型, 而是把物理上不可能的事直接写进提示词:
- 8 个月大,不会走路站立!只会躺、坐、爬、趴
- 站着走路的都不是他,是大人或其他小孩
- 婴儿床上被子有隆起 / 小鼓包 = 在被子里睡觉
- 彩色画面 = 开灯;黑白画面 = 关灯 / 夜视模式
这本质上是用领域先验去收窄模型的搜索空间。「会站立行走」是一条极强的排除条件, 写进去之后误识别明显下降。这部分没有捷径,就是看它出错、把错法补进 prompt、再看。
每次调用会把最近 6 条日志和当前状态一起送进去。不然模型每次都是从零开始看图, 没法给出「前 5 分钟在客厅,后来被抱回卧室」这种带转场的描述—— 而转场恰恰是状态机最需要的信息。
模型给的是自然语言,状态机要的是枚举值。中间这层是纯规则的关键词映射:
if any(w in desc for w in ["睡", "sleep", "休息", "闭眼"]):
status = "sleeping"
elif any(w in desc for w in ["玩", "爬", "坐", "play", "活动", "翻"]):
status = "alone_awake" if "无人" in companion else "playing"
elif any(w in desc for w in ["抱", "held", "怀里"]):
status = "held"
elif any(w in desc for w in ["吃", "奶", "eat", "喂", "餐椅"]):
status = "eating"
为什么不让模型直接输出枚举值?因为
alone_awake 是整个系统里唯一会真正吵醒人的状态。
它的判据我希望留在代码里,而不是留在模型输出里——出了问题可以直接读日志复现,
改判据也不用重新调 prompt。其它状态只是记录,错了影响不大;这一个错了会半夜把人吓醒,
或者更糟,该响的时候没响。
这层是目前最脆的一环。
关键词匹配撑不住自然语言的多样性:「在婴儿床里安静地待着」既不含「睡」也不含「玩」,
会直接落到 unknown。更合理的做法是让模型同时输出一个受限枚举和一段自然语言描述,
枚举进状态机、描述进日志,代价是 prompt 变长、输出变长。这个改动还没做。
单帧的「在睡觉」几乎没有价值,「已经连续睡了 182 分钟」才有。所以状态机存的不只是当前状态, 还有进入这个状态的时间戳,告警全部建立在时长之上。
这里有一个不起眼但很关键的设计:unknown 不写入状态。
if new_status == "unknown":
consecutive_unknown += 1 # 只计数
else:
consecutive_unknown = 0
if new_status != old_status and new_status != "unknown":
status = new_status # 只有确定的状态才改写
status_since = now # 计时从这一刻开始
如果 unknown 也覆盖状态并重置 status_since,那么「连续睡了 3 小时」这个判断
永远不可能成立——三小时里只要有一次看不清(夜视、遮挡、模型抽风),计时就清零了。
让不确定的观测不破坏已有结论,是这类系统里反复会遇到的模式。
连续看不清本身是另一个信号,单独计数、单独告警:
| 条件 | 级别 | 含义 |
|---|---|---|
alone_awake ≥ 5 分钟 | URGENT | 醒着且身边没人,需要立刻处理 |
sleeping ≥ 180 分钟 | WATCH | 睡得偏久,提醒看一眼 |
连续 unknown ≥ 2 次 | WATCH / ALERT | 大概率是摄像头或取流出了问题 |
转入 alone_awake | ALERT | 状态刚发生变化 |
注意最后一条和第一条的区别:刚醒和醒了五分钟还没人管是两件事, 前者提个醒,后者才升级。把「转换」和「持续」分成两类判据,是状态机相比逐帧判断真正多出来的能力。
最初门口猫眼和室内摄像头一样每轮截图,每次往批次里塞 2 张。两个问题: 每批从 10 张涨到 12 张,输入 token 多了约 15%,而门口画面在绝大多数时间里和目标毫无关系—— 它只在被抱出门和回来这两个瞬间有信息量。
于是改成由状态机的边沿触发:
visible = status in ("sleeping", "playing", "held", "eating", "alone_awake")
if was_visible and not visible: # 从室内画面消失
check_door("out") # 是被抱出门了吗?
elif not was_visible and visible: # 重新出现
check_door("in") # 是回来了吗?
检查本身也尽量便宜:不自己截图,而是拉设备自己存的移动侦测告警截图 (猫眼本来就有这个功能,有人经过会自动存图),最近 15 分钟最多取 3 张, 然后只问模型一个二值问题:
请判断画面中是否有婴儿车 / 推车 / 伞车。
只输出 YES 或 NO,不要多余文字。
二值问题的输出 token 几乎可以忽略,而且答错了也只影响一条事件记录,不会污染状态机。 同一事件 30 分钟内不重复通知,避免门口来回走动刷屏。
这个改动的本质是把一个「传感器」重新归类成了「探针」。 轮询适合持续有信息的源,边沿触发适合只在状态跳变时才有信息的源。 一开始把三个摄像头平等对待,是没想清楚这件事。
每个摄像头独立重试 3 次,退避 2 / 5 / 10 秒;一个失败不影响另一个。取流服务先做健康检查,不在线就直接跳过,不去耗那 30 秒超时。
重试 2 次,退避 5 / 15 秒。最终失败不是静默跳过,而是把 consecutive_unknown 加一——失败会自己走到告警链路上去。
群机器人 → 本地 webhook → 落盘 pending_alerts.txt。告警系统自己挂掉却无声无息,是这类系统最坏的失效方式。
每次采集写心跳文件供外部监控;截图只保留 30 分钟,超时自动清理,磁盘占用有上界。
一条贯穿的原则:失败要能被看见。 系统里绝大多数 bug 不是「算错了」,而是「悄悄什么都没做」。 强制分析、连续 unknown 计数、心跳文件、落盘兜底,针对的都是同一类问题。
unknown。alone_awake 之后那条「如果有大人陪同则降级为 playing」的兜底,因为上游分支的条件已经保证了陪伴字段含「无」,实际永远不会触发。留着无害,但它不是它看起来的那道保险。| 参数 | 值 | 作用 |
|---|---|---|
| 采集周期 | 1 分钟 | 每个室内摄像头各抓 1 帧 |
| 分析周期 | 10 分钟 | 取最近 12 分钟的帧做一批 |
| 帧差比较尺寸 | 160 × 120 灰度 | 缩放即低通滤波,压掉噪点 |
| 帧差阈值 | 8.0 | 0-255 平均绝对差,超过才调用模型 |
| 强制分析间隔 | 30 分钟 | 沉默上限,兼作卡死探测 |
| 单摄像头采样 | 5 张 | 均匀采样,最多 10 张 / 次 |
| 送分析前缩放 | 宽 800px | 质量 85 的 JPEG |
| 独自清醒告警 | 5 分钟 | URGENT |
| 长睡提醒 | 180 分钟 | WATCH |
| 事件去重窗口 | 30 分钟 | 同一出门 / 回家事件不重复推送 |
| 猫眼告警回溯 | 15 分钟 / 最多 3 张 | 仅在状态边沿触发时查 |
| 运行时段 | 07:00 - 22:00 | 时段外不采集也不分析 |
| 截图保留 | 30 分钟 | 超时自动清理 |
回头看,这个系统里真正起作用的不是模型,是模型外面那几层: 一道便宜的闸门决定要不要花钱,一个宽松的输出格式决定解析失败时会发生什么, 一条「不确定不写入」的规则决定时长统计能不能成立,一次从轮询到边沿触发的改写决定第三个摄像头值不值得接。 模型本身是可替换的——换一个更强的,上面这些取舍一条都不会变。
另外必须说清楚:它替代不了人看孩子,一秒都不能。 它解决的是我上班时的信息缺口,不是看护本身。做这类东西的时候,把边界说清楚比把指标做高更重要。