""" 定时任务调度模块 """ import threading import time import datetime from camera import CameraCapture from analyzer import ImageAnalyzer from local_analyzer import LocalAnalyzer from database import db from config import config_mgr class VisionScheduler: """视觉记录调度器""" def __init__(self): self.camera = CameraCapture() self.vision_analyzer = ImageAnalyzer() # 大模型分析器 self.local_analyzer = LocalAnalyzer() # 本地分析器 self.running = False self.timer = None self.prev_image_path = None # 保存前一张图片路径 # 统计 self.capture_count = 0 self.last_capture_time = None self.last_analyze_time = None self.errors = [] self.model_calls = 0 # 大模型调用次数 self.local_analyses = 0 # 本地分析次数 def start(self): """启动定时拍照""" if self.running: return {'success': False, 'error': '已在运行中'} self.running = True interval = config_mgr.get('capture_interval', 60) self._schedule_next() return { 'success': True, 'interval': interval, 'auto_analyze': config_mgr.get('auto_analyze', True) } def stop(self): """停止定时拍照""" self.running = False if self.timer: self.timer.cancel() self.timer = None self.camera.close() return {'success': True} def _schedule_next(self): """安排下一次拍照""" if not self.running: return interval = config_mgr.get('capture_interval', 60) self.timer = threading.Timer(interval, self._capture_task) self.timer.start() def _capture_task(self): """拍照任务""" if not self.running: return try: # 拍照 result = self.camera.capture() if result['success']: # 记录到数据库 image_id = db.add_image( result['path'], date_folder=result.get('date_folder') ) self.capture_count += 1 self.last_capture_time = datetime.datetime.now().isoformat() # 自动分析 if config_mgr.get('auto_analyze', True): self._analyze_task(image_id, result['path']) else: self.errors.append({ 'time': datetime.datetime.now().isoformat(), 'error': result['error'] }) except Exception as e: self.errors.append({ 'time': datetime.datetime.now().isoformat(), 'error': str(e) }) # 安排下一次 self._schedule_next() def _analyze_task(self, image_id, image_path): """分析任务 - 先本地分析，再决定是否调用大模型""" try: print(f"[Scheduler] ===== Analyzing image {image_id} =====") print(f"[Scheduler] Image path: {image_path}") self.local_analyses += 1 # 1. 本地快速分析 print(f"[Scheduler] Starting local analysis...") local_result = self.local_analyzer.analyze(image_path, self.prev_image_path) # 保存当前图片路径供下次对比 self.prev_image_path = image_path if local_result['success']: # 记录本地检测到的事件 for event in local_result['events']: db.add_event( image_id, event['event_type'] + '(本地)', event['description'], event['confidence'] ) # 2. 判断是否需要大模型分析 if local_result['need_model'] and config_mgr.get('auto_analyze', True): print(f"[Scheduler] Local analysis triggered model call for image {image_id}") self._call_vision_api(image_id, image_path) else: # 不需要大模型，直接标记已分析 db.mark_image_analyzed(image_id) print(f"[Scheduler] Local analysis sufficient for image {image_id}") print(f" - Motion: {local_result['metrics'].get('motion_ratio', 0):.2%}") print(f" - Human: {local_result['metrics'].get('human_count', 0)}") print(f" - Need model: {local_result['need_model']}") self.last_analyze_time = datetime.datetime.now().isoformat() else: self.errors.append({ 'time': datetime.datetime.now().isoformat(), 'error': f"本地分析失败: {local_result['error']}" }) # 本地分析失败，尝试直接调用大模型 if config_mgr.get('auto_analyze', True): self._call_vision_api(image_id, image_path) except Exception as e: self.errors.append({ 'time': datetime.datetime.now().isoformat(), 'error': str(e) }) def _call_vision_api(self, image_id, image_path): """调用大模型 Vision API""" try: self.model_calls += 1 print(f"[Scheduler] Calling Vision API for image {image_id}") result = self.vision_analyzer.analyze(image_path) if result['success']: for event in result['events']: db.add_event( image_id, event['event_type'] + '(AI)', event['description'], event['confidence'] ) db.mark_image_analyzed(image_id) print(f"[Scheduler] Vision API analysis complete for image {image_id}") else: print(f"[Scheduler] Vision API failed: {result['error']}") self.errors.append({ 'time': datetime.datetime.now().isoformat(), 'error': f"Vision API失败: {result['error']}" }) # 即使失败也标记已分析（避免重复调用） db.mark_image_analyzed(image_id) except Exception as e: print(f"[Scheduler] Vision API exception: {e}") self.errors.append({ 'time': datetime.datetime.now().isoformat(), 'error': str(e) }) def capture_now(self): """立即拍照""" result = self.camera.capture() if result['success']: image_id = db.add_image( result['path'], date_folder=result.get('date_folder') ) self.capture_count += 1 self.last_capture_time = datetime.datetime.now().isoformat() # 如果自动分析开启，立即分析 if config_mgr.get('auto_analyze', True): threading.Thread( target=self._analyze_task, args=(image_id, result['path']) ).start() return { 'success': True, 'image_id': image_id, 'path': result['path'], 'timestamp': result['timestamp'], 'date_folder': result.get('date_folder') } return result def analyze_now(self, image_id): """立即分析指定图片""" try: image = db.get_image_by_id(image_id) if not image: return {'success': False, 'error': '图片不存在'} # 获取前一张图片 prev_images = db.get_images(limit=1, offset=1) prev_path = prev_images[0]['path'] if prev_images else None # 先本地分析 local_result = self.local_analyzer.analyze(image['path'], prev_path) if local_result['success']: # 记录本地事件 for event in local_result['events']: db.add_event( image_id, event['event_type'] + '(本地)', event['description'], event['confidence'] ) # 再调用大模型（强制调用，用户手动点击） vision_result = self.vision_analyzer.analyze(image['path']) if vision_result['success']: for event in vision_result['events']: db.add_event( image_id, event['event_type'] + '(AI)', event['description'], event['confidence'] ) db.mark_image_analyzed(image_id) self.last_analyze_time = datetime.datetime.now().isoformat() return {'success': True, 'events': local_result['events'] + vision_result['events']} else: db.mark_image_analyzed(image_id) return {'success': True, 'events': local_result['events'], 'vision_error': vision_result['error']} return local_result except Exception as e: return {'success': False, 'error': str(e)} def analyze_unanalyzed(self): """分析所有未分析的图片""" images = db.get_unanalyzed_images(limit=10) results = [] for image in images: result = self.analyzer.analyze(image['path']) if result['success']: for event in result['events']: db.add_event( image['id'], event['event_type'], event['description'], event['confidence'] ) db.mark_image_analyzed(image['id']) results.append({'image_id': image['id'], 'success': True}) else: results.append({'image_id': image['id'], 'success': False, 'error': result['error']}) return results def get_status(self): """获取调度器状态""" return { 'running': self.running, 'interval': config_mgr.get('capture_interval', 60), 'auto_analyze': config_mgr.get('auto_analyze', True), 'capture_count': self.capture_count, 'last_capture_time': self.last_capture_time, 'last_analyze_time': self.last_analyze_time, 'model_calls': self.model_calls, 'local_analyses': self.local_analyses, 'local_stats': self.local_analyzer.get_stats(), 'recent_errors': self.errors[-5:] if self.errors else [] } def set_interval(self, interval): """设置拍照间隔""" config_mgr.set('capture_interval', interval) if self.running: # 重启定时器 self.stop() self.start() return {'success': True, 'interval': interval} # 全局实例 scheduler = VisionScheduler()