diff --git a/fam-edge/config/config.yaml.example b/fam-edge/config/config.yaml.example new file mode 100644 index 0000000..7a70613 --- /dev/null +++ b/fam-edge/config/config.yaml.example @@ -0,0 +1,75 @@ +# FAM-Edge 配置文件 (Oracle 端) +# 复制此文件为 config.yaml 并修改实际值 + +# NAS 端回调地址 +nas: + webhook_url: "http://100.x.x.10:8000/api/core/callback/event" + media_base_url: "http://100.x.x.10:8000/media" + media_token: "xxx" + +# Oracle 端服务 +server: + host: "0.0.0.0" + port: 5000 + max_concurrent_tasks: 1 + +# 关键帧筛选参数 +video: + candidate_frames: 30 # 粗抽候选帧数 + min_key_frames: 5 # 最少关键帧 + max_key_frames: 8 # 最多关键帧 + mse_threshold: 500 # 帧差阈值 + jpeg_quality: 80 + max_long_edge: 1024 + +# 超时（秒） +timeout: + download: 60 + vlm_visual: 240 # 单模型视觉分析超时 + vlm_fusion: 120 + callback: 30 + overall: 600 + +# 模型清单（可扩展，新增模型只需在此数组加一项 + 实现适配器） +models: + - provider: "ollama" + enabled: true + model_name: "llava-phi3" + base_url: "http://localhost:11434" + timeout: 240 + circuit_breaker: + enabled: false # 本地模型不启用熔断 + threshold: 5 + cooldown: 900 + + - provider: "gemini" + enabled: true + model_name: "gemini-1.5-flash" + api_key: "${GEMINI_API_KEY}" # 从环境变量读取 + timeout: 8 + circuit_breaker: + enabled: true + threshold: 5 + cooldown: 900 + + # v1.1 扩展示例（取消注释并填入 API Key 即启用） + # - provider: "openai" + # enabled: false + # model_name: "gpt-4o" + # api_key: "${OPENAI_API_KEY}" + # timeout: 30 + # circuit_breaker: + # enabled: true + # threshold: 5 + # cooldown: 900 + + # - provider: "nvidia" + # enabled: false + # model_name: "nvidia/llama-3.1-nemotron-70b-instruct" + # api_key: "${NVIDIA_API_KEY}" + # base_url: "https://integrate.api.nvidia.com/v1" + # timeout: 30 + # circuit_breaker: + # enabled: true + # threshold: 5 + # cooldown: 900 diff --git a/fam-edge/requirements.txt b/fam-edge/requirements.txt new file mode 100644 index 0000000..ed946c7 --- /dev/null +++ b/fam-edge/requirements.txt @@ -0,0 +1,8 @@ +flask>=3.0.0 +gunicorn>=21.2.0 +requests>=2.31.0 +PyYAML>=6.0 +opencv-python>=4.8.0 +numpy>=1.24.0 +google-generativeai>=0.5.0 +openai>=1.10.0 diff --git a/fam-edge/src/fam_edge/__init__.py b/fam-edge/src/fam_edge/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/fam-edge/src/fam_edge/ai_orchestrator/__init__.py b/fam-edge/src/fam_edge/ai_orchestrator/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/fam-edge/src/fam_edge/ai_orchestrator/json_parser.py b/fam-edge/src/fam_edge/ai_orchestrator/json_parser.py new file mode 100644 index 0000000..ae0e7a8 --- /dev/null +++ b/fam-edge/src/fam_edge/ai_orchestrator/json_parser.py @@ -0,0 +1,104 @@ +""" +VLM JSON 解析容错 + Schema 校验 + +三层容错策略: +1. 直接 json.loads +2. 提取 markdown fence 内容 +3. 贪婪匹配最大的 {...} + +validate_schema: 校验 + 脏数据清洗 +""" +import re +import json +from typing import Dict + + +class VLMOutputInvalidError(Exception): + """VLM 输出无法解析为合法 JSON""" + pass + + +def parse_vlm_json(raw: str) -> dict: + """三层容错解析 VLM 输出的 JSON""" + # 第 1 层：直接 json.loads + try: + return validate_schema(json.loads(raw.strip())) + except json.JSONDecodeError: + pass + + # 第 2 层：提取 markdown fence 内容 + fence_match = re.search(r'```(?:json)?\s*(\{.*?\})\s*```', raw, re.DOTALL) + if fence_match: + try: + return validate_schema(json.loads(fence_match.group(1))) + except json.JSONDecodeError: + pass + + # 第 3 层：贪婪匹配最大的 {...} + brace_match = re.search(r'\{.*\}', raw, re.DOTALL) + if brace_match: + try: + return validate_schema(json.loads(brace_match.group(0))) + except json.JSONDecodeError: + pass + + raise VLMOutputInvalidError(f"无法从 VLM 输出中解析 JSON: {raw[:200]}") + + +def validate_schema(data: dict) -> dict: + """Schema 校验 + 脏数据清洗""" + required = ["global_summary", "entities_json", "frame_details", "compute_provider"] + for k in required: + if k not in data: + raise VLMOutputInvalidError(f"缺失字段: {k}") + + # entities_json 结构校验 + if not isinstance(data["entities_json"], list): + raise VLMOutputInvalidError("entities_json 必须为数组") + + cleaned_entities = [] + for ent in data["entities_json"]: + if not isinstance(ent, dict): + continue + if "person" not in ent or "action" not in ent: + raise VLMOutputInvalidError("entity 缺少 person 或 action 字段") + cleaned_entities.append({ + "person": str(ent["person"]), + "action": str(ent["action"]), + "clothing": str(ent.get("clothing", "")) + }) + data["entities_json"] = cleaned_entities + + # frame_details 结构校验 + if not isinstance(data["frame_details"], list): + raise VLMOutputInvalidError("frame_details 必须为数组") + + cleaned_frames = [] + for frame in data["frame_details"]: + if not isinstance(frame, dict): + continue + for k in ["frame_index", "frame_timestamp", "person", "action", "source_providers"]: + if k not in frame: + raise VLMOutputInvalidError(f"frame_details 缺少字段: {k}") + sp = frame["source_providers"] + if not isinstance(sp, list) or len(sp) == 0: + raise VLMOutputInvalidError("frame_details.source_providers 必须为非空数组") + cleaned_frames.append({ + "frame_index": int(frame["frame_index"]), + "frame_timestamp": str(frame["frame_timestamp"]), + "person": str(frame["person"]), + "action": str(frame["action"]), + "clothing": str(frame.get("clothing", "")), + "is_attention_event": bool(frame.get("is_attention_event", False)), + "source_providers": [str(p) for p in sp] + }) + data["frame_details"] = cleaned_frames + + # compute_provider 校验为数组 + if not isinstance(data["compute_provider"], list): + raise VLMOutputInvalidError("compute_provider 必须为数组") + if len(data["compute_provider"]) == 0: + raise VLMOutputInvalidError("compute_provider 不能为空数组") + data["compute_provider"] = [str(p) for p in data["compute_provider"]] + + return data diff --git a/fam-edge/src/fam_edge/ai_orchestrator/orchestrator.py b/fam-edge/src/fam_edge/ai_orchestrator/orchestrator.py new file mode 100644 index 0000000..18d5a0e --- /dev/null +++ b/fam-edge/src/fam_edge/ai_orchestrator/orchestrator.py @@ -0,0 +1,327 @@ +""" +AI-Orchestrator - 多模型并行编排 + +流程: +1. 加载所有启用的模型适配器 +2. 健康检查 +3. 抽帧 + 关键帧筛选 + 压缩 +4. 并行调用所有健康模型（ThreadPoolExecutor） +5. 文本融合（多模型输出平等交叉验证） +6. 回调 NAS +7. 清理临时文件 +""" +import time +import json +import base64 +import requests +from concurrent.futures import ThreadPoolExecutor, as_completed, TimeoutError as FuturesTimeout +from typing import Dict, List, Optional + +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 parse_vlm_json, VLMOutputInvalidError + +logger = setup_logger('fam-edge.orchestrator') + +FUSION_SYSTEM_PROMPT = """你是一个无情的数据提取器。不要输出任何思考过程，只输出合法 JSON。 + +输入参数: +- 多模型视觉分析日志（每个模型独立输出，平等对待，交叉验证）: +{model_outputs} +- 已知成员清单: {known_members} + +执行规则: +1. 多个模型的输出平等对待，交叉验证： + - 多个模型一致描述的内容 → 可信度高，必须纳入 frame_details，source_providers 列出所有一致的模型 + - 仅单一模型描述的内容 → 纳入 frame_details，source_providers 仅含该模型 + - 多个模型冲突时（如人物动作描述不一致）→ 以多数模型一致为准，source_providers 列出多数派模型 +2. 画面人物按特征匹配已知成员清单： + - 匹配到已命名成员（real_name 非空）→ person 字段填 real_name + - 匹配到未命名成员（real_name 为空）→ person 字段填 abstract_label + - 都不匹配 → 按出现顺序赋予新标识"人物B"、"人物C"... +3. 提取每张关键帧对应的时间点、人物、动作、衣着，输出到 frame_details 数组。 +4. frame_details 每条必须包含 source_providers 数组。 +5. compute_provider 字段填入本次实际成功调用的所有模型标识数组（去重）。 +6. 仅输出合法 JSON，不输出任何思考过程、markdown 标记或注释。 + +输出 JSON 结构: +{{ + "global_summary": "字符串，整个时段的整体摘要，简体中文", + "entities_json": [ + {{ + "person": "字符串", + "action": "字符串", + "clothing": "字符串" + }} + ], + "frame_details": [ + {{ + "frame_index": "数字", + "frame_timestamp": "字符串，ISO 8601 格式时间戳", + "person": "字符串", + "action": "字符串", + "clothing": "字符串", + "is_attention_event": "布尔值", + "source_providers": "数组" + }} + ], + "compute_provider": "数组" +}} +""" + + +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) -> Dict[str, str]: + """并行调用所有健康模型进行视觉分析""" + model_outputs = {} + max_timeout = max((a.get_timeout() for a in adapters), default=240) + + with ThreadPoolExecutor(max_workers=len(adapters)) as pool: + futures = {} + for adapter in adapters: + if adapter.get_circuit_breaker().is_open(): + logger.warning(f"[{adapter.provider_name}] 熔断器 OPEN，跳过") + 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 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 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 adapters if a.provider_name == provider) + adapter.get_circuit_breaker().record_failure() + + return model_outputs + + def run_text_fusion(self, model_outputs: Dict[str, str], + known_members_context: str, + task_id: int) -> dict: + """文本融合阶段 - 多模型输出平等交叉验证""" + # 构建 model_outputs 文本 + outputs_text = '\n'.join( + f" - {provider} 输出: {output}" for provider, output in model_outputs.items() + ) + + prompt = FUSION_SYSTEM_PROMPT.format( + model_outputs=outputs_text, + known_members=known_members_context or '（暂无已知成员）' + ) + + # 调用 Ollama 纯文本模式 + ollama_cfg = next( + (cfg for cfg in self.config.get('models', []) if cfg.get('provider') == 'ollama'), + None + ) + if not ollama_cfg: + raise VLMOutputInvalidError("没有 Ollama 配置，无法执行文本融合") + + base_url = ollama_cfg.get('base_url', 'http://localhost:11434') + model_name = ollama_cfg.get('model_name', 'llava-phi3') + fusion_timeout = self.timeout_cfg.get('vlm_fusion', 120) + + start = time.time() + resp = requests.post( + f"{base_url}/api/generate", + json={ + "model": model_name, + "prompt": prompt, + "stream": False, + "format": "json", + "options": {"temperature": 0.0} + }, + timeout=fusion_timeout + ) + + if resp.status_code != 200: + raise VLMOutputInvalidError(f"融合阶段 Ollama 调用失败: {resp.status_code}") + + raw_output = resp.json().get('response', '') + duration_ms = int((time.time() - start) * 1000) + log_task(logger, task_id, 'vlm_fusion', f'融合完成，原始输出长度={len(raw_output)}', duration_ms=duration_ms) + + # 解析 JSON（三层容错） + result = parse_vlm_json(raw_output) + + # 确保 compute_provider 与实际调用的模型一致 + result['compute_provider'] = list(model_outputs.keys()) + + # 确保 frame_details 的 source_providers 只包含实际成功的模型 + valid_providers = set(model_outputs.keys()) + for frame in result.get('frame_details', []): + frame['source_providers'] = [ + p for p in frame.get('source_providers', []) if p in valid_providers + ] or list(valid_providers) + + return result + + 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): + """端到端处理任务""" + 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 + ) + + if not model_outputs: + raise Exception('All models failed in visual analysis') + + # 4. 文本融合 + fusion_result = self.run_text_fusion(model_outputs, 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}] VLM 输出解析失败: {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 diff --git a/fam-edge/src/fam_edge/api_gateway/__init__.py b/fam-edge/src/fam_edge/api_gateway/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/fam-edge/src/fam_edge/api_gateway/api_gateway.py b/fam-edge/src/fam_edge/api_gateway/api_gateway.py new file mode 100644 index 0000000..4a0794f --- /dev/null +++ b/fam-edge/src/fam_edge/api_gateway/api_gateway.py @@ -0,0 +1,89 @@ +""" +API-Gateway - Flask 蓝图，接收任务 + +同时只允许 1 个任务在处理；新任务到达时若当前有任务处理中，返回 429 +""" +import threading +from flask import Blueprint, request, jsonify + +from ..logger import setup_logger +from ..ai_orchestrator.orchestrator import AIOrchestrator + +logger = setup_logger('fam-edge.api_gateway') + +api_bp = Blueprint('api_gateway', __name__) + +# 并发控制：同时只允许 1 个任务 +_current_task_lock = threading.Lock() +_currently_processing = False + +_orchestrator = None + + +def get_orchestrator(): + global _orchestrator + if _orchestrator is None: + _orchestrator = AIOrchestrator() + return _orchestrator + + +@api_bp.route('/api/edge/video/analyze', methods=['POST']) +def receive_task(): + """接收分析任务""" + global _currently_processing + + 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('/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 diff --git a/fam-edge/src/fam_edge/app.py b/fam-edge/src/fam_edge/app.py new file mode 100644 index 0000000..031c131 --- /dev/null +++ b/fam-edge/src/fam_edge/app.py @@ -0,0 +1,30 @@ +""" +FAM-Edge 主应用 - Flask 单进程 + +承载: API-Gateway / Video-Preprocessor / AI-Orchestrator / Storage-Cleaner +""" +import os +import sys +from flask import Flask, jsonify + +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 + +logger = setup_logger('fam-edge.app') + +app = Flask(__name__) +app.register_blueprint(api_bp) + + +@app.route('/', methods=['GET']) +def root(): + return jsonify({"service": "fam-edge", "version": "1.0"}), 200 + + +if __name__ == '__main__': + cfg = load_config() + port = cfg.get('server', {}).get('port', 5000) + app.run(host='0.0.0.0', port=port, debug=False, threaded=True) diff --git a/fam-edge/src/fam_edge/config_loader.py b/fam-edge/src/fam_edge/config_loader.py new file mode 100644 index 0000000..6caa9cc --- /dev/null +++ b/fam-edge/src/fam_edge/config_loader.py @@ -0,0 +1,30 @@ +""" +配置加载器 - 从 config.yaml 读取配置，支持 ${ENV_VAR} 解析 +""" +import os +import re +import yaml + + +def _resolve_env_vars(value): + if isinstance(value, str): + def replace_env(match): + env_name = match.group(1) + return os.environ.get(env_name, match.group(0)) + return re.sub(r'\$\{(\w+)\}', replace_env, value) + elif isinstance(value, dict): + return {k: _resolve_env_vars(v) for k, v in value.items()} + elif isinstance(value, list): + return [_resolve_env_vars(item) for item in value] + return value + + +def load_config(config_path=None): + if config_path is None: + config_path = os.path.join( + os.path.dirname(os.path.dirname(os.path.abspath(__file__))), + 'config', 'config.yaml' + ) + with open(config_path, 'r', encoding='utf-8') as f: + raw = yaml.safe_load(f) + return _resolve_env_vars(raw) diff --git a/fam-edge/src/fam_edge/logger.py b/fam-edge/src/fam_edge/logger.py new file mode 100644 index 0000000..bfd3faa --- /dev/null +++ b/fam-edge/src/fam_edge/logger.py @@ -0,0 +1,28 @@ +""" +日志工具 - 统一格式，带 task_id 作为 trace_id +""" +import logging +import sys + + +def setup_logger(name='fam-edge', level=logging.INFO): + logger = logging.getLogger(name) + if logger.handlers: + return logger + logger.setLevel(level) + handler = logging.StreamHandler(sys.stdout) + formatter = logging.Formatter( + '%(asctime)s [%(name)s] %(levelname)s %(message)s', + datefmt='%Y-%m-%d %H:%M:%S' + ) + handler.setFormatter(formatter) + logger.addHandler(handler) + return logger + + +def log_task(logger, task_id, stage, message, level=logging.INFO, duration_ms=None): + parts = [f"[task_id={task_id}]", stage] + if duration_ms is not None: + parts.append(f"done in {duration_ms}ms") + parts.append(message) + logger.log(level, ' '.join(parts)) diff --git a/fam-edge/src/fam_edge/model_adapters/__init__.py b/fam-edge/src/fam_edge/model_adapters/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/fam-edge/src/fam_edge/model_adapters/adapter_factory.py b/fam-edge/src/fam_edge/model_adapters/adapter_factory.py new file mode 100644 index 0000000..fb9b4b4 --- /dev/null +++ b/fam-edge/src/fam_edge/model_adapters/adapter_factory.py @@ -0,0 +1,54 @@ +""" +适配器工厂 - 根据 config 创建适配器实例 + +新增模型只需: +1. 实现适配器类（继承 BaseModelAdapter） +2. 在此工厂注册 +3. 在 config.yaml 的 models 数组加一项 +""" +from typing import List + +from .base_adapter import BaseModelAdapter +from .ollama_adapter import OllamaAdapter +from .gemini_adapter import GeminiAdapter +from ..logger import setup_logger + +logger = setup_logger('fam-edge.adapter_factory') + +_ADAPTER_REGISTRY = { + "ollama": OllamaAdapter, + "gemini": GeminiAdapter, + # v1.1 扩展: + # "openai": OpenAIAdapter, + # "nvidia": NvidiaAdapter, +} + + +def build_adapter(config: dict) -> BaseModelAdapter: + """根据 config 中的 provider 字段创建适配器""" + provider = config.get('provider', '') + adapter_cls = _ADAPTER_REGISTRY.get(provider) + if adapter_cls is None: + raise ValueError(f"未知的模型 provider: {provider}，请先注册适配器") + return adapter_cls(config) + + +def build_adapters(configs: List[dict]) -> List[BaseModelAdapter]: + """批量创建适配器（仅 enabled 的）""" + adapters = [] + for cfg in configs: + if not cfg.get('enabled', False): + continue + try: + adapter = build_adapter(cfg) + adapters.append(adapter) + logger.info(f"适配器已创建: {adapter.provider_name} ({cfg.get('model_name', '?')})") + except Exception as e: + logger.error(f"创建适配器失败 ({cfg.get('provider', '?')}): {e}") + return adapters + + +def register_adapter(provider_name: str, adapter_cls): + """注册新适配器（供扩展使用）""" + _ADAPTER_REGISTRY[provider_name] = adapter_cls + logger.info(f"适配器已注册: {provider_name}") diff --git a/fam-edge/src/fam_edge/model_adapters/base_adapter.py b/fam-edge/src/fam_edge/model_adapters/base_adapter.py new file mode 100644 index 0000000..8c3c956 --- /dev/null +++ b/fam-edge/src/fam_edge/model_adapters/base_adapter.py @@ -0,0 +1,42 @@ +""" +模型适配器基类 - 所有模型适配器的抽象基类 + +新增模型只需继承此类并实现 4 个方法: +1. health_check() -> bool +2. analyze_frames(frame_paths, frame_timestamps, known_members_context) -> Optional[str] +3. get_timeout() -> int +4. get_circuit_breaker() -> CircuitBreaker +""" +from abc import ABC, abstractmethod +from typing import List, Optional + + +class BaseModelAdapter(ABC): + """所有模型适配器的抽象基类""" + + def __init__(self, provider_name: str, config: dict): + self.provider_name = provider_name # 如 "ollama", "gemini" + self.config = config + + @abstractmethod + def health_check(self) -> bool: + """健康检查，返回 True/False""" + pass + + @abstractmethod + def analyze_frames(self, frame_paths: List[str], + frame_timestamps: List[str], + known_members_context: str) -> Optional[str]: + """视觉分析：输入帧图片路径 + 时间戳 + 成员清单，输出自然语言描述。 + 失败/超时返回 None。""" + pass + + @abstractmethod + def get_timeout(self) -> int: + """该模型的调用超时秒数""" + pass + + @abstractmethod + def get_circuit_breaker(self): + """返回该模型专属的熔断器实例""" + pass diff --git a/fam-edge/src/fam_edge/model_adapters/circuit_breaker.py b/fam-edge/src/fam_edge/model_adapters/circuit_breaker.py new file mode 100644 index 0000000..85fee74 --- /dev/null +++ b/fam-edge/src/fam_edge/model_adapters/circuit_breaker.py @@ -0,0 +1,47 @@ +""" +熔断器 - 每个云端模型独立实例 + +状态机: CLOSED -> OPEN -> HALF_OPEN -> CLOSED/OPEN +- 连续 threshold 次失败 -> OPEN +- 冷却 cooldown 秒后 -> HALF_OPEN（允许一次探测） +- 探测成功 -> CLOSED；探测失败 -> 重新 OPEN +""" +from collections import deque +import time + + +class CircuitBreaker: + def __init__(self, threshold: int = 5, cooldown: int = 900, enabled: bool = True): + self.enabled = enabled + if enabled: + self.failures = deque(maxlen=threshold) + else: + self.failures = None + self.threshold = threshold + self.cooldown = cooldown + self.state = 'CLOSED' + self.last_failure = None + + def record_failure(self): + if not self.enabled: + return + self.failures.append(time.time()) + if len(self.failures) >= self.threshold: + self.state = 'OPEN' + self.last_failure = time.time() + + def record_success(self): + if not self.enabled: + return + self.failures.clear() + self.state = 'CLOSED' + + def is_open(self): + if not self.enabled: + return False + if self.state == 'OPEN' and self.last_failure and time.time() - self.last_failure > self.cooldown: + self.state = 'HALF_OPEN' + return self.state == 'OPEN' + + def __repr__(self): + return f"CircuitBreaker(state={self.state}, enabled={self.enabled})" diff --git a/fam-edge/src/fam_edge/model_adapters/gemini_adapter.py b/fam-edge/src/fam_edge/model_adapters/gemini_adapter.py new file mode 100644 index 0000000..2de8ae5 --- /dev/null +++ b/fam-edge/src/fam_edge/model_adapters/gemini_adapter.py @@ -0,0 +1,156 @@ +""" +GeminiAdapter - Google Gemini 云端模型适配器 + +provider_name = "gemini" +模型: gemini-1.5-flash +健康检查: GET models API +熔断器: 启用，连续 5 次失败 -> OPEN 15 分钟 +""" +import os +import base64 +import requests +from typing import List, Optional + +from .base_adapter import BaseModelAdapter +from .circuit_breaker import CircuitBreaker +from ..logger import setup_logger + +logger = setup_logger('fam-edge.gemini_adapter') + + +class GeminiAdapter(BaseModelAdapter): + """Gemini 云端 VLM 适配器""" + + def __init__(self, config: dict): + super().__init__("gemini", config) + self.model_name = config.get('model_name', 'gemini-1.5-flash') + self.api_key = config.get('api_key', '') + self.timeout = config.get('timeout', 8) + cb_cfg = config.get('circuit_breaker', {}) + self._cb = CircuitBreaker( + threshold=cb_cfg.get('threshold', 5), + cooldown=cb_cfg.get('cooldown', 900), + enabled=cb_cfg.get('enabled', True) # 云端默认启用 + ) + self._base_url = "https://generativelanguage.googleapis.com/v1beta" + + def health_check(self) -> bool: + """GET models API，检查可用性""" + if not self.api_key: + logger.warning("Gemini API Key 未配置，健康检查失败") + return False + try: + resp = requests.get( + f"{self._base_url}/models?key={self.api_key}", + timeout=10 + ) + if resp.status_code == 200: + models = resp.json().get('models', []) + model_names = [m.get('name', '') for m in models] + has_model = any(self.model_name in name for name in model_names) + if has_model: + logger.info(f"Gemini 健康检查通过: 模型 {self.model_name} 可用") + return True + else: + logger.warning(f"Gemini 健康检查失败: 模型 {self.model_name} 未找到") + return False + return False + except Exception as e: + logger.error(f"Gemini 健康检查异常: {e}") + return False + + def analyze_frames(self, frame_paths: List[str], + frame_timestamps: List[str], + known_members_context: str) -> Optional[str]: + """调用 Gemini 视觉分析""" + if self._cb.is_open(): + logger.warning("Gemini 熔断器 OPEN，跳过调用") + return None + + if not self.api_key: + logger.warning("Gemini API Key 未配置，跳过调用") + return None + + # 构建 Prompt + n = len(frame_paths) + prompt = self._build_visual_prompt(n, frame_timestamps, known_members_context) + + # 构建 inline_data + parts = [{"text": prompt}] + for path in frame_paths: + try: + with open(path, 'rb') as f: + img_data = base64.b64encode(f.read()).decode('utf-8') + parts.append({ + "inline_data": { + "mime_type": "image/jpeg", + "data": img_data + } + }) + except Exception as e: + logger.error(f"读取图片失败 {path}: {e}") + + try: + resp = requests.post( + f"{self._base_url}/models/{self.model_name}:generateContent?key={self.api_key}", + json={ + "contents": [{"parts": parts}], + "generationConfig": {"temperature": 0.2, "topP": 0.8} + }, + timeout=self.timeout + ) + + if resp.status_code == 200: + data = resp.json() + candidates = data.get('candidates', []) + if candidates: + output = candidates[0].get('content', {}).get('parts', [{}])[0].get('text', '') + self._cb.record_success() + logger.info(f"Gemini 视觉分析完成，输出长度={len(output)}") + return output + else: + logger.warning("Gemini 返回空 candidates") + self._cb.record_failure() + return None + else: + logger.error(f"Gemini 调用失败: {resp.status_code} {resp.text[:200]}") + self._cb.record_failure() + return None + + except requests.Timeout: + logger.warning(f"Gemini 调用超时 ({self.timeout}s)，降级跳过") + self._cb.record_failure() + return None + except Exception as e: + logger.error(f"Gemini 调用异常: {e}") + self._cb.record_failure() + return None + + def get_timeout(self) -> int: + return self.timeout + + def get_circuit_breaker(self) -> CircuitBreaker: + return self._cb + + def _build_visual_prompt(self, n: int, timestamps: List[str], known_members: str) -> str: + ts_lines = '\n'.join( + f"[Image {i+1}] Time: {ts}" for i, ts in enumerate(timestamps) + ) + return f"""You are a home surveillance video analysis assistant. Describe what you see in the following {n} images chronologically. Be objective. + +Timestamps: +{ts_lines} + +For each image, report: +1. People: count, clothing (color + type), visible actions +2. Objects: toys, bottles, furniture, etc. +3. Interactions: between people or people and objects + +Known family members (match by features, use real_name if matched, otherwise "PersonX"): +{known_members or 'None'} + +Output format (plain text, one paragraph per image, keep timestamp markers): +[Image 1] Time: {timestamps[0] if timestamps else ''} +Description: ... + +Be concise and objective. Do not output JSON or markdown.""" diff --git a/fam-edge/src/fam_edge/model_adapters/ollama_adapter.py b/fam-edge/src/fam_edge/model_adapters/ollama_adapter.py new file mode 100644 index 0000000..2b846a9 --- /dev/null +++ b/fam-edge/src/fam_edge/model_adapters/ollama_adapter.py @@ -0,0 +1,139 @@ +""" +OllamaAdapter - 本地 VLM 模型适配器 + +provider_name = "ollama" +模型: llava-phi3 +健康检查: GET /api/tags +""" +import base64 +import requests +from typing import List, Optional + +from .base_adapter import BaseModelAdapter +from .circuit_breaker import CircuitBreaker +from ..logger import setup_logger + +logger = setup_logger('fam-edge.ollama_adapter') + + +class OllamaAdapter(BaseModelAdapter): + """Ollama 本地 VLM 适配器""" + + def __init__(self, config: dict): + super().__init__("ollama", config) + self.base_url = config.get('base_url', 'http://localhost:11434') + self.model_name = config.get('model_name', 'llava-phi3') + self.timeout = config.get('timeout', 240) + cb_cfg = config.get('circuit_breaker', {}) + self._cb = CircuitBreaker( + threshold=cb_cfg.get('threshold', 5), + cooldown=cb_cfg.get('cooldown', 900), + enabled=cb_cfg.get('enabled', False) # 本地模型默认不启用 + ) + + def health_check(self) -> bool: + """GET /api/tags，检查模型是否可用""" + try: + resp = requests.get(f"{self.base_url}/api/tags", timeout=10) + if resp.status_code == 200: + models = resp.json().get('models', []) + model_names = [m.get('name', '') for m in models] + # 兼容 llava-phi3:latest 等后缀 + has_model = any(self.model_name in name for name in model_names) + if has_model: + logger.info(f"Ollama 健康检查通过: 模型 {self.model_name} 可用") + return True + else: + logger.warning(f"Ollama 健康检查失败: 模型 {self.model_name} 未找到，可用模型: {model_names}") + return False + return False + except Exception as e: + logger.error(f"Ollama 健康检查异常: {e}") + return False + + def analyze_frames(self, frame_paths: List[str], + frame_timestamps: List[str], + known_members_context: str) -> Optional[str]: + """调用 Ollama 视觉分析""" + if self._cb.is_open(): + logger.warning("Ollama 熔断器 OPEN，跳过调用") + return None + + # 构建 Prompt + n = len(frame_paths) + prompt = self._build_visual_prompt(n, frame_timestamps, known_members_context) + + # 读取图片并 Base64 编码 + images = [] + for path in frame_paths: + try: + with open(path, 'rb') as f: + images.append(base64.b64encode(f.read()).decode('utf-8')) + except Exception as e: + logger.error(f"读取图片失败 {path}: {e}") + + if not images: + logger.error("没有可用的图片帧") + return None + + try: + resp = requests.post( + f"{self.base_url}/api/generate", + json={ + "model": self.model_name, + "prompt": prompt, + "images": images, + "stream": False, + "options": {"temperature": 0.2, "top_p": 0.8} + }, + timeout=self.timeout + ) + + if resp.status_code == 200: + output = resp.json().get('response', '') + self._cb.record_success() + logger.info(f"Ollama 视觉分析完成，输出长度={len(output)}") + return output + else: + logger.error(f"Ollama 调用失败: {resp.status_code} {resp.text[:200]}") + self._cb.record_failure() + return None + + except requests.Timeout: + logger.error(f"Ollama 调用超时 ({self.timeout}s)") + self._cb.record_failure() + return None + except Exception as e: + logger.error(f"Ollama 调用异常: {e}") + self._cb.record_failure() + return None + + def get_timeout(self) -> int: + return self.timeout + + def get_circuit_breaker(self) -> CircuitBreaker: + return self._cb + + def _build_visual_prompt(self, n: int, timestamps: List[str], known_members: str) -> str: + """构建视觉分析 Prompt""" + ts_lines = '\n'.join( + f"[图片{i+1}] 时间: {ts}" for i, ts in enumerate(timestamps) + ) + return f"""你是家庭监控视频分析助手。请按时间顺序描述下列 {n} 张图片中可见的内容，只描述客观画面，不要猜测或推测。 + +每张图片对应的时间戳如下： +{ts_lines} + +每张图片需报告： +1. 人物：数量、衣着（颜色+类型）、可见动作 +2. 物品：玩具、奶瓶、家具等显眼物品 +3. 互动：人与人、人与物品之间的互动 + +已知家庭成员清单（按特征匹配，匹配成功用 real_name，未匹配用"人物X"标识）： +{known_members or '（暂无已知成员）'} + +输出格式（纯文本，每张图片一段，保留时间戳标记）： +[图片1] 时间: {timestamps[0] if timestamps else ''} +内容: ... + +要求简洁、客观。不要输出 JSON，不要输出 markdown。""" diff --git a/fam-edge/src/fam_edge/storage_cleaner/__init__.py b/fam-edge/src/fam_edge/storage_cleaner/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/fam-edge/src/fam_edge/video_preprocessor/__init__.py b/fam-edge/src/fam_edge/video_preprocessor/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/fam-edge/src/fam_edge/video_preprocessor/preprocessor.py b/fam-edge/src/fam_edge/video_preprocessor/preprocessor.py new file mode 100644 index 0000000..cd6d243 --- /dev/null +++ b/fam-edge/src/fam_edge/video_preprocessor/preprocessor.py @@ -0,0 +1,255 @@ +""" +Video-Preprocessor - 视频预处理 + +流程: +1. 下载视频（超时 60s） +2. FFmpeg 等距粗抽 30 张候选帧 +3. OpenCV 帧差分析筛选 5-8 张关键帧（MSE 阈值） +4. 压缩（长边 ≤ 1024px，JPEG 质量 80） + +异常兜底: +- ffprobe 失败 -> 退化为按 60s 间隔抽帧 +- 帧差分析异常 -> 退化为等距抽 5 帧 +- 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_frames = video_cfg.get('candidate_frames', 30) + self.min_key_frames = video_cfg.get('min_key_frames', 5) + self.max_key_frames = video_cfg.get('max_key_frames', 8) + 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) + + # 临时目录 + 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 _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]: + """等距粗抽候选帧""" + os.makedirs(self.frames_dir, exist_ok=True) + duration = self._get_video_duration(video_path) + + if duration > 0: + interval = duration / self.candidate_frames + else: + # 兜底: 每 60s 抽一帧 + interval = 60 + logger.warning(f"[task_id={self.task_id}] ffprobe 失败，退化为 60s 间隔抽帧") + + cmd = [ + 'ffmpeg', '-i', video_path, + '-vf', f'fps=1/{interval}', + '-q:v', '2', + os.path.join(self.frames_dir, 'frame_%04d.jpg') + ] + try: + subprocess.run(cmd, capture_output=True, timeout=120, check=True) + except subprocess.CalledProcessError as e: + logger.error(f"[task_id={self.task_id}] FFmpeg 抽帧失败: {e}") + raise + + # 收集候选帧路径 + 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'粗抽 {len(frames)} 张候选帧') + return frames + + def select_key_frames(self, candidate_frames: List[str]) -> List[str]: + """帧差分析筛选关键帧""" + if len(candidate_frames) <= self.min_key_frames: + return candidate_frames[:self.max_key_frames] + + 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[:self.max_key_frames] + + # 计算每帧与前一关键帧的 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_key_frames，从剩余中均匀补足 + if len(key_indices) < self.min_key_frames: + remaining = [i for i in range(len(images)) if i not in key_indices] + step = max(1, len(remaining) // (self.min_key_frames - len(key_indices))) + for i in range(0, len(remaining), step): + if len(key_indices) >= self.min_key_frames: + break + key_indices.append(remaining[i]) + key_indices.sort() + + # 若 > max_key_frames，按差异值降序取前 N + if len(key_indices) > self.max_key_frames: + # 计算每个关键帧与前一帧的差异 + 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[:self.max_key_frames - 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}，退化为等距抽 5 帧") + step = max(1, len(candidate_frames) // self.min_key_frames) + return candidate_frames[::step][:self.min_key_frames] + + 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 + try: + start_dt = datetime.fromisoformat(event_start_time.replace('Z', '+00:00')) + except Exception: + start_dt = datetime.now() + + 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}")