salmanabjam/deepvision-prompt-builder
0
1"""2Core Analysis Engine3 4Main orchestration engine for DeepVision Prompt Builder.5Manages image/video processing, plugin execution, and result generation.6"""7 8from datetime import datetime9from pathlib import Path10from typing import Dict, List, Any, Optional, Union11from loguru import logger12 13from core.config import config14from core.image_processor import ImageProcessor15from core.video_processor import VideoProcessor16from core.result_manager import ResultManager17from core.exceptions import DeepVisionError18 19 20class AnalysisEngine:21 """22 Main analysis engine for processing images and videos.23 24 Orchestrates the complete analysis pipeline:25 1. File validation and preprocessing26 2. Plugin execution27 3. Result aggregation28 4. JSON output generation29 """30 31 def __init__(self):32 """Initialize AnalysisEngine."""33 self.image_processor = ImageProcessor()34 self.video_processor = VideoProcessor()35 self.result_manager = ResultManager()36 self.plugins: Dict[str, Any] = {}37 self.plugin_order: List[str] = []38 39 logger.info(f"AnalysisEngine initialized - {config.APP_NAME} v{config.APP_VERSION}")40 41 def register_plugin(self, plugin_name: str, plugin_instance: Any) -> None:42 """43 Register a plugin for analysis.44 45 Args:46 plugin_name: Unique name for the plugin47 plugin_instance: Instance of the plugin class48 """49 if plugin_name in self.plugins:50 logger.warning(f"Plugin '{plugin_name}' already registered, replacing")51 52 self.plugins[plugin_name] = plugin_instance53 54 # Maintain execution order55 if plugin_name not in self.plugin_order:56 self.plugin_order.append(plugin_name)57 58 logger.info(f"Registered plugin: {plugin_name}")59 60 def unregister_plugin(self, plugin_name: str) -> None:61 """62 Unregister a plugin.63 64 Args:65 plugin_name: Name of plugin to remove66 """67 if plugin_name in self.plugins:68 del self.plugins[plugin_name]69 70 if plugin_name in self.plugin_order:71 self.plugin_order.remove(plugin_name)72 73 logger.info(f"Unregistered plugin: {plugin_name}")74 75 def get_registered_plugins(self) -> List[str]:76 """77 Get list of registered plugins.78 79 Returns:80 List of plugin names81 """82 return list(self.plugins.keys())83 84 def analyze_image(85 self,86 image_path: Union[str, Path],87 plugins: Optional[List[str]] = None,88 **kwargs89 ) -> Dict[str, Any]:90 """91 Analyze a single image.92 93 Args:94 image_path: Path to image file95 plugins: List of plugin names to use (None for all)96 **kwargs: Additional arguments for processing97 98 Returns:99 Analysis results dictionary100 """101 start_time = datetime.now()102 image_path = Path(image_path)103 104 logger.info(f"Starting image analysis: {image_path.name}")105 106 try:107 # Clear previous results108 self.result_manager.clear()109 110 # Process image111 image = self.image_processor.process(112 image_path,113 resize=kwargs.get("resize", True),114 normalize=kwargs.get("normalize", False)115 )116 117 # Get image info118 image_info = self.image_processor.get_image_info(image_path)119 120 # Set file metadata121 self.result_manager.set_file_info(122 filename=image_info["filename"],123 file_type="image",124 file_size=image_info["file_size"],125 width=image_info["width"],126 height=image_info["height"],127 format=image_info["format"],128 hash=image_info["hash"],129 )130 131 # Execute plugins132 plugins_used = self._execute_plugins(133 image,134 image_path,135 plugins,136 media_type="image"137 )138 139 # Set processing metadata140 end_time = datetime.now()141 self.result_manager.set_processing_info(142 start_time=start_time,143 end_time=end_time,144 plugins_used=plugins_used145 )146 147 # Get final results148 results = self.result_manager.to_dict(149 include_metadata=config.INCLUDE_METADATA150 )151 152 logger.info(f"Image analysis completed: {image_path.name} "153 f"({len(plugins_used)} plugins)")154 155 return results156 157 except Exception as e:158 logger.error(f"Image analysis failed: {e}")159 raise DeepVisionError(160 f"Analysis failed for {image_path.name}: {str(e)}",161 {"path": str(image_path), "error": str(e)}162 )163 164 def analyze_video(165 self,166 video_path: Union[str, Path],167 plugins: Optional[List[str]] = None,168 extract_method: str = "keyframes",169 num_frames: int = 5,170 **kwargs171 ) -> Dict[str, Any]:172 """173 Analyze a video by extracting and analyzing frames.174 175 Args:176 video_path: Path to video file177 plugins: List of plugin names to use178 extract_method: Frame extraction method ("fps" or "keyframes")179 num_frames: Number of frames to extract180 **kwargs: Additional arguments181 182 Returns:183 Analysis results dictionary184 """185 start_time = datetime.now()186 video_path = Path(video_path)187 188 logger.info(f"Starting video analysis: {video_path.name}")189 190 try:191 # Clear previous results192 self.result_manager.clear()193 194 # Get video info195 video_info = self.video_processor.get_video_info(video_path)196 197 # Set file metadata198 self.result_manager.set_file_info(199 filename=video_info["filename"],200 file_type="video",201 file_size=video_info["file_size"],202 width=video_info["width"],203 height=video_info["height"],204 fps=video_info["fps"],205 duration=video_info["duration"],206 frame_count=video_info["frame_count"],207 )208 209 # Extract frames210 if extract_method == "keyframes":211 frame_paths = self.video_processor.extract_key_frames(212 video_path,213 num_frames=num_frames214 )215 else:216 frame_paths = self.video_processor.extract_frames(217 video_path,218 max_frames=num_frames,219 **kwargs220 )221 222 logger.info(f"Extracted {len(frame_paths)} frames from video")223 224 # Analyze each frame225 frame_results = []226 for idx, frame_path in enumerate(frame_paths):227 logger.info(f"Analyzing frame {idx + 1}/{len(frame_paths)}")228 229 # Process frame230 image = self.image_processor.process(frame_path, resize=True)231 232 # Execute plugins on frame233 plugins_used = self._execute_plugins(234 image,235 frame_path,236 plugins,237 media_type="video_frame"238 )239 240 # Get frame results241 frame_result = {242 "frame_index": idx,243 "frame_path": str(frame_path.name),244 "results": dict(self.result_manager.results)245 }246 frame_results.append(frame_result)247 248 # Clear for next frame249 self.result_manager.results.clear()250 251 # Aggregate frame results252 aggregated = self._aggregate_video_results(frame_results)253 254 # Set aggregated results255 self.result_manager.results = aggregated256 257 # Set processing metadata258 end_time = datetime.now()259 self.result_manager.set_processing_info(260 start_time=start_time,261 end_time=end_time,262 plugins_used=plugins_used263 )264 265 # Add video-specific metadata266 self.result_manager.add_metadata({267 "frames_analyzed": len(frame_paths),268 "extraction_method": extract_method,269 })270 271 # Get final results272 results = self.result_manager.to_dict(273 include_metadata=config.INCLUDE_METADATA274 )275 276 logger.info(f"Video analysis completed: {video_path.name} "277 f"({len(frame_paths)} frames, {len(plugins_used)} plugins)")278 279 return results280 281 except Exception as e:282 logger.error(f"Video analysis failed: {e}")283 raise DeepVisionError(284 f"Analysis failed for {video_path.name}: {str(e)}",285 {"path": str(video_path), "error": str(e)}286 )287 288 def _execute_plugins(289 self,290 media,291 media_path: Path,292 plugin_names: Optional[List[str]] = None,293 media_type: str = "image"294 ) -> List[str]:295 """296 Execute registered plugins on media.297 298 Args:299 media: Processed media (image or frame)300 media_path: Path to media file301 plugin_names: List of plugins to execute (None for all)302 media_type: Type of media being processed303 304 Returns:305 List of executed plugin names306 """307 # Determine which plugins to execute308 if plugin_names is None:309 plugins_to_run = self.plugin_order310 else:311 plugins_to_run = [312 p for p in self.plugin_order if p in plugin_names313 ]314 315 executed = []316 317 for plugin_name in plugins_to_run:318 if plugin_name not in self.plugins:319 logger.warning(f"Plugin '{plugin_name}' not found, skipping")320 continue321 322 try:323 logger.debug(f"Executing plugin: {plugin_name}")324 325 plugin = self.plugins[plugin_name]326 327 # Execute plugin328 result = plugin.analyze(media, media_path)329 330 # Add result331 self.result_manager.add_result(plugin_name, result)332 333 executed.append(plugin_name)334 335 logger.debug(f"Plugin '{plugin_name}' completed successfully")336 337 except Exception as e:338 logger.error(f"Plugin '{plugin_name}' failed: {e}")339 340 # Add error to results341 self.result_manager.add_result(342 plugin_name,343 {344 "error": str(e),345 "status": "failed"346 }347 )348 349 return executed350 351 def _aggregate_video_results(352 self,353 frame_results: List[Dict[str, Any]]354 ) -> Dict[str, Any]:355 """356 Aggregate results from multiple video frames.357 358 Args:359 frame_results: List of results from each frame360 361 Returns:362 Aggregated results dictionary363 """364 aggregated = {365 "frames": frame_results,366 "summary": {}367 }368 369 # For each plugin, aggregate results across frames370 if not frame_results:371 return aggregated372 373 # Get plugin names from first frame374 first_frame = frame_results[0]["results"]375 376 for plugin_name in first_frame.keys():377 plugin_summary = self._aggregate_plugin_results(378 plugin_name,379 [f["results"].get(plugin_name, {}) for f in frame_results]380 )381 aggregated["summary"][plugin_name] = plugin_summary382 383 return aggregated384 385 def _aggregate_plugin_results(386 self,387 plugin_name: str,388 results: List[Dict[str, Any]]389 ) -> Dict[str, Any]:390 """391 Aggregate results for a specific plugin across frames.392 393 Args:394 plugin_name: Name of the plugin395 results: List of results from each frame396 397 Returns:398 Aggregated result for the plugin399 """400 # Default aggregation: collect all unique values401 aggregated = {402 "frames_processed": len(results),403 }404 405 # Plugin-specific aggregation logic406 if plugin_name == "object_detector":407 all_objects = []408 for result in results:409 all_objects.extend(result.get("objects", []))410 411 # Count object occurrences412 object_counts = {}413 for obj in all_objects:414 name = obj["name"]415 object_counts[name] = object_counts.get(name, 0) + 1416 417 aggregated["total_objects"] = len(all_objects)418 aggregated["unique_objects"] = len(object_counts)419 aggregated["object_frequency"] = object_counts420 421 elif plugin_name == "caption_generator":422 captions = [r.get("caption", "") for r in results if r.get("caption")]423 aggregated["captions"] = captions424 aggregated["caption_count"] = len(captions)425 426 elif plugin_name == "color_analyzer":427 all_colors = []428 for result in results:429 all_colors.extend(result.get("dominant_colors", []))430 431 # Get most frequent colors432 color_counts = {}433 for color in all_colors:434 name = color["name"]435 color_counts[name] = color_counts.get(name, 0) + 1436 437 aggregated["color_frequency"] = color_counts438 439 return aggregated440 441 def analyze(442 self,443 file_path: Union[str, Path],444 **kwargs445 ) -> Dict[str, Any]:446 """447 Automatically detect file type and analyze.448 449 Args:450 file_path: Path to image or video file451 **kwargs: Additional arguments452 453 Returns:454 Analysis results455 """456 file_path = Path(file_path)457 458 # Detect file type459 ext = file_path.suffix.lower()460 461 if ext in config.ALLOWED_IMAGE_FORMATS:462 return self.analyze_image(file_path, **kwargs)463 elif ext in config.ALLOWED_VIDEO_FORMATS:464 return self.analyze_video(file_path, **kwargs)465 else:466 raise ValueError(f"Unsupported file format: {ext}")467 468 def __repr__(self) -> str:469 """Object representation."""470 return (f"AnalysisEngine(plugins={len(self.plugins)}, "471 f"registered={self.get_registered_plugins()})")472