ericwyuan/sentinel-home-ai
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

[阶段2] FAM-Edge 重构为整视频分析+同步接口+人物服务 - 移除切片/抽帧/队列，新增 oracle_db/person_service/qa/watch_processor/video_processor，api_gateway 提供 /api/oracle/sync 与 /api/oracle/people/correct

Browse Source
This commit is contained in:
ericwyuan
2026-08-21 10:38:15 +08:00
parent ae1c13589f
commit 9b1cc8f93b
23 changed files with 1097 additions and 2335 deletions
Download Patch File Download Diff File Expand all files Collapse all files

101
fam-edge/config/config.yaml
View File

@@ -1,78 +1,58 @@
# FAM-Edge 配置文件 (Oracle 端) - 多模型池配置 # FAM-Edge 配置文件 (Oracle 端) - 新架构 v2
# Tailscale: Oracle=100.74.137.126, NAS=100.70.234.39
# #
# 异步队列模式: # 新架构（2026-08-21 重构）:
# NAS 上传视频 → /api/edge/video/enqueue 入 SQLite 队列 → 消费者线程异步处理 # 1. 不再切片/抽帧：整视频直传云端 VLM（Gemini 用 Files API，NVIDIA 用整视频 video_url）
# → NAS Poller 从 /api/edge/results 拉取结果 # 2. 视频来源：rclone 从 Google 硬盘实时同步到本地 local_dir，监听目录处理新视频
# 速率限制: Gemini 1000RPM x2 burst, NVIDIA 40RPM x2 burst # 3. Oracle 自建 SQLite 库存储所有视频摘要/事件/人物，并对外提供同步接口供 NAS 拉取
# 4. 独立 person_service 汇总全量人物 -> LLM 合并为规范人物表 -> 回灌视频提示
# 5. NAS 仅作管理后台，每 30 分钟从甲骨文拉增量镜像到本地 MariaDB
# NAS 端回调地址（旧 webhook 模式保留，异步模式不使用） # Oracle 端 HTTP 服务
nas:
webhook_url: "http://100.70.234.39:8000/api/core/callback/event"
media_base_url: "http://100.70.234.39:8000/media"
media_token: "sentinel-media-2026"
# Oracle 端服务
server: server:
host: "0.0.0.0" host: "0.0.0.0"
port: 5000 port: 5000
max_concurrent_tasks: 1 max_concurrent_tasks: 1
# 异步任务队列 # Google 硬盘同步（rclone 负责同步落地，本段仅描述监听行为）
queue: gdrive_sync:
db_path: "/opt/fam-edge/data/fam_queue.db" enabled: true
upload_dir: "/tmp/fam_uploads" local_dir: "/opt/fam-edge/gdrive_videos" # rclone 同步落地目录（video_processing 监听此目录）
poll_interval: 10 # 消费者轮询间隔（秒） watch_interval_sec: 30 # 监听新视频的轮询间隔
# API 速率限制 (RPM)，burst_factor=2 表示突发容量为 2 倍 RPM camera_name: "客厅" # 摄像头名称（注入视频提示）
rate_limit: # 文件名解析开始时间：监控文件名含时间戳时使用（如 2026-08-21_081500.mp4）
gemini_rpm: 1000 parse_start_from_filename: true
nvidia_rpm: 40
burst_factor: 2
# 编排调度模式: fallback(顺序降级, 默认) | ensemble(并行交叉验证) # Oracle 本地库（视频摘要/事件/人物）
orchestrator: oracle_db:
mode: "fallback" path: "/opt/fam-edge/data/oracle.db"
overall_timeout: 600
# 关键帧筛选参数（自适应：帧数随视频时长动态计算） # NAS 拉取同步接口鉴权 token（与 NAS oracle_sync.token 一致）
video: sync_api:
candidate_per_minute: 2 # 每分钟粗抽候选帧数 token: "${ORACLE_SYNC_TOKEN}"
candidate_min: 30 # 候选帧下限（短视频保底）
candidate_max: 120 # 候选帧上限（超长视频截断）
key_frame_interval_sec: 150 # 关键帧间隔（秒），每2.5分钟1张
min_key_frames: 5 # 关键帧下限（帧差不足时补足到此数）
max_key_frames_floor: 8 # 关键帧上限的下限（短视频保底）
max_key_frames_cap: 30 # 关键帧上限（超长视频截断）
mse_threshold: 500
jpeg_quality: 80
max_long_edge: 1024
# 超时（秒） # 人物识别服务
timeout: person_service:
download: 60 enabled: true
vlm_visual: 600 schedule_interval_sec: 1800 # 每 30 分钟重新汇总一次人物
vlm_fusion: 300 model: "gemini" # 用哪个模型做人物合并（vision 模型也支持纯文本）
callback: 30
overall: 1800
# 多模型池配置（新框架：本地大模型不参与视频分析，仅智能问答兜底） # 视频处理
# video_processing:
# 视频分析链路（推送模式）: max_concurrent: 1
# 云端 VLM 直接产出结构化 JSON (global_summary / entities_json / frame_details) timeout: 900 # 单视频分析超时（整视频上云较慢）
# -> Edge 仅做格式化/校验 (format_cloud_result) -> 直接回写 NAS，无本地融合步骤 # 降级顺序：先 gemini 整视频，失败再 nvidia 整视频；两者都失败 -> 标记 failed
# 视觉角色: Gemini(主) -> NVIDIA NIM(备) 顺序降级; 两云端全失败 -> 任务 FAILED 走重试 vision_order: ["gemini", "nvidia"]
#
# 智能问答链路: # 智能问答降级链（与视频分析独立）：Gemini -> NVIDIA -> 本地 Ollama
# Gemini -> NVIDIA -> 本地 Ollama (仅当两云端都失败才启用本地兜底)
models: models:
- provider: "gemini" - provider: "gemini"
role: "vision" role: "vision"
enabled: true enabled: true
model_name: "gemini-flash-latest" # 主模型（每日免费配额 20 请求，按模型独立） model_name: "gemini-flash-latest" # 主模型（每日免费配额 20 请求，按模型独立）
fallback_models: # 429 配额耗尽/503 过载时依次切换 fallback_models:
- "gemini-flash-lite-latest" - "gemini-flash-lite-latest"
api_key: "${GEMINI_API_KEY}" api_key: "${GEMINI_API_KEY}"
timeout: 90 timeout: 600
circuit_breaker: circuit_breaker:
enabled: true enabled: true
threshold: 5 threshold: 5
@@ -81,16 +61,17 @@ models:
- provider: "nvidia" - provider: "nvidia"
role: "vision" role: "vision"
enabled: true enabled: true
model_name: "nvidia/nemotron-3-nano-omni-30b-a3b-reasoning" # Omni 原生视频输入；llama-3.2-11b-vision 仅逐帧 # Nemotron Nano 12B v2 VL：NIM 官方支持整视频 video_url 输入（内部自行采样帧）
model_name: "nvidia/nemotron-nano-12b-v2-vl"
base_url: "https://integrate.api.nvidia.com/v1" base_url: "https://integrate.api.nvidia.com/v1"
api_key: "${NVIDIA_API_KEY}" api_key: "${NVIDIA_API_KEY}"
timeout: 120 timeout: 600
circuit_breaker: circuit_breaker:
enabled: true enabled: true
threshold: 5 threshold: 5
cooldown: 300 cooldown: 300
# 本地模型：纯文本 qwen2.5:7b，仅参与智能问答，作为 Gemini/NVIDIA 都失败时的兜底 # 本地模型：纯文本 qwen2.5:7b，仅参与智能问答兜底
- provider: "ollama" - provider: "ollama"
role: "text" role: "text"
usage: "qa_fallback" usage: "qa_fallback"

534
fam-edge/src/fam_edge/ai_orchestrator/orchestrator.py
View File

@@ -1,534 +0,0 @@
"""
AI-Orchestrator - 多模型编排
视频分析链路（新框架）:
1. 加载所有启用的模型适配器
2. 健康检查
3. 抽帧 + 关键帧筛选 + 压缩
4. 云端 VLM 视觉分析（Gemini 主 / NVIDIA 兜底），直出结构化 JSON
5. format_cloud_result：对云端结果做**格式化/校验**（无本地模型调用，不汇总摘要）
6. 同步返回 NAS → 落库
智能问答链路（新框架）:
- run_qa：Gemini → NVIDIA → 本地 Ollama（仅当两云端都失败才用本地兜底）
"""
import time
import json
import base64
import requests
from datetime import datetime, timedelta
from concurrent.futures import ThreadPoolExecutor, as_completed, TimeoutError as FuturesTimeout
from typing import Dict, List, Optional, Tuple
from ..logger import setup_logger, log_task
from ..config_loader import load_config
from ..model_adapters.adapter_factory import build_adapters
from ..model_adapters.base_adapter import BaseModelAdapter
from ..video_preprocessor.preprocessor import VideoPreprocessor
from .json_parser import VLMOutputInvalidError, validate_schema
logger = setup_logger('fam-edge.orchestrator')
class AIOrchestrator:
"""AI 编排器"""
def __init__(self):
self.config = load_config()
self.adapters: List[BaseModelAdapter] = build_adapters(self.config.get('models', []))
self.timeout_cfg = self.config.get('timeout', {})
def health_check_all(self) -> List[BaseModelAdapter]:
"""健康检查，返回健康的适配器列表"""
healthy = []
for adapter in self.adapters:
try:
if adapter.health_check():
healthy.append(adapter)
except Exception as e:
logger.error(f"适配器 {adapter.provider_name} 健康检查异常: {e}")
return healthy
def run_visual_analysis(self, adapters: List[BaseModelAdapter],
frame_paths: List[str],
frame_timestamps: List[str],
known_members_context: str,
rate_limiter=None,
video_path: str = None,
event_start_time: str = '') -> Dict[str, dict]:
"""视觉分析阶段：仅 role=vision 的适配器参与
支持 analyze_video 的适配器（如 NVIDIA Omni）优先走原生视频输入，
失败自动降级回逐帧图片模式。
orchestrator.mode:
- fallback (默认): 按 config 顺序依次尝试，首个成功即采用（单元素 dict）
- ensemble: 并行所有健康 vision 模型，全部成功结果都保留（交叉验证）
rate_limiter: 可选 RateLimiter 实例，按 provider 限速（2x burst）
"""
vision_adapters = [a for a in adapters if getattr(a, 'role', 'vision') == 'vision']
if not vision_adapters:
logger.error("没有 vision 角色的可用适配器")
return {}
mode = self.config.get('orchestrator', {}).get('mode', 'fallback')
if mode == 'ensemble':
return self._run_visual_ensemble(
vision_adapters, frame_paths, frame_timestamps,
known_members_context, rate_limiter)
# fallback: 顺序降级，首个成功即采用
model_outputs = {}
for adapter in vision_adapters:
if adapter.get_circuit_breaker().is_open():
logger.warning(f"[{adapter.provider_name}] 熔断器 OPEN，跳过")
continue
# 速率限制：按 provider 获取 token（2x burst）
if rate_limiter:
acquired = rate_limiter.acquire(adapter.provider_name, timeout=300)
if not acquired:
logger.warning(f"[{adapter.provider_name}] 速率限制超时，跳过")
continue
start = time.time()
try:
output = None
if video_path and hasattr(adapter, 'analyze_video'):
try:
logger.info(f"[{adapter.provider_name}] 尝试原生视频输入分析")
output = adapter.analyze_video(
video_path, frame_timestamps, known_members_context,
event_start_time=event_start_time)
if not output:
logger.warning(f"[{adapter.provider_name}] 视频模式失败，降级逐帧模式")
except Exception as ve:
logger.warning(f"[{adapter.provider_name}] 视频模式异常: {ve}，降级逐帧模式")
output = None
if not output:
output = adapter.analyze_frames(
frame_paths, frame_timestamps, known_members_context)
duration_ms = int((time.time() - start) * 1000)
if output:
adapter.get_circuit_breaker().record_success()
log_task(logger, 0, f'model_{adapter.provider_name}',
f'视觉分析成功', duration_ms=duration_ms)
model_outputs[adapter.provider_name] = output
logger.info(f"fallback 采用 [{adapter.provider_name}]，停止降级")
break
else:
adapter.get_circuit_breaker().record_failure()
logger.warning(f"[{adapter.provider_name}] 视觉分析返回空，降级下一模型")
except Exception as e:
logger.error(f"[{adapter.provider_name}] 视觉分析异常: {e}")
adapter.get_circuit_breaker().record_failure()
return model_outputs
def _run_visual_ensemble(self, vision_adapters, frame_paths,
frame_timestamps, known_members_context,
rate_limiter=None) -> Dict[str, dict]:
"""并行调用所有健康 vision 模型，保留全部成功结果（交叉验证）"""
model_outputs = {}
max_timeout = max((a.get_timeout() for a in vision_adapters), default=240)
with ThreadPoolExecutor(max_workers=len(vision_adapters)) as pool:
futures = {}
for adapter in vision_adapters:
if adapter.get_circuit_breaker().is_open():
logger.warning(f"[{adapter.provider_name}] 熔断器 OPEN，跳过")
continue
# 速率限制：按 provider 获取 token（2x burst）
if rate_limiter:
acquired = rate_limiter.acquire(adapter.provider_name, timeout=300)
if not acquired:
logger.warning(f"[{adapter.provider_name}] 速率限制超时，跳过")
continue
future = pool.submit(
adapter.analyze_frames,
frame_paths, frame_timestamps, known_members_context)
futures[future] = adapter.provider_name
for future in as_completed(futures, timeout=max_timeout + 10):
provider = futures[future]
start = time.time()
try:
adapter = next(a for a in vision_adapters if a.provider_name == provider)
output = future.result(timeout=adapter.get_timeout())
duration_ms = int((time.time() - start) * 1000)
if output:
model_outputs[provider] = output
adapter.get_circuit_breaker().record_success()
log_task(logger, 0, f'model_{provider}',
f'视觉分析成功，输出长度={len(output)}', duration_ms=duration_ms)
else:
adapter.get_circuit_breaker().record_failure()
logger.warning(f"[{provider}] 视觉分析返回空")
except FuturesTimeout:
logger.warning(f"[{provider}] 视觉分析超时")
adapter = next(a for a in vision_adapters if a.provider_name == provider)
adapter.get_circuit_breaker().record_failure()
except Exception as e:
logger.error(f"[{provider}] 视觉分析异常: {e}")
adapter = next(a for a in vision_adapters if a.provider_name == provider)
adapter.get_circuit_breaker().record_failure()
return model_outputs
@staticmethod
def _attach_frame_images(frame_details: List[dict], frame_paths: List[str]) -> None:
"""把关键帧图片 base64 附加到 frame_details（按位置对齐视觉分析输入帧）
附带人脸红框标记与 face_count（NAS 落盘 meta.json，UI 据此挑有人像的头像）
"""
from ..frame_marker import mark_jpeg
for i, fd in enumerate(frame_details):
if i >= len(frame_paths):
break
try:
with open(frame_paths[i], 'rb') as f:
raw = f.read()
marked, faces = mark_jpeg(raw)
fd['frame_image'] = base64.b64encode(marked).decode('ascii')
fd['face_count'] = faces
except OSError as e:
logger.warning(f"关键帧图片读取失败: {frame_paths[i]}: {e}")
def format_cloud_result(self, provider: str, raw_result: dict,
known_members_context: str = '',
task_id: int = 0) -> dict:
"""格式化云端 VLM 直出的结构化结果（**无本地模型调用**）。
- 云端模型已产出结构化数据（frame_details / 可选 global_summary / entities_json）
- 本方法仅做：字段归一化、source_providers 与 compute_provider 填充、
entities 推导、global_summary 缺失时格式化生成
- 解析/校验失败抛 VLMOutputInvalidError
"""
if not isinstance(raw_result, dict):
raise VLMOutputInvalidError("云端视觉模型未返回结构化数据(dict)")
data = dict(raw_result)
frame_details = data.get('frame_details')
if not isinstance(frame_details, list) or not frame_details:
raise VLMOutputInvalidError("云端结果缺少非空的 frame_details")
# 归一化每条 frame_detail
normalized = []
for f in frame_details:
if not isinstance(f, dict):
continue
sp = f.get('source_providers')
if not isinstance(sp, list) or not sp:
sp = [provider]
normalized.append({
"frame_index": int(f.get("frame_index", len(normalized) + 1)),
"frame_timestamp": str(f.get("frame_timestamp", "")),
"person": str(f.get("person", "无人")),
"action": str(f.get("action", "")),
"clothing": str(f.get("clothing", "")),
"is_attention_event": bool(f.get("is_attention_event", False)),
"source_providers": [str(p) for p in sp],
})
if not normalized:
raise VLMOutputInvalidError("frame_details 解析后为空")
data['frame_details'] = normalized
# compute_provider：本次实际成功的云端模型
data['compute_provider'] = [provider]
# entities_json：缺失时由 frame_details 推导（按人物去重）
if not data.get('entities_json'):
seen = set()
ents = []
for f in normalized:
p = f['person']
if p and p != '无人' and p not in seen:
seen.add(p)
ents.append({
"person": p,
"action": f['action'],
"clothing": f['clothing'],
})
data['entities_json'] = ents
# global_summary：云端未给则格式化生成（非 LLM 汇总，仅拼接事实）
if not data.get('global_summary'):
data['global_summary'] = self._build_summary_from_frames(normalized)
return validate_schema(data)
def _build_summary_from_frames(self, frame_details: List[dict]) -> str:
"""当云端模型未提供 global_summary 时，由 frame_details 格式化生成摘要。
注意：这是确定性事实拼接，非 LLM 二次汇总。"""
persons = {}
has_attention = False
for f in frame_details:
p = f['person']
if p and p != '无人':
persons.setdefault(p, set()).add(f['action'])
if f.get('is_attention_event'):
has_attention = True
if not persons:
summary = "整个时段内画面中未检测到人物出现，主要为环境静态画面。"
else:
parts = []
for p, acts in persons.items():
acts_desc = "、".join(sorted(a for a in acts if a)) or "无明显动作"
parts.append(f"{p}（{acts_desc}）")
summary = f"时段内检测到：{'；'.join(parts)}。"
if has_attention:
summary += " ⚠️ 存在需关注的异常事件。"
return summary
def run_qa(self, prompt: str, max_tokens: int = 512) -> Tuple[Optional[str], Optional[str]]:
"""智能问答编排：Gemini → NVIDIA → 本地 Ollama（仅当两云端都失败才用本地兜底）。
返回 (answer, provider)；全部失败返回 (None, None)。
"""
qa_order = ['gemini', 'nvidia', 'ollama']
for name in qa_order:
adapter = next((a for a in self.adapters if a.provider_name == name), None)
if adapter is None:
logger.warning(f"[qa] 未配置模型 {name}，跳过")
continue
try:
if adapter.get_circuit_breaker().is_open():
logger.warning(f"[qa] {name} 熔断器 OPEN，跳过")
continue
answer = adapter.chat(prompt, max_tokens=max_tokens)
if answer:
logger.info(f"[qa] 由 {name} 回答（长度={len(answer)}）")
return answer, name
logger.warning(f"[qa] {name} 返回空")
except Exception as e:
logger.error(f"[qa] {name} 调用异常: {e}")
return None, None
def send_callback(self, webhook_url: str, task_id: int,
result: dict, camera_name: str = '',
event_start_time: str = '', event_end_time: str = ''):
"""回调 NAS"""
payload = {
"task_id": task_id,
"status": "success",
"event_start_time": event_start_time,
"event_end_time": event_end_time,
"camera_name": camera_name,
"global_summary": result.get('global_summary', ''),
"entities_json": result.get('entities_json', []),
"frame_details": result.get('frame_details', []),
"compute_provider": result.get('compute_provider', []),
"error_message": None
}
callback_timeout = self.timeout_cfg.get('callback', 30)
max_retries = 3
for attempt in range(max_retries):
try:
resp = requests.post(webhook_url, json=payload, timeout=callback_timeout)
if resp.status_code == 200:
log_task(logger, task_id, 'callback', '回调成功')
return
else:
logger.warning(f"[task_id={task_id}] 回调返回 {resp.status_code}，重试 {attempt+1}/{max_retries}")
except Exception as e:
logger.warning(f"[task_id={task_id}] 回调异常: {e}，重试 {attempt+1}/{max_retries}")
raise Exception(f"回调失败，已重试 {max_retries} 次")
def send_failure_callback(self, webhook_url: str, task_id: int,
failure_stage: str, error_message: str):
"""发送失败回调"""
payload = {
"task_id": task_id,
"status": "failed",
"failure_stage": failure_stage,
"error_message": error_message
}
try:
requests.post(webhook_url, json=payload, timeout=30)
except Exception as e:
logger.error(f"[task_id={task_id}] 失败回调也失败: {e}")
def process_task(self, task_data: dict):
"""端到端处理任务（拉取模式，webhook 回调）"""
task_id = task_data.get('task_id')
video_url = task_data.get('video_url')
webhook_url = task_data.get('webhook_url')
known_members = task_data.get('known_members_context', '')
logger.info(f"[task_id={task_id}] ====== 开始处理任务 ======")
start_time = time.time()
# 1. 健康检查
healthy_adapters = self.health_check_all()
if not healthy_adapters:
logger.error(f"[task_id={task_id}] 所有模型不健康，返回 503")
self.send_failure_callback(webhook_url, task_id, 'vlm_visual', 'All models unhealthy')
return 503
# 2. 下载 + 抽帧
preprocessor = VideoPreprocessor(task_id)
try:
# 下载
video_path = preprocessor.download_video(video_url)
# 抽帧
candidate_frames = preprocessor.extract_candidate_frames(video_path)
if not candidate_frames:
raise Exception("抽帧失败，无候选帧")
# 关键帧筛选
key_frames = preprocessor.select_key_frames(candidate_frames)
# 压缩
compressed_frames = preprocessor.compress_frames(key_frames)
if not compressed_frames:
raise Exception("压缩后无可用帧")
# 计算时间戳
event_start_time = task_data.get('event_start_time', '')
frame_timestamps = preprocessor.compute_timestamps(
video_path, len(compressed_frames), event_start_time
)
# 3. 并行视觉分析
model_outputs = self.run_visual_analysis(
healthy_adapters, compressed_frames, frame_timestamps,
known_members, video_path=video_path,
event_start_time=event_start_time
)
if not model_outputs:
raise Exception('All models failed in visual analysis')
# 4. 云端直出结果格式化（无本地融合）
provider = next(iter(model_outputs))
fusion_result = self.format_cloud_result(
provider, model_outputs[provider], known_members, task_id)
# 5. 回调
# 从视频文件名推断 camera_name
camera_name = task_data.get('camera_name', '')
event_end_time = task_data.get('event_end_time', '')
self.send_callback(
webhook_url, task_id, fusion_result,
camera_name=camera_name,
event_start_time=event_start_time,
event_end_time=event_end_time
)
total_ms = int((time.time() - start_time) * 1000)
log_task(logger, task_id, 'overall', f'任务完成', duration_ms=total_ms)
except VLMOutputInvalidError as e:
logger.error(f"[task_id={task_id}] 云端结果格式化失败: {e}")
self.send_failure_callback(webhook_url, task_id, 'vlm_fusion', str(e))
except Exception as e:
logger.error(f"[task_id={task_id}] 任务处理失败: {e}", exc_info=True)
self.send_failure_callback(webhook_url, task_id, 'download', str(e))
finally:
# 6. 清理
if 'preprocessor' in locals():
preprocessor.cleanup()
return 200
def process_push_task(self, task_data: dict, video_path: str,
preprocessor: 'VideoPreprocessor',
rate_limiter=None) -> dict:
"""推送模式：同步处理上传的视频，结果直接返回（无 webhook 回调）
rate_limiter: 可选 RateLimiter 实例，按 provider 限速（2x burst）
返回 payload 结构与原 webhook 回调一致:
- 成功: {task_id, status: "success", event_start_time, ..., frame_details, ...}
- 失败: {task_id, status: "failed", failure_stage, error_message}
"""
task_id = task_data.get('task_id')
known_members = task_data.get('known_members_context', '')
event_start_time = task_data.get('event_start_time', '')
logger.info(f"[task_id={task_id}] ====== 开始处理推送任务 ======")
start_time = time.time()
try:
# 1. 健康检查
healthy_adapters = self.health_check_all()
if not healthy_adapters:
logger.error(f"[task_id={task_id}] 所有模型不健康")
return {
"task_id": task_id, "status": "failed",
"failure_stage": "vlm_visual",
"error_message": "All models unhealthy"
}
# 2. 抽帧（视频已由调用方保存到本地，无需下载）
candidate_frames = preprocessor.extract_candidate_frames(video_path)
if not candidate_frames:
raise Exception("抽帧失败，无候选帧")
key_frames = preprocessor.select_key_frames(candidate_frames)
compressed_frames = preprocessor.compress_frames(key_frames)
if not compressed_frames:
raise Exception("压缩后无可用帧")
frame_timestamps = preprocessor.compute_timestamps(
video_path, len(compressed_frames), event_start_time
)
# event_end_time 未提供时，用 start + 视频时长推算（DB 列 NOT NULL）
event_end_time = task_data.get('event_end_time', '')
if not event_end_time and event_start_time and preprocessor.video_duration > 0:
try:
start_dt = datetime.strptime(event_start_time, '%Y-%m-%d %H:%M:%S')
event_end_time = (
start_dt + timedelta(seconds=int(preprocessor.video_duration))
).strftime('%Y-%m-%d %H:%M:%S')
except ValueError:
pass
# 3. 并行视觉分析
model_outputs = self.run_visual_analysis(
healthy_adapters, compressed_frames, frame_timestamps,
known_members, rate_limiter, video_path=video_path,
event_start_time=event_start_time
)
if not model_outputs:
raise Exception('All models failed in visual analysis')
# 4. 云端直出结果格式化（无本地融合）
provider = next(iter(model_outputs))
fusion_result = self.format_cloud_result(
provider, model_outputs[provider], known_members, task_id)
# 5. 附加关键帧图片（NAS 落盘后供 UI 时间轴展示）
frame_details = fusion_result.get('frame_details', [])
self._attach_frame_images(frame_details, compressed_frames)
total_ms = int((time.time() - start_time) * 1000)
log_task(logger, task_id, 'overall', '推送任务完成', duration_ms=total_ms)
return {
"task_id": task_id,
"status": "success",
"event_start_time": event_start_time,
"event_end_time": event_end_time,
"camera_name": task_data.get('camera_name', ''),
"global_summary": fusion_result.get('global_summary', ''),
"entities_json": fusion_result.get('entities_json', []),
"frame_details": frame_details,
"compute_provider": fusion_result.get('compute_provider', []),
"error_message": None
}
except VLMOutputInvalidError as e:
logger.error(f"[task_id={task_id}] VLM 输出解析失败: {e}")
return {
"task_id": task_id, "status": "failed",
"failure_stage": "vlm_fusion", "error_message": str(e)
}
except Exception as e:
logger.error(f"[task_id={task_id}] 推送任务处理失败: {e}", exc_info=True)
return {
"task_id": task_id, "status": "failed",
"failure_stage": "process", "error_message": str(e)
}

572
fam-edge/src/fam_edge/api_gateway/api_gateway.py
View File

@@ -1,517 +1,103 @@
""" """
API-Gateway - Flask 蓝图，接收任务 API-Gateway - Flask 蓝图（新架构 v2）
模式: 端点:
1. enqueue (异步): NAS 上传视频 → Edge 入队 → 立即返回 → 消费者异步处理 → NAS 轮询拉取结果 GET /api/oracle/sync NAS 每 30 分钟拉取增量（since + token 校验）
2. push (同步, 兼容保留): NAS 上传 → Edge 同步处理 → 结果随响应返回 POST /api/oracle/people/correct NAS 推送手动命名校正（label -> canonical_name）
3. analyze (旧拉取模式, 兼容保留) POST /api/edge/chat/ask 智能问答编排（Gemini -> NVIDIA -> Ollama）
GET /health 健康检查
已移除（旧推送/分块/队列模式）: /video/push, /enqueue, /chunk, /assemble,
/results, /queue/stats, /mark_frames
""" """
import os import os
import base64
import threading
import requests
from flask import Blueprint, request, jsonify from flask import Blueprint, request, jsonify
from ..logger import setup_logger from ..logger import setup_logger
from ..ai_orchestrator.orchestrator import AIOrchestrator from .. import state
from ..video_preprocessor.preprocessor import VideoPreprocessor from ..qa import QAOrchestrator
from ..queue import queue_manager
logger = setup_logger('fam-edge.api_gateway') logger = setup_logger('fam-edge.api_gateway')
api_bp = Blueprint('api_gateway', __name__) api_bp = Blueprint('api_gateway', __name__)
_current_task_lock = threading.Lock() _qa = None
_currently_processing = False
_orchestrator = None
def get_orchestrator(): def get_qa():
global _orchestrator global _qa
if _orchestrator is None: if _qa is None:
_orchestrator = AIOrchestrator() _qa = QAOrchestrator()
return _orchestrator return _qa
@api_bp.route('/api/edge/video/analyze', methods=['POST']) def _check_token() -> bool:
def receive_task(): expected = _sync_token()
"""接收分析任务""" token = request.args.get('token') or request.form.get('token') or \
global _currently_processing (request.get_json(silent=True) or {}).get('token', '')
return bool(expected) and token == expected
_SYNC_TOK = None
def _sync_token():
global _SYNC_TOK
if _SYNC_TOK is None:
from ..config_loader import load_config
_SYNC_TOK = load_config().get('sync_api', {}).get('token', '${ORACLE_SYNC_TOKEN}')
if _SYNC_TOK.startswith('${') and _SYNC_TOK.endswith('}'):
_SYNC_TOK = os.environ.get(_SYNC_TOK[2:-1], '')
return _SYNC_TOK or ''
@api_bp.route('/api/oracle/sync', methods=['GET'])
def sync_pull():
"""NAS 拉取增量数据。since=ISO 时间字符串（默认 '' 拉全量）。
返回: {videos:[...], events:[...], people:[...], server_time}
"""
if not _check_token():
return jsonify({"error": "unauthorized"}), 401
since = request.args.get('since', '')
try:
delta = state.get_db().get_sync_delta(since)
except Exception as e:
logger.error(f"sync_pull 异常: {e}")
return jsonify({"error": str(e)}), 500
return jsonify(delta), 200
@api_bp.route('/api/oracle/people/correct', methods=['POST'])
def people_correct():
"""NAS 手动命名校正推送。
请求: {"label": "人物A", "canonical_name": "张三", "token": "..."}
更新 people 表（manual 优先，不被 LLM 覆盖），立即重算 known_members_context。
"""
if not _check_token():
return jsonify({"error": "unauthorized"}), 401
data = request.get_json(silent=True) data = request.get_json(silent=True)
if not data: if not data:
return jsonify({"error": "Invalid JSON"}), 400 return jsonify({"error": "Invalid JSON"}), 400
label = (data.get('label') or '').strip()
task_id = data.get('task_id') canonical = (data.get('canonical_name') or '').strip()
video_url = data.get('video_url') if not label or not canonical:
webhook_url = data.get('webhook_url') return jsonify({"error": "缺少 label / canonical_name"}), 400
if not task_id or not video_url or not webhook_url:
return jsonify({"error": "缺少必填字段: task_id, video_url, webhook_url"}), 400
logger.info(f"[task_id={task_id}] 收到任务: {video_url}")
# 并发控制
with _current_task_lock:
if _currently_processing:
logger.warning(f"[task_id={task_id}] 队列已满 (当前有任务处理中)，返回 429")
return jsonify({"error": "Queue full", "retry_after": 60}), 429
_currently_processing = True
# 异步处理
def _process():
global _currently_processing
try:
orch = get_orchestrator()
orch.process_task(data)
except Exception as e:
logger.error(f"[task_id={task_id}] 处理异常: {e}", exc_info=True)
finally:
with _current_task_lock:
_currently_processing = False
thread = threading.Thread(target=_process, daemon=True, name=f'task-{task_id}')
thread.start()
return jsonify({"status": "accepted", "task_id": task_id}), 202
@api_bp.route('/api/edge/video/enqueue', methods=['POST'])
def enqueue_task():
"""异步模式：接收 multipart 视频上传，入队后立即返回
NAS 上传视频 → Edge 保存到磁盘 + 入 SQLite 队列 → 返回 task_id
消费者线程异步处理，NAS 通过 /api/edge/results 拉取结果
"""
task_id_raw = request.form.get('task_id')
file = request.files.get('video')
if not task_id_raw or not file:
return jsonify({"error": "缺少必填字段: task_id, video"}), 400
try: try:
task_id = int(task_id_raw) state.get_db().set_canonical(label, canonical, source='manual')
except ValueError:
return jsonify({"error": "task_id 必须是整数"}), 400
camera_name = request.form.get('camera_name', '')
event_start_time = request.form.get('event_start_time', '')
known_members_context = request.form.get('known_members_context', '')
upload_dir = os.environ.get('FAM_UPLOAD_DIR', '/tmp/fam_uploads')
os.makedirs(upload_dir, exist_ok=True)
video_filename = f"task_{task_id}_{file.filename}"
video_path = os.path.join(upload_dir, video_filename)
try:
file.save(video_path)
size_mb = os.path.getsize(video_path) / 1024 / 1024
logger.info(f"[task_id={task_id}] 入队: {file.filename} ({size_mb:.1f}MB)")
queue_id = queue_manager.enqueue(
nas_task_id=task_id,
video_filename=file.filename,
video_path=video_path,
camera_name=camera_name,
event_start_time=event_start_time,
known_members_context=known_members_context,
)
return jsonify({
"status": "queued",
"task_id": task_id,
"queue_id": queue_id,
}), 202
except Exception as e: except Exception as e:
logger.error(f"[task_id={task_id}] 入队失败: {e}", exc_info=True) logger.error(f"people_correct 异常: {e}")
if os.path.exists(video_path):
os.remove(video_path)
return jsonify({"error": str(e)}), 500 return jsonify({"error": str(e)}), 500
return jsonify({"status": "ok", "label": label, "canonical_name": canonical}), 200
# ========== 分块上传（断点续传）==========
CHUNK_SIZE = 20 * 1024 * 1024 # 20MB per chunk
def _chunk_dir(task_id: int) -> str:
upload_dir = os.environ.get('FAM_UPLOAD_DIR', '/tmp/fam_uploads')
d = os.path.join(upload_dir, f"task_{task_id}")
os.makedirs(d, exist_ok=True)
return d
@api_bp.route('/api/edge/video/chunk', methods=['POST'])
def upload_chunk():
"""接收单个分块，保存到 task_{id}/chunk_{index:04d}
断点续传：同一 task_id + chunk_index 重复上传会覆盖，
NAS 端可通过 /chunks 查询已上传分块，跳过已有的。
"""
task_id_raw = request.form.get('task_id')
chunk_index_raw = request.form.get('chunk_index')
total_chunks_raw = request.form.get('total_chunks')
filename = request.form.get('filename', 'video.mp4')
chunk_file = request.files.get('chunk')
if not task_id_raw or not chunk_index_raw or not chunk_file:
return jsonify({"error": "缺少必填字段: task_id, chunk_index, chunk"}), 400
try:
task_id = int(task_id_raw)
chunk_index = int(chunk_index_raw)
total_chunks = int(total_chunks_raw) if total_chunks_raw else 0
except ValueError:
return jsonify({"error": "task_id/chunk_index 必须是整数"}), 400
d = _chunk_dir(task_id)
chunk_path = os.path.join(d, f"chunk_{chunk_index:04d}")
try:
# 检查 total_chunks 是否变化（chunk_size 变更导致），自动清理旧分块并更新元数据
import json
meta_path = os.path.join(d, "meta.json")
if total_chunks and os.path.exists(meta_path):
try:
with open(meta_path) as mf:
old_meta = json.load(mf)
if old_meta.get('total_chunks') and old_meta['total_chunks'] != total_chunks:
logger.warning(f"[task_id={task_id}] total_chunks 变更 "
f"({old_meta['total_chunks']}→{total_chunks})，清理旧分块")
for fn in os.listdir(d):
if fn.startswith('chunk_'):
os.remove(os.path.join(d, fn))
with open(meta_path, 'w') as f:
json.dump({"filename": filename, "total_chunks": total_chunks}, f)
except (json.JSONDecodeError, IOError):
pass
chunk_file.save(chunk_path)
size_kb = os.path.getsize(chunk_path) / 1024
# 写元数据（首次上传时）
if not os.path.exists(meta_path):
meta = {"filename": filename, "total_chunks": total_chunks}
with open(meta_path, 'w') as f:
json.dump(meta, f)
# 统计已上传分块
uploaded = sorted([
int(fn.split('_')[1]) for fn in os.listdir(d)
if fn.startswith('chunk_') and len(fn.split('_')) == 2
])
logger.info(f"[task_id={task_id}] 分块 {chunk_index}/{total_chunks} 上传成功 "
f"({size_kb:.0f}KB, 已上传 {len(uploaded)}/{total_chunks})")
return jsonify({
"status": "ok",
"task_id": task_id,
"chunk_index": chunk_index,
"uploaded_count": len(uploaded),
"total_chunks": total_chunks,
}), 200
except Exception as e:
logger.error(f"[task_id={task_id}] 分块上传失败: {e}", exc_info=True)
return jsonify({"error": str(e)}), 500
@api_bp.route('/api/edge/video/chunks', methods=['GET'])
def query_chunks():
"""查询已上传分块列表（断点续传：NAS 重启后查询跳过已有分块）"""
task_id_raw = request.args.get('task_id')
if not task_id_raw:
return jsonify({"error": "缺少 task_id"}), 400
try:
task_id = int(task_id_raw)
except ValueError:
return jsonify({"error": "task_id 必须是整数"}), 400
d = _chunk_dir(task_id)
uploaded = sorted([
int(fn.split('_')[1]) for fn in os.listdir(d)
if fn.startswith('chunk_') and len(fn.split('_')) == 2
]) if os.path.isdir(d) else []
total = 0
meta_path = os.path.join(d, "meta.json")
if os.path.exists(meta_path):
import json
try:
with open(meta_path) as f:
total = json.load(f).get('total_chunks', 0)
except (json.JSONDecodeError, IOError):
pass
return jsonify({
"task_id": task_id,
"uploaded_chunks": uploaded,
"uploaded_count": len(uploaded),
"total_chunks": total,
}), 200
@api_bp.route('/api/edge/video/assemble', methods=['POST'])
def assemble_chunks():
"""合并所有分块为完整视频文件，入 SQLite 队列
NAS 上传完全部分块后调用此端点触发合并 + 入队。
"""
task_id_raw = request.form.get('task_id')
if not task_id_raw:
return jsonify({"error": "缺少 task_id"}), 400
try:
task_id = int(task_id_raw)
except ValueError:
return jsonify({"error": "task_id 必须是整数"}), 400
camera_name = request.form.get('camera_name', '')
event_start_time = request.form.get('event_start_time', '')
known_members_context = request.form.get('known_members_context', '')
d = _chunk_dir(task_id)
# 读取元数据
import json
meta_path = os.path.join(d, "meta.json")
if not os.path.exists(meta_path):
return jsonify({"error": "元数据不存在，请先上传分块"}), 400
try:
with open(meta_path) as f:
meta = json.load(f)
except json.JSONDecodeError:
return jsonify({"error": "元数据损坏"}), 500
filename = meta.get('filename', 'video.mp4')
total_chunks = meta.get('total_chunks', 0)
# 检查分块完整性
chunk_files = sorted([
fn for fn in os.listdir(d)
if fn.startswith('chunk_') and len(fn.split('_')) == 2
])
if total_chunks and len(chunk_files) < total_chunks:
missing = total_chunks - len(chunk_files)
return jsonify({
"error": f"分块不完整: {len(chunk_files)}/{total_chunks}，缺 {missing} 块",
"uploaded_count": len(chunk_files),
"total_chunks": total_chunks,
}), 400
# 合并分块
upload_dir = os.environ.get('FAM_UPLOAD_DIR', '/tmp/fam_uploads')
video_filename = f"task_{task_id}_{filename}"
video_path = os.path.join(upload_dir, video_filename)
try:
with open(video_path, 'wb') as out:
for cf in chunk_files:
chunk_path = os.path.join(d, cf)
with open(chunk_path, 'rb') as chunk_f:
out.write(chunk_f.read())
size_mb = os.path.getsize(video_path) / 1024 / 1024
logger.info(f"[task_id={task_id}] 分块合并完成: {filename} ({size_mb:.1f}MB, {len(chunk_files)} 块)")
# 清理分块目录
import shutil
shutil.rmtree(d, ignore_errors=True)
# 入队
queue_id = queue_manager.enqueue(
nas_task_id=task_id,
video_filename=filename,
video_path=video_path,
camera_name=camera_name,
event_start_time=event_start_time,
known_members_context=known_members_context,
)
return jsonify({
"status": "queued",
"task_id": task_id,
"queue_id": queue_id,
"size_mb": round(size_mb, 1),
}), 202
except Exception as e:
logger.error(f"[task_id={task_id}] 合并失败: {e}", exc_info=True)
return jsonify({"error": str(e)}), 500
@api_bp.route('/api/edge/results', methods=['GET'])
def get_results():
"""返回已完成但未拉取的结果，标记为已交付"""
limit = int(request.args.get('limit', 10))
results = queue_manager.get_undelivered_results(limit=limit)
import json
payload = []
task_ids = []
for r in results:
try:
result_json = json.loads(r['result_json']) if r['result_json'] else None
except json.JSONDecodeError:
result_json = None
if r['status'] == 'FAILED':
result_json = {
"status": "failed",
"error_message": r['error_message'] or 'unknown',
"failure_stage": r['failure_stage'] or '',
}
payload.append({
"nas_task_id": r['nas_task_id'],
"result": result_json,
})
task_ids.append(r['id'])
if task_ids:
queue_manager.mark_delivered(task_ids)
return jsonify({"results": payload, "count": len(payload)}), 200
@api_bp.route('/api/edge/queue/stats', methods=['GET'])
def queue_stats():
"""队列状态统计"""
stats = queue_manager.get_queue_stats()
return jsonify(stats), 200
@api_bp.route('/api/edge/video/push', methods=['POST'])
def receive_push_task():
"""推送模式：接收 multipart 视频上传，同步分析，结果随 HTTP 响应返回
NAS 无法被 Oracle 反向访问（Tailscale 不通），因此改为 NAS 主动上传视频，
Edge 用 OpenCV 场景变化检测抽帧后分析，摘要直接放在响应里带回。
"""
global _currently_processing
task_id_raw = request.form.get('task_id')
file = request.files.get('video')
if not task_id_raw or not file:
return jsonify({"error": "缺少必填字段: task_id, video"}), 400
try:
task_id = int(task_id_raw)
except ValueError:
return jsonify({"error": "task_id 必须是整数"}), 400
logger.info(f"[task_id={task_id}] 收到推送任务: {file.filename}")
# 并发控制（同步处理，占用整个请求周期）
with _current_task_lock:
if _currently_processing:
logger.warning(f"[task_id={task_id}] 已有任务处理中，返回 429")
return jsonify({"error": "Queue full", "retry_after": 60}), 429
_currently_processing = True
preprocessor = None
try:
preprocessor = VideoPreprocessor(task_id)
video_path = preprocessor.save_upload(file)
task_data = {
"task_id": task_id,
"camera_name": request.form.get('camera_name', ''),
"event_start_time": request.form.get('event_start_time', ''),
"event_end_time": request.form.get('event_end_time', ''),
"known_members_context": request.form.get('known_members_context', ''),
}
result = get_orchestrator().process_push_task(task_data, video_path, preprocessor)
return jsonify(result), 200
except Exception as e:
logger.error(f"[task_id={task_id}] 推送任务异常: {e}", exc_info=True)
return jsonify({
"task_id": task_id, "status": "failed",
"failure_stage": "upload", "error_message": str(e)
}), 200
finally:
if preprocessor is not None:
preprocessor.cleanup()
with _current_task_lock:
_currently_processing = False
@api_bp.route('/api/edge/mark_frames', methods=['POST'])
def mark_frames():
"""NAS 存量关键帧批量补红框（检测计算在 Edge，NAS 只存图）"""
data = request.get_json(silent=True)
if not data:
return jsonify({"error": "Invalid JSON"}), 400
images = data.get('images')
if not isinstance(images, list) or not images or len(images) > 12:
return jsonify({"error": "images 需要 1-12 项 [{key, data}]"}), 400
from ..frame_marker import mark_jpeg
results = []
for item in images:
key = item.get('key', '')
b64 = item.get('data', '')
try:
marked, faces = mark_jpeg(base64.b64decode(b64))
results.append({
"key": key,
"data": base64.b64encode(marked).decode('ascii'),
"faces": faces
})
except Exception as e:
logger.warning(f"补标失败 {key}: {e}")
results.append({"key": key, "data": None, "faces": 0, "error": str(e)})
return jsonify({"results": results}), 200
@api_bp.route('/health', methods=['GET'])
def health():
"""健康检查"""
global _currently_processing
orch = get_orchestrator()
healthy = orch.health_check_all()
if not healthy:
return jsonify({
"status": "unavailable",
"healthy_models": [],
"processing": _currently_processing
}), 503
return jsonify({
"status": "ok",
"healthy_models": [a.provider_name for a in healthy],
"processing": _currently_processing
}), 200
@api_bp.route('/api/edge/chat', methods=['POST'])
def chat_proxy():
"""代理转发至本地 Ollama /api/generate（兼容旧调用，Ollama 未对外暴露）"""
data = request.get_json(silent=True)
if not data:
return jsonify({"error": "Invalid JSON"}), 400
try:
resp = requests.post(
'http://127.0.0.1:11434/api/generate',
json=data,
timeout=data.get('options', {}).get('timeout', 120)
)
return jsonify(resp.json()), resp.status_code
except requests.RequestException as e:
logger.error(f"Chat proxy error: {e}")
return jsonify({"error": f"Ollama unreachable: {e}"}), 502
@api_bp.route('/api/edge/chat/ask', methods=['POST']) @api_bp.route('/api/edge/chat/ask', methods=['POST'])
def chat_ask(): def chat_ask():
"""智能问答编排：Gemini → NVIDIA → 本地 Ollama（两云端都失败才用本地兜底） """智能问答编排：Gemini → NVIDIA → 本地 Ollama（两云端都失败才用本地兜底）
请求: {"prompt": "..."} 请求: {"prompt": "...", "max_tokens": 512}
响应: {"answer": "...", "provider": "gemini"|"nvidia"|"ollama"} 响应: {"answer": "...", "provider": "gemini"|"nvidia"|"ollama"}
""" """
data = request.get_json(silent=True) data = request.get_json(silent=True)
@@ -519,12 +105,24 @@ def chat_ask():
return jsonify({"error": "缺少必填字段: prompt"}), 400 return jsonify({"error": "缺少必填字段: prompt"}), 400
prompt = data['prompt'] prompt = data['prompt']
max_tokens = int(data.get('max_tokens', 512)) max_tokens = int(data.get('max_tokens', 1024))
answer, provider = get_orchestrator().run_qa(prompt, max_tokens=max_tokens) answer, provider = get_qa().run_qa(prompt, max_tokens=max_tokens)
if answer is None: if answer is None:
return jsonify({ return jsonify({
"error": "所有模型均不可用（Gemini / NVIDIA / Ollama 全部失败）" "error": "所有模型均不可用（Gemini / NVIDIA / Ollama 全部失败）"
}), 503 }), 503
return jsonify({"answer": answer, "provider": provider}), 200 return jsonify({"answer": answer, "provider": provider}), 200
@api_bp.route('/health', methods=['GET'])
def health():
"""健康检查"""
try:
db = state.get_db()
vids = db._conn.execute(
"SELECT COUNT(*) c FROM videos WHERE status='done'").fetchone()['c']
return jsonify({"status": "ok", "processed_videos": vids}), 200
except Exception as e:
return jsonify({"status": "error", "error": str(e)}), 500

32
fam-edge/src/fam_edge/app.py
View File

@@ -1,7 +1,10 @@
""" """
FAM-Edge 主应用 - Flask 单进程 FAM-Edge 主应用 - Flask 单进程（新架构 v2）
承载: API-Gateway / Video-Preprocessor / AI-Orchestrator / Storage-Cleaner / Queue-Consumer 承载:
- API-Gateway（同步拉取 / 命名校正 / 智能问答）
- WatchProcessor（监听 Google 硬盘同步落地目录，整视频分析）
- PersonService（人物汇总合并，定时）
""" """
import os import os
import sys import sys
@@ -12,7 +15,9 @@ sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
from .config_loader import load_config from .config_loader import load_config
from .logger import setup_logger from .logger import setup_logger
from .api_gateway.api_gateway import api_bp from .api_gateway.api_gateway import api_bp
from .queue.consumer import get_consumer from . import state
from .watch_processor import WatchProcessor
from .person_service import PersonService
logger = setup_logger('fam-edge.app') logger = setup_logger('fam-edge.app')
@@ -22,17 +27,24 @@ app.register_blueprint(api_bp)
@app.route('/', methods=['GET']) @app.route('/', methods=['GET'])
def root(): def root():
return jsonify({"service": "fam-edge", "version": "2.0"}), 200 return jsonify({"service": "fam-edge", "version": "2.0",
"mode": "drive-sync + whole-video analysis"}), 200
# 启动消费者线程（异步任务队列） # 启动监听处理器 + 人物服务
_consumer = None _watch = None
_person = None
try: try:
_consumer = get_consumer() db = state.get_db()
_consumer.start() _watch = WatchProcessor(db)
logger.info("Queue-Consumer 已启动") _watch.start()
logger.info("WatchProcessor 已启动")
_person = PersonService(db)
_person.start()
logger.info("PersonService 已启动")
except Exception as e: except Exception as e:
logger.error(f"Queue-Consumer 启动失败: {e}") logger.error(f"后台服务启动失败: {e}", exc_info=True)
if __name__ == '__main__': if __name__ == '__main__':

71
fam-edge/src/fam_edge/frame_marker.py
View File

@@ -1,71 +0,0 @@
"""
Frame-Marker - 关键帧人脸红框标记
Edge 端统一做检测计算（NAS ARM 太弱），NAS 只存图零计算:
- orchestrator 分析后、回传前: 画红框 + 统计人脸数
- /api/edge/mark_frames: NAS 存量帧批量补标
"""
import os
import threading
import cv2
import numpy as np
from .logger import setup_logger
from .config_loader import load_config
logger = setup_logger('fam-edge.frame_marker')
_lock = threading.Lock()
_detector = None
DEFAULT_MODEL = '/opt/fam-edge/models/yunet.onnx'
def _get_detector():
global _detector
if _detector is not None:
return _detector
with _lock:
if _detector is not None:
return _detector
path = load_config().get('frame_marker', {}).get('model_path', DEFAULT_MODEL)
if not os.path.isfile(path):
logger.warning(f"YuNet 模型不存在，跳过红框标记: {path}")
return None
det = cv2.FaceDetectorYN_create(path, '', (320, 320), score_threshold=0.6)
_detector = det
logger.info(f"YuNet 人脸检测器就绪: {path}")
return det
def mark_jpeg(jpeg_bytes: bytes):
"""在 JPEG 帧图上画人脸红框
返回 (标记后的 JPEG bytes, 人脸数)。检测失败/无模型时原样返回。
"""
det = _get_detector()
if det is None:
return jpeg_bytes, 0
img = cv2.imdecode(np.frombuffer(jpeg_bytes, np.uint8), cv2.IMREAD_COLOR)
if img is None:
return jpeg_bytes, 0
h, w = img.shape[:2]
with _lock:
det.setInputSize((w, h))
_, faces = det.detect(img)
if faces is None or len(faces) == 0:
return jpeg_bytes, 0
for f in faces:
x, y, fw, fh = int(f[0]), int(f[1]), int(f[2]), int(f[3])
# 人脸框外扩 40%，远处小脸也能看清
pad_w, pad_h = int(fw * 0.4), int(fh * 0.4)
x1 = max(0, x - pad_w)
y1 = max(0, y - pad_h)
x2 = min(w, x + fw + pad_w)
y2 = min(h, y + fh + pad_h)
cv2.rectangle(img, (x1, y1), (x2, y2), (0, 0, 255), 2)
ok, buf = cv2.imencode('.jpg', img, [cv2.IMWRITE_JPEG_QUALITY, 85])
if not ok:
return jpeg_bytes, 0
return buf.tobytes(), len(faces)

39
fam-edge/src/fam_edge/model_adapters/base_adapter.py
View File

@@ -3,13 +3,21 @@
新增模型只需继承此类并实现方法: 新增模型只需继承此类并实现方法:
1. health_check() -> bool 1. health_check() -> bool
2. analyze_frames(frame_paths, frame_timestamps, known_members_context) -> Optional[dict] 2. analyze_video(video_path, known_members_context, event_start_time) -> Optional[dict]
- 视觉分析：输入帧图片路径 + 时间戳 + 成员清单，直接输出**结构化结果 dict** - 整视频分析：直接把完整视频交给云端 VLM（本地不切片、不抽帧）。
(含 frame_details 等，详见 format_cloud_result 约定)。 - 模型内部自行采样帧，输出结构化结果 dict。失败/超时返回 None。
- 失败/超时返回 None。 - 返回约定:
{
"global_summary": str, # 整段视频摘要
"events": [ # 有用时间点 + 画面信息
{"timestamp": "2026-08-21 08:15:30", # 绝对北京时间（event_start_time 推算）
"description": str,
"people": [str],
"is_attention_event": bool}, ...],
"people_mentioned": [str], # 本视频出现的人物标识/真名
}
3. chat(prompt) -> Optional[str] 3. chat(prompt) -> Optional[str]
- 纯文本问答（智能问答场景），返回文本或 None。 - 纯文本问答（智能问答场景），返回文本或 None。
- 默认实现抛 NotImplementedError；文本/视觉模型按需实现。
4. get_timeout() -> int 4. get_timeout() -> int
5. get_circuit_breaker() -> CircuitBreaker 5. get_circuit_breaker() -> CircuitBreaker
""" """
@@ -36,22 +44,13 @@ class BaseModelAdapter(ABC):
pass pass
@abstractmethod @abstractmethod
def analyze_frames(self, frame_paths: List[str], def analyze_video(self, video_path: str,
frame_timestamps: List[str], known_members_context: str,
known_members_context: str) -> Optional[Dict]: event_start_time: str = '') -> Optional[Dict]:
"""视觉分析：输入帧图片路径 + 时间戳 + 成员清单， """整视频分析：把完整视频交给云端 VLM，输出结构化结果 dict。
直接输出结构化结果 dict（含 frame_details 等）。失败/超时返回 None。
约定返回结构（云端模型直出，Edge 仅做格式化校验，不再本地融合）: 本地不切片、不抽帧；模型内部自行采样帧。
{ 失败/超时返回 None。
"global_summary": "整个时段整体摘要(可选，缺失时由 Edge 格式化生成)",
"entities_json": [{"person","action","clothing"}] (可选，缺失时由 frame_details 推导),
"frame_details": [
{"frame_index":int, "frame_timestamp":str, "person":str,
"action":str, "clothing":str, "is_attention_event":bool,
"source_providers":[provider]}
]
}
""" """
pass pass

260
fam-edge/src/fam_edge/model_adapters/gemini_adapter.py
View File

@@ -2,15 +2,16 @@
GeminiAdapter - Google Gemini 云端 VLM 适配器 GeminiAdapter - Google Gemini 云端 VLM 适配器
provider_name = "gemini" provider_name = "gemini"
模型: gemini-flash-latest (v1beta 下 gemini-1.5-flash 会 404，用 flash-latest 别名) 模型: gemini-flash-latest
角色: vision (视觉分析直出结构化 JSON) + 智能问答 角色: vision (整视频直出结构化 JSON) + 智能问答
健康检查: GET /v1beta/models?key=... 健康检查: GET /v1beta/models?key=...
熔断器: 启用 熔断器: 启用
视觉分析: 多图单请求直出结构化 JSON（global_summary/entities_json/frame_details） 整视频分析: 用 Files API 上传完整视频 -> generateContent 直出结构化 JSON
（本地不切片、不抽帧；Gemini 原生支持长视频）
""" """
import os import os
import time import time
import base64 import json
import requests import requests
from typing import Dict, List, Optional from typing import Dict, List, Optional
@@ -23,16 +24,15 @@ logger = setup_logger('fam-edge.gemini_adapter')
class GeminiAdapter(BaseModelAdapter): class GeminiAdapter(BaseModelAdapter):
"""Gemini 云端 VLM 适配器 (视觉直出结构化 JSON + 文本问答)""" """Gemini 云端 VLM 适配器 (整视频直出结构化 JSON + 文本问答)"""
def __init__(self, config: dict): def __init__(self, config: dict):
super().__init__("gemini", config) super().__init__("gemini", config)
self.model_name = config.get('model_name', 'gemini-flash-latest') self.model_name = config.get('model_name', 'gemini-flash-latest')
# 免费层配额按模型独立（20 请求/天/模型），fallback 链用于跨模型借用配额
self.model_chain = [self.model_name] + [ self.model_chain = [self.model_name] + [
m for m in config.get('fallback_models', []) if m and m != self.model_name] m for m in config.get('fallback_models', []) if m and m != self.model_name]
self.api_key = self._resolve_key(config.get('api_key', '')) self.api_key = self._resolve_key(config.get('api_key', ''))
self.timeout = config.get('timeout', 30) self.timeout = config.get('timeout', 600)
cb_cfg = config.get('circuit_breaker', {}) cb_cfg = config.get('circuit_breaker', {})
self._cb = CircuitBreaker( self._cb = CircuitBreaker(
threshold=cb_cfg.get('threshold', 3), threshold=cb_cfg.get('threshold', 3),
@@ -69,76 +69,131 @@ class GeminiAdapter(BaseModelAdapter):
return False return False
# ------------------------------------------------------------------ # ------------------------------------------------------------------
# 视觉分析：多图单请求，直出结构化 JSON # 整视频分析：Files API 上传 -> generateContent
# ------------------------------------------------------------------ # ------------------------------------------------------------------
def analyze_frames(self, frame_paths: List[str], def analyze_video(self, video_path: str,
frame_timestamps: List[str], known_members_context: str,
known_members_context: str) -> Optional[Dict]: event_start_time: str = '') -> Optional[Dict]:
if self._cb.is_open(): if self._cb.is_open():
logger.warning("Gemini 熔断器 OPEN，跳过调用") logger.warning("Gemini 熔断器 OPEN，跳过视频分析")
return None return None
if not self.api_key: if not self.api_key:
logger.warning("Gemini API Key 未配置，跳过调用") logger.warning("Gemini API Key 未配置，跳过视频分析")
return None return None
if not frame_paths: if not os.path.isfile(video_path):
logger.warning("Gemini 无帧可分析") logger.warning(f"Gemini 视频文件不存在: {video_path}")
return None return None
parts = [] file_uri = self._upload_file(video_path)
ts_map = {} if not file_uri:
for i, (path, ts) in enumerate(zip(frame_paths, frame_timestamps), 1): self._cb.record_failure()
try:
with open(path, 'rb') as f:
img = base64.b64encode(f.read()).decode('utf-8')
except Exception as e:
logger.error(f"读取图片失败 {path}: {e}")
continue
parts.append({"inline_data": {"mime_type": "image/jpeg", "data": img}})
parts.append({"text": f"[图片{i}] 时间: {ts}"})
ts_map[i] = ts
if not parts:
return None return None
parts.insert(0, {"text": self._build_structured_prompt(known_members_context)}) prompt = self._build_video_prompt(known_members_context, event_start_time)
try: try:
text = self._generate(parts, max_tokens=2048, temperature=0.2) text = self._generate_video(file_uri, prompt, max_tokens=4096, temperature=0.2)
if text is None: if text is None:
self._cb.record_failure() self._cb.record_failure()
return None return None
try: try:
result = parse_vlm_json(text) result = parse_vlm_json(text)
# 确保 frame_details 的 frame_timestamp 与标注一致 result = self._normalize(result)
for f in result.get('frame_details', []): if not result or 'events' not in result:
idx = f.get('frame_index') logger.error(f"Gemini 视频输出缺少 events: {text[:150]}")
if isinstance(idx, int) and idx in ts_map and not f.get('frame_timestamp'): self._cb.record_failure()
f['frame_timestamp'] = ts_map[idx] return None
for f in result.get('frame_details', []): result['compute_provider'] = 'gemini'
if 'source_providers' not in f or not f.get('source_providers'):
f['source_providers'] = ['gemini']
self._cb.record_success() self._cb.record_success()
logger.info(f"Gemini 视觉分析完成，frame_details={len(result.get('frame_details', []))}") logger.info(f"Gemini 整视频分析完成，events={len(result.get('events', []))}")
return result return result
except VLMOutputInvalidError as e: except VLMOutputInvalidError as e:
logger.error(f"Gemini 输出无法解析为 JSON: {e}") logger.error(f"Gemini 视频输出无法解析为 JSON: {e}")
self._cb.record_failure() self._cb.record_failure()
return None return None
except requests.Timeout: except requests.Timeout:
logger.warning(f"Gemini 视觉分析超时 ({self.timeout}s)") logger.warning(f"Gemini 视频分析超时 ({self.timeout}s)")
self._cb.record_failure() self._cb.record_failure()
return None
except Exception as e: except Exception as e:
logger.error(f"Gemini 视觉分析异常: {e}") logger.error(f"Gemini 视频分析异常: {e}")
self._cb.record_failure() self._cb.record_failure()
return None
finally:
self._delete_file(file_uri)
def _upload_file(self, video_path: str) -> Optional[str]:
"""用 Files API 上传完整视频，返回可引用 URI。"""
name = os.path.basename(video_path)
upload_url = f"{self._base_url}/files?key={self.api_key}"
try:
with open(video_path, 'rb') as f:
data = f.read()
except OSError as e:
logger.error(f"读取视频失败 {video_path}: {e}")
return None
headers = {
"X-Goog-Upload-Protocol": "raw",
"X-Goog-Upload-File-Name": name,
"Content-Type": "video/mp4",
}
try:
resp = requests.post(upload_url, headers=headers, data=data, timeout=300)
except requests.Timeout:
logger.warning("Gemini 文件上传超时 (300s)")
return None
except Exception as e:
logger.error(f"Gemini 文件上传异常: {e}")
return None
if resp.status_code not in (200, 201):
logger.warning(f"Gemini 文件上传失败 HTTP {resp.status_code}: {resp.text[:200]}")
return None
try:
info = resp.json().get('file', {})
uri = info.get('uri')
file_name = info.get('name')
state = info.get('state')
except (ValueError, KeyError):
logger.warning("Gemini 文件上传响应解析失败")
return None
if not uri:
return None
# 等待 ACTIVE（大文件可能还在处理）
if state != 'ACTIVE' and file_name:
uri = self._wait_active(file_name)
return uri
def _wait_active(self, file_name: str, max_wait: int = 120) -> Optional[str]:
url = f"{self._base_url}/{file_name}?key={self.api_key}"
deadline = time.time() + max_wait
while time.time() < deadline:
try:
r = requests.get(url, timeout=15)
if r.status_code == 200:
j = r.json()
if j.get('state') == 'ACTIVE':
return j.get('uri')
except Exception:
pass
time.sleep(5)
logger.warning(f"Gemini 文件 {file_name} 未在 {max_wait}s 内 ACTIVE")
return None return None
def _generate(self, parts: List[dict], max_tokens: int, def _delete_file(self, file_uri: str):
temperature: float) -> Optional[str]: if not file_uri or 'files/' not in file_uri:
"""带模型 fallback 链的 generateContent 调用 return
name = file_uri.split('files/', 1)[-1]
try:
requests.delete(f"{self._base_url}/files/{name}?key={self.api_key}", timeout=15)
except Exception:
pass
- 429（每日免费配额耗尽，按模型独立）→ 立即换下一个模型，不重试 def _generate_video(self, file_uri: str, prompt: str,
- 503（模型过载，临时性）→ 同模型退避 3s 重试一次，仍失败换下一个 max_tokens: int, temperature: float) -> Optional[str]:
""" """带模型 fallback 链的 generateContent（视频文件引用）调用。"""
parts = [
{"file_data": {"mime_type": "video/mp4", "file_uri": file_uri}},
{"text": prompt},
]
for model in self.model_chain: for model in self.model_chain:
for attempt in range(2): for attempt in range(2):
try: try:
@@ -146,14 +201,15 @@ class GeminiAdapter(BaseModelAdapter):
f"{self._base_url}/models/{model}:generateContent?key={self.api_key}", f"{self._base_url}/models/{model}:generateContent?key={self.api_key}",
json={"contents": [{"parts": parts}], json={"contents": [{"parts": parts}],
"generationConfig": { "generationConfig": {
"temperature": temperature, "maxOutputTokens": max_tokens}}, "temperature": temperature,
"maxOutputTokens": max_tokens}},
timeout=self.timeout timeout=self.timeout
) )
except requests.Timeout: except requests.Timeout:
logger.warning(f"Gemini [{model}] 请求超时 ({self.timeout}s)") logger.warning(f"Gemini [{model}] 视频请求超时 ({self.timeout}s)")
break break
except Exception as e: except Exception as e:
logger.error(f"Gemini [{model}] 请求异常: {e}") logger.error(f"Gemini [{model}] 视频请求异常: {e}")
break break
if resp.status_code == 200: if resp.status_code == 200:
@@ -164,55 +220,76 @@ class GeminiAdapter(BaseModelAdapter):
).strip() if cands else '' ).strip() if cands else ''
if text: if text:
if model != self.model_name: if model != self.model_name:
logger.info(f"Gemini 主模型不可用，由 fallback 模型 [{model}] 出结果") logger.info(f"Gemini 主模型不可用，由 fallback [{model}] 出结果")
return text return text
logger.warning(f"Gemini [{model}] 返回空文本") logger.warning(f"Gemini [{model}] 返回空文本")
continue continue
detail = resp.text[:150].replace('\n', ' ') detail = resp.text[:150].replace('\n', ' ')
if resp.status_code == 429: if resp.status_code == 429:
logger.warning(f"Gemini [{model}] 429 每日免费配额耗尽，切换下一模型") logger.warning(f"Gemini [{model}] 429 配额耗尽，切换下一模型")
break break
if resp.status_code == 503: if resp.status_code == 503:
if attempt == 0: if attempt == 0:
logger.warning(f"Gemini [{model}] 503 过载，3s 后重试") logger.warning(f"Gemini [{model}] 503 过载，3s 后重试")
time.sleep(3) time.sleep(3)
continue continue
logger.warning(f"Gemini [{model}] 503 重试仍失败，切换下一模型")
break break
logger.warning(f"Gemini [{model}] HTTP {resp.status_code}: {detail}") logger.warning(f"Gemini [{model}] HTTP {resp.status_code}: {detail}")
break break
return None return None
def _build_structured_prompt(self, known_members: str) -> str: @staticmethod
return f"""你是家庭监控视频分析助手。下面按时间顺序排列了多张监控截图。 def _normalize(result: dict) -> dict:
请分析整个时段，只输出合法 JSON（不要 markdown、不要任何解释文字），结构如下： """统一字段名：frame_details -> events（兼容旧结构）。"""
events = result.get('events')
if events is None and 'frame_details' in result:
events = []
for f in result['frame_details']:
events.append({
"timestamp": f.get('frame_timestamp', ''),
"description": f.get('action', ''),
"people": [f.get('person', '')] if f.get('person') else [],
"is_attention_event": bool(f.get('is_attention_event', False)),
})
if events is None:
events = []
people = result.get('people_mentioned') or result.get('entities_json') or []
if isinstance(people, list) and people and isinstance(people[0], dict):
people = [p.get('person', '') for p in people]
people = [p for p in people if p]
return {
"global_summary": result.get('global_summary', ''),
"events": events,
"people_mentioned": people,
}
def _build_video_prompt(self, known_members: str, event_start_time: str) -> str:
start_hint = ""
if event_start_time:
start_hint = f"\n视频开始时间（北京时间）约为：{event_start_time}。请据此推算每个事件的绝对时间戳。"
return f"""你是家庭监控视频分析助手。下面是一段完整监控录像（已整段上传）。
请观看整段视频，提取其中有用的信息，只输出合法 JSON（不要 markdown、不要任何解释文字），结构如下：
{{ {{
"global_summary": "整个时段的整体摘要，简体中文，2-4 句，客观描述人物与主要活动", "global_summary": "整个时段的整体摘要，简体中文，2-4 句，客观描述人物与主要活动",
"entities_json": [ "events": [
{{"person": "人物标识(匹配已知成员用真名，否则用'人物A'/'人物B'...)", "action": "主要动作", "clothing": "衣着"}}
],
"frame_details": [
{{ {{
"frame_index": 图片序号(从1开始，与[图片N]标注对应), "timestamp": "事件发生时的绝对北京时间(格式 YYYY-MM-DD HH:MM:SS)",
"frame_timestamp": "该帧的时间戳(用[图片N]标注里的时间)", "description": "该时间点的画面/动作信息摘要（谁、在做什么、位置）",
"person": "该帧画面中的人物或'无人'", "people": ["出现在该时刻的人物，用已知成员真名或'人物A'/'人物B'"],
"action": "该帧可见动作", "is_attention_event": false
"clothing": "该帧衣着(颜色+类型)",
"is_attention_event": false,
"source_providers": ["gemini"]
}} }}
] ],
}} "people_mentioned": ["本视频出现过的所有人物标识/真名"]
}}{start_hint}
规则： 规则：
1. 只描述客观画面，不要猜测或想象。 1. 只描述客观画面，不要猜测或想象。
2. frame_details 每帧一条，frame_index 与上方[图片N]序号对应，frame_timestamp 用标注时间。 2. events 提取视频中"有意义的时间点"（人物出现/动作变化/异常），不要逐秒罗列；timestamp 用绝对北京时间。
3. 已知家庭成员（按特征匹配，匹配到用 real_name，否则用"人物X"）： 3. 已知家庭成员（按特征匹配，匹配到用 real_name，否则用"人物X"）：
{known_members or '（暂无已知成员）'} {known_members or '（暂无已知成员）'}
4. is_attention_event：是否为跌倒、危险、异常哭闹等需关注事件（没有则为 false）。 4. is_attention_event：是否为跌倒、危险、异常哭闹等需关注事件（没有则为 false）。
5. 没有人物出现的帧 person 填"无人"，action 填""。""" 5. 没有人物出现的时段不要单独成 event；people 留空数组。"""
# ------------------------------------------------------------------ # ------------------------------------------------------------------
# 智能问答：纯文本 # 智能问答：纯文本
@@ -222,11 +299,42 @@ class GeminiAdapter(BaseModelAdapter):
logger.warning("Gemini API Key 未配置，跳过问答") logger.warning("Gemini API Key 未配置，跳过问答")
return None return None
try: try:
return self._generate([{"text": prompt}], max_tokens=max_tokens, temperature=0.3) return self._generate_text(prompt, max_tokens=max_tokens, temperature=0.3)
except Exception as e: except Exception as e:
logger.error(f"Gemini 问答异常: {e}") logger.error(f"Gemini 问答异常: {e}")
return None return None
def _generate_text(self, text: str, max_tokens: int, temperature: float) -> Optional[str]:
"""纯文本 generateContent（复用模型 fallback 链）。"""
for model in self.model_chain:
try:
resp = requests.post(
f"{self._base_url}/models/{model}:generateContent?key={self.api_key}",
json={"contents": [{"parts": [{"text": text}]}],
"generationConfig": {
"temperature": temperature,
"maxOutputTokens": max_tokens}},
timeout=self.timeout
)
except requests.Timeout:
logger.warning(f"Gemini [{model}] 问答超时")
continue
except Exception as e:
logger.error(f"Gemini [{model}] 问答异常: {e}")
continue
if resp.status_code == 200:
cands = resp.json().get('candidates', [])
out = ''.join(
p.get('text', '')
for p in (cands[0].get('content', {}) if cands else {}).get('parts', [])
).strip() if cands else ''
if out:
return out
elif resp.status_code == 429:
logger.warning(f"Gemini [{model}] 429，切换模型")
continue
return None
def get_timeout(self) -> int: def get_timeout(self) -> int:
return self.timeout return self.timeout

333
fam-edge/src/fam_edge/model_adapters/nvidia_adapter.py
View File

@@ -2,23 +2,15 @@
NvidiaVisionAdapter - NVIDIA NIM 云端 VLM 适配器 NvidiaVisionAdapter - NVIDIA NIM 云端 VLM 适配器
provider_name = "nvidia" provider_name = "nvidia"
模型: nvidia/nemotron-3-nano-omni-30b-a3b-reasoning (Omni, 原生视频输入) 模型: nvidia/nemotron-nano-12b-v2-vl（NIM 官方支持整视频 video_url 输入，内部自行采样帧）
角色: vision (视觉分析直出结构化 JSON) + 智能问答 角色: vision (整视频直出结构化 JSON) + 智能问答
SDK: openai (NIM 兼容 OpenAI API 规范) SDK: openai (NIM 兼容 OpenAI API 规范)
整视频分析: 整视频 base64 经 video_url 单次调用 —— 本地不切片、不抽帧
视频模式 (analyze_video): 按关键帧时间点截取 ±1.5s 片段拼接集锦视频
(片段左上角叠加原始时间戳)，base64 后经 video_url 单次调用 —
模型看到动态画面而非静态帧，动作/轨迹识别显著优于逐帧图片。
图片模式 (analyze_frames): 逐帧 image_url 调用（无视频文件时的降级路径）。
注意: nemotron-omni 是 reasoning 模型，max_tokens 需给足（reasoning 消耗 token）。
""" """
import os import os
import base64 import base64
import json import json
import re import re
import subprocess
import tempfile
from typing import Dict, List, Optional from typing import Dict, List, Optional
from .base_adapter import BaseModelAdapter from .base_adapter import BaseModelAdapter
@@ -32,20 +24,17 @@ try:
except ImportError: except ImportError:
OpenAI = None OpenAI = None
VIDEO_SEGMENT_PAD = 1.5 # 关键帧前后各截取秒数
HIGHLIGHT_WIDTH = 640 # 集锦视频宽度（保持宽高比）
class NvidiaVisionAdapter(BaseModelAdapter): class NvidiaVisionAdapter(BaseModelAdapter):
"""NVIDIA NIM 云端 VLM 适配器 (视频集锦单次调用; 逐帧降级; 文本问答)""" """NVIDIA NIM 云端 VLM 适配器 (整视频单次调用; 文本问答)"""
def __init__(self, config: dict): def __init__(self, config: dict):
super().__init__("nvidia", config) super().__init__("nvidia", config)
self.model_name = config.get( self.model_name = config.get(
'model_name', 'nvidia/nemotron-3-nano-omni-30b-a3b-reasoning') 'model_name', 'nvidia/nemotron-nano-12b-v2-vl')
self.api_key = self._resolve_key(config.get('api_key', '')) self.api_key = self._resolve_key(config.get('api_key', ''))
self.base_url = config.get('base_url', 'https://integrate.api.nvidia.com/v1') self.base_url = config.get('base_url', 'https://integrate.api.nvidia.com/v1')
self.timeout = config.get('timeout', 120) self.timeout = config.get('timeout', 600)
cb_cfg = config.get('circuit_breaker', {}) cb_cfg = config.get('circuit_breaker', {})
self._cb = CircuitBreaker( self._cb = CircuitBreaker(
threshold=cb_cfg.get('threshold', 3), threshold=cb_cfg.get('threshold', 3),
@@ -78,127 +67,29 @@ class NvidiaVisionAdapter(BaseModelAdapter):
return False return False
# ------------------------------------------------------------------ # ------------------------------------------------------------------
# 视频模式：集锦视频 + video_url 单次调用（主路径） # 整视频分析：base64 整视频 -> video_url 单次调用
# ------------------------------------------------------------------ # ------------------------------------------------------------------
@staticmethod
def _ts_to_seconds(ts: str) -> float:
"""'2026-08-20 04:34:10'（生产格式）/ 'HH:MM:SS' -> 当日秒偏移"""
s = str(ts).strip()
# 绝对时间格式: 取时间部分（同一天 30min 片段内足够）
date_part, _, time_part = s.partition(' ')
if time_part and '-' in date_part:
s = time_part
parts = s.split(':')
try:
if len(parts) == 3:
return int(parts[0]) * 3600 + int(parts[1]) * 60 + float(parts[2])
if len(parts) == 2:
return int(parts[0]) * 60 + float(parts[1])
return float(ts)
except ValueError:
return -1.0
@staticmethod
def _probe_duration(video_path: str) -> float:
"""ffprobe 解析视频时长，失败返回 0"""
try:
r = subprocess.run(
['ffprobe', '-v', 'quiet', '-show_entries', 'format=duration',
'-of', 'csv=p=0', video_path],
capture_output=True, timeout=30)
return float(r.stdout.decode().strip() or 0)
except (subprocess.TimeoutExpired, OSError, ValueError):
return 0.0
def _build_highlight_video(self, video_path: str, frame_timestamps: List[str],
event_start_time: str = '') -> Optional[str]:
"""按关键帧时间点截取 ±pad 秒片段，叠加时间戳后拼接集锦视频
时间戳以 2026-08-20 04-34-10 形式叠加（连字符避免 ffmpeg drawtext 冒号转义）。
偏移换算: 绝对时间戳 - 视频开始时间（event_start_time 缺失时，
时间戳值本身须已是视频内偏移，如 HH:MM:SS 相对时间）。
"""
start_sec = self._ts_to_seconds(event_start_time) if event_start_time else 0.0
duration = self._probe_duration(video_path)
clips = []
for ts in frame_timestamps:
sec = self._ts_to_seconds(ts)
if sec < 0:
continue
if start_sec > 0:
sec -= start_sec
if sec < 0:
sec += 86400 # 跨午夜
if duration > 0 and (sec < -VIDEO_SEGMENT_PAD
or sec > duration - 0.5):
logger.info(f"NVIDIA 跳过超界片段: {ts} -> {sec:.1f}s (视频 {duration:.0f}s)")
continue
clips.append((sec, str(ts).replace(':', '-')))
if not clips:
return None
out_path = os.path.join(
tempfile.mkdtemp(prefix='nim_highlight_'), 'highlight.mp4')
cmd = ['ffmpeg', '-y', '-loglevel', 'error']
for start, _ in clips:
cmd += ['-ss', f'{start:.2f}', '-t', f'{VIDEO_SEGMENT_PAD * 2}', '-i', video_path]
parts = []
for i, (_, label) in enumerate(clips):
parts.append(
f"[{i}:v]fps=15,scale={HIGHLIGHT_WIDTH}:-2,"
f"drawtext=text='ts {label}':x=8:y=8:fontsize=22:"
f"fontcolor=white:box=1:boxcolor=black@0.6[v{i}]")
concat_in = ''.join(f'[v{i}]' for i in range(len(clips)))
parts.append(f'{concat_in}concat=n={len(clips)}:v=1:a=0[out]')
cmd += ['-filter_complex', ';'.join(parts), '-map', '[out]',
'-r', '15',
'-c:v', 'libx264', '-preset', 'veryfast', '-crf', '28',
'-an', out_path]
try:
subprocess.run(cmd, check=True, capture_output=True, timeout=120)
except subprocess.TimeoutExpired:
logger.warning("NVIDIA 集锦视频生成超时")
return None
except subprocess.CalledProcessError as e:
logger.warning(f"NVIDIA 集锦视频生成失败: {e.stderr.decode()[:200] if e.stderr else e}")
return None
size = os.path.getsize(out_path)
logger.info(f"NVIDIA 集锦视频生成: {len(clips)} 片段, {size // 1024}KB")
if size < 1024:
return None
return out_path
def analyze_video(self, video_path: str, def analyze_video(self, video_path: str,
frame_timestamps: List[str], known_members_context: str,
known_members_context: str, event_start_time: str = '') -> Optional[Dict]:
event_start_time: str = '') -> Optional[Dict]:
"""原生视频输入分析: 集锦片段 -> video_url 单次调用"""
if self._cb.is_open(): if self._cb.is_open():
logger.warning("NVIDIA 熔断器 OPEN，跳过视频分析") logger.warning("NVIDIA 熔断器 OPEN，跳过视频分析")
return None return None
if self._client is None: if self._client is None:
logger.warning("NVIDIA 客户端未初始化，跳过视频分析") logger.warning("NVIDIA 客户端未初始化，跳过视频分析")
return None return None
if not os.path.isfile(video_path):
highlight = self._build_highlight_video( logger.warning(f"NVIDIA 视频文件不存在: {video_path}")
video_path, frame_timestamps, event_start_time)
if not highlight:
logger.warning("NVIDIA 集锦视频不可用，降级逐帧模式")
return None return None
try: try:
with open(highlight, 'rb') as f: with open(video_path, 'rb') as f:
b64 = base64.b64encode(f.read()).decode('utf-8') b64 = base64.b64encode(f.read()).decode('utf-8')
except Exception as e: except Exception as e:
logger.warning(f"NVIDIA 读取集锦视频失败: {e}") logger.warning(f"NVIDIA 读取视频失败: {e}")
return None return None
finally:
try:
os.remove(highlight)
os.rmdir(os.path.dirname(highlight))
except OSError:
pass
prompt = self._build_video_prompt(frame_timestamps, known_members_context) prompt = self._build_video_prompt(known_members_context, event_start_time)
try: try:
resp = self._client.chat.completions.create( resp = self._client.chat.completions.create(
model=self.model_name, model=self.model_name,
@@ -208,134 +99,47 @@ class NvidiaVisionAdapter(BaseModelAdapter):
"url": f"data:video/mp4;base64,{b64}"}} "url": f"data:video/mp4;base64,{b64}"}}
]}], ]}],
temperature=0.2, temperature=0.2,
max_tokens=3072, max_tokens=4096,
# NIM 扩展：控制视频采样帧数（模型上限 128 帧）
extra_body={"media_io_kwargs": {"video": {"num_frames": 128}}},
timeout=self.timeout timeout=self.timeout
) )
content = resp.choices[0].message.content content = resp.choices[0].message.content
if not content: if not content:
logger.warning("NVIDIA 视频分析返回空 content") logger.warning("NVIDIA 视频分析返回空 content")
self._cb.record_failure()
return None return None
data = self._parse_single_frame_json(content) data = self._parse_json(content)
if not data or 'frame_details' not in data: if not data or 'events' not in data:
logger.warning(f"NVIDIA 视频 JSON 解析失败: {content[:150]}") logger.warning(f"NVIDIA 视频 JSON 解析失败: {content[:150]}")
return None self._cb.record_failure()
frame_details = self._normalize_frame_details(data, frame_timestamps)
if not frame_details:
return None return None
self._cb.record_success() self._cb.record_success()
logger.info(f"NVIDIA 视频分析完成，frame_details={len(frame_details)}") logger.info(f"NVIDIA 整视频分析完成，events={len(data.get('events', []))}")
result = {"frame_details": frame_details} return {
if data.get('global_summary'): "global_summary": str(data.get('global_summary', '')),
result['global_summary'] = str(data['global_summary']) "events": data.get('events', []),
if data.get('entities_json'): "people_mentioned": data.get('people_mentioned', []),
result['entities_json'] = data['entities_json'] "compute_provider": "nvidia",
return result }
except Exception as e: except Exception as e:
self._cb.record_failure() self._cb.record_failure()
logger.warning(f"NVIDIA 视频分析异常: {e}") logger.warning(f"NVIDIA 视频分析异常: {e}")
return None return None
def _normalize_frame_details(self, data: dict, @staticmethod
frame_timestamps: List[str]) -> List[Dict]: def _parse_json(content: str) -> Optional[dict]:
"""归一化模型输出的 frame_details，按已知时间戳对齐"""
details = []
for i, item in enumerate(data.get('frame_details', []), 1):
if not isinstance(item, dict):
continue
ts = str(item.get('frame_timestamp',
frame_timestamps[i - 1] if i <= len(frame_timestamps) else ''))
details.append({
"frame_index": i,
"frame_timestamp": ts,
"person": str(item.get('person', '无人')),
"action": str(item.get('action', '')),
"clothing": str(item.get('clothing', '')),
"is_attention_event": bool(item.get('is_attention_event', False)),
"source_providers": ["nvidia"],
})
return details
def _build_video_prompt(self, frame_timestamps: List[str],
known_members: str) -> str:
ts_list = '\n'.join(f' 片段{i}: 原始时间 {ts}' for i, ts in enumerate(frame_timestamps, 1))
return f"""你是家庭监控视频分析助手。下面的视频是由一段长时间监控录像中抽取的片段集锦，
共 {len(frame_timestamps)} 个片段（每个约 3 秒），按顺序拼接。每个片段左上角叠加了
原始时间戳（ts 后的 2026-08-20 04-34-10 表示北京时间 2026年8月20日 04:34:10）。
片段时间对照：
{ts_list}
只输出合法 JSON（不要 markdown、不要解释），结构如下：
{{
"frame_details": [
{{
"frame_timestamp": "<片段原始时间>",
"person": "片段中的人物或'无人'",
"action": "片段中人物的动作（动态观察，如走动/跑动/坐下）",
"clothing": "衣着(颜色+类型)",
"is_attention_event": false
}}
],
"global_summary": "整段录像的综合摘要",
"entities_json": [{{"person": "人物名或人物X", "action": "行为概括", "clothing": "衣着"}}]
}}
规则：
1. 只描述客观画面，不猜测。
2. 已知家庭成员（按特征匹配，匹配到用 real_name，否则用"人物X"）：
{known_members or '（暂无已知成员）'}
3. is_attention_event：跌倒、危险、异常哭闹等需关注事件（没有则为 false）。
4. 无人出现的片段 person 填"无人"，action 填""。"""
# ------------------------------------------------------------------
# 图片模式：逐帧调用（降级路径，无视频文件时使用）
# ------------------------------------------------------------------
def analyze_frames(self, frame_paths: List[str],
frame_timestamps: List[str],
known_members_context: str) -> Optional[Dict]:
if self._cb.is_open():
logger.warning("NVIDIA 熔断器 OPEN，跳过调用")
return None
if self._client is None:
logger.warning("NVIDIA 客户端未初始化，跳过调用")
return None
if not frame_paths:
logger.warning("NVIDIA 无帧可分析")
return None
frame_details = []
ok = False
for i, (path, ts) in enumerate(zip(frame_paths, frame_timestamps), 1):
detail = self._analyze_one_structured(path, ts, i, known_members_context)
if detail:
frame_details.append(detail)
ok = True
if not ok:
self._cb.record_failure()
return None
self._cb.record_success()
logger.info(f"NVIDIA 视觉分析完成，frame_details={len(frame_details)}")
# NVIDIA 单帧无法跨帧综合 global_summary，交由 Edge format_cloud_result 格式化生成
return {"frame_details": frame_details}
def _parse_single_frame_json(self, content: str) -> Optional[dict]:
"""轻量解析单帧 JSON（不要求全 schema，仅提取字段）"""
content = content.strip() content = content.strip()
# 直接解析
try: try:
return json.loads(content) return json.loads(content)
except json.JSONDecodeError: except json.JSONDecodeError:
pass pass
# 提取 markdown fence
fence = re.search(r'```(?:json)?\s*(\{.*?\})\s*```', content, re.DOTALL) fence = re.search(r'```(?:json)?\s*(\{.*?\})\s*```', content, re.DOTALL)
if fence: if fence:
try: try:
return json.loads(fence.group(1)) return json.loads(fence.group(1))
except json.JSONDecodeError: except json.JSONDecodeError:
pass pass
# 贪婪匹配最大 {...}
brace = re.search(r'\{.*\}', content, re.DOTALL) brace = re.search(r'\{.*\}', content, re.DOTALL)
if brace: if brace:
try: try:
@@ -344,68 +148,35 @@ class NvidiaVisionAdapter(BaseModelAdapter):
pass pass
return None return None
def _analyze_one_structured(self, path: str, ts: str, idx: int, def _build_video_prompt(self, known_members: str, event_start_time: str) -> str:
known_members: str) -> Optional[Dict]: start_hint = ""
try: if event_start_time:
with open(path, 'rb') as f: start_hint = f"\n视频开始时间（北京时间）约为：{event_start_time}。请据此推算每个事件的绝对时间戳。"
b64 = base64.b64encode(f.read()).decode('utf-8') return f"""你是家庭监控视频分析助手。下面是一段完整监控录像（已整段上传）。
except Exception as e: 请观看整段视频，提取其中有用的信息，只输出合法 JSON（不要 markdown、不要解释），结构如下：
logger.error(f"读取图片失败 {path}: {e}")
return None
prompt = self._build_structured_prompt(ts, idx, known_members)
try:
resp = self._client.chat.completions.create(
model=self.model_name,
messages=[{"role": "user", "content": [
{"type": "text", "text": prompt},
{"type": "image_url", "image_url": {"url": f"data:image/jpeg;base64,{b64}"}}
]}],
temperature=0.2,
max_tokens=512,
timeout=self.timeout
)
content = resp.choices[0].message.content
if not content:
return None
data = self._parse_single_frame_json(content)
if not data:
logger.warning(f"NVIDIA 单帧 JSON 解析失败: {content[:120]}")
return None
return {
"frame_index": idx,
"frame_timestamp": str(data.get("frame_timestamp", ts)),
"person": str(data.get("person", "无人")),
"action": str(data.get("action", "")),
"clothing": str(data.get("clothing", "")),
"is_attention_event": bool(data.get("is_attention_event", False)),
"source_providers": ["nvidia"],
}
except Exception as e:
logger.warning(f"NVIDIA 单帧异常: {e}")
return None
def _build_structured_prompt(self, ts: str, idx: int, known_members: str) -> str:
return f"""你是家庭监控视频分析助手。请看这张监控截图（拍摄时间 {ts}）。
只输出合法 JSON（不要 markdown、不要解释），结构如下：
{{ {{
"frame_timestamp": "{ts}", "global_summary": "整个时段的整体摘要，简体中文，2-4 句",
"person": "该帧画面中的人物或'无人'", "events": [
"action": "该帧可见动作", {{
"clothing": "该帧衣着(颜色+类型)", "timestamp": "事件发生时的绝对北京时间(YYYY-MM-DD HH:MM:SS)",
"is_attention_event": false "description": "该时刻画面/动作信息摘要",
}} "people": ["出现在该时刻的人物，用已知成员真名或'人物A'"],
"is_attention_event": false
}}
],
"people_mentioned": ["本视频出现过的所有人物标识/真名"]
}}{start_hint}
规则： 规则：
1. 只描述客观画面，不猜测。 1. 只描述客观画面，不猜测。
2. 已知家庭成员（按特征匹配，匹配到用 real_name，否则用"人物X"）： 2. events 提取有意义的时间点（人物出现/动作变化/异常），timestamp 用绝对北京时间。
3. 已知家庭成员（按特征匹配，匹配到用 real_name，否则用"人物X"）：
{known_members or '（暂无已知成员）'} {known_members or '（暂无已知成员）'}
3. is_attention_event：是否为跌倒、危险、异常哭闹等需关注事件（没有则为 false）。 4. is_attention_event：跌倒、危险、异常哭闹等需关注事件（没有则为 false）。"""
4. 没有人物出现的帧 person 填"无人"，action 填""。"""
# ------------------------------------------------------------------ # ------------------------------------------------------------------
# 智能问答：纯文本（reasoning 模型，max_tokens 需给足） # 智能问答：纯文本
# ------------------------------------------------------------------ # ------------------------------------------------------------------
def chat(self, prompt: str, max_tokens: int = 2048) -> Optional[str]: def chat(self, prompt: str, max_tokens: int = 2048) -> Optional[str]:
if self._client is None: if self._client is None:

7
fam-edge/src/fam_edge/model_adapters/ollama_adapter.py
View File

@@ -111,6 +111,13 @@ class OllamaAdapter(BaseModelAdapter):
self._cb.record_failure() self._cb.record_failure()
return None return None
def analyze_video(self, video_path: str,
known_members_context: str,
event_start_time: str = '') -> Optional[Dict]:
"""Ollama 为纯文本模型，不参与视频分析，返回 None（降级链不会选它做视频）。"""
logger.info("Ollama 为纯文本模型，跳过视频分析")
return None
def get_timeout(self) -> int: def get_timeout(self) -> int:
return self.timeout return self.timeout

238
fam-edge/src/fam_edge/oracle_db.py Normal file
View File

@@ -0,0 +1,238 @@
"""
Oracle 本地库（SQLite） - 视频摘要 / 事件 / 人物 存储
表结构:
videos : 每个被处理的视频一个记录（含全局摘要 + 事件列表 + 人物列表，JSON 冗余存储便于查询）
events : 视频拆出的事件（时间点 + 描述 + 涉及人物）
people : 规范人物表（canonical_name + 别名），由 person_service 维护
sync_cursor: 同步游标（NAS 拉取用，记录最后成功同步时间）
对外提供:
- upsert_video / get_pending_videos / mark_video_processed
- upsert_event
- upsert_person / get_known_members_context
- get_sync_delta(since_iso) -> 增量数据（供 NAS 拉取）
- set_cursor / get_cursor
"""
import os
import json
import sqlite3
from datetime import datetime, timezone, timedelta
from typing import Dict, List, Optional
logger = None # 延迟注入，避免循环 import
def _now_iso() -> str:
return datetime.now(timezone(timedelta(hours=8))).strftime('%Y-%m-%d %H:%M:%S')
class OracleDB:
def __init__(self, db_path: str):
os.makedirs(os.path.dirname(db_path), exist_ok=True)
self.db_path = db_path
self._conn = sqlite3.connect(db_path, check_same_thread=False)
self._conn.row_factory = sqlite3.Row
self._conn.execute("PRAGMA journal_mode=WAL")
self._init_schema()
# ------------------------------------------------------------------
def _init_schema(self):
c = self._conn
c.executescript("""
CREATE TABLE IF NOT EXISTS videos (
id INTEGER PRIMARY KEY AUTOINCREMENT,
drive_file_id TEXT,
filename TEXT UNIQUE,
local_path TEXT,
camera_name TEXT,
duration_sec REAL,
event_start_time TEXT,
status TEXT DEFAULT 'pending',
summary_json TEXT,
events_json TEXT,
people_json TEXT,
compute_provider TEXT,
created_at TEXT,
updated_at TEXT,
processed_at TEXT
);
CREATE TABLE IF NOT EXISTS events (
id INTEGER PRIMARY KEY AUTOINCREMENT,
video_id INTEGER,
ts TEXT,
description TEXT,
person_list_json TEXT,
is_attention_event INTEGER DEFAULT 0,
FOREIGN KEY(video_id) REFERENCES videos(id)
);
CREATE TABLE IF NOT EXISTS people (
id INTEGER PRIMARY KEY AUTOINCREMENT,
label TEXT UNIQUE,
canonical_name TEXT,
first_seen TEXT,
appearances INTEGER DEFAULT 0,
source TEXT DEFAULT 'llm',
updated_at TEXT
);
CREATE TABLE IF NOT EXISTS sync_cursor (
key TEXT PRIMARY KEY,
value TEXT
);
CREATE INDEX IF NOT EXISTS idx_videos_updated ON videos(updated_at);
CREATE INDEX IF NOT EXISTS idx_events_video ON events(video_id);
""")
self._conn.commit()
# ------------------------------------------------------------------
# videos
# ------------------------------------------------------------------
def get_video_by_filename(self, filename: str) -> Optional[sqlite3.Row]:
cur = self._conn.execute("SELECT * FROM videos WHERE filename=?", (filename,))
return cur.fetchone()
def ensure_video(self, filename: str, local_path: str,
camera_name: str = '', event_start_time: str = '',
duration_sec: float = 0.0, drive_file_id: str = '') -> int:
"""视频进入监听目录时登记；已存在则更新路径。返回 video_id。"""
now = _now_iso()
row = self.get_video_by_filename(filename)
if row:
self._conn.execute(
"UPDATE videos SET local_path=?, camera_name=?, event_start_time=?, "
"duration_sec=?, updated_at=? WHERE id=?",
(local_path, camera_name, event_start_time, duration_sec, now, row['id']))
self._conn.commit()
return row['id']
cur = self._conn.execute(
"INSERT INTO videos (drive_file_id, filename, local_path, camera_name, "
"duration_sec, event_start_time, status, created_at, updated_at) "
"VALUES (?,?,?,?,?,?, 'pending', ?, ?)",
(drive_file_id, filename, local_path, camera_name, duration_sec,
event_start_time, now, now))
self._conn.commit()
return cur.lastrowid
def get_pending_videos(self, limit: int = 1) -> List[sqlite3.Row]:
cur = self._conn.execute(
"SELECT * FROM videos WHERE status IN ('pending','failed') "
"ORDER BY id ASC LIMIT ?", (limit,))
return cur.fetchall()
def mark_video_processed(self, video_id: int, summary: str, events: List[dict],
people: List[str], compute_provider: str):
now = _now_iso()
self._conn.execute(
"UPDATE videos SET status='done', summary_json=?, events_json=?, "
"people_json=?, compute_provider=?, updated_at=?, processed_at=? WHERE id=?",
(summary, json.dumps(events, ensure_ascii=False), json.dumps(people, ensure_ascii=False),
compute_provider, now, now, video_id))
# 事件落独立表，便于 NAS 拉取
self._conn.execute("DELETE FROM events WHERE video_id=?", (video_id,))
for ev in events:
self._conn.execute(
"INSERT INTO events (video_id, ts, description, person_list_json, "
"is_attention_event) VALUES (?,?,?,?,?)",
(video_id, ev.get('timestamp', ''), ev.get('description', ''),
json.dumps(ev.get('people', []), ensure_ascii=False),
1 if ev.get('is_attention_event') else 0))
self._conn.commit()
def mark_video_failed(self, video_id: int, error: str = ''):
now = _now_iso()
self._conn.execute(
"UPDATE videos SET status='failed', summary_json=?, updated_at=? WHERE id=?",
(error, now, video_id))
self._conn.commit()
def get_all_videos(self) -> List[sqlite3.Row]:
return self._conn.execute(
"SELECT * FROM videos WHERE status='done' ORDER BY id ASC").fetchall()
# ------------------------------------------------------------------
# people
# ------------------------------------------------------------------
def upsert_person(self, label: str, canonical_name: str = '', source: str = 'llm',
first_seen: str = ''):
now = _now_iso()
row = self._conn.execute("SELECT * FROM people WHERE label=?", (label,)).fetchone()
if row:
# manual 覆盖 llm；llm 不覆盖 manual
if source == 'manual' or row['source'] != 'manual':
self._conn.execute(
"UPDATE people SET canonical_name=?, source=?, appearances=appearances+1, "
"updated_at=? WHERE label=?",
(canonical_name or row['canonical_name'], source, now, label))
else:
self._conn.execute(
"UPDATE people SET appearances=appearances+1, updated_at=? WHERE label=?",
(now, label))
else:
self._conn.execute(
"INSERT INTO people (label, canonical_name, first_seen, appearances, "
"source, updated_at) VALUES (?,?,?,1,?,?)",
(label, canonical_name, first_seen or now, source, now))
self._conn.commit()
def set_canonical(self, label: str, canonical_name: str, source: str = 'manual'):
"""手动命名：设置规范名（label 可视为别名）。"""
self.upsert_person(label, canonical_name, source='manual')
def get_people(self) -> List[sqlite3.Row]:
return self._conn.execute("SELECT * FROM people ORDER BY id ASC").fetchall()
def get_known_members_context(self) -> str:
"""生成 known_members_context 文本，注入视频提示让模型用真名。"""
rows = self.get_people()
lines = []
for r in rows:
name = r['canonical_name'] or r['label']
if name and name != r['label']:
lines.append(f"- {name}（别名/标识：{r['label']}）")
else:
lines.append(f"- {name}")
return '\n'.join(lines) if lines else ''
# ------------------------------------------------------------------
# 同步导出（供 NAS 拉取）
# ------------------------------------------------------------------
def get_sync_delta(self, since_iso: str) -> Dict:
"""返回 since 之后变更的 videos / events / people。"""
videos = self._conn.execute(
"SELECT * FROM videos WHERE updated_at > ? ORDER BY id ASC", (since_iso,)
).fetchall()
events = self._conn.execute(
"SELECT * FROM events WHERE updated_at > ? ORDER BY id ASC", (since_iso,)
).fetchall() if False else self._conn.execute(
"SELECT e.* FROM events e JOIN videos v ON e.video_id=v.id "
"WHERE v.updated_at > ? ORDER BY e.id ASC", (since_iso,)).fetchall()
people = self._conn.execute(
"SELECT * FROM people WHERE updated_at > ? ORDER BY id ASC", (since_iso,)
).fetchall()
def _ser(row):
d = dict(row)
return d
return {
"videos": [_ser(v) for v in videos],
"events": [_ser(e) for e in events],
"people": [_ser(p) for p in people],
"server_time": _now_iso(),
}
# ------------------------------------------------------------------
# 同步游标
# ------------------------------------------------------------------
def get_cursor(self, key: str) -> str:
row = self._conn.execute("SELECT value FROM sync_cursor WHERE key=?", (key,)).fetchone()
return row['value'] if row else ''
def set_cursor(self, key: str, value: str):
self._conn.execute(
"INSERT INTO sync_cursor (key, value) VALUES (?, ?) "
"ON CONFLICT(key) DO UPDATE SET value=excluded.value", (key, value))
self._conn.commit()
def close(self):
self._conn.close()

174
fam-edge/src/fam_edge/person_service.py Normal file
View File

@@ -0,0 +1,174 @@
"""
PersonService - 独立人物识别/汇总服务
职责:
1. 汇总所有视频中出现的人物（来自 OracleDB.people / videos.people_json）
2. 用 LLM（Gemini）将跨视频的人物标签合并为规范身份集
（不再依赖 OpenCV 人脸，纯靠视频 LLM 输出的人物标签 + 上下文由大模型判断合并）
3. 维护 people 表 canonical_name，生成 known_members_context
4. 回灌给视频分析提示（video_processor 每次分析前读取 known_members_context）
5. 接收 NAS 手动命名校正（source='manual' 优先，不被 LLM 覆盖）
注意: 无 face embedding，合并基于标签文本 + 描述上下文的大模型判断，保守合并。
"""
import json
import re
import threading
import time
from typing import Dict, List, Optional
from .logger import setup_logger
from .config_loader import load_config
from .model_adapters.adapter_factory import build_adapters
from . import oracle_db
logger = setup_logger('fam-edge.person_service')
class PersonService:
def __init__(self, db: oracle_db.OracleDB):
self.config = load_config()
self.db = db
self.interval = self.config.get('person_service', {}).get('schedule_interval_sec', 1800)
self.enabled = self.config.get('person_service', {}).get('enabled', True)
self._timer: Optional[threading.Timer] = None
self._stop = False
# 选一个 vision 模型做合并（通常 gemini）
adapters = build_adapters(self.config.get('models', []))
model_name = self.config.get('person_service', {}).get('model', 'gemini')
self._llm = next((a for a in adapters if a.provider_name == model_name), None)
if self._llm is None and adapters:
self._llm = adapters[0]
# ------------------------------------------------------------------
def reconcile(self):
"""汇总 + LLM 合并一次。可由定时或手动触发。"""
# 1. 先把所有视频的 people_mentioned 同步进 people 表（标签级）
for v in self.db.get_all_videos():
try:
people = json.loads(v['people_json'] or '[]')
except (ValueError, TypeError):
people = []
for p in people:
if p and p not in ('无人', '无'):
self.db.upsert_person(p, source='llm')
# 2. 收集未命名（无 canonical 或 canonical==label）的标签 + 描述样本
rows = self.db.get_people()
manual = {r['label']: r['canonical_name'] for r in rows if r['source'] == 'manual' and r['canonical_name']}
unnamed = [r for r in rows if not r['canonical_name'] or r['canonical_name'] == r['label']]
if not unnamed:
logger.info("PersonService: 无待合并人物，跳过 LLM 合并")
return
samples = self._collect_descriptions([r['label'] for r in unnamed])
mapping = self._llm_merge(unnamed, samples)
if not mapping:
return
for label, canonical in mapping.items():
if label in manual:
continue # 手动命名优先
if canonical and canonical != label:
self.db.set_canonical(label, canonical, source='llm')
logger.info(f"PersonService: LLM 合并完成，更新 {len(mapping)} 条")
def _collect_descriptions(self, labels: List[str]) -> Dict[str, List[str]]:
"""从 events 表收集每个标签出现时的描述样本。"""
samples: Dict[str, List[str]] = {l: [] for l in labels}
rows = self.db._conn.execute(
"SELECT description, person_list_json FROM events").fetchall()
for r in rows:
try:
plist = json.loads(r['person_list_json'] or '[]')
except (ValueError, TypeError):
plist = []
for p in plist:
if p in samples and len(samples[p]) < 3 and r['description']:
samples[p].append(r['description'])
return samples
def _llm_merge(self, unnamed: List, samples: Dict[str, List[str]]) -> Dict[str, str]:
"""请 LLM 把标签合并为规范名。返回 {label: canonical}。"""
if self._llm is None:
logger.warning("PersonService: 无可用的 LLM 适配器，跳过合并")
return {}
lines = []
for r in unnamed:
label = r['label']
desc = '；'.join(samples.get(label, [])) or '（无描述）'
lines.append(f"- {label}：出现场景 {desc}")
prompt = f"""你是家庭监控人物汇总助手。下面是若干人物标识及其出现场景描述。
请判断哪些标识指向同一个人，并为每个人输出一个稳定的规范名（用'人物A'/'人物B'这类占位，
或若场景描述足以区分则保留原标识）。只输出 JSON，格式：
{{"<原标识>": "<规范名>", ...}}
不要编造真实姓名，仅做去重/合并。
待处理人物：
{chr(10).join(lines)}"""
try:
text = self._llm.chat(prompt, max_tokens=1024)
except Exception as e:
logger.warning(f"PersonService: LLM 调用失败: {e}")
return {}
if not text:
return {}
return self._parse_mapping(text, {r['label'] for r in unnamed})
@staticmethod
def _parse_mapping(text: str, valid_labels) -> Dict[str, str]:
text = text.strip()
try:
data = json.loads(text)
except json.JSONDecodeError:
fence = re.search(r'```(?:json)?\s*(\{.*?\})\s*```', text, re.DOTALL)
if fence:
try:
data = json.loads(fence.group(1))
except json.JSONDecodeError:
return {}
else:
brace = re.search(r'\{.*\}', text, re.DOTALL)
if brace:
try:
data = json.loads(brace.group(0))
except json.JSONDecodeError:
return {}
else:
return {}
out = {}
for k, v in data.items():
if k in valid_labels and v and isinstance(v, str):
out[k] = v
return out
# ------------------------------------------------------------------
# 定时循环
# ------------------------------------------------------------------
def start(self):
if not self.enabled:
logger.info("PersonService 未启用")
return
self.reconcile() # 启动即跑一次
self._schedule_next()
def _schedule_next(self):
if self._stop:
return
self._timer = threading.Timer(self.interval, self._tick)
self._timer.daemon = True
self._timer.start()
def _tick(self):
if self._stop:
return
try:
self.reconcile()
except Exception as e:
logger.error(f"PersonService tick 异常: {e}")
self._schedule_next()
def stop(self):
self._stop = True
if self._timer:
self._timer.cancel()

35
fam-edge/src/fam_edge/qa.py Normal file
View File

@@ -0,0 +1,35 @@
"""
QA - 智能问答编排
run_qa(prompt): 按 models 顺序尝试 chat()，首个成功返回 (answer, provider)。
顺序 = vision 模型(Gemini -> NVIDIA) + text 模型(Ollama 兜底)。
即 Gemini -> NVIDIA -> Ollama 三级降级。
"""
from typing import Optional, Tuple
from .logger import setup_logger
from .config_loader import load_config
from .model_adapters.adapter_factory import build_adapters
logger = setup_logger('fam-edge.qa')
class QAOrchestrator:
def __init__(self):
self.config = load_config()
self.adapters = build_adapters(self.config.get('models', []))
def run_qa(self, prompt: str,
max_tokens: int = 1024) -> Tuple[Optional[str], Optional[str]]:
"""依次尝试各适配器的 chat()，返回 (answer, provider)。"""
for adapter in self.adapters:
try:
answer = adapter.chat(prompt, max_tokens=max_tokens)
except Exception as e:
logger.warning(f"QA {adapter.provider_name} 异常: {e}")
continue
if answer:
logger.info(f"QA 命中 provider={adapter.provider_name}")
return answer, adapter.provider_name
logger.info(f"QA {adapter.provider_name} 无返回，降级下一模型")
return None, None

5
fam-edge/src/fam_edge/queue/__init__.py
View File

@@ -1,5 +0,0 @@
"""
SQLite 异步任务队列 (Oracle 端)
"""
from . import queue_manager
from .consumer import get_consumer

128
fam-edge/src/fam_edge/queue/consumer.py
View File

@@ -1,128 +0,0 @@
"""
消费者线程 - 从 SQLite 队列消费任务，限速处理
策略：Gemini 优先 → NVIDIA 兜底（与现有 orchestrator 一致）
速率：按 API 限制速度的 2 倍设置突发容量
"""
import os
import time
import threading
from typing import Optional
from ..logger import setup_logger
from ..config_loader import load_config
from ..rate_limiter import RateLimiter
from ..ai_orchestrator.orchestrator import AIOrchestrator
from ..video_preprocessor.preprocessor import VideoPreprocessor
from . import queue_manager
logger = setup_logger('fam-edge.consumer')
# 从配置加载速率限制参数
_cfg = load_config()
_queue_cfg = _cfg.get('queue', {})
_rate_cfg = _queue_cfg.get('rate_limit', {})
GEMINI_RPM = _rate_cfg.get('gemini_rpm', 1000)
NVIDIA_RPM = _rate_cfg.get('nvidia_rpm', 40)
BURST_FACTOR = _rate_cfg.get('burst_factor', 2)
POLL_INTERVAL = _queue_cfg.get('poll_interval', 10)
# 同步 SQLite DB 路径到环境变量（供 queue_manager 读取）
os.environ.setdefault('FAM_QUEUE_DB', _queue_cfg.get('db_path', '/opt/fam-edge/data/fam_queue.db'))
os.environ.setdefault('FAM_UPLOAD_DIR', _queue_cfg.get('upload_dir', '/tmp/fam_uploads'))
class Consumer:
def __init__(self):
self._running = False
self._thread = None
self._orchestrator = AIOrchestrator()
self._rate_limiter = RateLimiter()
self._rate_limiter.register('gemini', GEMINI_RPM, burst_factor=BURST_FACTOR)
self._rate_limiter.register('nvidia', NVIDIA_RPM, burst_factor=BURST_FACTOR)
self._poll_interval = POLL_INTERVAL
def _process_one(self, task: dict) -> bool:
task_id = task['id']
nas_task_id = task['nas_task_id']
video_path = task['video_path']
logger.info(f"[nas_task={nas_task_id}] 消费者开始处理")
preprocessor = None
try:
preprocessor = VideoPreprocessor(nas_task_id)
task_data = {
"task_id": nas_task_id,
"camera_name": task.get('camera_name', ''),
"event_start_time": task.get('event_start_time', ''),
"event_end_time": "",
"known_members_context": task.get('known_members_context', ''),
}
result = self._orchestrator.process_push_task(
task_data, video_path, preprocessor, self._rate_limiter
)
if result.get('status') == 'success':
import json
queue_manager.mark_success(task_id, json.dumps(result, ensure_ascii=False))
logger.info(f"[nas_task={nas_task_id}] 消费者处理成功")
return True
else:
error = result.get('error_message', 'unknown')
stage = result.get('failure_stage', '')
queue_manager.mark_failed(task_id, error, stage)
logger.error(f"[nas_task={nas_task_id}] 消费者处理失败: {error}")
return False
except Exception as e:
logger.error(f"[nas_task={nas_task_id}] 消费者异常: {e}", exc_info=True)
queue_manager.mark_failed(task_id, str(e), 'process')
return False
finally:
if preprocessor is not None:
preprocessor.cleanup()
try:
if video_path and __import__('os').path.exists(video_path):
__import__('os').remove(video_path)
logger.info(f"[nas_task={nas_task_id}] 清理视频文件: {video_path}")
except Exception:
pass
def _run(self):
logger.info(f"消费者线程启动，轮询间隔 {self._poll_interval}s")
logger.info(f"速率限制: Gemini {GEMINI_RPM}RPM x2 burst, NVIDIA {NVIDIA_RPM}RPM x2 burst")
while self._running:
try:
task = queue_manager.claim_next()
if task is None:
time.sleep(self._poll_interval)
continue
self._process_one(task)
except Exception as e:
logger.error(f"消费者循环异常: {e}", exc_info=True)
time.sleep(self._poll_interval)
def start(self):
if self._running:
return
self._running = True
self._thread = threading.Thread(target=self._run, daemon=True, name='consumer')
self._thread.start()
logger.info("消费者线程已启动")
def stop(self):
self._running = False
logger.info("消费者线程已停止")
_consumer: Optional[Consumer] = None
def get_consumer() -> Consumer:
global _consumer
if _consumer is None:
_consumer = Consumer()
return _consumer

209
fam-edge/src/fam_edge/queue/queue_manager.py
View File

@@ -1,209 +0,0 @@
"""
SQLite 队列管理器 - Oracle 端异步任务队列
表结构:
- task_queue: 任务队列 (PENDING → PROCESSING → SUCCESS/FAILED)
- 元数据: delivered 标记 NAS 是否已拉取结果
"""
import os
import sqlite3
import json
import threading
from typing import Optional, List, Dict
DB_PATH = os.environ.get('FAM_QUEUE_DB', '/opt/fam-edge/data/fam_queue.db')
_init_lock = threading.Lock()
_initialized = False
def _get_conn() -> sqlite3.Connection:
global _initialized
if not _initialized:
with _init_lock:
if not _initialized:
_init_db()
_initialized = True
conn = sqlite3.connect(DB_PATH, timeout=30)
conn.row_factory = sqlite3.Row
conn.execute("PRAGMA journal_mode=WAL")
return conn
_TASK_QUEUE_DDL = """
CREATE TABLE {name} (
id INTEGER PRIMARY KEY AUTOINCREMENT,
nas_task_id INTEGER NOT NULL,
video_filename TEXT NOT NULL,
video_path TEXT NOT NULL,
camera_name TEXT DEFAULT '',
event_start_time TEXT DEFAULT '',
known_members_context TEXT DEFAULT '',
status TEXT DEFAULT 'PENDING',
result_json TEXT,
error_message TEXT,
failure_stage TEXT,
retry_count INTEGER DEFAULT 0,
created_at TEXT DEFAULT (datetime('now', '+8 hours')),
updated_at TEXT DEFAULT (datetime('now', '+8 hours')),
delivered INTEGER DEFAULT 0,
UNIQUE(nas_task_id)
)
"""
def _init_db():
os.makedirs(os.path.dirname(DB_PATH), exist_ok=True)
conn = sqlite3.connect(DB_PATH)
# DEFAULT 约束固化在表 schema 中，CREATE TABLE IF NOT EXISTS 不会更新已存在的旧表
# （旧表 DEFAULT 是 localtime，UTC 机器上=UTC）。检测到旧 schema 时重建表迁移。
row = conn.execute(
"SELECT sql FROM sqlite_master WHERE type='table' AND name='task_queue'"
).fetchone()
if row is not None and 'localtime' in (row[0] or ''):
conn.execute("BEGIN IMMEDIATE")
conn.execute(_TASK_QUEUE_DDL.format(name='task_queue_new'))
conn.execute("INSERT INTO task_queue_new SELECT * FROM task_queue")
conn.execute("DROP TABLE task_queue")
conn.execute("ALTER TABLE task_queue_new RENAME TO task_queue")
conn.commit()
conn.execute(_TASK_QUEUE_DDL.format(name='task_queue').replace(
'CREATE TABLE task_queue', 'CREATE TABLE IF NOT EXISTS task_queue'))
conn.execute("CREATE INDEX IF NOT EXISTS idx_status ON task_queue(status)")
conn.execute("CREATE INDEX IF NOT EXISTS idx_delivered ON task_queue(delivered)")
conn.commit()
conn.close()
def enqueue(nas_task_id: int, video_filename: str, video_path: str,
camera_name: str, event_start_time: str,
known_members_context: str) -> int:
conn = _get_conn()
try:
cur = conn.execute(
"INSERT OR IGNORE INTO task_queue "
"(nas_task_id, video_filename, video_path, camera_name, event_start_time, "
"known_members_context, created_at, updated_at) "
"VALUES (?, ?, ?, ?, ?, ?, datetime('now','+8 hours'), datetime('now','+8 hours'))",
(nas_task_id, video_filename, video_path, camera_name, event_start_time, known_members_context)
)
conn.commit()
if cur.rowcount == 0:
row = conn.execute(
"SELECT id, status, delivered FROM task_queue WHERE nas_task_id=?",
(nas_task_id,)
).fetchone()
if row is None:
return 0
# NAS 重新派发: FAILED（或已交付的 SUCCESS）重置为 PENDING，用新上传的视频重跑。
# SUCCESS 且未交付的行不动，避免丢失待 Poller 拉取的结果。
if row['status'] == 'FAILED' or (row['status'] == 'SUCCESS' and row['delivered']):
conn.execute(
"UPDATE task_queue SET status='PENDING', result_json=NULL, error_message=NULL, "
"failure_stage=NULL, retry_count=0, delivered=0, video_filename=?, video_path=?, "
"camera_name=?, event_start_time=?, known_members_context=?, "
"updated_at=datetime('now','+8 hours') WHERE id=?",
(video_filename, video_path, camera_name, event_start_time,
known_members_context, row['id'])
)
conn.commit()
return row['id']
return cur.lastrowid
finally:
conn.close()
def claim_next() -> Optional[Dict]:
conn = _get_conn()
try:
conn.execute("BEGIN IMMEDIATE")
row = conn.execute(
"SELECT * FROM task_queue WHERE status='PENDING' ORDER BY id LIMIT 1"
).fetchone()
if row:
conn.execute(
"UPDATE task_queue SET status='PROCESSING', updated_at=datetime('now','+8 hours') WHERE id=?",
(row['id'],)
)
conn.commit()
return dict(row)
conn.rollback()
return None
except Exception:
conn.rollback()
return None
finally:
conn.close()
def mark_success(task_id: int, result_json: str):
conn = _get_conn()
try:
conn.execute(
"UPDATE task_queue SET status='SUCCESS', result_json=?, updated_at=datetime('now','+8 hours') WHERE id=?",
(result_json, task_id)
)
conn.commit()
finally:
conn.close()
def mark_failed(task_id: int, error_message: str, failure_stage: str = ''):
conn = _get_conn()
try:
conn.execute(
"UPDATE task_queue SET status='FAILED', error_message=?, failure_stage=?, "
"updated_at=datetime('now','+8 hours') WHERE id=?",
(error_message, failure_stage, task_id)
)
conn.commit()
finally:
conn.close()
def get_undelivered_results(limit: int = 10) -> List[Dict]:
conn = _get_conn()
try:
# FAILED 也需交付: 否则 NAS 永远收不到失败结果，任务卡 PROCESSING
# 直至僵尸回收后无意义地重传 22MB 视频
rows = conn.execute(
"SELECT * FROM task_queue WHERE status IN ('SUCCESS','FAILED') AND delivered=0 "
"ORDER BY id LIMIT ?", (limit,)
).fetchall()
return [dict(r) for r in rows]
finally:
conn.close()
def mark_delivered(task_ids: List[int]):
if not task_ids:
return
conn = _get_conn()
try:
placeholders = ','.join('?' * len(task_ids))
conn.execute(
f"UPDATE task_queue SET delivered=1, updated_at=datetime('now','+8 hours') "
f"WHERE id IN ({placeholders})", task_ids
)
conn.commit()
finally:
conn.close()
def get_queue_stats() -> Dict:
conn = _get_conn()
try:
stats = {}
for status in ['PENDING', 'PROCESSING', 'SUCCESS', 'FAILED']:
row = conn.execute(
"SELECT COUNT(*) as cnt FROM task_queue WHERE status=?", (status,)
).fetchone()
stats[status] = row['cnt']
row = conn.execute(
"SELECT COUNT(*) as cnt FROM task_queue "
"WHERE status IN ('SUCCESS','FAILED') AND delivered=0"
).fetchone()
stats['UNDELIVERED'] = row['cnt']
return stats
finally:
conn.close()

48
fam-edge/src/fam_edge/rate_limiter.py
View File

@@ -1,48 +0,0 @@
"""
Rate Limiter - Token Bucket 算法
按 API 限制速度的 2 倍设置突发容量，按 API 限制速度持续补充。
"""
import time
import threading
class TokenBucket:
def __init__(self, rpm: int, burst_factor: int = 2):
self.capacity = rpm * burst_factor
self.refill_rate = rpm / 60.0
self.tokens = float(self.capacity)
self.last_refill = time.monotonic()
self._lock = threading.Lock()
def acquire(self, tokens: int = 1, timeout: float = 300.0) -> bool:
deadline = time.monotonic() + timeout
while True:
with self._lock:
now = time.monotonic()
elapsed = now - self.last_refill
self.tokens = min(self.capacity, self.tokens + elapsed * self.refill_rate)
self.last_refill = now
if self.tokens >= tokens:
self.tokens -= tokens
return True
wait = (tokens - self.tokens) / self.refill_rate
if time.monotonic() + wait > deadline:
return False
time.sleep(min(wait, 1.0))
class RateLimiter:
"""多 API 速率限制管理"""
def __init__(self):
self._buckets = {}
def register(self, name: str, rpm: int, burst_factor: int = 2):
self._buckets[name] = TokenBucket(rpm, burst_factor)
def acquire(self, name: str, tokens: int = 1, timeout: float = 300.0) -> bool:
bucket = self._buckets.get(name)
if bucket is None:
return True
return bucket.acquire(tokens, timeout)

19
fam-edge/src/fam_edge/state.py Normal file
View File

@@ -0,0 +1,19 @@
"""
state - 进程内共享单例（OracleDB 实例）
watch_processor / person_service / api_gateway 都通过 get_db() 访问同一个 SQLite 连接，
避免重复打开与循环 import。
"""
from . import oracle_db
from .config_loader import load_config
_db = None
def get_db() -> oracle_db.OracleDB:
global _db
if _db is None:
cfg = load_config()
path = cfg.get('oracle_db', {}).get('path', '/opt/fam-edge/data/oracle.db')
_db = oracle_db.OracleDB(path)
return _db

4
fam-edge/src/fam_edge/storage_cleaner/__init__.py
View File

@@ -1,4 +0,0 @@
"""Storage-Cleaner 包"""
from .cleaner import StorageCleaner
__all__ = ["StorageCleaner"]

82
fam-edge/src/fam_edge/storage_cleaner/cleaner.py
View File

@@ -1,82 +0,0 @@
"""
Storage-Cleaner - 临时文件清理
首期仅 finally 清理（不做 Cron 兜底）
- 删除下载的视频文件
- 删除粗抽候选帧
- 删除压缩关键帧
- 清理任务工作目录
"""
import os
import shutil
from ..logger import setup_logger, log_task
logger = setup_logger('fam-edge.storage_cleaner')
class StorageCleaner:
"""临时文件清理器"""
def __init__(self, task_id: int, work_dir: str):
self.task_id = task_id
self.work_dir = work_dir
def cleanup(self):
"""清理整个工作目录"""
try:
if os.path.exists(self.work_dir):
# 统计清理前大小
total_size = 0
for dirpath, dirnames, filenames in os.walk(self.work_dir):
for f in filenames:
fp = os.path.join(dirpath, f)
try:
total_size += os.path.getsize(fp)
except OSError:
pass
shutil.rmtree(self.work_dir)
size_mb = total_size / (1024 * 1024)
log_task(logger, self.task_id, 'cleanup',
f'已清理工作目录: {self.work_dir} ({size_mb:.1f}MB)')
else:
log_task(logger, self.task_id, 'cleanup',
f'工作目录不存在，无需清理: {self.work_dir}')
except PermissionError as e:
logger.warning(f"[task_id={self.task_id}] 清理权限不足: {e}")
except Exception as e:
logger.warning(f"[task_id={self.task_id}] 清理异常: {e}")
def cleanup_file(self, filepath: str):
"""清理单个文件"""
try:
if os.path.exists(filepath):
os.remove(filepath)
logger.info(f"[task_id={self.task_id}] 已删除: {filepath}")
except Exception as e:
logger.warning(f"[task_id={self.task_id}] 删除文件失败 {filepath}: {e}")
@staticmethod
def cleanup_stale_dirs(base_dir='/tmp/fam_media', max_age_hours=24):
"""清理超期的残留目录（超过 max_age_hours 的 task_* 目录）
首期不通过 Cron 调用，可在进程启动时手动执行一次。
"""
if not os.path.isdir(base_dir):
return
import time
now = time.time()
max_age_seconds = max_age_hours * 3600
for entry in os.listdir(base_dir):
entry_path = os.path.join(base_dir, entry)
if not os.path.isdir(entry_path) or not entry.startswith('task_'):
continue
try:
dir_mtime = os.path.getmtime(entry_path)
if now - dir_mtime > max_age_seconds:
shutil.rmtree(entry_path)
logger.info(f"清理超期残留目录: {entry_path}")
except Exception as e:
logger.warning(f"清理残留目录失败 {entry_path}: {e}")

4
fam-edge/src/fam_edge/video_preprocessor/__init__.py
View File

@@ -1,4 +0,0 @@
"""Video-Preprocessor 包"""
from .preprocessor import VideoPreprocessor
__all__ = ["VideoPreprocessor"]

316
fam-edge/src/fam_edge/video_preprocessor/preprocessor.py
View File

@@ -1,316 +0,0 @@
"""
Video-Preprocessor - 视频预处理
流程:
1. 下载视频（超时 60s）
2. 根据视频时长自适应计算候选帧数，FFmpeg 等距粗抽
3. 根据视频时长自适应计算关键帧数，OpenCV 帧差分析筛选（MSE 阈值）
4. 压缩（长边 ≤ 1024px，JPEG 质量 80）
自适应规则:
- 候选帧: max(candidate_min, duration_min * candidate_per_minute), 上限 candidate_max
- 关键帧: max(min_key_frames, duration / key_frame_interval_sec), 上限 max_key_frames_cap
例: 30分钟视频 → 候选60张 → 关键帧12张（每2.5分钟1张）
例: 3分钟视频 → 候选30张 → 关键帧8张（保底）
异常兜底:
- ffprobe 失败 -> 退化为按 60s 间隔抽帧
- 帧差分析异常 -> 退化为等距抽 min_key_frames 帧
- OpenCV 压缩失败 -> 跳过该帧，记录 WARN
"""
import os
import time
import subprocess
import requests
import cv2
import numpy as np
from typing import List, Tuple, Optional
from ..logger import setup_logger, log_task
from ..config_loader import load_config
logger = setup_logger('fam-edge.preprocessor')
class VideoPreprocessor:
"""视频预处理器"""
def __init__(self, task_id: int):
self.task_id = task_id
cfg = load_config()
video_cfg = cfg.get('video', {})
self.candidate_per_minute = video_cfg.get('candidate_per_minute', 2)
self.candidate_min = video_cfg.get('candidate_min', 30)
self.candidate_max = video_cfg.get('candidate_max', 120)
self.key_frame_interval_sec = video_cfg.get('key_frame_interval_sec', 150)
self.min_key_frames = video_cfg.get('min_key_frames', 5)
self.max_key_frames_floor = video_cfg.get('max_key_frames_floor', 8)
self.max_key_frames_cap = video_cfg.get('max_key_frames_cap', 30)
self.mse_threshold = video_cfg.get('mse_threshold', 500)
self.jpeg_quality = video_cfg.get('jpeg_quality', 80)
self.max_long_edge = video_cfg.get('max_long_edge', 1024)
timeout_cfg = cfg.get('timeout', {})
self.download_timeout = timeout_cfg.get('download', 60)
# 视频时长（秒），在 extract_candidate_frames 中填充
self.video_duration = 0.0
# 临时目录
self.work_dir = f"/tmp/fam_media/task_{task_id}"
self.video_path = os.path.join(self.work_dir, f"video_{task_id}.mp4")
self.frames_dir = os.path.join(self.work_dir, "frames")
self.keyframes_dir = os.path.join(self.work_dir, "keyframes")
def download_video(self, video_url: str) -> str:
"""下载视频"""
os.makedirs(self.work_dir, exist_ok=True)
start = time.time()
log_task(logger, self.task_id, 'download', f'开始下载: {video_url}')
resp = requests.get(video_url, stream=True, timeout=self.download_timeout)
if resp.status_code != 200:
raise Exception(f"下载失败: HTTP {resp.status_code}")
with open(self.video_path, 'wb') as f:
for chunk in resp.iter_content(chunk_size=8192):
f.write(chunk)
duration_ms = int((time.time() - start) * 1000)
size_mb = os.path.getsize(self.video_path) / (1024 * 1024)
log_task(logger, self.task_id, 'download', f'下载完成: {size_mb:.1f}MB', duration_ms=duration_ms)
return self.video_path
def save_upload(self, file_storage) -> str:
"""保存推送模式上传的视频文件（multipart），替代 download_video"""
os.makedirs(self.work_dir, exist_ok=True)
start = time.time()
file_storage.save(self.video_path)
duration_ms = int((time.time() - start) * 1000)
size_mb = os.path.getsize(self.video_path) / (1024 * 1024)
log_task(logger, self.task_id, 'upload',
f'保存上传视频: {size_mb:.1f}MB', duration_ms=duration_ms)
return self.video_path
def _get_video_duration(self, video_path: str) -> float:
"""用 ffprobe 获取视频时长（秒）"""
try:
cmd = [
'ffprobe', '-v', 'error',
'-show_entries', 'format=duration',
'-of', 'default=noprint_wrappers=1:nokey=1',
video_path
]
result = subprocess.run(cmd, capture_output=True, text=True, timeout=30)
if result.returncode == 0:
return float(result.stdout.strip())
except Exception as e:
logger.warning(f"[task_id={self.task_id}] ffprobe 失败: {e}")
return 0.0
def extract_candidate_frames(self, video_path: str) -> List[str]:
"""等距粗抽候选帧（数量随视频时长自适应，使用快速 seek）"""
os.makedirs(self.frames_dir, exist_ok=True)
duration = self._get_video_duration(video_path)
self.video_duration = duration
if duration > 0:
duration_min = duration / 60
# 自适应候选帧数：每分钟 candidate_per_minute 张，保底 candidate_min，上限 candidate_max
candidate_count = min(
max(self.candidate_min, int(duration_min * self.candidate_per_minute)),
self.candidate_max
)
interval = duration / candidate_count
else:
# 兜底: 每 60s 抽一帧
interval = 60
candidate_count = 0
logger.warning(f"[task_id={self.task_id}] ffprobe 失败，退化为 60s 间隔抽帧")
# 快速 seek 逐帧提取（比 fps 滤镜快 6-8 倍，ARM CPU 上尤甚）
timestamps = [i * interval for i in range(candidate_count)] if candidate_count > 0 else []
if not timestamps:
# 兜底: 未知时长，用 ffprobe 不可用时按 60s 间隔
timestamps = [i * 60 for i in range(30)]
for i, ts in enumerate(timestamps):
output_path = os.path.join(self.frames_dir, f'frame_{i+1:04d}.jpg')
cmd = [
'ffmpeg', '-ss', f'{ts:.1f}',
'-i', video_path,
'-frames:v', '1',
'-q:v', '2',
output_path
]
try:
subprocess.run(cmd, capture_output=True, timeout=30, check=True)
except (subprocess.CalledProcessError, subprocess.TimeoutExpired) as e:
logger.warning(f"[task_id={self.task_id}] seek 到 {ts:.1f}s 失败: {e}")
# 收集候选帧路径
frames = sorted([
os.path.join(self.frames_dir, f)
for f in os.listdir(self.frames_dir)
if f.endswith('.jpg')
])
log_task(logger, self.task_id, 'extract',
f'视频时长 {duration:.0f}s, 快速 seek 粗抽 {len(frames)} 张候选帧 (目标 {candidate_count})')
return frames
def _compute_adaptive_key_frame_counts(self) -> Tuple[int, int]:
"""根据视频时长自适应计算关键帧下限和上限"""
if self.video_duration > 0:
# 每隔 key_frame_interval_sec 秒 1 张关键帧
adaptive = int(self.video_duration / self.key_frame_interval_sec)
max_kf = min(max(self.max_key_frames_floor, adaptive), self.max_key_frames_cap)
else:
max_kf = self.max_key_frames_floor
min_kf = max(self.min_key_frames, max_kf // 2)
return min_kf, max_kf
def select_key_frames(self, candidate_frames: List[str]) -> List[str]:
"""帧差分析筛选关键帧（数量随视频时长自适应）"""
min_kf, max_kf = self._compute_adaptive_key_frame_counts()
log_task(logger, self.task_id, 'select_keyframes',
f'自适应关键帧: min={min_kf}, max={max_kf} (视频时长 {self.video_duration:.0f}s)')
if len(candidate_frames) <= min_kf:
return candidate_frames[:max_kf]
try:
# 加载所有候选帧
images = []
for path in candidate_frames:
img = cv2.imread(path)
if img is not None:
images.append((path, img))
if len(images) < 2:
return candidate_frames[:max_kf]
# 计算每帧与前一关键帧的 MSE
key_indices = [0] # 首帧必选
last_key_img = images[0][1]
for i in range(1, len(images)):
mse = self._compute_mse(last_key_img, images[i][1])
if mse > self.mse_threshold:
key_indices.append(i)
last_key_img = images[i][1]
# 末帧必选
if key_indices[-1] != len(images) - 1:
key_indices.append(len(images) - 1)
# 若 < min_kf，从剩余中均匀补足
if len(key_indices) < min_kf:
remaining = [i for i in range(len(images)) if i not in key_indices]
step = max(1, len(remaining) // (min_kf - len(key_indices)))
for i in range(0, len(remaining), step):
if len(key_indices) >= min_kf:
break
key_indices.append(remaining[i])
key_indices.sort()
# 若 > max_kf，按差异值降序取前 N
if len(key_indices) > max_kf:
# 计算每个关键帧与前一帧的差异
diffs = []
for idx in key_indices[1:-1]: # 不含首末帧
diff = self._compute_mse(images[idx-1][1], images[idx][1])
diffs.append((idx, diff))
diffs.sort(key=lambda x: x[1], reverse=True)
# 保留首末帧 + 差异最大的
keep = {0, len(images)-1}
for idx, _ in diffs[:max_kf - 2]:
keep.add(idx)
key_indices = sorted(keep)
key_frames = [images[i][0] for i in key_indices]
log_task(logger, self.task_id, 'select_keyframes', f'筛选 {len(key_frames)} 张关键帧')
return key_frames
except Exception as e:
logger.warning(f"[task_id={self.task_id}] 帧差分析异常: {e}，退化为等距抽 {min_kf} 帧")
step = max(1, len(candidate_frames) // min_kf)
return candidate_frames[::step][:min_kf]
def _compute_mse(self, img1, img2) -> float:
"""计算两帧的 MSE"""
# 转灰度并统一尺寸
h = min(img1.shape[0], img2.shape[0])
w = min(img1.shape[1], img2.shape[1])
g1 = cv2.cvtColor(img1, cv2.COLOR_BGR2GRAY)
g2 = cv2.cvtColor(img2, cv2.COLOR_BGR2GRAY)
g1 = cv2.resize(g1, (w, h))
g2 = cv2.resize(g2, (w, h))
diff = g1.astype(np.float64) - g2.astype(np.float64)
mse = np.mean(diff ** 2)
return float(mse)
def compress_frames(self, frame_paths: List[str]) -> List[str]:
"""压缩关键帧（长边 ≤ max_long_edge，JPEG 质量 80）"""
os.makedirs(self.keyframes_dir, exist_ok=True)
compressed = []
for i, path in enumerate(frame_paths):
out_path = os.path.join(self.keyframes_dir, f"keyframe_{i+1:02d}.jpg")
try:
img = cv2.imread(path)
if img is None:
logger.warning(f"[task_id={self.task_id}] 读取图片失败: {path}")
continue
h, w = img.shape[:2]
if max(h, w) > self.max_long_edge:
scale = self.max_long_edge / max(h, w)
img = cv2.resize(img, (int(w * scale), int(h * scale)))
cv2.imwrite(out_path, img, [cv2.IMWRITE_JPEG_QUALITY, self.jpeg_quality])
compressed.append(out_path)
except Exception as e:
logger.warning(f"[task_id={self.task_id}] 压缩失败 {path}: {e}")
continue
log_task(logger, self.task_id, 'compress', f'压缩 {len(compressed)} 张关键帧')
return compressed
def compute_timestamps(self, video_path: str, frame_count: int,
event_start_time: str) -> List[str]:
"""计算每帧的绝对时间戳 = 视频开始时间 + 帧偏移"""
from datetime import datetime, timedelta
duration = self._get_video_duration(video_path)
if duration <= 0:
duration = frame_count * 60 # 兜底
interval = duration / frame_count
from datetime import timedelta, timezone
# 统一北京时区: 视频均为北京时间录制，Edge 机器是 UTC，
# fallback 不能用本地 datetime.now()
try:
start_dt = datetime.fromisoformat(event_start_time.replace('Z', '+00:00'))
if start_dt.tzinfo is not None:
start_dt = start_dt.astimezone(timezone(timedelta(hours=8))).replace(tzinfo=None)
except Exception:
start_dt = datetime.now(timezone(timedelta(hours=8))).replace(tzinfo=None)
timestamps = []
for i in range(frame_count):
offset = interval * i
ts = start_dt + timedelta(seconds=offset)
timestamps.append(ts.strftime('%Y-%m-%d %H:%M:%S'))
return timestamps
def cleanup(self):
"""清理临时文件"""
import shutil
try:
if os.path.exists(self.work_dir):
shutil.rmtree(self.work_dir)
log_task(logger, self.task_id, 'cleanup', f'清理临时目录: {self.work_dir}')
except Exception as e:
logger.warning(f"[task_id={self.task_id}] 清理失败: {e}")

132
fam-edge/src/fam_edge/video_processor.py Normal file
View File

@@ -0,0 +1,132 @@
"""
VideoProcessor - 整视频分析编排
流程（不再切片/抽帧）:
1. 从 OracleDB 取当前 known_members_context（已命名/合并的人物）
2. 按 vision_order 依次调适配器的 analyze_video（Gemini 整视频 -> NVIDIA 整视频）
3. 首个成功结果 -> 归一化 -> 写 OracleDB（videos + events 表）
4. 把本视频 people_mentioned 更新进 people 表（供 person_service 后续合并）
降级: 全部视觉模型失败 -> 标记视频 failed（不再本地融合）
"""
import os
import re
from datetime import datetime, timedelta, timezone
from typing import Dict, List, Optional
from .logger import setup_logger
from .config_loader import load_config
from .model_adapters.adapter_factory import build_adapters
from .model_adapters.base_adapter import BaseModelAdapter
from . import oracle_db
logger = setup_logger('fam-edge.video_processor')
def _parse_event_start_from_filename(filename: str) -> str:
"""从监控文件名解析开始时间（北京时间）。示例: 2026-08-21_081500.mp4"""
m = re.search(r'(\d{4})[-_](\d{2})[-_](\d{2})[_-]?(\d{2})(\d{2})(\d{2})', filename)
if m:
y, mo, d, hh, mm, ss = m.groups()
try:
dt = datetime(int(y), int(mo), int(d), int(hh), int(mm), int(ss))
return dt.strftime('%Y-%m-%d %H:%M:%S')
except ValueError:
pass
# 退而求其次: 2026-08-21 08-15-00 等
m2 = re.search(r'(\d{4}-\d{2}-\d{2})[ _T-]+(\d{2})[-:](\d{2})[-:](\d{2})', filename)
if m2:
return f"{m2.group(1)} {m2.group(2)}:{m2.group(3)}:{m2.group(4)}"
return ''
class VideoProcessor:
def __init__(self, db: oracle_db.OracleDB):
self.config = load_config()
self.db = db
self.vision_order = self.config.get('video_processing', {}).get(
'vision_order', ['gemini', 'nvidia'])
self.vision_timeout = self.config.get('video_processing', {}).get('timeout', 900)
self.parse_start = self.config.get('gdrive_sync', {}).get(
'parse_start_from_filename', True)
adapters = build_adapters(self.config.get('models', []))
self.vision_adapters: Dict[str, BaseModelAdapter] = {
a.provider_name: a for a in adapters if a.get_role() == 'vision'}
def _ordered_vision_adapters(self) -> List[BaseModelAdapter]:
ordered = []
for name in self.vision_order:
if name in self.vision_adapters:
ordered.append(self.vision_adapters[name])
# 追加未在顺序里但启用的视觉适配器
for name, a in self.vision_adapters.items():
if name not in self.vision_order:
ordered.append(a)
return ordered
def process_video(self, video_id: int, filename: str, local_path: str) -> bool:
"""处理一个视频记录，返回是否成功。"""
if not os.path.isfile(local_path):
logger.error(f"[video_id={video_id}] 文件不存在，跳过: {local_path}")
self.db.mark_video_failed(video_id, "file_missing")
return False
camera_name = self.db.get_video_by_filename(filename)['camera_name'] or ''
event_start = ''
if self.parse_start:
event_start = _parse_event_start_from_filename(filename)
# 回写解析到的开始时间
if event_start:
self.db._conn.execute(
"UPDATE videos SET event_start_time=? WHERE id=?",
(event_start, video_id))
self.db._conn.commit()
known = self.db.get_known_members_context()
logger.info(f"[video_id={video_id}] 开始整视频分析: {filename} "
f"(event_start={event_start}, known_members={'有' if known else '无'})")
last_err = "no_vision_adapter"
for adapter in self._ordered_vision_adapters():
try:
logger.info(f"[video_id={video_id}] 尝试 {adapter.provider_name} 整视频分析")
result = adapter.analyze_video(local_path, known, event_start)
except Exception as e:
logger.error(f"[video_id={video_id}] {adapter.provider_name} 异常: {e}")
last_err = str(e)
continue
if result:
self._store_result(video_id, result)
return True
else:
last_err = f"{adapter.provider_name}_failed"
logger.warning(f"[video_id={video_id}] {adapter.provider_name} 未返回结果，降级下一模型")
logger.error(f"[video_id={video_id}] 所有视觉模型失败，标记 failed: {last_err}")
self.db.mark_video_failed(video_id, last_err)
return False
def _store_result(self, video_id: int, result: Dict):
events = result.get('events', [])
people = result.get('people_mentioned', [])
summary = result.get('global_summary', '')
provider = result.get('compute_provider', 'unknown')
# 归一化 events 时间戳（若模型给的是相对偏移，这里不强制；以模型输出为准）
norm_events = []
for ev in events:
norm_events.append({
"timestamp": str(ev.get('timestamp', '')),
"description": str(ev.get('description', '')),
"people": [str(p) for p in ev.get('people', []) if p],
"is_attention_event": bool(ev.get('is_attention_event', False)),
})
self.db.mark_video_processed(video_id, summary, norm_events, people, provider)
# 更新 people 表（标签级，待 person_service 合并）
for p in people:
if p and p not in ('无人', '无'):
self.db.upsert_person(p, source='llm')
logger.info(f"[video_id={video_id}] 已落库: summary={len(summary)}字, "
f"events={len(norm_events)}, people={people}")

89
fam-edge/src/fam_edge/watch_processor.py Normal file
View File

@@ -0,0 +1,89 @@
"""
WatchProcessor - 监听 Google 硬盘同步落地目录，处理新视频
流程:
1. rclone 已把 Google 硬盘目录实时同步到 local_dir（视频文件）
2. 每 watch_interval_sec 轮询一次 local_dir
3. 发现未在 videos 表登记的文件 -> ensure_video 登记
4. 取 pending/failed 的视频，逐个整视频分析（max_concurrent=1，串行避免过载）
"""
import os
import time
import threading
from typing import List
from .logger import setup_logger
from .config_loader import load_config
from . import oracle_db
from .video_processor import VideoProcessor
logger = setup_logger('fam-edge.watch_processor')
VIDEO_EXTS = ('.mp4', '.mkv', '.avi', '.mov', '.ts')
class WatchProcessor:
def __init__(self, db: oracle_db.OracleDB):
self.config = load_config()
self.db = db
self.local_dir = self.config.get('gdrive_sync', {}).get('local_dir', '/opt/fam-edge/gdrive_videos')
self.interval = self.config.get('gdrive_sync', {}).get('watch_interval_sec', 30)
self.camera_name = self.config.get('gdrive_sync', {}).get('camera_name', '摄像头')
self.max_concurrent = self.config.get('video_processing', {}).get('max_concurrent', 1)
self.processor = VideoProcessor(db)
self._running = False
self._thread = None
def _scan_files(self) -> List[str]:
if not os.path.isdir(self.local_dir):
logger.warning(f"监听目录不存在: {self.local_dir}")
return []
out = []
for fn in sorted(os.listdir(self.local_dir)):
if fn.lower().endswith(VIDEO_EXTS):
out.append(fn)
return out
def _register_new(self, files: List[str]):
for fn in files:
if self.db.get_video_by_filename(fn) is None:
path = os.path.join(self.local_dir, fn)
self.db.ensure_video(fn, path, camera_name=self.camera_name)
logger.info(f"登记新视频: {fn}")
def _process_pending(self):
pending = self.db.get_pending_videos(limit=self.max_concurrent)
for row in pending:
try:
self.processor.process_video(row['id'], row['filename'], row['local_path'])
except Exception as e:
logger.error(f"处理视频 {row['filename']} 异常: {e}", exc_info=True)
self.db.mark_video_failed(row['id'], f"watch_error: {e}")
def _run(self):
logger.info(f"WatchProcessor 启动，监听 {self.local_dir}，间隔 {self.interval}s")
while self._running:
try:
files = self._scan_files()
self._register_new(files)
self._process_pending()
except Exception as e:
logger.error(f"WatchProcessor 轮询异常: {e}", exc_info=True)
# 处理完一小批后休眠
for _ in range(self.interval):
if not self._running:
break
time.sleep(1)
def start(self):
if self._running:
return
self._running = True
self._thread = threading.Thread(target=self._run, daemon=True, name='watch')
self._thread.start()
def is_alive(self):
return self._thread is not None and self._thread.is_alive()
def stop(self):
self._running = False