Initial FastAPI backend for HF Spaces

README.md

@@ -0,0 +1,83 @@
+---
+title: Text to 3D Backend
+emoji: 🎨
+colorFrom: blue
+colorTo: purple
+sdk: docker
+pinned: false
+license: mit
+app_port: 7860
+---
+
+# Text to 3D Model Converter - Backend
+
+This is the backend API for the Text-to-3D Model Converter application. It provides FastAPI endpoints for:
+
+- **Text-to-3D**: Generate 3D models from text descriptions
+- **Image-to-3D**: Convert uploaded images to 3D models
+- **Progress Tracking**: Real-time job progress monitoring
+
+## Features
+
+- 🎨 **Direct Model Loading**: Stable Diffusion and DPT models loaded locally for fast inference
+- ⚡ **Async Processing**: Background job processing with progress tracking
+- 🔄 **Job Management**: Cancel jobs, track progress, and get results
+- ☁️ **Cloud Storage**: Automatic upload to Cloudinary for results
+- 🚀 **FastAPI**: High-performance API with automatic docs
+
+## API Endpoints
+
+- `GET /` - Health check and model status
+- `POST /generate` - Generate 3D model from text prompt
+- `POST /upload` - Convert uploaded image to 3D model
+- `GET /progress/{job_id}` - Get job progress
+- `POST /cancel` - Cancel a running job
+
+## Models Used
+
+- **Image Generation**: Stable Diffusion v1.5 (runwayml/stable-diffusion-v1-5)
+- **Depth Estimation**: DPT (Intel/dpt-beit-large-512)
+- **3D Reconstruction**: Open3D Poisson surface reconstruction
+
+## Environment Variables
+
+Set these in the Space settings:
+
+```
+CLOUDINARY_CLOUD_NAME=your_cloud_name
+CLOUDINARY_API_KEY=your_api_key
+CLOUDINARY_API_SECRET=your_api_secret
+```
+
+## Usage
+
+The API is designed to work with the frontend application deployed on Render. CORS is configured to allow requests from the frontend domain.
+
+### Example Request
+
+```python
+import requests
+
+# Generate 3D model from text
+response = requests.post(
+    "https://your-space-url/generate",
+    json={"prompt": "a red sports car"}
+)
+
+job_id = response.json()["job_id"]
+
+# Check progress
+progress = requests.get(f"https://your-space-url/progress/{job_id}")
+print(progress.json())
+```
+
+## Development
+
+To run locally:
+
+```bash
+pip install -r requirements.txt
+python app.py
+```
+
+The API will be available at `http://localhost:7860`

app.py

@@ -0,0 +1,365 @@
+"""
+FastAPI Backend for Text-to-3D Model Converter
+Deployed on Hugging Face Spaces with direct model loading
+"""
+
+import os
+import logging
+import time
+import uuid
+import asyncio
+from typing import Optional
+from contextlib import asynccontextmanager
+
+import uvicorn
+from fastapi import FastAPI, File, UploadFile, HTTPException, BackgroundTasks
+from fastapi.middleware.cors import CORSMiddleware
+from fastapi.responses import JSONResponse
+from pydantic import BaseModel
+
+from models.depth_processor import DepthProcessor
+from models.image_generator import ImageGenerator
+from utils.job_manager import JobManager
+from utils.cloudinary_client import CloudinaryClient
+
+# Configure logging
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger(__name__)
+
+# Global variables for models
+depth_processor = None
+image_generator = None
+job_manager = None
+cloudinary_client = None
+
+@asynccontextmanager
+async def lifespan(app: FastAPI):
+    """Initialize models on startup"""
+    global depth_processor, image_generator, job_manager, cloudinary_client
+    
+    logger.info("🚀 Starting Text-to-3D Backend...")
+    
+    # Initialize utilities
+    job_manager = JobManager()
+    cloudinary_client = CloudinaryClient()
+    
+    # Initialize models
+    logger.info("📦 Loading AI models...")
+    try:
+        # Initialize depth processor
+        depth_processor = DepthProcessor()
+        await asyncio.to_thread(depth_processor.load_model)
+        logger.info("✅ Depth estimation model loaded")
+        
+        # Initialize image generator
+        image_generator = ImageGenerator()
+        await asyncio.to_thread(image_generator.load_model)
+        logger.info("✅ Image generation model loaded")
+        
+        logger.info("🎉 All models loaded successfully!")
+        
+    except Exception as e:
+        logger.error(f"❌ Failed to load models: {str(e)}")
+        raise e
+    
+    yield
+    
+    # Cleanup on shutdown
+    logger.info("🔄 Shutting down...")
+
+# Initialize FastAPI app
+app = FastAPI(
+    title="Text-to-3D Backend",
+    description="Convert text prompts and images to 3D models",
+    version="1.0.0",
+    lifespan=lifespan
+)
+
+# Configure CORS
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=[
+        "http://localhost:3000",  # Local development
+        "https://*.render.com",   # Render deployment
+        "*"  # Allow all for now, restrict in production
+    ],
+    allow_credentials=True,
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
+
+# Request/Response models
+class GenerateRequest(BaseModel):
+    prompt: str
+    user_id: Optional[str] = None
+
+class GenerateResponse(BaseModel):
+    success: bool
+    job_id: str
+    image_url: Optional[str] = None
+    model_url: Optional[str] = None
+    depth_map_url: Optional[str] = None
+    error: Optional[str] = None
+
+class ProgressResponse(BaseModel):
+    stage: str
+    progress: int
+    message: str
+    timestamp: Optional[float] = None
+
+@app.get("/")
+async def root():
+    """Health check endpoint"""
+    return {
+        "status": "Text-to-3D Backend is running! 🚀",
+        "version": "1.0.0",
+        "models_loaded": {
+            "depth_processor": depth_processor is not None,
+            "image_generator": image_generator is not None
+        },
+        "gpu_available": depth_processor.device.type == "cuda" if depth_processor else False
+    }
+
+@app.get("/health")
+async def health_check():
+    """Detailed health check"""
+    return {
+        "status": "healthy",
+        "models": {
+            "depth_estimation": "loaded" if depth_processor else "not_loaded",
+            "image_generation": "loaded" if image_generator else "not_loaded"
+        },
+        "device": str(depth_processor.device) if depth_processor else "unknown",
+        "active_jobs": job_manager.get_active_job_count() if job_manager else 0
+    }
+
+@app.post("/generate", response_model=GenerateResponse)
+async def generate_from_text(
+    request: GenerateRequest,
+    background_tasks: BackgroundTasks
+):
+    """Generate 3D model from text prompt"""
+    try:
+        if not request.prompt.strip():
+            raise HTTPException(status_code=400, detail="Prompt cannot be empty")
+        
+        # Create job ID
+        job_id = str(uuid.uuid4())
+        job_manager.register_job(job_id)
+        
+        logger.info(f"🎨 Starting text-to-3D generation: '{request.prompt}' (Job: {job_id})")
+        
+        # Start background processing
+        background_tasks.add_task(
+            process_text_to_3d,
+            job_id, 
+            request.prompt, 
+            request.user_id
+        )
+        
+        return GenerateResponse(
+            success=True,
+            job_id=job_id,
+            message="Generation started"
+        )
+        
+    except Exception as e:
+        logger.error(f"❌ Error in generate endpoint: {str(e)}")
+        raise HTTPException(status_code=500, detail=str(e))
+
+@app.post("/upload")
+async def upload_image(
+    file: UploadFile = File(...),
+    background_tasks: BackgroundTasks = None,
+    user_id: Optional[str] = None
+):
+    """Convert uploaded image to 3D model"""
+    try:
+        # Validate file type
+        if not file.content_type.startswith('image/'):
+            raise HTTPException(status_code=400, detail="File must be an image")
+        
+        # Create job ID
+        job_id = str(uuid.uuid4())
+        job_manager.register_job(job_id)
+        
+        logger.info(f"📤 Processing uploaded image: {file.filename} (Job: {job_id})")
+        
+        # Read file content
+        file_content = await file.read()
+        
+        # Start background processing
+        background_tasks.add_task(
+            process_upload_to_3d,
+            job_id,
+            file_content,
+            file.filename,
+            user_id
+        )
+        
+        return {
+            "success": True,
+            "job_id": job_id,
+            "message": "Upload processing started"
+        }
+        
+    except Exception as e:
+        logger.error(f"❌ Error in upload endpoint: {str(e)}")
+        raise HTTPException(status_code=500, detail=str(e))
+
+@app.get("/progress/{job_id}", response_model=ProgressResponse)
+async def get_progress(job_id: str):
+    """Get job progress"""
+    try:
+        progress = job_manager.get_job_progress(job_id)
+        if not progress:
+            raise HTTPException(status_code=404, detail="Job not found")
+        
+        return ProgressResponse(**progress)
+        
+    except Exception as e:
+        logger.error(f"❌ Error getting progress: {str(e)}")
+        raise HTTPException(status_code=500, detail=str(e))
+
+@app.post("/cancel")
+async def cancel_job(job_id: str):
+    """Cancel a running job"""
+    try:
+        success = job_manager.cancel_job(job_id)
+        if success:
+            return {"success": True, "message": f"Job {job_id} cancelled"}
+        else:
+            raise HTTPException(status_code=404, detail="Job not found")
+            
+    except Exception as e:
+        logger.error(f"❌ Error cancelling job: {str(e)}")
+        raise HTTPException(status_code=500, detail=str(e))
+
+async def process_text_to_3d(job_id: str, prompt: str, user_id: Optional[str]):
+    """Background task to process text to 3D"""
+    try:
+        # Update progress
+        job_manager.update_job_progress(job_id, "generating_image", 10, "Generating image from text...")
+        
+        # Generate image from text
+        image_result = await asyncio.to_thread(
+            image_generator.generate_image, 
+            prompt
+        )
+        
+        if job_manager.is_job_cancelled(job_id):
+            return
+        
+        job_manager.update_job_progress(job_id, "uploading_image", 40, "Uploading generated image...")
+        
+        # Upload image to Cloudinary
+        image_url = await asyncio.to_thread(
+            cloudinary_client.upload_image_from_bytes,
+            image_result['image_bytes'],
+            f"generated_{job_id}"
+        )
+        
+        if job_manager.is_job_cancelled(job_id):
+            return
+        
+        job_manager.update_job_progress(job_id, "creating_depth", 60, "Creating depth map...")
+        
+        # Generate depth map and 3D model
+        depth_result = await asyncio.to_thread(
+            depth_processor.process_image_to_3d,
+            image_result['image_pil'],
+            job_id
+        )
+        
+        if job_manager.is_job_cancelled(job_id):
+            return
+        
+        job_manager.update_job_progress(job_id, "uploading_results", 90, "Uploading 3D model...")
+        
+        # Upload results
+        model_url = await asyncio.to_thread(
+            cloudinary_client.upload_file,
+            depth_result['obj_path'],
+            f"model_{job_id}.obj"
+        )
+        
+        depth_map_url = await asyncio.to_thread(
+            cloudinary_client.upload_image_from_path,
+            depth_result['depth_map_path'],
+            f"depth_{job_id}"
+        )
+        
+        # Complete job
+        job_manager.complete_job(job_id, {
+            "image_url": image_url,
+            "model_url": model_url,
+            "depth_map_url": depth_map_url
+        })
+        
+        logger.info(f"✅ Text-to-3D generation completed: {job_id}")
+        
+    except Exception as e:
+        logger.error(f"❌ Error in text-to-3D processing: {str(e)}")
+        job_manager.fail_job(job_id, str(e))
+
+async def process_upload_to_3d(job_id: str, file_content: bytes, filename: str, user_id: Optional[str]):
+    """Background task to process uploaded image to 3D"""
+    try:
+        job_manager.update_job_progress(job_id, "uploading", 20, "Uploading image to cloud...")
+        
+        # Upload original image
+        image_url = await asyncio.to_thread(
+            cloudinary_client.upload_image_from_bytes,
+            file_content,
+            f"upload_{job_id}_{filename}"
+        )
+        
+        if job_manager.is_job_cancelled(job_id):
+            return
+        
+        job_manager.update_job_progress(job_id, "processing", 50, "Processing image to 3D...")
+        
+        # Convert to PIL Image
+        from PIL import Image
+        image_pil = Image.open(io.BytesIO(file_content))
+        
+        # Generate depth map and 3D model
+        depth_result = await asyncio.to_thread(
+            depth_processor.process_image_to_3d,
+            image_pil,
+            job_id
+        )
+        
+        if job_manager.is_job_cancelled(job_id):
+            return
+        
+        job_manager.update_job_progress(job_id, "uploading_results", 90, "Uploading 3D model...")
+        
+        # Upload results
+        model_url = await asyncio.to_thread(
+            cloudinary_client.upload_file,
+            depth_result['obj_path'],
+            f"model_{job_id}.obj"
+        )
+        
+        depth_map_url = await asyncio.to_thread(
+            cloudinary_client.upload_image_from_path,
+            depth_result['depth_map_path'],
+            f"depth_{job_id}"
+        )
+        
+        # Complete job
+        job_manager.complete_job(job_id, {
+            "image_url": image_url,
+            "model_url": model_url,
+            "depth_map_url": depth_map_url
+        })
+        
+        logger.info(f"✅ Upload-to-3D processing completed: {job_id}")
+        
+    except Exception as e:
+        logger.error(f"❌ Error in upload-to-3D processing: {str(e)}")
+        job_manager.fail_job(job_id, str(e))
+
+if __name__ == "__main__":
+    uvicorn.run(app, host="0.0.0.0", port=7860)

models/__init__.py

@@ -0,0 +1 @@
+# Models package

models/depth_processor.py

@@ -0,0 +1,195 @@
+"""
+Depth processing module for converting 2D images to depth maps and 3D models
+"""
+
+import os
+import logging
+import tempfile
+import numpy as np
+import torch
+from PIL import Image
+from transformers import DPTImageProcessor, DPTForDepthEstimation
+import open3d as o3d
+import matplotlib.pyplot as plt
+
+logger = logging.getLogger(__name__)
+
+class DepthProcessor:
+    """Handles depth estimation and 3D model generation"""
+    
+    def __init__(self):
+        self.device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
+        self.processor = None
+        self.model = None
+        self.temp_dir = tempfile.mkdtemp()
+        
+    def load_model(self):
+        """Load the DPT depth estimation model"""
+        try:
+            logger.info(f"🔄 Loading DPT model on {self.device}...")
+            
+            # Load processor and model
+            self.processor = DPTImageProcessor.from_pretrained("Intel/dpt-beit-large-512")
+            self.model = DPTForDepthEstimation.from_pretrained("Intel/dpt-beit-large-512")
+            self.model.to(self.device)
+            self.model.eval()
+            
+            if self.device.type == "cuda":
+                logger.info(f"✅ DPT model loaded on GPU: {torch.cuda.get_device_name(0)}")
+            else:
+                logger.info("✅ DPT model loaded on CPU")
+                
+        except Exception as e:
+            logger.error(f"❌ Failed to load DPT model: {str(e)}")
+            raise e
+    
+    def generate_depth_map(self, image: Image.Image) -> np.ndarray:
+        """Generate depth map from PIL Image"""
+        try:
+            # Prepare image for model
+            inputs = self.processor(images=image, return_tensors="pt")
+            inputs = {k: v.to(self.device) for k, v in inputs.items()}
+            
+            # Generate depth map
+            with torch.no_grad():
+                outputs = self.model(**inputs)
+                predicted_depth = outputs.predicted_depth
+            
+            # Convert to numpy and normalize
+            depth = predicted_depth.squeeze().cpu().numpy()
+            depth_normalized = (depth - depth.min()) / (depth.max() - depth.min())
+            
+            return depth_normalized
+            
+        except Exception as e:
+            logger.error(f"❌ Error generating depth map: {str(e)}")
+            raise e
+    
+    def save_depth_map_image(self, depth_map: np.ndarray, job_id: str) -> str:
+        """Save depth map as image file"""
+        try:
+            # Create colorized depth map
+            plt.figure(figsize=(10, 10))
+            plt.imshow(depth_map, cmap='plasma')
+            plt.axis('off')
+            plt.tight_layout()
+            
+            # Save image
+            depth_path = os.path.join(self.temp_dir, f"depth_{job_id}.png")
+            plt.savefig(depth_path, bbox_inches='tight', pad_inches=0, dpi=150)
+            plt.close()
+            
+            return depth_path
+            
+        except Exception as e:
+            logger.error(f"❌ Error saving depth map image: {str(e)}")
+            raise e
+    
+    def create_3d_model(self, image: Image.Image, depth_map: np.ndarray, job_id: str) -> str:
+        """Create 3D OBJ model from image and depth map"""
+        try:
+            # Convert image to numpy array
+            img_array = np.array(image)
+            h, w = depth_map.shape
+            
+            # Create point cloud
+            points = []
+            colors = []
+            
+            # Sample points (reduce resolution for performance)
+            step = max(1, min(h, w) // 200)  # Target ~200x200 points max
+            
+            for y in range(0, h, step):
+                for x in range(0, w, step):
+                    # Get depth value (invert for proper 3D orientation)
+                    z = (1.0 - depth_map[y, x]) * 50.0  # Scale depth
+                    
+                    # Skip points that are too far
+                    if z > 45.0:
+                        continue
+                    
+                    # Add point
+                    points.append([x / w - 0.5, (h - y) / h - 0.5, z])
+                    
+                    # Add color
+                    if len(img_array.shape) == 3:
+                        colors.append(img_array[y, x] / 255.0)
+                    else:
+                        colors.append([0.7, 0.7, 0.7])  # Gray for grayscale
+            
+            if not points:
+                raise ValueError("No valid points generated for 3D model")
+            
+            # Create Open3D point cloud
+            pcd = o3d.geometry.PointCloud()
+            pcd.points = o3d.utility.Vector3dVector(np.array(points))
+            pcd.colors = o3d.utility.Vector3dVector(np.array(colors))
+            
+            # Estimate normals
+            pcd.estimate_normals()
+            
+            # Create mesh using Poisson reconstruction
+            mesh, _ = o3d.geometry.TriangleMesh.create_from_point_cloud_poisson(
+                pcd, depth=8, width=0, scale=1.1, linear_fit=False
+            )
+            
+            # Remove degenerate triangles and unreferenced vertices
+            mesh.remove_degenerate_triangles()
+            mesh.remove_duplicated_triangles()
+            mesh.remove_duplicated_vertices()
+            mesh.remove_non_manifold_edges()
+            
+            # Smooth the mesh
+            mesh = mesh.filter_smooth_simple(number_of_iterations=2)
+            
+            # Save as OBJ file
+            obj_path = os.path.join(self.temp_dir, f"model_{job_id}.obj")
+            o3d.io.write_triangle_mesh(obj_path, mesh)
+            
+            logger.info(f"✅ 3D model created: {len(mesh.vertices)} vertices, {len(mesh.triangles)} triangles")
+            
+            return obj_path
+            
+        except Exception as e:
+            logger.error(f"❌ Error creating 3D model: {str(e)}")
+            raise e
+    
+    def process_image_to_3d(self, image: Image.Image, job_id: str) -> dict:
+        """Complete pipeline: image -> depth map -> 3D model"""
+        try:
+            logger.info(f"🔄 Processing image to 3D model (Job: {job_id})")
+            
+            # Resize image if too large (for performance)
+            max_size = 512
+            if max(image.size) > max_size:
+                ratio = max_size / max(image.size)
+                new_size = (int(image.size[0] * ratio), int(image.size[1] * ratio))
+                image = image.resize(new_size, Image.Resampling.LANCZOS)
+                logger.info(f"📐 Resized image to {new_size}")
+            
+            # Convert to RGB if needed
+            if image.mode != 'RGB':
+                image = image.convert('RGB')
+            
+            # Generate depth map
+            depth_map = self.generate_depth_map(image)
+            
+            # Save depth map as image
+            depth_map_path = self.save_depth_map_image(depth_map, job_id)
+            
+            # Create 3D model
+            obj_path = self.create_3d_model(image, depth_map, job_id)
+            
+            return {
+                'depth_map': depth_map,
+                'depth_map_path': depth_map_path,
+                'obj_path': obj_path,
+                'success': True
+            }
+            
+        except Exception as e:
+            logger.error(f"❌ Error in image-to-3D pipeline: {str(e)}")
+            return {
+                'success': False,
+                'error': str(e)
+            }

models/image_generator.py

@@ -0,0 +1,140 @@
+"""
+Image generation module using Stable Diffusion
+"""
+
+import os
+import logging
+import tempfile
+import io
+from PIL import Image
+import torch
+from diffusers import StableDiffusionPipeline
+import gc
+
+logger = logging.getLogger(__name__)
+
+class ImageGenerator:
+    """Handles text-to-image generation using Stable Diffusion"""
+    
+    def __init__(self):
+        self.device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
+        self.pipeline = None
+        self.temp_dir = tempfile.mkdtemp()
+        
+    def load_model(self):
+        """Load the Stable Diffusion model"""
+        try:
+            logger.info(f"🔄 Loading Stable Diffusion model on {self.device}...")
+            
+            # Use a smaller, faster model for better performance on free tier
+            model_id = "runwayml/stable-diffusion-v1-5"
+            
+            # Load pipeline
+            self.pipeline = StableDiffusionPipeline.from_pretrained(
+                model_id,
+                torch_dtype=torch.float16 if self.device.type == "cuda" else torch.float32,
+                safety_checker=None,  # Disable safety checker for faster inference
+                requires_safety_checker=False
+            )
+            
+            self.pipeline.to(self.device)
+            
+            # Enable memory efficient attention if available
+            if hasattr(self.pipeline, "enable_attention_slicing"):
+                self.pipeline.enable_attention_slicing()
+            
+            # Enable model offloading for CPU
+            if self.device.type == "cpu":
+                self.pipeline.enable_sequential_cpu_offload()
+            
+            if self.device.type == "cuda":
+                logger.info(f"✅ Stable Diffusion loaded on GPU: {torch.cuda.get_device_name(0)}")
+            else:
+                logger.info("✅ Stable Diffusion loaded on CPU")
+                
+        except Exception as e:
+            logger.error(f"❌ Failed to load Stable Diffusion model: {str(e)}")
+            raise e
+    
+    def generate_image(self, prompt: str, negative_prompt: str = None) -> dict:
+        """Generate image from text prompt"""
+        try:
+            logger.info(f"🎨 Generating image for prompt: '{prompt}'")
+            
+            # Default negative prompt for better quality
+            if negative_prompt is None:
+                negative_prompt = "blurry, low quality, distorted, deformed, ugly, bad anatomy"
+            
+            # Enhanced prompt for 3D-suitable images
+            enhanced_prompt = f"{prompt}, high quality, detailed, clear lighting, suitable for 3D modeling"
+            
+            # Generation parameters
+            generator = torch.Generator(device=self.device).manual_seed(42)  # Fixed seed for consistency
+            
+            # Generate image
+            with torch.no_grad():
+                result = self.pipeline(
+                    prompt=enhanced_prompt,
+                    negative_prompt=negative_prompt,
+                    num_inference_steps=20,  # Reduced for faster inference
+                    guidance_scale=7.5,
+                    width=512,
+                    height=512,
+                    generator=generator
+                )
+            
+            image = result.images[0]
+            
+            # Convert to bytes for storage
+            img_bytes = io.BytesIO()
+            image.save(img_bytes, format='PNG', quality=95)
+            img_bytes.seek(0)
+            
+            # Clean up GPU memory
+            if self.device.type == "cuda":
+                torch.cuda.empty_cache()
+                gc.collect()
+            
+            logger.info("✅ Image generated successfully")
+            
+            return {
+                'image_pil': image,
+                'image_bytes': img_bytes.getvalue(),
+                'success': True
+            }
+            
+        except Exception as e:
+            logger.error(f"❌ Error generating image: {str(e)}")
+            
+            # Clean up memory on error
+            if self.device.type == "cuda":
+                torch.cuda.empty_cache()
+            gc.collect()
+            
+            return {
+                'success': False,
+                'error': str(e)
+            }
+    
+    def enhance_prompt_for_3d(self, prompt: str) -> str:
+        """Enhance prompt to be more suitable for 3D modeling"""
+        # Add keywords that typically produce good depth information
+        enhancement_keywords = [
+            "3D rendering",
+            "detailed texture",
+            "clear lighting",
+            "high contrast",
+            "depth",
+            "dimensional"
+        ]
+        
+        # Check if prompt already contains 3D-related terms
+        has_3d_terms = any(term in prompt.lower() for term in ["3d", "render", "model", "dimensional"])
+        
+        if not has_3d_terms:
+            # Add one enhancement keyword
+            enhanced = f"{prompt}, 3D rendering style"
+        else:
+            enhanced = prompt
+        
+        return enhanced

requirements.txt

@@ -0,0 +1,19 @@
+fastapi==0.104.1
+uvicorn[standard]==0.24.0
+python-multipart==0.0.6
+pydantic==2.5.0
+torch==2.1.1
+torchvision==0.16.1
+torchaudio==2.1.1
+transformers==4.39.3
+diffusers==0.27.0
+accelerate==0.27.0
+Pillow==10.3.0
+numpy==1.24.3
+open3d==0.18.0
+matplotlib==3.7.2
+cloudinary==1.37.0
+python-dotenv==1.0.0
+safetensors==0.4.2
+huggingface_hub==0.20.2
+requests==2.31.0

utils/__init__.py

@@ -0,0 +1 @@
+# Utils package

utils/cloudinary_client.py

@@ -0,0 +1,119 @@
+"""
+Cloudinary client for file uploads
+"""
+
+import os
+import logging
+import cloudinary
+import cloudinary.uploader
+from typing import Union
+
+logger = logging.getLogger(__name__)
+
+class CloudinaryClient:
+    """Handles file uploads to Cloudinary"""
+    
+    def __init__(self):
+        # Configure Cloudinary
+        cloudinary.config(
+            cloud_name=os.environ.get("CLOUDINARY_CLOUD_NAME"),
+            api_key=os.environ.get("CLOUDINARY_API_KEY"),
+            api_secret=os.environ.get("CLOUDINARY_API_SECRET")
+        )
+        
+        # Verify configuration
+        if not all([
+            os.environ.get("CLOUDINARY_CLOUD_NAME"),
+            os.environ.get("CLOUDINARY_API_KEY"),
+            os.environ.get("CLOUDINARY_API_SECRET")
+        ]):
+            logger.warning("⚠️ Cloudinary credentials not fully configured")
+        else:
+            logger.info("✅ Cloudinary client initialized")
+    
+    def upload_image_from_bytes(self, image_bytes: bytes, public_id: str) -> str:
+        """Upload image from bytes to Cloudinary"""
+        try:
+            logger.info(f"☁️ Uploading image to Cloudinary: {public_id}")
+            
+            result = cloudinary.uploader.upload(
+                image_bytes,
+                public_id=f"text-to-3d/{public_id}",
+                resource_type="image",
+                unique_filename=True,
+                overwrite=True,
+                quality="auto"
+            )
+            
+            url = result["secure_url"]
+            logger.info(f"✅ Image uploaded: {url}")
+            return url
+            
+        except Exception as e:
+            logger.error(f"❌ Error uploading image to Cloudinary: {str(e)}")
+            raise e
+    
+    def upload_image_from_path(self, file_path: str, public_id: str) -> str:
+        """Upload image from file path to Cloudinary"""
+        try:
+            logger.info(f"☁️ Uploading image file to Cloudinary: {public_id}")
+            
+            result = cloudinary.uploader.upload(
+                file_path,
+                public_id=f"text-to-3d/{public_id}",
+                resource_type="image",
+                unique_filename=True,
+                overwrite=True,
+                quality="auto"
+            )
+            
+            url = result["secure_url"]
+            logger.info(f"✅ Image file uploaded: {url}")
+            return url
+            
+        except Exception as e:
+            logger.error(f"❌ Error uploading image file to Cloudinary: {str(e)}")
+            raise e
+    
+    def upload_file(self, file_path: str, public_id: str) -> str:
+        """Upload any file to Cloudinary"""
+        try:
+            logger.info(f"☁️ Uploading file to Cloudinary: {public_id}")
+            
+            result = cloudinary.uploader.upload(
+                file_path,
+                public_id=f"text-to-3d/{public_id}",
+                resource_type="raw",  # For non-image files
+                unique_filename=True,
+                overwrite=True
+            )
+            
+            url = result["secure_url"]
+            logger.info(f"✅ File uploaded: {url}")
+            return url
+            
+        except Exception as e:
+            logger.error(f"❌ Error uploading file to Cloudinary: {str(e)}")
+            raise e
+    
+    def delete_file(self, public_id: str, resource_type: str = "image") -> bool:
+        """Delete file from Cloudinary"""
+        try:
+            logger.info(f"🗑️ Deleting file from Cloudinary: {public_id}")
+            
+            result = cloudinary.uploader.destroy(
+                f"text-to-3d/{public_id}",
+                resource_type=resource_type
+            )
+            
+            success = result.get("result") == "ok"
+            if success:
+                logger.info(f"✅ File deleted: {public_id}")
+            else:
+                logger.warning(f"⚠️ File deletion may have failed: {public_id}")
+            
+            return success
+            
+        except Exception as e:
+            logger.error(f"❌ Error deleting file from Cloudinary: {str(e)}")
+            return False

utils/job_manager.py

@@ -0,0 +1,164 @@
+"""
+Job management for tracking async tasks
+"""
+
+import time
+import threading
+import logging
+from typing import Dict, Optional, Any
+
+logger = logging.getLogger(__name__)
+
+class JobManager:
+    """Manages background job tracking and progress"""
+    
+    def __init__(self):
+        self.active_jobs: Dict[str, Dict] = {}
+        self.job_progress: Dict[str, Dict] = {}
+        self.job_results: Dict[str, Dict] = {}
+        self.jobs_lock = threading.Lock()
+        self.progress_lock = threading.Lock()
+        self.results_lock = threading.Lock()
+        
+        # Start cleanup task
+        self._start_cleanup_task()
+    
+    def register_job(self, job_id: str):
+        """Register a new job"""
+        with self.jobs_lock:
+            self.active_jobs[job_id] = {
+                'cancelled': False,
+                'created_at': time.time(),
+                'status': 'active'
+            }
+        
+        with self.progress_lock:
+            self.job_progress[job_id] = {
+                'stage': 'starting',
+                'progress': 0,
+                'message': 'Job started...',
+                'timestamp': time.time()
+            }
+        
+        logger.info(f"📝 Job registered: {job_id}")
+    
+    def is_job_cancelled(self, job_id: str) -> bool:
+        """Check if a job has been cancelled"""
+        with self.jobs_lock:
+            return self.active_jobs.get(job_id, {}).get('cancelled', False)
+    
+    def cancel_job(self, job_id: str) -> bool:
+        """Cancel a job"""
+        with self.jobs_lock:
+            if job_id in self.active_jobs:
+                self.active_jobs[job_id]['cancelled'] = True
+                self.active_jobs[job_id]['status'] = 'cancelled'
+                logger.info(f"❌ Job cancelled: {job_id}")
+                return True
+        return False
+    
+    def update_job_progress(self, job_id: str, stage: str, progress: int, message: str):
+        """Update job progress"""
+        with self.progress_lock:
+            if job_id in self.job_progress:
+                self.job_progress[job_id] = {
+                    'stage': stage,
+                    'progress': progress,
+                    'message': message,
+                    'timestamp': time.time()
+                }
+                logger.info(f"📊 Job {job_id}: {stage} - {progress}% - {message}")
+    
+    def get_job_progress(self, job_id: str) -> Optional[Dict]:
+        """Get current job progress"""
+        with self.progress_lock:
+            return self.job_progress.get(job_id)
+    
+    def complete_job(self, job_id: str, results: Dict[str, Any]):
+        """Mark job as completed with results"""
+        with self.jobs_lock:
+            if job_id in self.active_jobs:
+                self.active_jobs[job_id]['status'] = 'completed'
+        
+        with self.progress_lock:
+            self.job_progress[job_id] = {
+                'stage': 'completed',
+                'progress': 100,
+                'message': 'Job completed successfully!',
+                'timestamp': time.time()
+            }
+        
+        with self.results_lock:
+            self.job_results[job_id] = {
+                **results,
+                'completed_at': time.time()
+            }
+        
+        logger.info(f"✅ Job completed: {job_id}")
+    
+    def fail_job(self, job_id: str, error_message: str):
+        """Mark job as failed"""
+        with self.jobs_lock:
+            if job_id in self.active_jobs:
+                self.active_jobs[job_id]['status'] = 'failed'
+        
+        with self.progress_lock:
+            self.job_progress[job_id] = {
+                'stage': 'error',
+                'progress': 0,
+                'message': f'Error: {error_message}',
+                'timestamp': time.time()
+            }
+        
+        logger.error(f"❌ Job failed: {job_id} - {error_message}")
+    
+    def get_job_results(self, job_id: str) -> Optional[Dict]:
+        """Get job results if completed"""
+        with self.results_lock:
+            return self.job_results.get(job_id)
+    
+    def get_active_job_count(self) -> int:
+        """Get number of active jobs"""
+        with self.jobs_lock:
+            return len([j for j in self.active_jobs.values() if j['status'] == 'active'])
+    
+    def cleanup_old_jobs(self):
+        """Clean up jobs older than 30 minutes"""
+        current_time = time.time()
+        cleanup_age = 1800  # 30 minutes
+        
+        jobs_to_remove = []
+        
+        with self.jobs_lock:
+            for job_id, job_data in self.active_jobs.items():
+                if current_time - job_data['created_at'] > cleanup_age:
+                    jobs_to_remove.append(job_id)
+        
+        for job_id in jobs_to_remove:
+            self._remove_job(job_id)
+            logger.info(f"🧹 Cleaned up old job: {job_id}")
+    
+    def _remove_job(self, job_id: str):
+        """Remove job from all tracking dictionaries"""
+        with self.jobs_lock:
+            self.active_jobs.pop(job_id, None)
+        
+        with self.progress_lock:
+            self.job_progress.pop(job_id, None)
+        
+        with self.results_lock:
+            self.job_results.pop(job_id, None)
+    
+    def _start_cleanup_task(self):
+        """Start background cleanup task"""
+        def cleanup_worker():
+            while True:
+                time.sleep(300)  # Run every 5 minutes
+                try:
+                    self.cleanup_old_jobs()
+                except Exception as e:
+                    logger.error(f"❌ Error in cleanup task: {str(e)}")
+        
+        cleanup_thread = threading.Thread(target=cleanup_worker, daemon=True)
+        cleanup_thread.start()
+        logger.info("🧹 Cleanup task started")