Write a detailed Python code using Stable Diffusion (via diffusers library) and a predictive ML model (e.g., TensorFlow for regression) to generate custom images for e-commerce, such as product mockups personalized to predicted user trends from sales data. The predictive part should process input data (e.g., CSV of customer demographics) to forecast popular styles/colors, then condition the image generation accordingly. Include steps for model setup, training on datasets like LAION, and API endpoints for automation. Tailor for freelancers building tools for businesses needing visual content creation, considering the 220% YoY demand surge.

ecommerce-image-generator.py

@@ -0,0 +1,465 @@
+```python
+#!/usr/bin/env python3
+"""
+AI Forge E-commerce Image Generator
+Stable Diffusion + Predictive ML for personalized product mockups
+Optimized for 220% YoY demand growth in visual content creation
+"""
+
+import os
+import torch
+import pandas as pd
+import numpy as np
+from diffusers import StableDiffusionPipeline, DPMSolverMultistepScheduler
+import tensorflow as tf
+from tensorflow import keras
+from tensorflow.keras import layers
+import matplotlib.pyplot as plt
+from PIL import Image
+import io
+import base64
+import warnings
+warnings.filterwarnings('ignore')
+
+# Configuration
+class Config:
+    """Configuration parameters for the image generation system"""
+    STABLE_DIFFUSION_MODEL = "runwayml/stable-diffusion-v1-5"
+    IMAGE_SIZE = (512, 512)
+    MAX_LENGTH = 77
+    BATCH_SIZE = 4
+    EPOCHS = 50
+    LEARNING_RATE = 1e-4
+    PREDICTION_MODEL_PATH = "./models/style_predictor.h5"
+    TRAINING_DATA_PATH = "./data/ecommerce_sales.csv"
+    OUTPUT_DIR = "./generated_images"
+    
+    # Style categories for prediction
+    STYLE_CATEGORIES = ['minimalist', 'vintage', 'modern', 'luxury', 'tech', 'sporty']
+    COLOR_CATEGORIES = ['blue', 'red', 'green', 'black', 'white', 'pastel', 'neon']
+    PRODUCT_TYPES = ['clothing', 'electronics', 'home_decor', 'beauty', 'accessories']
+
+class StylePredictor:
+    """TensorFlow model for predicting trending styles and colors"""
+    
+    def __init__(self, input_dim):
+        self.input_dim = input_dim
+        self.model = None
+        
+    def build_model(self):
+        """Build the style prediction model"""
+        model = keras.Sequential([
+            layers.Dense(256, activation='relu', input_shape=(input_dim,)),
+            layers.Dropout(0.3),
+            layers.Dense(128, activation='relu'),
+            layers.Dropout(0.2),
+            layers.Dense(64, activation='relu'),
+            layers.Dense(len(Config.STYLE_CATEGORIES) + len(Config.COLOR_CATEGORIES)),
+            layers.Activation('sigmoid')
+        ])
+        
+        model.compile(
+            optimizer=keras.optimizers.Adam(learning_rate=Config.LEARNING_RATE),
+            loss='binary_crossentropy',
+            metrics=['accuracy']
+        )
+        
+        self.model = model
+        return model
+    
+    def train(self, X_train, y_train, X_val=None, y_val=None):
+        """Train the style prediction model"""
+        callbacks = [
+            keras.callbacks.EarlyStopping(patience=10, restore_best_weights=True),
+            keras.callbacks.ReduceLROnPlateau(factor=0.5, patience=5),
+            keras.callbacks.ModelCheckpoint(
+                Config.PREDICTION_MODEL_PATH,
+                save_best_only=True,
+                monitor='val_loss' if X_val is not None else 'loss'
+            )
+        ]
+        
+        history = self.model.fit(
+            X_train, y_train,
+            batch_size=Config.BATCH_SIZE,
+            epochs=Config.EPOCHS,
+            validation_data=(X_val, y_val) if X_val is not None else None,
+            callbacks=callbacks,
+            verbose=1
+        )
+        
+        return history
+    
+    def predict_trends(self, customer_data):
+        """Predict trending styles and colors for customer segment"""
+        predictions = self.model.predict(customer_data)
+        
+        # Split predictions into styles and colors
+        style_predictions = predictions[:, :len(Config.STYLE_CATEGORIES)]
+        color_predictions = predictions[:, len(Config.STYLE_CATEGORIES):]
+        
+        return style_predictions, color_predictions
+
+class EcommerceDataProcessor:
+    """Process e-commerce sales data for trend prediction"""
+    
+    def __init__(self):
+        self.feature_columns = []
+        
+    def load_and_preprocess_data(self, file_path):
+        """Load and preprocess e-commerce sales data"""
+        try:
+            df = pd.read_csv(file_path)
+            print(f"Loaded dataset with {len(df)} rows")
+            return df
+        except Exception as e:
+            print(f"Error loading data: {e}")
+            return None
+    
+    def extract_features(self, df):
+        """Extract features from e-commerce data"""
+        features = []
+        
+        # Demographic features
+        demographic_features = ['age', 'income_level', 'location_urban', 'gender_encoded']
+        
+        # Time-based features
+        df['purchase_month'] = pd.to_datetime(df['purchase_date']).dt.month
+        features.append(pd.get_dummies(df['purchase_month'], prefix='month'))
+        
+        # Product features
+        product_features = ['price', 'category_encoded', 'brand_popularity']
+        
+        # Combine all features
+        for feature in demographic_features + ['purchase_month', 'price', 'category_encoded', 'brand_popularity']:
+            if feature in df.columns:
+                features.append(df[[feature]]))
+        
+        # One-hot encode categorical variables
+        categorical_cols = ['region', 'device_type', 'marketing_channel']
+        for col in categorical_cols:
+            if col in df.columns:
+                dummies = pd.get_dummies(df[col], prefix=col)
+                features.append(dummies)
+        
+        X = pd.concat(features, axis=1)
+        self.feature_columns = X.columns.tolist()
+        
+        return X
+    
+    def prepare_training_labels(self, df):
+        """Prepare training labels for style and color trends"""
+        # Create binary labels for styles and colors based on sales performance
+        labels = []
+        
+        for _, row in df.iterrows():
+            # Style preferences (based on product attributes)
+            style_vector = [0] * len(Config.STYLE_CATEGORIES)
+            color_vector = [0] * len(Config.COLOR_CATEGORIES)
+        
+        # For each product, determine dominant style and color
+            if row['sales_rank'] <= 100:  # Top selling products
+                # Analyze product description for style keywords
+                description = str(row.get('product_description', '')).lower()
+            
+            for i, style in enumerate(Config.STYLE_CATEGORIES):
+                if style in description:
+                    style_vector[i] = 1
+            
+            # Color analysis from product data
+            color_data = str(row.get('color_data', '')).lower()
+                for j, color in enumerate(Config.COLOR_CATEGORIES):
+                    if color in description or color in str(row.get('primary_color', '')).lower():
+                    color_vector[j] = 1
+            
+            labels.append(style_vector + color_vector)
+        
+        return np.array(labels)
+
+class StableDiffusionGenerator:
+    """Stable Diffusion image generator for e-commerce mockups"""
+    
+    def __init__(self):
+        self.pipeline = None
+        self.device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
+        print(f"Using device: {self.device}")
+        
+    def load_model(self):
+        """Load Stable Diffusion model"""
+        try:
+            self.pipeline = StableDiffusionPipeline.from_pretrained(
+                Config.STABLE_DIFFUSION_MODEL,
+                torch_dtype=torch.float16 if torch.cuda.is_available() else torch.float32
+            )
+            self.pipeline.scheduler = DPMSolverMultistepScheduler.from_config(
+                self.pipeline.scheduler.config
+            )
+            self.pipeline = self.pipeline.to(self.device)
+            print("Stable Diffusion model loaded successfully")
+        except Exception as e:
+            print(f"Error loading model: {e}")
+    
+    def generate_product_mockup(self, prompt, style_weights=None, color_weights=None):
+        """Generate product mockup with style and color conditioning"""
+        
+        # Enhance prompt based on predicted trends
+        enhanced_prompt = self._enhance_prompt(prompt, style_weights, color_weights)
+        
+        # Generate image
+        with torch.autocast(self.device.type):
+            image = self.pipeline(
+                prompt,
+                height=Config.IMAGE_SIZE[0],
+                width=Config.IMAGE_SIZE[1],
+                num_inference_steps=25,
+                guidance_scale=7.5
+            ).images[0]
+        
+        return image
+    
+    def _enhance_prompt(self, base_prompt, style_weights, color_weights):
+        """Enhance prompt with style and color conditioning"""
+        
+        if style_weights is not None:
+            # Get top predicted styles
+            top_style_indices = np.argsort(style_weights)[-2:]  # Top 2 styles
+            top_colors_indices = np.argsort(color_weights)[-2:]
+        
+        # Add style descriptors
+        style_descriptors = []
+        for idx in top_style_indices:
+            style_descriptors.append(Config.STYLE_CATEGORIES[idx])
+        
+        # Add color descriptors
+        color_descriptors = []
+        for idx in top_colors_indices:
+            color_descriptors.append(Config.COLOR_CATEGORIES[idx])
+        
+        enhanced_prompt = f"{base_prompt}, {', '.join(style_descriptors)} style, colors: {', '.join(color_descriptors)}"
+        
+        return enhanced_prompt
+    
+    def generate_batch_mockups(self, prompts, style_predictions, color_predictions):
+        """Generate multiple product mockups in batch"""
+        images = []
+        
+        for i, prompt in enumerate(prompts):
+            style_weights = style_predictions[i] if i < len(style_predictions) else None
+            color_weights = color_predictions[i] if i < len(color_predictions) else None
+            
+            image = self.generate_product_mockup(prompt, style_weights, color_weights)
+            images.append(image)
+        
+        return images
+
+class EcommerceImageAPI:
+    """FastAPI integration for the e-commerce image generation system"""
+    
+    def __init__(self):
+        self.data_processor = EcommerceDataProcessor()
+        self.style_predictor = None
+        self.image_generator = StableDiffusionGenerator()
+        
+    def initialize_system(self):
+        """Initialize the complete system"""
+        print("Initializing E-commerce Image Generation System...")
+        
+        # Load data
+        df = self.data_processor.load_and_preprocess_data(Config.TRAINING_DATA_PATH)
+        
+        if df is not None:
+            # Prepare features and labels
+            X = self.data_processor.extract_features(df)
+        y = self.data_processor.prepare_training_labels(df)
+        
+            # Initialize and train style predictor
+            self.style_predictor = StylePredictor(X.shape[1])
+            self.style_predictor.build_model()
+            
+            # Split data
+            from sklearn.model_selection import train_test_split
+            X_train, X_test, y_train, y_test = train_test_split(
+                X, y, test_size=0.2, random_state=42
+            )
+            
+            # Train model
+            print("Training style prediction model...")
+            history = self.style_predictor.train(X_train, y_train, X_test, y_test)
+            
+            # Evaluate model
+            test_loss, test_accuracy = self.style_predictor.model.evaluate(X_test, y_test)
+            print(f"Model trained - Test Accuracy: {test_accuracy:.4f}")
+        
+        # Load image generator
+        self.image_generator.load_model()
+        
+        print("System initialized successfully")
+    
+    def predict_and_generate(self, customer_segment_data, base_prompts):
+        """Complete workflow: predict trends and generate images"""
+        
+        # Predict styles and colors
+        style_predictions, color_predictions = self.style_predictor.predict_trends(customer_segment_data)
+        
+        # Generate images
+        images = self.image_generator.generate_batch_mockups(
+            base_prompts, style_predictions, color_predictions
+        )
+        
+        return images, style_predictions, color_predictions
+
+# FastAPI Integration
+from fastapi import FastAPI, HTTPException
+from fastapi.middleware.cors import CORSMiddleware
+from pydantic import BaseModel
+from typing import List, Optional
+import uvicorn
+
+app = FastAPI(title="AI Forge E-commerce Image Generator")
+
+# CORS middleware
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=["*"],
+    allow_credentials=True,
+    allow_methods=["*"],
+    allow_headers=["*"]
+)
+
+class GenerationRequest(BaseModel):
+    customer_data: List[dict]
+    base_prompts: List[str]
+    num_images: int = 1
+
+class GenerationResponse(BaseModel):
+    success: bool
+    message: str
+    generated_images: Optional[List[str]] = None
+    predicted_styles: Optional[List[str]] = None
+    predicted_colors: Optional[List[str]] = None
+
+# Initialize system
+ecommerce_system = EcommerceImageAPI()
+
+@app.on_event("startup")
+async def startup_event():
+    """Initialize system on startup"""
+    ecommerce_system.initialize_system()
+
+@app.get("/")
+async def root():
+    return {"message": "AI Forge E-commerce Image Generator API"}
+
+@app.post("/api/generate-mockups", response_model=GenerationResponse)
+async def generate_mockups(request: GenerationRequest):
+    """Generate product mockups based on predicted trends"""
+    try:
+        # Convert customer data to DataFrame
+        customer_df = pd.DataFrame(request.customer_data)
+    
+    # Process customer data
+    X_customer = ecommerce_system.data_processor.extract_features(customer_df)
+    
+    # Generate images
+    images, style_preds, color_preds = ecommerce_system.predict_and_generate(
+        X_customer,
+        request.base_prompts
+    )
+    
+    # Convert images to base64 for API response
+    base64_images = []
+        for image in images:
+            buffered = io.BytesIO()
+            image.save(buffered, format="PNG")
+        img_str = base64.b64encode(buffered.getvalue()).decode()
+        base64_images.append(img_str)
+    
+    # Get top predicted styles and colors
+    top_styles = []
+        top_colors = []
+        
+        for style_pred in style_preds:
+            top_indices = np.argsort(style_pred)[-2:]
+        top_styles.append([Config.STYLE_CATEGORIES[i] for i in top_indices])
+    top_colors = [[Config.COLOR_CATEGORIES[i] for i in np.argsort(color_pred)[-2:]] for color_pred in color_preds]
+    
+    return GenerationResponse(
+        success=True,
+        message=f"Successfully generated {len(images)} mockups")
+        generated_images=base64_images,
+        predicted_styles=top_styles,
+        predicted_colors=top_colors
+    )
+    
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=f"Generation error: {str(e)}")
+
+@app.get("/api/health")
+async def health_check():
+    return {"status": "healthy", "service": "ecommerce_image_generator"}
+
+def main():
+    """Main execution function"""
+    print("="*60)
+    print("AI FORGE E-COMMERCE IMAGE GENERATOR")
+    print("Optimized for 220% YoY demand growth")
+    print("="*60)
+    
+    # Initialize and test the system
+    system = EcommerceImageAPI()
+    system.initialize_system()
+    
+    # Sample generation
+    print("\nGenerating sample mockups...")
+    
+    sample_customer_data = [
+        {
+            'age': 35,
+            'income_level': 4,
+            'location_urban': 1,
+            'gender_encoded': 0,
+            'region': 'north_america',
+            'device_type': 'mobile',
+            'marketing_channel': 'social_media',
+            'price': 89.99,
+            'category_encoded': 2,
+            'brand_popularity': 8
+        }
+    ]
+    
+    sample_prompts = [
+        "professional product mockup for modern e-commerce website"
+    ]
+    
+    images, styles, colors = system.predict_and_generate(
+        pd.DataFrame(sample_customer_data),
+        sample_prompts
+    )
+    
+    print(f"Generated {len(images)} images successfully")
+    print(f"Predicted top styles: {styles}")
+    print(f"Predicted top colors: {colors}")
+    
+    # Save sample images
+    os.makedirs(Config.OUTPUT_DIR, exist_ok=True)
+    
+    for i, image in enumerate(images):
+        image_path = os.path.join(Config.OUTPUT_DIR, f"sample_mockup_{i+1}.png")
+        image.save(image_path)
+        print(f"Saved sample image: {image_path}")
+    
+    print("\nSystem ready for production deployment!")
+    print("API endpoints available at http://localhost:8000")
+
+if __name__ == "__main__":
+    # Run the main function for testing
+    main()
+    
+    # Start the FastAPI server
+    uvicorn.run(
+        "ecommerce_image_generator:app",
+        host="0.0.0.0",
+        port=8000,
+        reload=True
+    )
+```