Spaces:
Sleeping
Sleeping
File size: 5,140 Bytes
5b6e956 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 |
"""
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
}
}
|