Benchmark: Stable Diffusion 3 vs. DALL-E 3 vs. Midjourney 6 for Generating Technical Diagram Images

import torch
from diffusers import StableDiffusion3Pipeline
from PIL import Image
import os
import json
from typing import List, Optional
import logging

# Configure logging for error tracking
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)

class SD3DiagramGenerator:
    \"\"\"Stable Diffusion 3 pipeline wrapper for technical diagram generation with benchmark tracking.\"\"\"

    def __init__(self, model_id: str = 'stabilityai/stable-diffusion-3-medium', device: str = 'cuda'):
        self.model_id = model_id
        self.device = device if torch.cuda.is_available() else 'cpu'
        self.pipe = None
        self.benchmark_metrics = []

        try:
            logger.info(f'Loading SD3 model: {model_id}')
            self.pipe = StableDiffusion3Pipeline.from_pretrained(
                model_id,
                torch_dtype=torch.float16 if self.device == 'cuda' else torch.float32
            )
            self.pipe.to(self.device)
            # Enable memory optimization for 48GB VRAM GPUs
            self.pipe.enable_model_cpu_offload()
            logger.info('SD3 model loaded successfully')
        except Exception as e:
            logger.error(f'Failed to load SD3 model: {str(e)}')
            raise RuntimeError(f'Model initialization failed: {str(e)}')

    def generate_diagram(
        self,
        prompt: str,
        output_path: str,
        negative_prompt: str = 'low quality, blurry, distorted text, incorrect labels, missing edges',
        num_inference_steps: int = 40,
        guidance_scale: float = 7.5
    ) -> Optional[Image.Image]:
        \"\"\"Generate a technical diagram with error handling and metric tracking.\"\"\"
        try:
            logger.info(f'Generating diagram for prompt: {prompt[:50]}...')
            # Track inference start time for latency metrics
            start_time = torch.cuda.Event(enable_timing=True)
            end_time = torch.cuda.Event(enable_timing=True)

            start_time.record()
            result = self.pipe(
                prompt=prompt,
                negative_prompt=negative_prompt,
                num_inference_steps=num_inference_steps,
                guidance_scale=guidance_scale,
                height=1024,
                width=1024
            ).images[0]
            end_time.record()

            torch.cuda.synchronize()
            latency_ms = start_time.elapsed_time(end_time)

            # Save output and track metrics
            os.makedirs(os.path.dirname(output_path), exist_ok=True)
            result.save(output_path)
            self.benchmark_metrics.append({
                'prompt': prompt,
                'latency_ms': latency_ms,
                'output_path': output_path
            })
            logger.info(f'Diagram saved to {output_path}, latency: {latency_ms:.2f}ms')
            return result
        except Exception as e:
            logger.error(f'Diagram generation failed: {str(e)}')
            return None

    def save_benchmark_metrics(self, path: str = 'sd3_benchmark.json'):
        \"\"\"Persist benchmark metrics to JSON.\"\"\"
        try:
            with open(path, 'w') as f:
                json.dump(self.benchmark_metrics, f, indent=2)
            logger.info(f'Benchmark metrics saved to {path}')
        except Exception as e:
            logger.error(f'Failed to save metrics: {str(e)}')

if __name__ == '__main__':
    # Example usage for generating a Kubernetes architecture diagram
    generator = SD3DiagramGenerator()
    diagram_prompt = '''Technical diagram of a 3-tier Kubernetes architecture: 
    - Frontend: 3 Nginx pods behind a Service
    - Backend: 5 Node.js pods with Redis cache sidecar
    - Database: PostgreSQL StatefulSet with 3 replicas
    - All components connected with labeled arrows, no distorted text, clean lines, white background'''

    output = generator.generate_diagram(
        prompt=diagram_prompt,
        output_path='outputs/k8s_architecture_sd3.png'
    )

    if output:
        print('Diagram generated successfully')
        generator.save_benchmark_metrics()
    else:
        print('Diagram generation failed')

import openai
import os
import json
import time
from typing import List, Dict, Optional
import logging
from dotenv import load_dotenv

# Load OpenAI API key from .env file
load_dotenv()
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class DALLE3DiagramGenerator:
    \"\"\"DALL-E 3 API wrapper for technical diagram generation with cost and rate limit tracking.\"\"\"

    def __init__(self, model: str = 'dall-e-3', size: str = '1024x1024'):
        self.model = model
        self.size = size
        self.api_key = os.getenv('OPENAI_API_KEY')
        self.total_cost = 0.0
        self.benchmark_metrics = []

        if not self.api_key:
            raise ValueError('OPENAI_API_KEY not found in environment variables')

        openai.api_key = self.api_key
        # DALL-E 3 pricing: $0.08 per 1024x1024 image
        self.cost_per_image = 0.08 if size == '1024x1024' else 0.04
        logger.info(f'Initialized DALL-E 3 generator: model={model}, size={size}, cost=${self.cost_per_image}/image')

    def generate_diagram(
        self,
        prompt: str,
        output_path: str,
        max_retries: int = 3,
        retry_delay: int = 5
    ) -> Optional[Dict]:
        \"\"\"Generate a diagram with retry logic for rate limits and API errors.\"\"\"
        retries = 0
        while retries < max_retries:
            try:
                logger.info(f'Generating DALL-E 3 diagram (attempt {retries+1}/{max_retries})')
                start_time = time.time()

                response = openai.images.generate(
                    model=self.model,
                    prompt=prompt,
                    size=self.size,
                    quality='standard',  # hd increases cost to $0.16/image
                    n=1
                )

                end_time = time.time()
                latency_s = end_time - start_time
                image_url = response.data[0].url

                # Track cost and metrics
                self.total_cost += self.cost_per_image
                metric = {
                    'prompt': prompt,
                    'latency_s': latency_s,
                    'cost_usd': self.cost_per_image,
                    'image_url': image_url,
                    'output_path': output_path
                }
                self.benchmark_metrics.append(metric)

                # Download and save image
                import requests
                img_data = requests.get(image_url).content
                os.makedirs(os.path.dirname(output_path), exist_ok=True)
                with open(output_path, 'wb') as f:
                    f.write(img_data)

                logger.info(f'Diagram saved to {output_path}, latency: {latency_s:.2f}s, total cost: ${self.total_cost:.2f}')
                return metric
            except openai.RateLimitError as e:
                logger.warning(f'Rate limit hit: {str(e)}. Retrying in {retry_delay}s...')
                time.sleep(retry_delay)
                retries += 1
            except Exception as e:
                logger.error(f'Generation failed: {str(e)}')
                retries += 1
                time.sleep(retry_delay)

        logger.error(f'Failed to generate diagram after {max_retries} retries')
        return None

    def generate_batch(
        self,
        prompts: List[str],
        output_dir: str = 'outputs/dalle3'
    ) -> List[Optional[Dict]]:
        \"\"\"Batch generate diagrams from a list of prompts.\"\"\"
        os.makedirs(output_dir, exist_ok=True)
        results = []
        for i, prompt in enumerate(prompts):
            output_path = os.path.join(output_dir, f'diagram_{i}.png')
            result = self.generate_diagram(prompt, output_path)
            results.append(result)
            # Respect rate limit: 1000 requests per minute = ~1 request per 60ms, but add buffer
            time.sleep(0.5)
        return results

    def save_metrics(self, path: str = 'dalle3_benchmark.json'):
        \"\"\"Save benchmark metrics and total cost to JSON.\"\"\"
        try:
            with open(path, 'w') as f:
                json.dump({
                    'total_cost_usd': self.total_cost,
                    'metrics': self.benchmark_metrics
                }, f, indent=2)
            logger.info(f'Metrics saved to {path}')
        except Exception as e:
            logger.error(f'Failed to save metrics: {str(e)}')

if __name__ == '__main__':
    generator = DALLE3DiagramGenerator()
    k8s_prompt = '''Technical diagram of a 3-tier Kubernetes architecture: 
    - Frontend: 3 Nginx pods behind a Service
    - Backend: 5 Node.js pods with Redis cache sidecar
    - Database: PostgreSQL StatefulSet with 3 replicas
    - All components connected with labeled arrows, no distorted text, clean lines, white background'''

    result = generator.generate_diagram(k8s_prompt, 'outputs/dalle3_k8s.png')
    if result:
        print(f'Diagram generated: {result[\"image_url\"]}')
        generator.save_metrics()
    else:
        print('Generation failed')

import discord
from discord.ext import commands
import os
import json
import time
import logging
from typing import List, Optional
import re

# Midjourney Discord bot configuration
DISCORD_TOKEN = os.getenv('MIDJOURNEY_DISCORD_TOKEN')
MIDJOURNEY_CHANNEL_ID = int(os.getenv('MIDJOURNEY_CHANNEL_ID', 0))
OUTPUT_DIR = 'outputs/midjourney'

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

class Midjourney6DiagramGenerator(commands.Bot):
    \"\"\"Discord bot wrapper for Midjourney 6 diagram generation with benchmark tracking.\"\"\"

    def __init__(self):
        intents = discord.Intents.default()
        intents.message_content = True
        super().__init__(command_prefix='!', intents=intents)

        self.channel = None
        self.benchmark_metrics = []
        self.total_cost = 0.0
        # Midjourney Mega tier: $120/month unlimited, ~$0.32 per image at 375 images/month
        self.cost_per_image = 0.32
        self.active_requests = {}

        if not DISCORD_TOKEN:
            raise ValueError('MIDJOURNEY_DISCORD_TOKEN not found in environment')
        if not MIDJOURNEY_CHANNEL_ID:
            raise ValueError('MIDJOURNEY_CHANNEL_ID not set')

        logger.info(f'Initialized Midjourney 6 generator, cost=${self.cost_per_image}/image')

    async def on_ready(self):
        \"\"\"Called when bot connects to Discord.\"\"\"
        logger.info(f'Logged in as {self.user} (ID: {self.user.id})')
        self.channel = self.get_channel(MIDJOURNEY_CHANNEL_ID)
        if not self.channel:
            raise ValueError(f'Channel {MIDJOURNEY_CHANNEL_ID} not found')
        logger.info(f'Connected to channel: {self.channel.name}')

    async def generate_diagram(
        self,
        prompt: str,
        output_path: str,
        timeout: int = 120
    ) -> Optional[dict]:
        \"\"\"Generate a diagram via Midjourney Discord API, wait for completion.\"\"\"
        try:
            logger.info(f'Generating Midjourney 6 diagram: {prompt[:50]}...')
            start_time = time.time()

            # Format Midjourney prompt with technical diagram parameters
            mj_prompt = f'/imagine prompt: {prompt} --ar 1:1 --style raw --v 6.0'
            msg = await self.channel.send(mj_prompt)
            self.active_requests[msg.id] = {'prompt': prompt, 'output_path': output_path, 'start_time': start_time}

            # Wait for Midjourney to process (poll for replies)
            elapsed = 0
            while elapsed < timeout:
                await discord.utils.sleep_until(time.time() + 5)
                elapsed = time.time() - start_time

                # Check for replies to the original message
                async for reply in self.channel.history(limit=10, after=msg):
                    if reply.author.id == 936929561302675456:  # Midjourney bot ID
                        if 'U1' in reply.content or 'U2' in reply.content:
                            # Extract image URL from embed
                            if reply.embeds:
                                image_url = reply.embeds[0].image.url
                                # Download image
                                import requests
                                img_data = requests.get(image_url).content
                                os.makedirs(os.path.dirname(output_path), exist_ok=True)
                                with open(output_path, 'wb') as f:
                                    f.write(img_data)

                                end_time = time.time()
                                latency_s = end_time - start_time
                                self.total_cost += self.cost_per_image

                                metric = {
                                    'prompt': prompt,
                                    'latency_s': latency_s,
                                    'cost_usd': self.cost_per_image,
                                    'image_url': image_url,
                                    'output_path': output_path
                                }
                                self.benchmark_metrics.append(metric)
                                logger.info(f'Diagram saved to {output_path}, latency: {latency_s:.2f}s')
                                return metric

            logger.error(f'Timeout waiting for Midjourney response after {timeout}s')
            return None
        except Exception as e:
            logger.error(f'Generation failed: {str(e)}')
            return None

    async def on_message(self, message):
        \"\"\"Handle incoming messages to track requests.\"\"\"
        if message.author.id == 936929561302675456:
            logger.debug(f'Midjourney message: {message.content}')
        await self.process_commands(message)

    def save_metrics(self, path: str = 'midjourney_benchmark.json'):
        \"\"\"Save benchmark metrics to JSON.\"\"\"
        try:
            with open(path, 'w') as f:
                json.dump({
                    'total_cost_usd': self.total_cost,
                    'metrics': self.benchmark_metrics
                }, f, indent=2)
            logger.info(f'Metrics saved to {path}')
        except Exception as e:
            logger.error(f'Failed to save metrics: {str(e)}')

if __name__ == '__main__':
    bot = Midjourney6DiagramGenerator()
    k8s_prompt = '''Technical diagram of a 3-tier Kubernetes architecture: 
    - Frontend: 3 Nginx pods behind a Service
    - Backend: 5 Node.js pods with Redis cache sidecar
    - Database: PostgreSQL StatefulSet with 3 replicas
    - All components connected with labeled arrows, no distorted text, clean lines, white background'''

    # Note: Bot runs asynchronously, this is a simplified example
    import asyncio
    async def main():
        await bot.login(DISCORD_TOKEN)
        await bot.connect()

    asyncio.run(main())

diagram_prompt = '''Technical diagram of [component name]: 
- [Component 1]: [quantity] [type], [connections]
- [Component 2]: [quantity] [type], [connections]
- All labels in Arial 12pt black text, no distorted characters
- Connection lines: solid 2px black, labeled with protocol/port
- White background, no gradients, no drop shadows
- UML 2.5/Kubernetes 1.28 compliant'''

from diffusers import StableDiffusion3Pipeline, LoRAAdapter

pipe = StableDiffusion3Pipeline.from_pretrained('stabilityai/stable-diffusion-3-medium')
lora_adapter = LoRAAdapter.from_pretrained('your-username/k8s-lora-adapter')
pipe.load_lora_adapter(lora_adapter)
pipe.to('cuda')

import hashlib

def get_prompt_hash(prompt: str) -> str:
    return hashlib.md5(prompt.encode()).hexdigest()

def generate_diagram_with_cache(self, prompt: str, output_path: str):
    prompt_hash = get_prompt_hash(prompt)
    cache_path = f'cache/{prompt_hash}.png'
    if os.path.exists(cache_path):
        import shutil
        shutil.copy(cache_path, output_path)
        return {'cached': True, 'output_path': output_path}
    # Generate new image if not cached
    result = self.generate_diagram(prompt, cache_path)
    if result:
        import shutil
        shutil.copy(cache_path, output_path)
    return result