""" Backend Configuration Defines backend location and connection types. Supports local, network, and cloud backends. """ from enum import Enum from typing import Optional, Dict, Any from dataclasses import dataclass, field class BackendLocation(Enum): """Backend deployment location.""" LOCAL = "local" # Running in project structure NETWORK = "network" # Running on LAN (IP:PORT) CLOUD = "cloud" # Commercial API over internet class BackendProtocol(Enum): """Communication protocol for backend.""" PYTHON = "python" # Direct Python import HTTP = "http" # HTTP REST API WEBSOCKET = "websocket" # WebSocket connection GRPC = "grpc" # gRPC @dataclass class BackendConnectionConfig: """ Configuration for connecting to a backend. Supports three deployment scenarios: 1. Local: Backend runs in project, direct Python import 2. Network: Backend runs on LAN, communicate via HTTP/WebSocket 3. Cloud: Commercial API, communicate via HTTPS """ # Backend identity name: str backend_type: str # e.g., "gemini", "omnigen2", "comfyui" # Location location: BackendLocation protocol: BackendProtocol # Connection details endpoint: Optional[str] = None # URL or IP:PORT api_key: Optional[str] = None # For authenticated APIs # Capabilities capabilities: Dict[str, Any] = field(default_factory=dict) # Timeouts and limits timeout: int = 120 # seconds max_retries: int = 3 # Health check health_check_endpoint: Optional[str] = None health_check_interval: int = 60 # seconds def __post_init__(self): """Validate configuration.""" if self.location == BackendLocation.LOCAL: # Local backends use Python protocol if self.protocol != BackendProtocol.PYTHON: raise ValueError("Local backends must use PYTHON protocol") elif self.location in [BackendLocation.NETWORK, BackendLocation.CLOUD]: # Network/Cloud backends need endpoint if not self.endpoint: raise ValueError(f"{self.location.value} backends require endpoint") # Network/Cloud use HTTP/WebSocket/gRPC if self.protocol == BackendProtocol.PYTHON: raise ValueError(f"{self.location.value} backends cannot use PYTHON protocol") def get_full_endpoint(self, path: str = "") -> str: """Get full endpoint URL with path.""" if not self.endpoint: return "" base = self.endpoint.rstrip('/') path = path.lstrip('/') return f"{base}/{path}" if path else base @classmethod def from_dict(cls, data: Dict[str, Any]) -> 'BackendConnectionConfig': """Create configuration from dictionary.""" return cls( name=data['name'], backend_type=data['backend_type'], location=BackendLocation(data['location']), protocol=BackendProtocol(data['protocol']), endpoint=data.get('endpoint'), api_key=data.get('api_key'), capabilities=data.get('capabilities', {}), timeout=data.get('timeout', 120), max_retries=data.get('max_retries', 3), health_check_endpoint=data.get('health_check_endpoint'), health_check_interval=data.get('health_check_interval', 60) ) def to_dict(self) -> Dict[str, Any]: """Convert configuration to dictionary.""" return { 'name': self.name, 'backend_type': self.backend_type, 'location': self.location.value, 'protocol': self.protocol.value, 'endpoint': self.endpoint, 'api_key': self.api_key, 'capabilities': self.capabilities, 'timeout': self.timeout, 'max_retries': self.max_retries, 'health_check_endpoint': self.health_check_endpoint, 'health_check_interval': self.health_check_interval } # Example configurations: # Local backend (running in project) EXAMPLE_LOCAL = { 'name': 'omnigen2_local', 'backend_type': 'omnigen2', 'location': 'local', 'protocol': 'python', 'capabilities': { 'supports_multi_image': True, 'max_resolution': 2048 } } # Network backend (running on LAN) EXAMPLE_NETWORK = { 'name': 'omnigen2_server', 'backend_type': 'omnigen2', 'location': 'network', 'protocol': 'http', 'endpoint': 'http://192.168.1.100:8000', 'health_check_endpoint': '/health', 'capabilities': { 'supports_multi_image': True, 'max_resolution': 2048 } } # Cloud backend (commercial API) EXAMPLE_CLOUD = { 'name': 'gemini_cloud', 'backend_type': 'gemini', 'location': 'cloud', 'protocol': 'http', 'endpoint': 'https://generativelanguage.googleapis.com/v1', 'api_key': 'YOUR_API_KEY', 'health_check_endpoint': '/models', 'capabilities': { 'supports_multi_image': True, 'max_input_images': 16, 'max_resolution': 4096 } }