2c5bf950c5
## 新架构：Oracle 集中计算 + NAS 代理展示 ### Oracle 端 (fam-edge) - 新增 frame_service: ffmpeg 视频抽帧 + VLM 人物定位裁剪头像（磁盘缓存） - 新增 /api/oracle/frame: 按 video_id+ts 抽帧返回 jpeg（带 token） - 新增 /api/oracle/avatar: 按 label 生成人物头像（VLM 定位人物 + 兜底整帧居中） - 新增 person_identifier: 人物身份识别模块 - Gemini 适配器支持 flash/flash-lite 双模型切换，429 自动降级 - frame_service VLM 全模型 429 时进入 10 分钟熔断，避免每次请求白打配额 - 兜底头像不落缓存，配额恢复后自动重试 VLM 精确定位 ### 人物合并硬规则校验（框架级修复） - person_service: LLM 合并结果落库前加硬冲突检测 - 性别冲突 → 绝不合并 - 年龄档跨未成年/成年 → 绝不合并（防止把爷爷/宝宝并进同一人） - oracle_db: upsert_person 入口剥离括号后缀（人物A（别名：人物B） → 人物A），消灭垃圾人物行 - 修复 set_canonical 丢弃 source 参数的 bug（旧代码硬编码 'manual' 导致错误合并被永久固化） - get_events_for_label: 只提取该身份组的特征文本，头像定位更精准 ### NAS 端 (fam-core) - 新增 img_proxy: /api/proxy/frame 和 /api/proxy/avatar 代理 Oracle 图片 - app.py 注册 img_bp 蓝图 - oracle_sync / db_layer / member_manager 同步人物表 ### UI 端 (fam-ui) - 事件时间轴: 每条事件卡片加时间点缩略帧 - 人物管理: 每人卡片加头像（150x150 圆角） - parse_persons: 剥离括号备注，与 Oracle 归一化一致 - 新增 EventItem 组件、Timeline 页改造 - Chat / ServiceStatus 页相应调整 ### 数据库 - scripts/ddl.sql: 同步表结构更新 - Oracle people 表: features_json / display_uid / source 字段完善
75 lines
3.0 KiB
Python
75 lines
3.0 KiB
Python
import os
|
||
|
||
from fam_edge import frame_service
|
||
from fam_edge.frame_service import _bbox_to_pixels
|
||
|
||
|
||
def test_bbox_to_pixels_basic():
|
||
# [ymin,xmin,ymax,xmax] 0-1000 归一化 -> 像素 (x1,y1,x2,y2)
|
||
# 实测样本：Gemini 对 440x248 帧返回 [0, 690, 203, 725]，
|
||
# 对应画面右上角门厅处的一个人（今天用真实截图验证过）。
|
||
x1, y1, x2, y2 = _bbox_to_pixels([0, 690, 203, 725], 440, 248)
|
||
assert x1 == int(690 / 1000 * 440)
|
||
assert y1 == 0
|
||
assert x2 == int(725 / 1000 * 440)
|
||
assert y2 == int(203 / 1000 * 248)
|
||
|
||
|
||
def test_bbox_to_pixels_full_frame():
|
||
x1, y1, x2, y2 = _bbox_to_pixels([0, 0, 1000, 1000], 400, 300)
|
||
assert (x1, y1, x2, y2) == (0, 0, 400, 300)
|
||
|
||
|
||
def test_bbox_to_pixels_zero_area():
|
||
x1, y1, x2, y2 = _bbox_to_pixels([500, 500, 500, 500], 400, 300)
|
||
assert (x1, y1) == (x2, y2)
|
||
|
||
|
||
class _FakeRow(dict):
|
||
"""支持 row['key'] 访问的假 sqlite3.Row。"""
|
||
def __getitem__(self, k):
|
||
return dict.get(self, k)
|
||
|
||
|
||
class _FakeDb:
|
||
def __init__(self, local_path, event_start_time):
|
||
self._row = _FakeRow(local_path=local_path, event_start_time=event_start_time)
|
||
|
||
def get_video_by_id(self, video_id):
|
||
return self._row
|
||
|
||
|
||
def test_extract_frame_cache_key_includes_width(tmp_path, monkeypatch):
|
||
"""核心诉求: 同一 (video_id, ts) 不同调用方要不同分辨率（时间轴缩略图/头像/
|
||
人物识别裁人脸），缓存 key 不带 width 会导致后来的高分辨率请求悄悄拿到早先
|
||
缓存的低分辨率帧——这里验证两次不同 width 请求各自落到独立的缓存文件。"""
|
||
monkeypatch.setattr(frame_service, "CACHE_DIR", str(tmp_path))
|
||
video_path = tmp_path / "fake_video.mp4"
|
||
video_path.write_bytes(b"not a real video, ffmpeg call is mocked")
|
||
db = _FakeDb(str(video_path), "2026-08-22 10:00:00")
|
||
|
||
written_widths = []
|
||
|
||
def fake_run_ffmpeg(args, timeout=60):
|
||
# 把请求的 -vf scale=WIDTH:-2 记下来，往输出路径写点假数据模拟成功
|
||
out_path = args[-1]
|
||
vf = next((a for a in args if a.startswith('scale=')), '')
|
||
written_widths.append(vf)
|
||
with open(out_path, 'wb') as f:
|
||
f.write(b'\xff\xd8fakejpeg')
|
||
return True
|
||
|
||
monkeypatch.setattr(frame_service, "_run_ffmpeg", fake_run_ffmpeg)
|
||
|
||
data_small = frame_service.extract_frame(db, 42, "2026-08-22 10:00:05", width=400)
|
||
data_large = frame_service.extract_frame(db, 42, "2026-08-22 10:00:05", width=2880)
|
||
|
||
assert data_small is not None and data_large is not None
|
||
cache_files = sorted(os.listdir(tmp_path))
|
||
frame_caches = [f for f in cache_files if f.startswith('frame_42_5_')]
|
||
assert len(frame_caches) == 2, f"expected 2 distinct cache files, got {frame_caches}"
|
||
assert 'frame_42_5_400.jpg' in frame_caches
|
||
assert 'frame_42_5_2880.jpg' in frame_caches
|
||
# 两次都真的各自调用了 ffmpeg（第二次没有因为撞到第一次的缓存而被跳过）
|
||
assert len(written_widths) == 2
|