diff --git a/fam-edge/config/config.yaml b/fam-edge/config/config.yaml index e8e2e9a..ece3294 100644 --- a/fam-edge/config/config.yaml +++ b/fam-edge/config/config.yaml @@ -1,78 +1,58 @@ -# FAM-Edge 配置文件 (Oracle 端) - 多模型池配置 -# Tailscale: Oracle=100.74.137.126, NAS=100.70.234.39 +# FAM-Edge 配置文件 (Oracle 端) - 新架构 v2 # -# 异步队列模式: -# NAS 上传视频 → /api/edge/video/enqueue 入 SQLite 队列 → 消费者线程异步处理 -# → NAS Poller 从 /api/edge/results 拉取结果 -# 速率限制: Gemini 1000RPM x2 burst, NVIDIA 40RPM x2 burst +# 新架构（2026-08-21 重构）: +# 1. 不再切片/抽帧：整视频直传云端 VLM（Gemini 用 Files API，NVIDIA 用整视频 video_url） +# 2. 视频来源：rclone 从 Google 硬盘实时同步到本地 local_dir，监听目录处理新视频 +# 3. Oracle 自建 SQLite 库存储所有视频摘要/事件/人物，并对外提供同步接口供 NAS 拉取 +# 4. 独立 person_service 汇总全量人物 -> LLM 合并为规范人物表 -> 回灌视频提示 +# 5. NAS 仅作管理后台，每 30 分钟从甲骨文拉增量镜像到本地 MariaDB -# NAS 端回调地址（旧 webhook 模式保留，异步模式不使用） -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 端服务 +# Oracle 端 HTTP 服务 server: host: "0.0.0.0" port: 5000 max_concurrent_tasks: 1 -# 异步任务队列 -queue: - db_path: "/opt/fam-edge/data/fam_queue.db" - upload_dir: "/tmp/fam_uploads" - poll_interval: 10 # 消费者轮询间隔（秒） - # API 速率限制 (RPM)，burst_factor=2 表示突发容量为 2 倍 RPM - rate_limit: - gemini_rpm: 1000 - nvidia_rpm: 40 - burst_factor: 2 +# Google 硬盘同步（rclone 负责同步落地，本段仅描述监听行为） +gdrive_sync: + enabled: true + local_dir: "/opt/fam-edge/gdrive_videos" # rclone 同步落地目录（video_processing 监听此目录） + watch_interval_sec: 30 # 监听新视频的轮询间隔 + camera_name: "客厅" # 摄像头名称（注入视频提示） + # 文件名解析开始时间：监控文件名含时间戳时使用（如 2026-08-21_081500.mp4） + parse_start_from_filename: true -# 编排调度模式: fallback(顺序降级, 默认) | ensemble(并行交叉验证) -orchestrator: - mode: "fallback" - overall_timeout: 600 +# Oracle 本地库（视频摘要/事件/人物） +oracle_db: + path: "/opt/fam-edge/data/oracle.db" -# 关键帧筛选参数（自适应：帧数随视频时长动态计算） -video: - candidate_per_minute: 2 # 每分钟粗抽候选帧数 - 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 +# NAS 拉取同步接口鉴权 token（与 NAS oracle_sync.token 一致） +sync_api: + token: "${ORACLE_SYNC_TOKEN}" -# 超时（秒） -timeout: - download: 60 - vlm_visual: 600 - vlm_fusion: 300 - callback: 30 - overall: 1800 +# 人物识别服务 +person_service: + enabled: true + schedule_interval_sec: 1800 # 每 30 分钟重新汇总一次人物 + model: "gemini" # 用哪个模型做人物合并（vision 模型也支持纯文本） -# 多模型池配置（新框架：本地大模型不参与视频分析，仅智能问答兜底） -# -# 视频分析链路（推送模式）: -# 云端 VLM 直接产出结构化 JSON (global_summary / entities_json / frame_details) -# -> Edge 仅做格式化/校验 (format_cloud_result) -> 直接回写 NAS，无本地融合步骤 -# 视觉角色: Gemini(主) -> NVIDIA NIM(备) 顺序降级; 两云端全失败 -> 任务 FAILED 走重试 -# -# 智能问答链路: -# Gemini -> NVIDIA -> 本地 Ollama (仅当两云端都失败才启用本地兜底) +# 视频处理 +video_processing: + max_concurrent: 1 + timeout: 900 # 单视频分析超时（整视频上云较慢） + # 降级顺序：先 gemini 整视频，失败再 nvidia 整视频；两者都失败 -> 标记 failed + vision_order: ["gemini", "nvidia"] + +# 智能问答降级链（与视频分析独立）：Gemini -> NVIDIA -> 本地 Ollama models: - provider: "gemini" role: "vision" enabled: true model_name: "gemini-flash-latest" # 主模型（每日免费配额 20 请求，按模型独立） - fallback_models: # 429 配额耗尽/503 过载时依次切换 + fallback_models: - "gemini-flash-lite-latest" api_key: "${GEMINI_API_KEY}" - timeout: 90 + timeout: 600 circuit_breaker: enabled: true threshold: 5 @@ -81,16 +61,17 @@ models: - provider: "nvidia" role: "vision" 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" api_key: "${NVIDIA_API_KEY}" - timeout: 120 + timeout: 600 circuit_breaker: enabled: true threshold: 5 cooldown: 300 - # 本地模型：纯文本 qwen2.5:7b，仅参与智能问答，作为 Gemini/NVIDIA 都失败时的兜底 + # 本地模型：纯文本 qwen2.5:7b，仅参与智能问答兜底 - provider: "ollama" role: "text" usage: "qa_fallback" diff --git a/fam-edge/src/fam_edge/ai_orchestrator/orchestrator.py b/fam-edge/src/fam_edge/ai_orchestrator/orchestrator.py deleted file mode 100644 index 42fa8d8..0000000 --- a/fam-edge/src/fam_edge/ai_orchestrator/orchestrator.py +++ /dev/null @@ -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) - } diff --git a/fam-edge/src/fam_edge/api_gateway/api_gateway.py b/fam-edge/src/fam_edge/api_gateway/api_gateway.py index d636d22..0a41f76 100644 --- a/fam-edge/src/fam_edge/api_gateway/api_gateway.py +++ b/fam-edge/src/fam_edge/api_gateway/api_gateway.py @@ -1,517 +1,103 @@ """ -API-Gateway - Flask 蓝图，接收任务 +API-Gateway - Flask 蓝图（新架构 v2） -模式: -1. enqueue (异步): NAS 上传视频 → Edge 入队 → 立即返回 → 消费者异步处理 → NAS 轮询拉取结果 -2. push (同步, 兼容保留): NAS 上传 → Edge 同步处理 → 结果随响应返回 -3. analyze (旧拉取模式, 兼容保留) +端点: + GET /api/oracle/sync NAS 每 30 分钟拉取增量（since + token 校验） + POST /api/oracle/people/correct NAS 推送手动命名校正（label -> canonical_name） + POST /api/edge/chat/ask 智能问答编排（Gemini -> NVIDIA -> Ollama） + GET /health 健康检查 + +已移除（旧推送/分块/队列模式）: /video/push, /enqueue, /chunk, /assemble, +/results, /queue/stats, /mark_frames """ import os -import base64 -import threading -import requests + from flask import Blueprint, request, jsonify from ..logger import setup_logger -from ..ai_orchestrator.orchestrator import AIOrchestrator -from ..video_preprocessor.preprocessor import VideoPreprocessor -from ..queue import queue_manager +from .. import state +from ..qa import QAOrchestrator logger = setup_logger('fam-edge.api_gateway') api_bp = Blueprint('api_gateway', __name__) -_current_task_lock = threading.Lock() -_currently_processing = False - -_orchestrator = None +_qa = None -def get_orchestrator(): - global _orchestrator - if _orchestrator is None: - _orchestrator = AIOrchestrator() - return _orchestrator +def get_qa(): + global _qa + if _qa is None: + _qa = QAOrchestrator() + return _qa -@api_bp.route('/api/edge/video/analyze', methods=['POST']) -def receive_task(): - """接收分析任务""" - global _currently_processing +def _check_token() -> bool: + expected = _sync_token() + token = request.args.get('token') or request.form.get('token') or \ + (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) if not data: return jsonify({"error": "Invalid JSON"}), 400 - - task_id = data.get('task_id') - video_url = data.get('video_url') - webhook_url = data.get('webhook_url') - - 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 - + label = (data.get('label') or '').strip() + canonical = (data.get('canonical_name') or '').strip() + if not label or not canonical: + return jsonify({"error": "缺少 label / canonical_name"}), 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', '') - - 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 - + state.get_db().set_canonical(label, canonical, source='manual') except Exception as e: - logger.error(f"[task_id={task_id}] 入队失败: {e}", exc_info=True) - if os.path.exists(video_path): - os.remove(video_path) + logger.error(f"people_correct 异常: {e}") return jsonify({"error": str(e)}), 500 - - -# ========== 分块上传（断点续传）========== - -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 + return jsonify({"status": "ok", "label": label, "canonical_name": canonical}), 200 @api_bp.route('/api/edge/chat/ask', methods=['POST']) def chat_ask(): """智能问答编排：Gemini → NVIDIA → 本地 Ollama（两云端都失败才用本地兜底） - 请求: {"prompt": "..."} + 请求: {"prompt": "...", "max_tokens": 512} 响应: {"answer": "...", "provider": "gemini"|"nvidia"|"ollama"} """ data = request.get_json(silent=True) @@ -519,12 +105,24 @@ def chat_ask(): return jsonify({"error": "缺少必填字段: prompt"}), 400 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: return jsonify({ "error": "所有模型均不可用（Gemini / NVIDIA / Ollama 全部失败）" }), 503 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 diff --git a/fam-edge/src/fam_edge/app.py b/fam-edge/src/fam_edge/app.py index 5e62ccd..fb58059 100644 --- a/fam-edge/src/fam_edge/app.py +++ b/fam-edge/src/fam_edge/app.py @@ -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 sys @@ -12,7 +15,9 @@ sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) from .config_loader import load_config from .logger import setup_logger 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') @@ -22,17 +27,24 @@ app.register_blueprint(api_bp) @app.route('/', methods=['GET']) 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: - _consumer = get_consumer() - _consumer.start() - logger.info("Queue-Consumer 已启动") + db = state.get_db() + _watch = WatchProcessor(db) + _watch.start() + logger.info("WatchProcessor 已启动") + + _person = PersonService(db) + _person.start() + logger.info("PersonService 已启动") except Exception as e: - logger.error(f"Queue-Consumer 启动失败: {e}") + logger.error(f"后台服务启动失败: {e}", exc_info=True) if __name__ == '__main__': diff --git a/fam-edge/src/fam_edge/frame_marker.py b/fam-edge/src/fam_edge/frame_marker.py deleted file mode 100644 index 021648a..0000000 --- a/fam-edge/src/fam_edge/frame_marker.py +++ /dev/null @@ -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) diff --git a/fam-edge/src/fam_edge/model_adapters/base_adapter.py b/fam-edge/src/fam_edge/model_adapters/base_adapter.py index c7e9eaf..7518cb7 100644 --- a/fam-edge/src/fam_edge/model_adapters/base_adapter.py +++ b/fam-edge/src/fam_edge/model_adapters/base_adapter.py @@ -3,13 +3,21 @@ 新增模型只需继承此类并实现方法: 1. health_check() -> bool -2. analyze_frames(frame_paths, frame_timestamps, known_members_context) -> Optional[dict] - - 视觉分析：输入帧图片路径 + 时间戳 + 成员清单，直接输出**结构化结果 dict** - (含 frame_details 等，详见 format_cloud_result 约定)。 - - 失败/超时返回 None。 +2. analyze_video(video_path, known_members_context, event_start_time) -> Optional[dict] + - 整视频分析：直接把完整视频交给云端 VLM（本地不切片、不抽帧）。 + - 模型内部自行采样帧，输出结构化结果 dict。失败/超时返回 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] - 纯文本问答（智能问答场景），返回文本或 None。 - - 默认实现抛 NotImplementedError；文本/视觉模型按需实现。 4. get_timeout() -> int 5. get_circuit_breaker() -> CircuitBreaker """ @@ -36,22 +44,13 @@ class BaseModelAdapter(ABC): pass @abstractmethod - def analyze_frames(self, frame_paths: List[str], - frame_timestamps: List[str], - known_members_context: str) -> Optional[Dict]: - """视觉分析：输入帧图片路径 + 时间戳 + 成员清单， - 直接输出结构化结果 dict（含 frame_details 等）。失败/超时返回 None。 + def analyze_video(self, video_path: str, + known_members_context: str, + event_start_time: str = '') -> Optional[Dict]: + """整视频分析：把完整视频交给云端 VLM，输出结构化结果 dict。 - 约定返回结构（云端模型直出，Edge 仅做格式化校验，不再本地融合）: - { - "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]} - ] - } + 本地不切片、不抽帧；模型内部自行采样帧。 + 失败/超时返回 None。 """ pass diff --git a/fam-edge/src/fam_edge/model_adapters/gemini_adapter.py b/fam-edge/src/fam_edge/model_adapters/gemini_adapter.py index 6a99e48..49ff714 100644 --- a/fam-edge/src/fam_edge/model_adapters/gemini_adapter.py +++ b/fam-edge/src/fam_edge/model_adapters/gemini_adapter.py @@ -2,15 +2,16 @@ GeminiAdapter - Google Gemini 云端 VLM 适配器 provider_name = "gemini" -模型: gemini-flash-latest (v1beta 下 gemini-1.5-flash 会 404，用 flash-latest 别名) -角色: vision (视觉分析直出结构化 JSON) + 智能问答 +模型: gemini-flash-latest +角色: vision (整视频直出结构化 JSON) + 智能问答 健康检查: GET /v1beta/models?key=... 熔断器: 启用 -视觉分析: 多图单请求直出结构化 JSON（global_summary/entities_json/frame_details） +整视频分析: 用 Files API 上传完整视频 -> generateContent 直出结构化 JSON + （本地不切片、不抽帧；Gemini 原生支持长视频） """ import os import time -import base64 +import json import requests from typing import Dict, List, Optional @@ -23,16 +24,15 @@ logger = setup_logger('fam-edge.gemini_adapter') class GeminiAdapter(BaseModelAdapter): - """Gemini 云端 VLM 适配器 (视觉直出结构化 JSON + 文本问答)""" + """Gemini 云端 VLM 适配器 (整视频直出结构化 JSON + 文本问答)""" def __init__(self, config: dict): super().__init__("gemini", config) self.model_name = config.get('model_name', 'gemini-flash-latest') - # 免费层配额按模型独立（20 请求/天/模型），fallback 链用于跨模型借用配额 self.model_chain = [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.timeout = config.get('timeout', 30) + self.timeout = config.get('timeout', 600) cb_cfg = config.get('circuit_breaker', {}) self._cb = CircuitBreaker( threshold=cb_cfg.get('threshold', 3), @@ -69,76 +69,131 @@ class GeminiAdapter(BaseModelAdapter): return False # ------------------------------------------------------------------ - # 视觉分析：多图单请求，直出结构化 JSON + # 整视频分析：Files API 上传 -> generateContent # ------------------------------------------------------------------ - def analyze_frames(self, frame_paths: List[str], - frame_timestamps: List[str], - known_members_context: str) -> Optional[Dict]: + def analyze_video(self, video_path: str, + known_members_context: str, + event_start_time: str = '') -> Optional[Dict]: if self._cb.is_open(): - logger.warning("Gemini 熔断器 OPEN，跳过调用") + logger.warning("Gemini 熔断器 OPEN，跳过视频分析") return None if not self.api_key: - logger.warning("Gemini API Key 未配置，跳过调用") + logger.warning("Gemini API Key 未配置，跳过视频分析") return None - if not frame_paths: - logger.warning("Gemini 无帧可分析") + if not os.path.isfile(video_path): + logger.warning(f"Gemini 视频文件不存在: {video_path}") return None - parts = [] - ts_map = {} - for i, (path, ts) in enumerate(zip(frame_paths, frame_timestamps), 1): - 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: + file_uri = self._upload_file(video_path) + if not file_uri: + self._cb.record_failure() 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: - 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: self._cb.record_failure() return None try: result = parse_vlm_json(text) - # 确保 frame_details 的 frame_timestamp 与标注一致 - for f in result.get('frame_details', []): - idx = f.get('frame_index') - if isinstance(idx, int) and idx in ts_map and not f.get('frame_timestamp'): - f['frame_timestamp'] = ts_map[idx] - for f in result.get('frame_details', []): - if 'source_providers' not in f or not f.get('source_providers'): - f['source_providers'] = ['gemini'] + result = self._normalize(result) + if not result or 'events' not in result: + logger.error(f"Gemini 视频输出缺少 events: {text[:150]}") + self._cb.record_failure() + return None + result['compute_provider'] = 'gemini' 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 except VLMOutputInvalidError as e: - logger.error(f"Gemini 输出无法解析为 JSON: {e}") + logger.error(f"Gemini 视频输出无法解析为 JSON: {e}") self._cb.record_failure() return None except requests.Timeout: - logger.warning(f"Gemini 视觉分析超时 ({self.timeout}s)") + logger.warning(f"Gemini 视频分析超时 ({self.timeout}s)") self._cb.record_failure() + return None except Exception as e: - logger.error(f"Gemini 视觉分析异常: {e}") + logger.error(f"Gemini 视频分析异常: {e}") 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 - def _generate(self, parts: List[dict], max_tokens: int, - temperature: float) -> Optional[str]: - """带模型 fallback 链的 generateContent 调用 + def _delete_file(self, file_uri: str): + if not file_uri or 'files/' not in file_uri: + 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（每日免费配额耗尽，按模型独立）→ 立即换下一个模型，不重试 - - 503（模型过载，临时性）→ 同模型退避 3s 重试一次，仍失败换下一个 - """ + def _generate_video(self, file_uri: str, prompt: str, + 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 attempt in range(2): try: @@ -146,14 +201,15 @@ class GeminiAdapter(BaseModelAdapter): f"{self._base_url}/models/{model}:generateContent?key={self.api_key}", json={"contents": [{"parts": parts}], "generationConfig": { - "temperature": temperature, "maxOutputTokens": max_tokens}}, + "temperature": temperature, + "maxOutputTokens": max_tokens}}, timeout=self.timeout ) except requests.Timeout: - logger.warning(f"Gemini [{model}] 请求超时 ({self.timeout}s)") + logger.warning(f"Gemini [{model}] 视频请求超时 ({self.timeout}s)") break except Exception as e: - logger.error(f"Gemini [{model}] 请求异常: {e}") + logger.error(f"Gemini [{model}] 视频请求异常: {e}") break if resp.status_code == 200: @@ -164,55 +220,76 @@ class GeminiAdapter(BaseModelAdapter): ).strip() if cands else '' if text: if model != self.model_name: - logger.info(f"Gemini 主模型不可用，由 fallback 模型 [{model}] 出结果") + logger.info(f"Gemini 主模型不可用，由 fallback [{model}] 出结果") return text logger.warning(f"Gemini [{model}] 返回空文本") continue - detail = resp.text[:150].replace('\n', ' ') if resp.status_code == 429: - logger.warning(f"Gemini [{model}] 429 每日免费配额耗尽，切换下一模型") + logger.warning(f"Gemini [{model}] 429 配额耗尽，切换下一模型") break if resp.status_code == 503: if attempt == 0: logger.warning(f"Gemini [{model}] 503 过载，3s 后重试") time.sleep(3) continue - logger.warning(f"Gemini [{model}] 503 重试仍失败，切换下一模型") break logger.warning(f"Gemini [{model}] HTTP {resp.status_code}: {detail}") break return None - def _build_structured_prompt(self, known_members: str) -> str: - return f"""你是家庭监控视频分析助手。下面按时间顺序排列了多张监控截图。 -请分析整个时段，只输出合法 JSON（不要 markdown、不要任何解释文字），结构如下： + @staticmethod + def _normalize(result: dict) -> dict: + """统一字段名：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 句，客观描述人物与主要活动", - "entities_json": [ - {{"person": "人物标识(匹配已知成员用真名，否则用'人物A'/'人物B'...)", "action": "主要动作", "clothing": "衣着"}} - ], - "frame_details": [ + "events": [ {{ - "frame_index": 图片序号(从1开始，与[图片N]标注对应), - "frame_timestamp": "该帧的时间戳(用[图片N]标注里的时间)", - "person": "该帧画面中的人物或'无人'", - "action": "该帧可见动作", - "clothing": "该帧衣着(颜色+类型)", - "is_attention_event": false, - "source_providers": ["gemini"] + "timestamp": "事件发生时的绝对北京时间(格式 YYYY-MM-DD HH:MM:SS)", + "description": "该时间点的画面/动作信息摘要（谁、在做什么、位置）", + "people": ["出现在该时刻的人物，用已知成员真名或'人物A'/'人物B'"], + "is_attention_event": false }} - ] -}} + ], + "people_mentioned": ["本视频出现过的所有人物标识/真名"] +}}{start_hint} 规则： 1. 只描述客观画面，不要猜测或想象。 -2. frame_details 每帧一条，frame_index 与上方[图片N]序号对应，frame_timestamp 用标注时间。 +2. events 提取视频中"有意义的时间点"（人物出现/动作变化/异常），不要逐秒罗列；timestamp 用绝对北京时间。 3. 已知家庭成员（按特征匹配，匹配到用 real_name，否则用"人物X"）： {known_members or '（暂无已知成员）'} 4. is_attention_event：是否为跌倒、危险、异常哭闹等需关注事件（没有则为 false）。 -5. 没有人物出现的帧 person 填"无人"，action 填""。""" +5. 没有人物出现的时段不要单独成 event；people 留空数组。""" # ------------------------------------------------------------------ # 智能问答：纯文本 @@ -222,11 +299,42 @@ class GeminiAdapter(BaseModelAdapter): logger.warning("Gemini API Key 未配置，跳过问答") return None 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: logger.error(f"Gemini 问答异常: {e}") 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: return self.timeout diff --git a/fam-edge/src/fam_edge/model_adapters/nvidia_adapter.py b/fam-edge/src/fam_edge/model_adapters/nvidia_adapter.py index 7898b29..6542d30 100644 --- a/fam-edge/src/fam_edge/model_adapters/nvidia_adapter.py +++ b/fam-edge/src/fam_edge/model_adapters/nvidia_adapter.py @@ -2,23 +2,15 @@ NvidiaVisionAdapter - NVIDIA NIM 云端 VLM 适配器 provider_name = "nvidia" -模型: nvidia/nemotron-3-nano-omni-30b-a3b-reasoning (Omni, 原生视频输入) -角色: vision (视觉分析直出结构化 JSON) + 智能问答 +模型: nvidia/nemotron-nano-12b-v2-vl（NIM 官方支持整视频 video_url 输入，内部自行采样帧） +角色: vision (整视频直出结构化 JSON) + 智能问答 SDK: openai (NIM 兼容 OpenAI API 规范) - -视频模式 (analyze_video): 按关键帧时间点截取 ±1.5s 片段拼接集锦视频 -(片段左上角叠加原始时间戳)，base64 后经 video_url 单次调用 — -模型看到动态画面而非静态帧，动作/轨迹识别显著优于逐帧图片。 - -图片模式 (analyze_frames): 逐帧 image_url 调用（无视频文件时的降级路径）。 -注意: nemotron-omni 是 reasoning 模型，max_tokens 需给足（reasoning 消耗 token）。 +整视频分析: 整视频 base64 经 video_url 单次调用 —— 本地不切片、不抽帧 """ import os import base64 import json import re -import subprocess -import tempfile from typing import Dict, List, Optional from .base_adapter import BaseModelAdapter @@ -32,20 +24,17 @@ try: except ImportError: OpenAI = None -VIDEO_SEGMENT_PAD = 1.5 # 关键帧前后各截取秒数 -HIGHLIGHT_WIDTH = 640 # 集锦视频宽度（保持宽高比） - class NvidiaVisionAdapter(BaseModelAdapter): - """NVIDIA NIM 云端 VLM 适配器 (视频集锦单次调用; 逐帧降级; 文本问答)""" + """NVIDIA NIM 云端 VLM 适配器 (整视频单次调用; 文本问答)""" def __init__(self, config: dict): super().__init__("nvidia", config) 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.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', {}) self._cb = CircuitBreaker( threshold=cb_cfg.get('threshold', 3), @@ -78,127 +67,29 @@ class NvidiaVisionAdapter(BaseModelAdapter): 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, - frame_timestamps: List[str], - known_members_context: str, - event_start_time: str = '') -> Optional[Dict]: - """原生视频输入分析: 集锦片段 -> video_url 单次调用""" + known_members_context: str, + event_start_time: str = '') -> Optional[Dict]: if self._cb.is_open(): logger.warning("NVIDIA 熔断器 OPEN，跳过视频分析") return None if self._client is None: logger.warning("NVIDIA 客户端未初始化，跳过视频分析") return None - - highlight = self._build_highlight_video( - video_path, frame_timestamps, event_start_time) - if not highlight: - logger.warning("NVIDIA 集锦视频不可用，降级逐帧模式") + if not os.path.isfile(video_path): + logger.warning(f"NVIDIA 视频文件不存在: {video_path}") return None + try: - with open(highlight, 'rb') as f: + with open(video_path, 'rb') as f: b64 = base64.b64encode(f.read()).decode('utf-8') except Exception as e: - logger.warning(f"NVIDIA 读取集锦视频失败: {e}") + logger.warning(f"NVIDIA 读取视频失败: {e}") 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: resp = self._client.chat.completions.create( model=self.model_name, @@ -208,134 +99,47 @@ class NvidiaVisionAdapter(BaseModelAdapter): "url": f"data:video/mp4;base64,{b64}"}} ]}], temperature=0.2, - max_tokens=3072, + max_tokens=4096, + # NIM 扩展：控制视频采样帧数（模型上限 128 帧） + extra_body={"media_io_kwargs": {"video": {"num_frames": 128}}}, timeout=self.timeout ) content = resp.choices[0].message.content if not content: logger.warning("NVIDIA 视频分析返回空 content") + self._cb.record_failure() return None - data = self._parse_single_frame_json(content) - if not data or 'frame_details' not in data: + data = self._parse_json(content) + if not data or 'events' not in data: logger.warning(f"NVIDIA 视频 JSON 解析失败: {content[:150]}") - return None - frame_details = self._normalize_frame_details(data, frame_timestamps) - if not frame_details: + self._cb.record_failure() return None self._cb.record_success() - logger.info(f"NVIDIA 视频分析完成，frame_details={len(frame_details)}") - result = {"frame_details": frame_details} - if data.get('global_summary'): - result['global_summary'] = str(data['global_summary']) - if data.get('entities_json'): - result['entities_json'] = data['entities_json'] - return result + logger.info(f"NVIDIA 整视频分析完成，events={len(data.get('events', []))}") + return { + "global_summary": str(data.get('global_summary', '')), + "events": data.get('events', []), + "people_mentioned": data.get('people_mentioned', []), + "compute_provider": "nvidia", + } except Exception as e: self._cb.record_failure() logger.warning(f"NVIDIA 视频分析异常: {e}") return None - def _normalize_frame_details(self, data: dict, - frame_timestamps: List[str]) -> List[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，仅提取字段）""" + @staticmethod + def _parse_json(content: str) -> Optional[dict]: content = content.strip() - # 直接解析 try: return json.loads(content) except json.JSONDecodeError: pass - # 提取 markdown fence fence = re.search(r'```(?:json)?\s*(\{.*?\})\s*```', content, re.DOTALL) if fence: try: return json.loads(fence.group(1)) except json.JSONDecodeError: pass - # 贪婪匹配最大 {...} brace = re.search(r'\{.*\}', content, re.DOTALL) if brace: try: @@ -344,68 +148,35 @@ class NvidiaVisionAdapter(BaseModelAdapter): pass return None - def _analyze_one_structured(self, path: str, ts: str, idx: int, - known_members: str) -> Optional[Dict]: - try: - with open(path, 'rb') as f: - b64 = base64.b64encode(f.read()).decode('utf-8') - except Exception as e: - 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、不要解释），结构如下： + 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、不要解释），结构如下： {{ - "frame_timestamp": "{ts}", - "person": "该帧画面中的人物或'无人'", - "action": "该帧可见动作", - "clothing": "该帧衣着(颜色+类型)", - "is_attention_event": false -}} + "global_summary": "整个时段的整体摘要，简体中文，2-4 句", + "events": [ + {{ + "timestamp": "事件发生时的绝对北京时间(YYYY-MM-DD HH:MM:SS)", + "description": "该时刻画面/动作信息摘要", + "people": ["出现在该时刻的人物，用已知成员真名或'人物A'"], + "is_attention_event": false + }} + ], + "people_mentioned": ["本视频出现过的所有人物标识/真名"] +}}{start_hint} 规则： 1. 只描述客观画面，不猜测。 -2. 已知家庭成员（按特征匹配，匹配到用 real_name，否则用"人物X"）： +2. events 提取有意义的时间点（人物出现/动作变化/异常），timestamp 用绝对北京时间。 +3. 已知家庭成员（按特征匹配，匹配到用 real_name，否则用"人物X"）： {known_members or '（暂无已知成员）'} -3. is_attention_event：是否为跌倒、危险、异常哭闹等需关注事件（没有则为 false）。 -4. 没有人物出现的帧 person 填"无人"，action 填""。""" +4. is_attention_event：跌倒、危险、异常哭闹等需关注事件（没有则为 false）。""" # ------------------------------------------------------------------ - # 智能问答：纯文本（reasoning 模型，max_tokens 需给足） + # 智能问答：纯文本 # ------------------------------------------------------------------ def chat(self, prompt: str, max_tokens: int = 2048) -> Optional[str]: if self._client is None: diff --git a/fam-edge/src/fam_edge/model_adapters/ollama_adapter.py b/fam-edge/src/fam_edge/model_adapters/ollama_adapter.py index b9664b6..44984ba 100644 --- a/fam-edge/src/fam_edge/model_adapters/ollama_adapter.py +++ b/fam-edge/src/fam_edge/model_adapters/ollama_adapter.py @@ -111,6 +111,13 @@ class OllamaAdapter(BaseModelAdapter): self._cb.record_failure() 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: return self.timeout diff --git a/fam-edge/src/fam_edge/oracle_db.py b/fam-edge/src/fam_edge/oracle_db.py new file mode 100644 index 0000000..1145fab --- /dev/null +++ b/fam-edge/src/fam_edge/oracle_db.py @@ -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() diff --git a/fam-edge/src/fam_edge/person_service.py b/fam-edge/src/fam_edge/person_service.py new file mode 100644 index 0000000..b0797d6 --- /dev/null +++ b/fam-edge/src/fam_edge/person_service.py @@ -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() diff --git a/fam-edge/src/fam_edge/qa.py b/fam-edge/src/fam_edge/qa.py new file mode 100644 index 0000000..9076b50 --- /dev/null +++ b/fam-edge/src/fam_edge/qa.py @@ -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 diff --git a/fam-edge/src/fam_edge/queue/__init__.py b/fam-edge/src/fam_edge/queue/__init__.py deleted file mode 100644 index 9f104b8..0000000 --- a/fam-edge/src/fam_edge/queue/__init__.py +++ /dev/null @@ -1,5 +0,0 @@ -""" -SQLite 异步任务队列 (Oracle 端) -""" -from . import queue_manager -from .consumer import get_consumer diff --git a/fam-edge/src/fam_edge/queue/consumer.py b/fam-edge/src/fam_edge/queue/consumer.py deleted file mode 100644 index 8e19bc9..0000000 --- a/fam-edge/src/fam_edge/queue/consumer.py +++ /dev/null @@ -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 diff --git a/fam-edge/src/fam_edge/queue/queue_manager.py b/fam-edge/src/fam_edge/queue/queue_manager.py deleted file mode 100644 index fec169f..0000000 --- a/fam-edge/src/fam_edge/queue/queue_manager.py +++ /dev/null @@ -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() diff --git a/fam-edge/src/fam_edge/rate_limiter.py b/fam-edge/src/fam_edge/rate_limiter.py deleted file mode 100644 index 24a9ec1..0000000 --- a/fam-edge/src/fam_edge/rate_limiter.py +++ /dev/null @@ -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) diff --git a/fam-edge/src/fam_edge/state.py b/fam-edge/src/fam_edge/state.py new file mode 100644 index 0000000..61c0059 --- /dev/null +++ b/fam-edge/src/fam_edge/state.py @@ -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 diff --git a/fam-edge/src/fam_edge/storage_cleaner/__init__.py b/fam-edge/src/fam_edge/storage_cleaner/__init__.py deleted file mode 100644 index ba7e140..0000000 --- a/fam-edge/src/fam_edge/storage_cleaner/__init__.py +++ /dev/null @@ -1,4 +0,0 @@ -"""Storage-Cleaner 包""" -from .cleaner import StorageCleaner - -__all__ = ["StorageCleaner"] diff --git a/fam-edge/src/fam_edge/storage_cleaner/cleaner.py b/fam-edge/src/fam_edge/storage_cleaner/cleaner.py deleted file mode 100644 index fe83957..0000000 --- a/fam-edge/src/fam_edge/storage_cleaner/cleaner.py +++ /dev/null @@ -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}") diff --git a/fam-edge/src/fam_edge/video_preprocessor/__init__.py b/fam-edge/src/fam_edge/video_preprocessor/__init__.py deleted file mode 100644 index 9a88b97..0000000 --- a/fam-edge/src/fam_edge/video_preprocessor/__init__.py +++ /dev/null @@ -1,4 +0,0 @@ -"""Video-Preprocessor 包""" -from .preprocessor import VideoPreprocessor - -__all__ = ["VideoPreprocessor"] diff --git a/fam-edge/src/fam_edge/video_preprocessor/preprocessor.py b/fam-edge/src/fam_edge/video_preprocessor/preprocessor.py deleted file mode 100644 index ec74d80..0000000 --- a/fam-edge/src/fam_edge/video_preprocessor/preprocessor.py +++ /dev/null @@ -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}") diff --git a/fam-edge/src/fam_edge/video_processor.py b/fam-edge/src/fam_edge/video_processor.py new file mode 100644 index 0000000..2b72afd --- /dev/null +++ b/fam-edge/src/fam_edge/video_processor.py @@ -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}") diff --git a/fam-edge/src/fam_edge/watch_processor.py b/fam-edge/src/fam_edge/watch_processor.py new file mode 100644 index 0000000..c79602f --- /dev/null +++ b/fam-edge/src/fam_edge/watch_processor.py @@ -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