548 lines
20 KiB
Python
548 lines
20 KiB
Python
"""Image Studio Manager - Main orchestration service for all image operations."""
|
|
|
|
from typing import Optional, Dict, Any, List
|
|
|
|
from .create_service import CreateStudioService, CreateStudioRequest
|
|
from .edit_service import EditStudioService, EditStudioRequest
|
|
from .upscale_service import UpscaleStudioService, UpscaleStudioRequest
|
|
from .control_service import ControlStudioService, ControlStudioRequest
|
|
from .social_optimizer_service import SocialOptimizerService, SocialOptimizerRequest
|
|
from .face_swap_service import FaceSwapService, FaceSwapStudioRequest
|
|
from .compression_service import ImageCompressionService, CompressionRequest, CompressionResult
|
|
from .format_converter_service import ImageFormatConverterService, FormatConversionRequest, FormatConversionResult
|
|
from .transform_service import (
|
|
TransformStudioService,
|
|
TransformImageToVideoRequest,
|
|
TalkingAvatarRequest,
|
|
)
|
|
from .templates import Platform, TemplateCategory, ImageTemplate
|
|
from utils.logger_utils import get_service_logger
|
|
|
|
|
|
logger = get_service_logger("image_studio.manager")
|
|
|
|
|
|
class ImageStudioManager:
|
|
"""Main manager for Image Studio operations."""
|
|
|
|
def __init__(self):
|
|
"""Initialize Image Studio Manager."""
|
|
self.create_service = CreateStudioService()
|
|
self.edit_service = EditStudioService()
|
|
self.upscale_service = UpscaleStudioService()
|
|
self.control_service = ControlStudioService()
|
|
self.social_optimizer_service = SocialOptimizerService()
|
|
self.face_swap_service = FaceSwapService()
|
|
self.compression_service = ImageCompressionService()
|
|
self.format_converter_service = ImageFormatConverterService()
|
|
self.transform_service = TransformStudioService()
|
|
logger.info("[Image Studio Manager] Initialized successfully")
|
|
|
|
# ====================
|
|
# CREATE STUDIO
|
|
# ====================
|
|
|
|
async def create_image(
|
|
self,
|
|
request: CreateStudioRequest,
|
|
user_id: Optional[str] = None
|
|
) -> Dict[str, Any]:
|
|
"""Create/generate image using Create Studio.
|
|
|
|
Args:
|
|
request: Create studio request
|
|
user_id: User ID for validation
|
|
|
|
Returns:
|
|
Dictionary with generation results
|
|
"""
|
|
logger.info("[Image Studio] Create image request from user: %s", user_id)
|
|
return await self.create_service.generate(request, user_id=user_id)
|
|
|
|
# ====================
|
|
# EDIT STUDIO
|
|
# ====================
|
|
|
|
async def edit_image(
|
|
self,
|
|
request: EditStudioRequest,
|
|
user_id: Optional[str] = None,
|
|
) -> Dict[str, Any]:
|
|
"""Run Edit Studio operations."""
|
|
logger.info("[Image Studio] Edit image request from user: %s", user_id)
|
|
return await self.edit_service.process_edit(request, user_id=user_id)
|
|
|
|
def get_edit_operations(self) -> Dict[str, Any]:
|
|
"""Expose edit operations for UI."""
|
|
return self.edit_service.list_operations()
|
|
|
|
def get_edit_models(
|
|
self,
|
|
operation: Optional[str] = None,
|
|
tier: Optional[str] = None,
|
|
) -> Dict[str, Any]:
|
|
"""Get available editing models.
|
|
|
|
Args:
|
|
operation: Filter by operation type
|
|
tier: Filter by tier (budget, mid, premium)
|
|
|
|
Returns:
|
|
Dictionary with models and metadata
|
|
"""
|
|
return self.edit_service.get_available_models(operation=operation, tier=tier)
|
|
|
|
def recommend_edit_model(
|
|
self,
|
|
operation: str,
|
|
image_resolution: Optional[Dict[str, int]] = None,
|
|
user_tier: Optional[str] = None,
|
|
preferences: Optional[Dict[str, Any]] = None,
|
|
) -> Dict[str, Any]:
|
|
"""Recommend best editing model for given context.
|
|
|
|
Args:
|
|
operation: Operation type
|
|
image_resolution: Image dimensions
|
|
user_tier: User subscription tier
|
|
preferences: User preferences (prioritize_cost, prioritize_quality)
|
|
|
|
Returns:
|
|
Dictionary with recommended model and alternatives
|
|
"""
|
|
return self.edit_service.recommend_model(
|
|
operation=operation,
|
|
image_resolution=image_resolution,
|
|
user_tier=user_tier,
|
|
preferences=preferences,
|
|
)
|
|
|
|
# ====================
|
|
# FACE SWAP STUDIO
|
|
# ====================
|
|
|
|
async def face_swap(
|
|
self,
|
|
request: FaceSwapStudioRequest,
|
|
user_id: Optional[str] = None,
|
|
) -> Dict[str, Any]:
|
|
"""Run Face Swap Studio operations."""
|
|
logger.info("[Image Studio] Face swap request from user: %s", user_id)
|
|
return await self.face_swap_service.process_face_swap(request, user_id=user_id)
|
|
|
|
def get_face_swap_models(
|
|
self,
|
|
tier: Optional[str] = None,
|
|
) -> Dict[str, Any]:
|
|
"""Get available face swap models.
|
|
|
|
Args:
|
|
tier: Filter by tier (budget, mid, premium)
|
|
|
|
Returns:
|
|
Dictionary with models and metadata
|
|
"""
|
|
return self.face_swap_service.get_available_models(tier=tier)
|
|
|
|
def recommend_face_swap_model(
|
|
self,
|
|
base_image_resolution: Optional[Dict[str, int]] = None,
|
|
face_image_resolution: Optional[Dict[str, int]] = None,
|
|
user_tier: Optional[str] = None,
|
|
preferences: Optional[Dict[str, Any]] = None,
|
|
) -> Dict[str, Any]:
|
|
"""Recommend best face swap model for given context.
|
|
|
|
Args:
|
|
base_image_resolution: Base image dimensions
|
|
face_image_resolution: Face image dimensions
|
|
user_tier: User subscription tier
|
|
preferences: User preferences (prioritize_cost, prioritize_quality)
|
|
|
|
Returns:
|
|
Dictionary with recommended model and alternatives
|
|
"""
|
|
return self.face_swap_service.recommend_model(
|
|
base_image_resolution=base_image_resolution,
|
|
face_image_resolution=face_image_resolution,
|
|
user_tier=user_tier,
|
|
preferences=preferences,
|
|
)
|
|
|
|
# ====================
|
|
# UPSCALE STUDIO
|
|
# ====================
|
|
|
|
async def upscale_image(
|
|
self,
|
|
request: UpscaleStudioRequest,
|
|
user_id: Optional[str] = None,
|
|
) -> Dict[str, Any]:
|
|
"""Run Upscale Studio operations."""
|
|
logger.info("[Image Studio] Upscale request from user: %s", user_id)
|
|
return await self.upscale_service.process_upscale(request, user_id=user_id)
|
|
|
|
def get_templates(
|
|
self,
|
|
platform: Optional[Platform] = None,
|
|
category: Optional[TemplateCategory] = None
|
|
) -> List[ImageTemplate]:
|
|
"""Get available templates.
|
|
|
|
Args:
|
|
platform: Filter by platform
|
|
category: Filter by category
|
|
|
|
Returns:
|
|
List of templates
|
|
"""
|
|
return self.create_service.get_templates(platform=platform, category=category)
|
|
|
|
def search_templates(self, query: str) -> List[ImageTemplate]:
|
|
"""Search templates by query.
|
|
|
|
Args:
|
|
query: Search query
|
|
|
|
Returns:
|
|
List of matching templates
|
|
"""
|
|
return self.create_service.search_templates(query)
|
|
|
|
def recommend_templates(
|
|
self,
|
|
use_case: str,
|
|
platform: Optional[Platform] = None
|
|
) -> List[ImageTemplate]:
|
|
"""Recommend templates based on use case.
|
|
|
|
Args:
|
|
use_case: Use case description
|
|
platform: Optional platform filter
|
|
|
|
Returns:
|
|
List of recommended templates
|
|
"""
|
|
return self.create_service.recommend_templates(use_case, platform)
|
|
|
|
def get_providers(self) -> Dict[str, Any]:
|
|
"""Get available image providers and their capabilities.
|
|
|
|
Returns:
|
|
Dictionary of providers with capabilities
|
|
"""
|
|
return {
|
|
"stability": {
|
|
"name": "Stability AI",
|
|
"models": ["ultra", "core", "sd3.5-large"],
|
|
"capabilities": ["text-to-image", "editing", "upscaling", "control", "3d"],
|
|
"max_resolution": (2048, 2048),
|
|
"cost_range": "3-8 credits per image",
|
|
},
|
|
"wavespeed": {
|
|
"name": "WaveSpeed AI",
|
|
"models": ["ideogram-v3-turbo", "qwen-image"],
|
|
"capabilities": ["text-to-image", "photorealistic", "fast-generation"],
|
|
"max_resolution": (1024, 1024),
|
|
"cost_range": "$0.05-$0.10 per image",
|
|
},
|
|
"huggingface": {
|
|
"name": "HuggingFace",
|
|
"models": ["FLUX.1-Krea-dev", "RunwayML"],
|
|
"capabilities": ["text-to-image", "image-to-image"],
|
|
"max_resolution": (1024, 1024),
|
|
"cost_range": "Free tier available",
|
|
},
|
|
"gemini": {
|
|
"name": "Google Gemini",
|
|
"models": ["imagen-3.0"],
|
|
"capabilities": ["text-to-image", "conversational-editing"],
|
|
"max_resolution": (1024, 1024),
|
|
"cost_range": "Free tier available",
|
|
}
|
|
}
|
|
|
|
# ====================
|
|
# COST ESTIMATION
|
|
# ====================
|
|
|
|
def estimate_cost(
|
|
self,
|
|
provider: str,
|
|
model: Optional[str],
|
|
operation: str,
|
|
num_images: int = 1,
|
|
resolution: Optional[tuple[int, int]] = None
|
|
) -> Dict[str, Any]:
|
|
"""Estimate cost for image operations.
|
|
|
|
Args:
|
|
provider: Provider name
|
|
model: Model name
|
|
operation: Operation type (generate, edit, upscale, etc.)
|
|
num_images: Number of images
|
|
resolution: Image resolution (width, height)
|
|
|
|
Returns:
|
|
Cost estimation details
|
|
"""
|
|
# Base costs (adjust based on actual pricing)
|
|
base_costs = {
|
|
"stability": {
|
|
"ultra": 0.08, # 8 credits
|
|
"core": 0.03, # 3 credits
|
|
"sd3": 0.065, # 6.5 credits
|
|
},
|
|
"wavespeed": {
|
|
"ideogram-v3-turbo": 0.10,
|
|
"qwen-image": 0.05,
|
|
},
|
|
"huggingface": {
|
|
"default": 0.0, # Free tier
|
|
},
|
|
"gemini": {
|
|
"default": 0.0, # Free tier
|
|
}
|
|
}
|
|
|
|
# Get base cost
|
|
provider_costs = base_costs.get(provider, {})
|
|
cost_per_image = provider_costs.get(model, provider_costs.get("default", 0.0))
|
|
|
|
# Calculate total
|
|
total_cost = cost_per_image * num_images
|
|
|
|
return {
|
|
"provider": provider,
|
|
"model": model,
|
|
"operation": operation,
|
|
"num_images": num_images,
|
|
"resolution": f"{resolution[0]}x{resolution[1]}" if resolution else "default",
|
|
"cost_per_image": cost_per_image,
|
|
"total_cost": total_cost,
|
|
"currency": "USD",
|
|
"estimated": True,
|
|
}
|
|
|
|
# ====================
|
|
# CONTROL STUDIO
|
|
# ====================
|
|
|
|
async def control_image(
|
|
self,
|
|
request: ControlStudioRequest,
|
|
user_id: Optional[str] = None,
|
|
) -> Dict[str, Any]:
|
|
"""Run Control Studio operations."""
|
|
logger.info("[Image Studio] Control request from user: %s", user_id)
|
|
return await self.control_service.process_control(request, user_id=user_id)
|
|
|
|
def get_control_operations(self) -> Dict[str, Any]:
|
|
"""Expose control operations for UI."""
|
|
return self.control_service.list_operations()
|
|
|
|
# ====================
|
|
# SOCIAL OPTIMIZER
|
|
# ====================
|
|
|
|
async def optimize_for_social(
|
|
self,
|
|
request: SocialOptimizerRequest,
|
|
user_id: Optional[str] = None,
|
|
) -> Dict[str, Any]:
|
|
"""Optimize image for social media platforms."""
|
|
logger.info("[Image Studio] Social optimization request from user: %s", user_id)
|
|
return self.social_optimizer_service.optimize_image(request)
|
|
|
|
def get_social_platform_formats(self, platform: Platform) -> List[Dict[str, Any]]:
|
|
"""Get available formats for a social platform."""
|
|
return self.social_optimizer_service.get_platform_formats(platform)
|
|
|
|
# ====================
|
|
# PLATFORM SPECS
|
|
# ====================
|
|
|
|
def get_platform_specs(self, platform: Platform) -> Dict[str, Any]:
|
|
"""Get platform specifications and requirements.
|
|
|
|
Args:
|
|
platform: Platform to get specs for
|
|
|
|
Returns:
|
|
Platform specifications
|
|
"""
|
|
specs = {
|
|
Platform.INSTAGRAM: {
|
|
"name": "Instagram",
|
|
"formats": [
|
|
{"name": "Feed Post (Square)", "ratio": "1:1", "size": "1080x1080"},
|
|
{"name": "Feed Post (Portrait)", "ratio": "4:5", "size": "1080x1350"},
|
|
{"name": "Story", "ratio": "9:16", "size": "1080x1920"},
|
|
{"name": "Reel", "ratio": "9:16", "size": "1080x1920"},
|
|
],
|
|
"file_types": ["JPG", "PNG"],
|
|
"max_file_size": "30MB",
|
|
},
|
|
Platform.FACEBOOK: {
|
|
"name": "Facebook",
|
|
"formats": [
|
|
{"name": "Feed Post", "ratio": "1.91:1", "size": "1200x630"},
|
|
{"name": "Feed Post (Square)", "ratio": "1:1", "size": "1080x1080"},
|
|
{"name": "Story", "ratio": "9:16", "size": "1080x1920"},
|
|
{"name": "Cover Photo", "ratio": "16:9", "size": "820x312"},
|
|
],
|
|
"file_types": ["JPG", "PNG"],
|
|
"max_file_size": "30MB",
|
|
},
|
|
Platform.TWITTER: {
|
|
"name": "Twitter/X",
|
|
"formats": [
|
|
{"name": "Post", "ratio": "16:9", "size": "1200x675"},
|
|
{"name": "Card", "ratio": "2:1", "size": "1200x600"},
|
|
{"name": "Header", "ratio": "3:1", "size": "1500x500"},
|
|
],
|
|
"file_types": ["JPG", "PNG", "GIF"],
|
|
"max_file_size": "5MB",
|
|
},
|
|
Platform.LINKEDIN: {
|
|
"name": "LinkedIn",
|
|
"formats": [
|
|
{"name": "Feed Post", "ratio": "1.91:1", "size": "1200x628"},
|
|
{"name": "Feed Post (Square)", "ratio": "1:1", "size": "1080x1080"},
|
|
{"name": "Article", "ratio": "2:1", "size": "1200x627"},
|
|
{"name": "Company Cover", "ratio": "4:1", "size": "1128x191"},
|
|
],
|
|
"file_types": ["JPG", "PNG"],
|
|
"max_file_size": "8MB",
|
|
},
|
|
Platform.YOUTUBE: {
|
|
"name": "YouTube",
|
|
"formats": [
|
|
{"name": "Thumbnail", "ratio": "16:9", "size": "1280x720"},
|
|
{"name": "Channel Art", "ratio": "16:9", "size": "2560x1440"},
|
|
],
|
|
"file_types": ["JPG", "PNG"],
|
|
"max_file_size": "2MB",
|
|
},
|
|
Platform.PINTEREST: {
|
|
"name": "Pinterest",
|
|
"formats": [
|
|
{"name": "Pin", "ratio": "2:3", "size": "1000x1500"},
|
|
{"name": "Story Pin", "ratio": "9:16", "size": "1080x1920"},
|
|
],
|
|
"file_types": ["JPG", "PNG"],
|
|
"max_file_size": "20MB",
|
|
},
|
|
Platform.TIKTOK: {
|
|
"name": "TikTok",
|
|
"formats": [
|
|
{"name": "Video Cover", "ratio": "9:16", "size": "1080x1920"},
|
|
],
|
|
"file_types": ["JPG", "PNG"],
|
|
"max_file_size": "10MB",
|
|
},
|
|
}
|
|
|
|
return specs.get(platform, {})
|
|
|
|
# ====================
|
|
# TRANSFORM STUDIO
|
|
# ====================
|
|
|
|
async def transform_image_to_video(
|
|
self,
|
|
request: TransformImageToVideoRequest,
|
|
user_id: Optional[str] = None,
|
|
) -> Dict[str, Any]:
|
|
"""Transform image to video using WAN 2.5."""
|
|
logger.info("[Image Studio] Transform image-to-video request from user: %s", user_id)
|
|
return await self.transform_service.transform_image_to_video(request, user_id=user_id or "anonymous")
|
|
|
|
async def create_talking_avatar(
|
|
self,
|
|
request: TalkingAvatarRequest,
|
|
user_id: Optional[str] = None,
|
|
) -> Dict[str, Any]:
|
|
"""Create talking avatar using InfiniteTalk."""
|
|
logger.info("[Image Studio] Talking avatar request from user: %s", user_id)
|
|
return await self.transform_service.create_talking_avatar(request, user_id=user_id or "anonymous")
|
|
|
|
def estimate_transform_cost(
|
|
self,
|
|
operation: str,
|
|
resolution: str,
|
|
duration: Optional[int] = None,
|
|
) -> Dict[str, Any]:
|
|
"""Estimate cost for transform operation."""
|
|
return self.transform_service.estimate_cost(operation, resolution, duration)
|
|
|
|
# ====================
|
|
# COMPRESSION STUDIO
|
|
# ====================
|
|
|
|
async def compress_image(
|
|
self,
|
|
request: CompressionRequest,
|
|
user_id: Optional[str] = None,
|
|
) -> CompressionResult:
|
|
"""Compress an image with specified settings."""
|
|
logger.info("[Image Studio] Compress image request from user: %s", user_id)
|
|
return await self.compression_service.compress(request, user_id=user_id)
|
|
|
|
async def compress_batch(
|
|
self,
|
|
requests: List[CompressionRequest],
|
|
user_id: Optional[str] = None,
|
|
) -> List[CompressionResult]:
|
|
"""Compress multiple images."""
|
|
logger.info("[Image Studio] Batch compress request (%d images) from user: %s", len(requests), user_id)
|
|
return await self.compression_service.compress_batch(requests, user_id=user_id)
|
|
|
|
async def estimate_compression(
|
|
self,
|
|
image_base64: str,
|
|
format: str = "jpeg",
|
|
quality: int = 85,
|
|
) -> Dict[str, Any]:
|
|
"""Estimate compression results without compressing."""
|
|
return await self.compression_service.estimate_compression(image_base64, format, quality)
|
|
|
|
def get_compression_formats(self) -> List[Dict[str, Any]]:
|
|
"""Get supported compression formats."""
|
|
return self.compression_service.get_supported_formats()
|
|
|
|
def get_compression_presets(self) -> List[Dict[str, Any]]:
|
|
"""Get compression presets for common use cases."""
|
|
return self.compression_service.get_presets()
|
|
|
|
# ====================
|
|
# FORMAT CONVERTER
|
|
# ====================
|
|
|
|
async def convert_format(
|
|
self,
|
|
request: FormatConversionRequest,
|
|
user_id: Optional[str] = None,
|
|
) -> FormatConversionResult:
|
|
"""Convert an image to target format."""
|
|
logger.info("[Image Studio] Convert format request from user: %s", user_id)
|
|
return await self.format_converter_service.convert(request, user_id=user_id)
|
|
|
|
async def convert_format_batch(
|
|
self,
|
|
requests: List[FormatConversionRequest],
|
|
user_id: Optional[str] = None,
|
|
) -> List[FormatConversionResult]:
|
|
"""Convert multiple images."""
|
|
logger.info("[Image Studio] Batch convert format request (%d images) from user: %s", len(requests), user_id)
|
|
return await self.format_converter_service.convert_batch(requests, user_id=user_id)
|
|
|
|
def get_supported_formats(self) -> List[Dict[str, Any]]:
|
|
"""Get supported conversion formats."""
|
|
return self.format_converter_service.get_supported_formats()
|
|
|
|
def get_format_recommendations(self, source_format: str) -> List[Dict[str, Any]]:
|
|
"""Get format recommendations based on source format."""
|
|
return self.format_converter_service.get_format_recommendations(source_format)
|
|
|