Spaces:
Sleeping
Sleeping
| """ | |
| Gemini Backend Plugin | |
| Plugin adapter for Gemini 2.5 Flash Image API backend. | |
| """ | |
| import sys | |
| from pathlib import Path | |
| from typing import Any, Dict, Optional, List | |
| from PIL import Image | |
| # Add parent directories to path for imports | |
| sys.path.insert(0, str(Path(__file__).parent.parent)) | |
| from core.gemini_client import GeminiClient | |
| from models.generation_request import GenerationRequest | |
| from config.settings import Settings | |
| # Import from shared plugin system | |
| sys.path.insert(0, str(Path(__file__).parent.parent.parent / 'shared')) | |
| from plugin_system.base_plugin import BaseBackendPlugin | |
| class GeminiPlugin(BaseBackendPlugin): | |
| """Plugin adapter for Gemini API backend.""" | |
| def __init__(self, config_path: Path): | |
| """Initialize Gemini plugin.""" | |
| super().__init__(config_path) | |
| # Get API key from settings | |
| settings = Settings() | |
| api_key = settings.get_api_key() | |
| if api_key: | |
| self.client = GeminiClient(api_key) | |
| self.available = True | |
| else: | |
| self.client = None | |
| self.available = False | |
| print("Warning: Gemini API key not found") | |
| def health_check(self) -> bool: | |
| """Check if Gemini backend is available.""" | |
| return self.available and self.client is not None | |
| def generate_image( | |
| self, | |
| prompt: str, | |
| input_images: Optional[List[Image.Image]] = None, | |
| **kwargs | |
| ) -> Image.Image: | |
| """ | |
| Generate image using Gemini backend. | |
| Args: | |
| prompt: Text prompt for generation | |
| input_images: Optional list of input images | |
| **kwargs: Additional generation parameters | |
| Returns: | |
| Generated PIL Image | |
| """ | |
| if not self.health_check(): | |
| raise RuntimeError("Gemini backend not available") | |
| # Create generation request | |
| request = GenerationRequest( | |
| prompt=prompt, | |
| input_images=input_images or [], | |
| aspect_ratio=kwargs.get('aspect_ratio', '1:1'), | |
| number_of_images=kwargs.get('number_of_images', 1), | |
| safety_filter_level=kwargs.get('safety_filter_level', 'block_some'), | |
| person_generation=kwargs.get('person_generation', 'allow_all') | |
| ) | |
| # Generate image | |
| result = self.client.generate(request) | |
| if result.images: | |
| return result.images[0] | |
| else: | |
| raise RuntimeError(f"Gemini generation failed: {result.error}") | |
| def get_capabilities(self) -> Dict[str, Any]: | |
| """Report Gemini backend capabilities.""" | |
| return { | |
| 'name': 'Gemini 2.5 Flash Image', | |
| 'type': 'cloud', | |
| 'supports_input_images': True, | |
| 'supports_multi_image': True, | |
| 'max_input_images': 16, | |
| 'supports_aspect_ratios': True, | |
| 'available_aspect_ratios': ['1:1', '3:4', '4:3', '9:16', '16:9'], | |
| 'supports_safety_filter': True, | |
| 'supports_person_generation': True, | |
| 'estimated_time_per_image': 3.0, # seconds | |
| 'cost_per_image': 0.02, # USD estimate | |
| } | |