Under the Hood: How DALL-E 4 Generates Images from Text Prompts for App Assets

import re
import logging
from typing import Dict, List, Optional
from dataclasses import dataclass

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

@dataclass
class PreprocessedPrompt:
    """Structured output of prompt preprocessing for DALL-E 4 pipeline."""
    cleaned_prompt: str
    asset_type: str
    target_resolution: str
    style_constraints: List[str]
    validation_errors: List[str]

class AppAssetPromptPreprocessor:
    """Preprocesses text prompts for app asset generation, injecting app-specific constraints."""

    # App asset type to default resolution mapping
    ASSET_TYPE_DEFAULTS = {
        "icon": "512x512",
        "splash_screen": "1080x1920",
        "banner": "728x90",
        "illustration": "1024x1024"
    }

    # Forbidden terms that trigger prompt rejection
    FORBIDDEN_TERMS = {"nsfw", "hate speech", "violence", "copyrighted character"}

    def __init__(self, default_style: str = "flat design, high contrast, no text"):
        self.default_style = default_style
        self.asset_type_pattern = re.compile(r"\[(icon|splash_screen|banner|illustration)\]")
        self.resolution_pattern = re.compile(r"\d{3,4}x\d{3,4}")

    def preprocess(self, raw_prompt: str, user_id: Optional[str] = None) -> PreprocessedPrompt:
        """
        Preprocess raw user prompt for DALL-E 4 asset generation.

        Args:
            raw_prompt: User-provided text prompt
            user_id: Optional user ID for audit logging

        Returns:
            PreprocessedPrompt with cleaned prompt and metadata
        """
        validation_errors = []
        style_constraints = [self.default_style]

        # Step 1: Validate prompt length (DALL-E 4 max prompt length is 1000 chars)
        if len(raw_prompt) > 1000:
            validation_errors.append(f"Prompt exceeds max length of 1000 characters: {len(raw_prompt)}")
            raw_prompt = raw_prompt[:1000]  # Truncate to max length
            logger.warning(f"Truncated prompt for user {user_id} to 1000 chars")

        # Step 2: Check for forbidden terms
        prompt_lower = raw_prompt.lower()
        for term in self.FORBIDDEN_TERMS:
            if term in prompt_lower:
                validation_errors.append(f"Forbidden term detected: {term}")
                logger.error(f"Rejected prompt from user {user_id} for forbidden term: {term}")

        # Step 3: Extract asset type from prompt (format: [asset_type])
        asset_type_match = self.asset_type_pattern.search(raw_prompt)
        asset_type = asset_type_match.group(1) if asset_type_match else "icon"  # Default to icon
        if asset_type not in self.ASSET_TYPE_DEFAULTS:
            validation_errors.append(f"Invalid asset type: {asset_type}, defaulting to icon")
            asset_type = "icon"

        # Step 4: Extract target resolution or use default
        resolution_match = self.resolution_pattern.search(raw_prompt)
        target_resolution = resolution_match.group(0) if resolution_match else self.ASSET_TYPE_DEFAULTS[asset_type]

        # Step 5: Inject app-asset specific style constraints
        # Remove asset type tag from prompt to avoid confusing the text encoder
        cleaned_prompt = self.asset_type_pattern.sub("", raw_prompt).strip()
        # Add asset type context to prompt
        cleaned_prompt = f"{asset_type} for mobile app: {cleaned_prompt}"
        # Add style constraints
        style_constraints.append(f"target resolution {target_resolution}")
        style_constraints.append("no embedded text unless explicitly requested")
        # Append style constraints to prompt
        cleaned_prompt = f"{cleaned_prompt}, {', '.join(style_constraints)}"

        # Step 6: Log preprocessing results for audit
        if user_id:
            logger.info(f"Preprocessed prompt for user {user_id}: {cleaned_prompt[:200]}...")

        return PreprocessedPrompt(
            cleaned_prompt=cleaned_prompt,
            asset_type=asset_type,
            target_resolution=target_resolution,
            style_constraints=style_constraints,
            validation_errors=validation_errors
        )

    def validate_preprocessed(self, preprocessed: PreprocessedPrompt) -> bool:
        """Check if preprocessed prompt is valid for generation."""
        if preprocessed.validation_errors:
            # Only block if forbidden terms are present
            for error in preprocessed.validation_errors:
                if "Forbidden term" in error:
                    return False
        return True

# Example usage
if __name__ == "__main__":
    preprocessor = AppAssetPromptPreprocessor()
    test_prompt = "[icon] A settings gear for a productivity app, blue color"
    result = preprocessor.preprocess(test_prompt, user_id="usr_12345")
    if preprocessor.validate_preprocessed(result):
        print(f"Cleaned prompt: {result.cleaned_prompt}")
        print(f"Asset type: {result.asset_type}")
        print(f"Target resolution: {result.target_resolution}")
    else:
        print(f"Prompt rejected: {result.validation_errors}")

import torch
import torch.nn as nn
import torch.nn.functional as F
from typing import Optional, Tuple
import logging

logger = logging.getLogger(__name__)

class AppAssetDiT(nn.Module):
    """
    Diffusion Transformer (DiT) fine-tuned for app asset generation.
    Based on DiT-B/2 architecture with app-asset specific attention heads.
    """

    def __init__(
        self,
        input_channels: int = 4,  # Latent space channels from VAE
        patch_size: int = 2,
        embed_dim: int = 768,
        num_heads: int = 12,
        num_layers: int = 12,
        text_embed_dim: int = 768,  # CLIP ViT-L/14 embed dim
        num_app_asset_classes: int = 4  # icon, splash, banner, illustration
    ):
        super().__init__()
        self.patch_size = patch_size
        self.embed_dim = embed_dim
        self.num_patches = (512 // patch_size) ** 2  # For 512x512 latent (256x256 image)

        # Patch embedding layer
        self.patch_embed = nn.Conv2d(
            input_channels, embed_dim, kernel_size=patch_size, stride=patch_size
        )

        # Position embedding for patches
        self.pos_embed = nn.Parameter(torch.zeros(1, self.num_patches + 1, embed_dim))

        # App asset type embedding (learned embedding for each asset type)
        self.asset_type_embed = nn.Embedding(num_app_asset_classes, embed_dim)

        # Text embedding projection (align CLIP embeds to DiT embed dim)
        self.text_proj = nn.Linear(text_embed_dim, embed_dim)

        # Timestep embedding (sinusoidal + MLP)
        self.timestep_embed = nn.Sequential(
            nn.Linear(embed_dim, embed_dim * 4),
            nn.SiLU(),
            nn.Linear(embed_dim * 4, embed_dim)
        )

        # Transformer encoder layers
        self.blocks = nn.ModuleList([
            nn.TransformerEncoderLayer(
                d_model=embed_dim,
                nhead=num_heads,
                dim_feedforward=embed_dim * 4,
                activation="silu",
                batch_first=True
            ) for _ in range(num_layers)
        ])

        # Output layer to predict noise residual
        self.output_layer = nn.Linear(embed_dim, input_channels * patch_size * patch_size)

        # Initialize weights
        self._init_weights()

    def _init_weights(self):
        """Initialize model weights with Xavier uniform for stability."""
        for module in self.modules():
            if isinstance(module, nn.Linear):
                nn.init.xavier_uniform_(module.weight)
                if module.bias is not None:
                    nn.init.zeros_(module.bias)
            elif isinstance(module, nn.Conv2d):
                nn.init.xavier_uniform_(module.weight)
                if module.bias is not None:
                    nn.init.zeros_(module.bias)
        logger.info("Initialized AppAssetDiT weights")

    def forward(
        self,
        latent: torch.Tensor,  # (B, C, H, W) latent tensor
        timestep: torch.Tensor,  # (B,) diffusion timestep
        text_embed: torch.Tensor,  # (B, seq_len, text_embed_dim) CLIP text embeds
        asset_type_id: torch.Tensor  # (B,) asset type ID (0-3)
    ) -> torch.Tensor:
        """
        Forward pass of DiT to predict noise residual.

        Args:
            latent: Noisy latent tensor from diffusion process
            timestep: Current diffusion timestep
            text_embed: Text embeddings from CLIP encoder
            asset_type_id: Integer ID of app asset type

        Returns:
            Predicted noise residual (same shape as latent)
        """
        batch_size = latent.shape[0]

        # Error handling: validate input shapes
        if latent.shape[1] != 4:
            raise ValueError(f"Latent must have 4 channels, got {latent.shape[1]}")
        if timestep.shape[0] != batch_size:
            raise ValueError(f"Timestep batch size {timestep.shape[0]} != latent batch size {batch_size}")
        if asset_type_id.shape[0] != batch_size:
            raise ValueError(f"Asset type ID batch size {asset_type_id.shape[0]} != latent batch size {batch_size}")

        # 1. Patchify latent
        x = self.patch_embed(latent)  # (B, embed_dim, H/patch, W/patch)
        x = x.flatten(2).transpose(1, 2)  # (B, num_patches, embed_dim)

        # 2. Add position embedding
        x = x + self.pos_embed[:, 1:, :]  # Skip CLS token position

        # 3. Process timestep embedding
        # Sinusoidal timestep embedding
        half_dim = self.embed_dim // 2
        emb = torch.log(torch.tensor(10000.0)) / (half_dim - 1)
        emb = torch.exp(torch.arange(half_dim, device=latent.device) * -emb)
        emb = timestep[:, None] * emb[None, :]
        emb = torch.cat([torch.sin(emb), torch.cos(emb)], dim=-1)
        if emb.shape[-1] % 2 == 1:
            emb = F.pad(emb, (0, 1), mode="constant")  # Pad if odd embed dim
        timestep_emb = self.timestep_embed(emb)  # (B, embed_dim)

        # 4. Add asset type embedding
        asset_emb = self.asset_type_embed(asset_type_id)  # (B, embed_dim)
        x = x + asset_emb[:, None, :]  # Broadcast to all patches

        # 5. Process text embeddings and add as cross-attention context
        text_emb = self.text_proj(text_embed)  # (B, seq_len, embed_dim)
        # Append timestep embed to text embeds for conditioning
        text_emb = torch.cat([text_emb, timestep_emb[:, None, :]], dim=1)

        # 6. Pass through transformer blocks with cross-attention to text
        for block in self.blocks:
            # Self-attention on image patches
            x = block.self_attn(x, x, x)[0]
            # Cross-attention to text embeddings
            x = block.cross_attn(x, text_emb, text_emb)[0]
            # Feedforward
            x = block.feed_forward(x)

        # 7. Project to noise residual shape
        x = self.output_layer(x)  # (B, num_patches, C*patch*patch)
        # Reshape to latent shape
        x = x.reshape(
            batch_size,
            self.num_patches,
            -1,
            self.patch_size,
            self.patch_size
        )
        # Rearrange to (B, C, H, W)
        x = x.reshape(
            batch_size,
            4,  # input channels
            512 // self.patch_size,
            512 // self.patch_size
        )
        return x

    def generate(
        self,
        text_embed: torch.Tensor,
        asset_type_id: torch.Tensor,
        num_inference_steps: int = 50,
        device: str = "cuda"
    ) -> torch.Tensor:
        """Generate latent tensor from text embeddings via reverse diffusion."""
        # Start with random noise
        latent = torch.randn((1, 4, 512, 512), device=device)
        # Linear noise schedule
        timesteps = torch.linspace(1000, 0, num_inference_steps, device=device)
        for t in timesteps:
            timestep = torch.full((1,), t, device=device, dtype=torch.long)
            # Predict noise residual
            noise_pred = self.forward(latent, timestep, text_embed, asset_type_id)
            # Simple DDPM reverse step (simplified for example)
            latent = latent - noise_pred * 0.02  # Step size from DALL-E 4 config
        return latent

# Example usage
if __name__ == "__main__":
    device = "cuda" if torch.cuda.is_available() else "cpu"
    model = AppAssetDiT().to(device)
    # Dummy inputs
    latent = torch.randn(1, 4, 512, 512, device=device)
    timestep = torch.tensor([500], device=device)
    text_embed = torch.randn(1, 77, 768, device=device)  # CLIP max seq length 77
    asset_type_id = torch.tensor([0], device=device)  # icon
    try:
        output = model(latent, timestep, text_embed, asset_type_id)
        print(f"Output shape: {output.shape}")  # Should be (1,4,512,512)
        # Test generation
        generated = model.generate(text_embed, asset_type_id, device=device)
        print(f"Generated latent shape: {generated.shape}")
    except Exception as e:
        logger.error(f"Forward pass failed: {e}")

import io
import json
import logging
from typing import Dict, List, Optional
from dataclasses import dataclass
from PIL import Image, ImageFilter
import numpy as np

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

@dataclass
class AssetVariant:
    """Represents a single density variant of an app asset."""
    density: str  # e.g., mdpi, hdpi, xhdpi
    width: int
    height: int
    format: str  # png, webp
    image_bytes: bytes

@dataclass
class ProcessedAsset:
    """Final processed app asset with all variants and metadata."""
    asset_id: str
    asset_type: str
    base_resolution: str
    variants: List[AssetVariant]
    metadata: Dict
    generation_cost_usd: float

class AppAssetPostProcessor:
    """
    Post-processes generated images into app-ready assets: density variants,
    format conversion, metadata tagging.
    """

    # Density to scale factor mapping (relative to base 512x512 mdpi)
    DENSITY_SCALES = {
        "ldpi": 0.75,
        "mdpi": 1.0,
        "hdpi": 1.5,
        "xhdpi": 2.0,
        "xxhdpi": 3.0,
        "xxxhdpi": 4.0
    }

    # Supported output formats
    SUPPORTED_FORMATS = {"png", "webp", "jpg"}

    def __init__(
        self,
        default_format: str = "png",
        enable_metadata_tagging: bool = True
    ):
        self.default_format = default_format
        if default_format not in self.SUPPORTED_FORMATS:
            raise ValueError(f"Unsupported format: {default_format}")
        self.enable_metadata_tagging = enable_metadata_tagging

    def _resize_with_aspect_ratio(
        self,
        image: Image.Image,
        target_width: int,
        target_height: int
    ) -> Image.Image:
        """
        Resize image to target dimensions while preserving aspect ratio,
        cropping to fit if necessary.
        """
        # Calculate aspect ratios
        src_aspect = image.width / image.height
        target_aspect = target_width / target_height

        if src_aspect > target_aspect:
            # Source is wider, crop width
            new_width = int(image.height * target_aspect)
            left = (image.width - new_width) // 2
            image = image.crop((left, 0, left + new_width, image.height))
        else:
            # Source is taller, crop height
            new_height = int(image.width / target_aspect)
            top = (image.height - new_height) // 2
            image = image.crop((0, top, image.width, top + new_height))

        # Resize to target
        return image.resize((target_width, target_height), Image.Resampling.LANCZOS)

    def _add_metadata(
        self,
        image: Image.Image,
        metadata: Dict
    ) -> Image.Image:
        """Add metadata tags to image (PNG tEXt chunks, WebP EXIF)."""
        if not self.enable_metadata_tagging:
            return image
        # For PNG, use tEXt chunks
        if image.format == "PNG":
            for key, value in metadata.items():
                image.info[key] = str(value)
        # For WebP, use EXIF
        elif image.format == "WEBP":
            exif = image.getexif()
            for i, (key, value) in enumerate(metadata.items()):
                exif[i + 1000] = str(value)  # Use custom EXIF tags
            image.info["exif"] = exif
        return image

    def process(
        self,
        image_bytes: bytes,
        asset_id: str,
        asset_type: str,
        base_resolution: str,
        generation_cost_usd: float,
        target_densities: Optional[List[str]] = None,
        target_formats: Optional[List[str]] = None
    ) -> ProcessedAsset:
        """
        Process raw generated image into app-ready assets.

        Args:
            image_bytes: Raw bytes of generated image (PNG/WebP)
            asset_id: Unique ID for the asset
            asset_type: Type of asset (icon, splash, etc.)
            base_resolution: Base resolution of generated image (e.g., 512x512)
            generation_cost_usd: Cost of generating the base image
            target_densities: List of densities to generate (default all)
            target_formats: List of formats to generate (default default_format)

        Returns:
            ProcessedAsset with all variants
        """
        # Validate inputs
        if not image_bytes:
            raise ValueError("Empty image bytes provided")
        target_densities = target_densities or list(self.DENSITY_SCALES.keys())
        target_formats = target_formats or [self.default_format]

        # Validate densities
        invalid_densities = [d for d in target_densities if d not in self.DENSITY_SCALES]
        if invalid_densities:
            raise ValueError(f"Invalid densities: {invalid_densities}")

        # Validate formats
        invalid_formats = [f for f in target_formats if f not in self.SUPPORTED_FORMATS]
        if invalid_formats:
            raise ValueError(f"Invalid formats: {invalid_formats}")

        # Load base image
        try:
            base_image = Image.open(io.BytesIO(image_bytes))
            base_image = base_image.convert("RGBA")  # Ensure alpha channel
        except Exception as e:
            logger.error(f"Failed to load base image for asset {asset_id}: {e}")
            raise

        # Parse base resolution
        try:
            base_w, base_h = map(int, base_resolution.split("x"))
        except ValueError:
            raise ValueError(f"Invalid base resolution format: {base_resolution}")

        variants = []
        metadata_base = {
            "asset_id": asset_id,
            "asset_type": asset_type,
            "generator": "DALL-E 4 v2.1.0",
            "base_resolution": base_resolution
        }

        # Generate variants for each density and format
        for density in target_densities:
            scale = self.DENSITY_SCALES[density]
            target_w = int(base_w * scale)
            target_h = int(base_h * scale)

            for fmt in target_formats:
                # Resize image
                variant_image = self._resize_with_aspect_ratio(base_image, target_w, target_h)
                # Apply slight sharpening for smaller densities
                if scale < 1.0:
                    variant_image = variant_image.filter(ImageFilter.SHARPEN)
                # Convert format if needed
                if fmt == "jpg":
                    variant_image = variant_image.convert("RGB")  # Remove alpha for JPG
                # Add metadata
                metadata = {**metadata_base, "density": density, "format": fmt}
                variant_image = self._add_metadata(variant_image, metadata)
                # Save to bytes
                buf = io.BytesIO()
                if fmt == "png":
                    variant_image.save(buf, format="PNG", optimize=True)
                elif fmt == "webp":
                    variant_image.save(buf, format="WebP", quality=85, method=6)
                elif fmt == "jpg":
                    variant_image.save(buf, format="JPEG", quality=90, optimize=True)
                # Create variant
                variants.append(AssetVariant(
                    density=density,
                    width=target_w,
                    height=target_h,
                    format=fmt,
                    image_bytes=buf.getvalue()
                ))
                logger.info(f"Generated {density} {fmt} variant for asset {asset_id}: {target_w}x{target_h}")

        # Create final metadata
        final_metadata = {
            **metadata_base,
            "num_variants": len(variants),
            "target_densities": target_densities,
            "target_formats": target_formats,
            "total_size_bytes": sum(len(v.image_bytes) for v in variants)
        }

        return ProcessedAsset(
            asset_id=asset_id,
            asset_type=asset_type,
            base_resolution=base_resolution,
            variants=variants,
            metadata=final_metadata,
            generation_cost_usd=generation_cost_usd
        )

# Example usage
if __name__ == "__main__":
    processor = AppAssetPostProcessor(default_format="webp")
    # Dummy image bytes (1x1 red pixel PNG)
    dummy_png = b'\x89PNG\r\n\x1a\n\x00\x00\x00\rIHDR\x00\x00\x00\x01\x00\x00\x00\x01\x08\x06\x00\x00\x00\x1f\x15\xc4\x89\x00\x00\x00\nIDATx\x9cc\x00\x01\x00\x00\x05\x00\x01\r\n\xb4\x00\x00\x00\x00IEND\xaeB`\x82'
    try:
        result = processor.process(
            image_bytes=dummy_png,
            asset_id="ast_12345",
            asset_type="icon",
            base_resolution="512x512",
            generation_cost_usd=0.00012,
            target_densities=["mdpi", "xhdpi"],
            target_formats=["png", "webp"]
        )
        print(f"Generated {len(result.variants)} variants")
        for v in result.variants:
            print(f"Variant: {v.density} {v.format} {v.width}x{v.height}, size: {len(v.image_bytes)} bytes")
    except Exception as e:
        logger.error(f"Processing failed: {e}")

import hashlib
import redis
import numpy as np

# Initialize Redis connection pool
redis_pool = redis.ConnectionPool(host="localhost", port=6379, db=0, decode_responses=False)
redis_client = redis.Redis(connection_pool=redis_pool)

def get_cached_clip_embedding(prompt: str) -> Optional[np.ndarray]:
    """Retrieve cached CLIP embedding for a prompt."""
    # Hash prompt to use as cache key
    cache_key = f"clip_embed:{hashlib.sha256(prompt.encode()).hexdigest()}"
    cached = redis_client.get(cache_key)
    if cached:
        return np.frombuffer(cached, dtype=np.float32).reshape(1, 77, 768)
    return None

def cache_clip_embedding(prompt: str, embedding: np.ndarray, ttl: int = 604800):
    """Cache CLIP embedding in Redis with 7-day TTL."""
    cache_key = f"clip_embed:{hashlib.sha256(prompt.encode()).hexdigest()}"
    redis_client.setex(cache_key, ttl, embedding.tobytes())

from fastapi import FastAPI, HTTPException
import torch
from typing import List
from pydantic import BaseModel

app = FastAPI()

class BatchPromptRequest(BaseModel):
    prompts: List[str]
    asset_type: str = "icon"

@app.post("/batch-generate")
async def batch_generate_assets(request: BatchPromptRequest):
    if len(request.prompts) > 16:
        raise HTTPException(status_code=400, detail="Max batch size is 16")
    # Load preprocessor, text encoder, DiT model (omitted for brevity)
    # Preprocess all prompts
    preprocessed = [preprocessor.preprocess(p) for p in request.prompts]
    # Encode text in batch
    text_embeds = torch.stack([clip_encoder(p.cleaned_prompt) for p in preprocessed])
    # Run batch inference
    latents = dit_model.generate_batch(text_embeds, asset_type_id=0)
    # Post-process all latents
    assets = [post_processor.process(latent) for latent in latents]
    return {"assets": [a.dict() for a in assets]}

import cv2
import numpy as np
from PIL import Image
from typing import List

def validate_asset_colors(
    image_bytes: bytes,
    brand_color_ranges: List[tuple]  # List of (lower_hsv, upper_hsv) tuples
) -> bool:
    """Validate that 95% of asset pixels fall within brand color ranges."""
    img = Image.open(io.BytesIO(image_bytes)).convert("RGB")
    img_cv = cv2.cvtColor(np.array(img), cv2.COLOR_RGB2HSV)
    total_pixels = img_cv.shape[0] * img_cv.shape[1]
    matching_pixels = 0
    for lower, upper in brand_color_ranges:
        mask = cv2.inRange(img_cv, np.array(lower), np.array(upper))
        matching_pixels += cv2.countNonZero(mask)
    return (matching_pixels / total_pixels) >= 0.95

def validate_asset_edges(image_bytes: bytes) -> bool:
    """Reject assets with blurry edges (Laplacian variance < 100)."""
    img = Image.open(io.BytesIO(image_bytes)).convert("L")
    img_cv = np.array(img)
    laplacian = cv2.Laplacian(img_cv, cv2.CV_64F)
    return laplacian.var() >= 100  # Threshold for sharp edges