接上文。这个项目最难的不是接 API,是在 EC600M 这种 RAM 只有几 MB、MicroPython 单 GIL 的环境里,同时跑:LCD 刷新、摄像头连拍、MQTT 通信、HTTP 上传、WebSocket 语音。任何一个阻塞都会连锁反应。
1. 四个线程各管一摊
主线程: drain 普通帧队列 + 每 5s 上报信号强度 + gc.collect()
LVGL 线程: UI 渲染 + 按键导航(lvgl_ui.run_lvgl_thread)
camera_flow 线程:manual/voice/cloud/interval 模式循环
定时帧上传线程:始终运行,队列空时 sleep 100ms 轻轮询
主线程和定时帧上传线程是双生产者消费者:
# 普通帧 → 主循环队列
_pending_photo = []
_pending_lock = _thread.allocate_lock()
# 定时帧 → 专属后台线程
_timed_queue = []
_timed_lock = _thread.allocate_lock()
def on_photo_done(photo_name, status):
"""camCapture 拍照完成回调——按文件名前缀分流"""
if photo_name.lower().startswith("timed_"):
with _timed_lock:
_timed_queue.append((photo_name, status))
else:
with _pending_lock:
_pending_photo.append((photo_name, status))
为什么要分两个队列?因为定时连拍的节奏是 5 秒一张,如果上传 HTTP 阻塞 10 秒,普通帧和定时帧挤在一个队列里,就会互相堵。定时帧走专属线程,上传再慢也不影响 camera_flow 的连拍 tick。
2. 退出 interval 模式必须 drain 队列
用户按 GPIO12 返回主菜单时,可能还有最后一张定时帧在队列里没传完。直接切模式会丢帧。所以:
def drain_timed_queue_for_interval_exit(timeout_ms=30000):
deadline = utime.ticks_add(utime.ticks_ms(), timeout_ms)
while True:
with _timed_lock:
remaining = len(_timed_queue)
if remaining == 0:
utime.sleep_ms(1500) # 多等一小段确认后台真处理完了
with _timed_lock:
if len(_timed_queue) == 0:
return True
if utime.ticks_diff(deadline, utime.ticks_ms()) <= 0:
return False
utime.sleep_ms(200)
注意那个 sleep_ms(1500):队列长度到 0 不代表后台线程真的把最后一张处理完了 —— 它可能刚 pop(0) 拿到帧,正在上传中。多等 1.5 秒再确认一次,这是实测出来的边界。
3. 阿里云 IoT:属性 vs 服务的选择
一开始我按教科书思路做了三个服务(TakePhoto / GetStatus / ClearOldPhotos),后来全删了,改用属性指令。原因:
-
服务调用是请求 - 响应模式,云端要等设备返回,超时逻辑麻烦;
-
属性写是单向推送,云端写 TakePhotoCommand=1,设备收到自己执行,执行完再通过事件上报结果;
-
设备端代码简单一个数量级,云侧物模型调试面板直接写个 1 就能触发拍照,不用搭服务调用 UI。
订阅主题只有一个:
/sys/{pk}/{dn}/thing/service/property/set
设备收到后判断 TakePhotoCommand==1 → 触发拍照 → 自动回写 0 → 上报 PhotoTaken 事件(带 photo_source 区分手动 / 云端 / 语音 / 定时四种来源)。
4. 语音 AI 链路:MCP tools 怎么把语音变成拍照
小智平台走 WebSocket:
按住 GPIO13 说话 → 录音上传 → 云端 STT → LLM
→ LLM 看到 tools 列表里有 camera.take_photo
→ 决定调用 → MCP call 消息下发
→ 设备端触发 cam.take_photo() → 入队列上传
→ 回执回云端 → TTS "已为你拍好"
MCP tools 定义在 protocol.py:
mcp_tools_list = [
{
"name": "camera.take_photo",
"description": "触发一次拍照",
"inputSchema": {"type": "object", "properties": {}} # ← 这个 type 必须有,坑篇讲
},
{
"name": "camera.set_lcd_backlight",
"description": "控制 LCD 背光 0/1",
"inputSchema": {
"type": "object",
"properties": {"level": {"type": "integer", "minimum": 0, "maximum": 1}}
}
}
]
设备端不做完整的 MCP Server,只做 MQTT/WebSocket 执行器 —— 这是方案文档里说的 “云端 MCP Bridge” 思路,把 JSON-RPC 的复杂度留给云端,板子只管干活。
5. Gitee 上传:400 冲突自动覆盖更新
Gitee Contents API 上传照片,如果文件已存在会返回 400(比如板子重启后编号回退到 photo_001)。处理流程:
status, body = self._post_form(api_url, payload)
if status == 400:
# 文件已存在,先 HEAD 拿 sha,再 PUT 覆盖
sha = self._get_sha(object_path)
if sha:
payload += "&sha={}&_method=PUT".format(sha)
status, body = self._post_form(api_url, payload)
多一次 HTTP 往返,但解决了重启撞名问题。配合 /usr/_timed_count.txt 持久化计数,编号跨重启不回退。
6. RAM 治理:每帧释放~70KB
这个是最关键的稳定性手段。上传一帧照片的内存足迹:
content (~16KB) 原始 JPEG
b64_data (~22KB) base64 编码
b64_str replace 去掉换行
b64_encoded URL-safe 替换
payload (~70KB) form-encoded 最终请求体
如果不主动释放,连拍十几次就 OOM。所以 photo_upload.py 的 finally 块:
finally:
try:
del content, b64_data, b64_str, b64_encoded, payload
except Exception:
pass
gc.collect()
主循环每轮也 gc.collect(),每 30 秒打一行 RAM free=N bytes 日志盯着趋势。微信语音 AI 推理期间 LCD 黑屏过一次,就是靠这个日志定位到 RAM 累积导致的。
下一篇全是踩坑 ——LCD 不初始化摄像头回调不触发、SPI 单例互斥、start_new_thread 返回 None、MCP schema 400、SIM 热插拔怎么让云侧秒判离线等等。