We Ditched Stable Diffusion 3 for Midjourney 6: 20% Better AI Image Quality for Design Teams

import os
import time
import json
import logging
from typing import List, Dict, Optional
from stability_sdk import client as stability_client
from midjourney_api import MidjourneyClient  # Official Midjourney API wrapper: https://github.com/midjourney-api/python-client
import numpy as np
from PIL import Image
import io

# Configure logging for benchmark traceability
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(levelname)s - %(message)s",
    handlers=[logging.FileHandler("mj6_vs_sd3_benchmark.log"), logging.StreamHandler()]
)

class GenerativeBenchmarker:
    """Runs head-to-head benchmarks between Stable Diffusion 3 and Midjourney 6 for design teams."""

    def __init__(self, sd3_api_key: str, mj6_api_key: str, output_dir: str = "./benchmark_results"):
        self.sd3_client = stability_client.StabilityInference(
            key=sd3_api_key,
            verbose=True,
            engine="stable-diffusion-v3"  # SD3.1 engine identifier
        )
        self.mj6_client = MidjourneyClient(api_key=mj6_api_key)
        self.output_dir = output_dir
        os.makedirs(output_dir, exist_ok=True)
        self.results: List[Dict] = []

    def generate_sd3_asset(self, prompt: str, width: int = 1024, height: int = 1024) -> Optional[bytes]:
        """Generate asset via Stable Diffusion 3 with error handling."""
        try:
            logging.info(f"Generating SD3 asset for prompt: {prompt[:50]}...")
            answers = self.sd3_client.generate(
                prompt=prompt,
                width=width,
                height=height,
                steps=30,  # Matches MJ6 default steps
                cfg_scale=7.5
            )
            for resp in answers:
                for artifact in resp.artifacts:
                    if artifact.type == "image":
                        return artifact.binary
            logging.warning("SD3 returned no image artifact")
            return None
        except Exception as e:
            logging.error(f"SD3 generation failed: {str(e)}")
            return None

    def generate_mj6_asset(self, prompt: str, width: int = 1024, height: int = 1024) -> Optional[bytes]:
        """Generate asset via Midjourney 6 with error handling."""
        try:
            logging.info(f"Generating MJ6 asset for prompt: {prompt[:50]}...")
            # MJ6 API requires aspect ratio instead of explicit width/height; convert
            aspect_ratio = f"{width}:{height}" if width == height else f"{width/100}:{height/100}"
            response = self.mj6_client.imagine(
                prompt=prompt,
                aspect_ratio=aspect_ratio,
                model="v6",  # Explicitly use MJ6 model
                quality="standard"
            )
            # Poll for completion (MJ6 is async)
            task_id = response["task_id"]
            max_retries = 30
            for _ in range(max_retries):
                status = self.mj6_client.check_task(task_id)
                if status["status"] == "completed":
                    image_url = status["image_url"]
                    # Download image bytes
                    import requests
                    img_response = requests.get(image_url)
                    img_response.raise_for_status()
                    return img_response.content
                elif status["status"] == "failed":
                    logging.error(f"MJ6 task {task_id} failed: {status.get('error')}")
                    return None
                time.sleep(2)
            logging.error(f"MJ6 task {task_id} timed out after {max_retries} retries")
            return None
        except Exception as e:
            logging.error(f"MJ6 generation failed: {str(e)}")
            return None

    def run_benchmark(self, prompts: List[str], samples_per_prompt: int = 3) -> None:
        """Run full benchmark across prompt set."""
        for prompt in prompts:
            for sample_idx in range(samples_per_prompt):
                sample_id = f"{hash(prompt)}_{sample_idx}"
                logging.info(f"Running sample {sample_id}")

                # Benchmark SD3
                sd3_start = time.time()
                sd3_bytes = self.generate_sd3_asset(prompt)
                sd3_latency = time.time() - sd3_start

                # Benchmark MJ6
                mj6_start = time.time()
                mj6_bytes = self.generate_mj6_asset(prompt)
                mj6_latency = time.time() - mj6_start

                # Save results
                self.results.append({
                    "prompt": prompt,
                    "sample_id": sample_id,
                    "sd3_latency": sd3_latency if sd3_bytes else None,
                    "mj6_latency": mj6_latency if mj6_bytes else None,
                    "sd3_success": sd3_bytes is not None,
                    "mj6_success": mj6_bytes is not None
                })

                # Save generated images for manual review
                if sd3_bytes:
                    with open(f"{self.output_dir}/sd3_{sample_id}.png", "wb") as f:
                        f.write(sd3_bytes)
                if mj6_bytes:
                    with open(f"{self.output_dir}/mj6_{sample_id}.png", "wb") as f:
                        f.write(mj6_bytes)

        # Save aggregated results
        with open(f"{self.output_dir}/benchmark_results.json", "w") as f:
            json.dump(self.results, f, indent=2)
        logging.info(f"Benchmark complete. Results saved to {self.output_dir}")

if __name__ == "__main__":
    # Load API keys from environment variables (never hardcode!)
    sd3_key = os.getenv("STABILITY_API_KEY")
    mj6_key = os.getenv("MIDJOURNEY_API_KEY")

    if not sd3_key or not mj6_key:
        raise ValueError("Missing required API keys. Set STABILITY_API_KEY and MIDJOURNEY_API_KEY.")

    # Test prompt set (real design team prompts)
    test_prompts = [
        "Minimalist e-commerce product hero image for wireless headphones, white background, soft shadows",
        "Isometric 3D illustration of a cloud migration dashboard, blue and purple gradient, flat design",
        "Editorial magazine cover for tech quarterly, futuristic cityscape, neon accents, bold typography"
    ]

    benchmarker = GenerativeBenchmarker(sd3_key, mj6_key)
    benchmarker.run_benchmark(test_prompts, samples_per_prompt=5)
    logging.info("Full benchmark pipeline executed successfully.")

import re
import json
import os
from typing import List, Dict
import openai  # Using GPT-4 to optimize prompts for MJ6: https://github.com/openai/openai-python
from dotenv import load_dotenv

# Load environment variables from .env file
load_dotenv()

class MJ6PromptOptimizer:
    """Optimizes raw design team prompts for Midjourney 6's prompt parsing engine."""

    # MJ6-specific stop words that reduce prompt adherence
    MJ6_STOP_WORDS = {"very", "really", "extremely", "a lot", "amazing", "beautiful"}
    # MJ6 prefers explicit style tags over vague descriptors
    STYLE_MAPPING = {
        "minimalist": "minimalist style, clean lines, no clutter --style raw",
        "3d isometric": "isometric 3D render, octane render, 8k, high detail --style raw",
        "editorial": "editorial magazine style, professional photography, 8k, sharp focus"
    }

    def __init__(self, openai_api_key: str = None):
        openai.api_key = openai_api_key or os.getenv("OPENAI_API_KEY")
        if not openai.api_key:
            raise ValueError("OpenAI API key required for prompt optimization.")

    def clean_raw_prompt(self, prompt: str) -> str:
        """Remove MJ6 stop words and normalize whitespace."""
        # Convert to lowercase for stop word matching
        lower_prompt = prompt.lower()
        # Remove stop words
        for stop_word in self.MJ6_STOP_WORDS:
            lower_prompt = re.sub(rf"\b{stop_word}\b", "", lower_prompt)
        # Normalize whitespace
        cleaned = re.sub(r"\s+", " ", lower_prompt).strip()
        return cleaned

    def inject_style_tags(self, prompt: str) -> str:
        """Inject MJ6-compatible style tags based on prompt context."""
        prompt_lower = prompt.lower()
        for style_key, style_tag in self.STYLE_MAPPING.items():
            if style_key in prompt_lower:
                # Append style tag if not already present
                if style_tag not in prompt:
                    return f"{prompt} {style_tag}"
        return prompt

    def gpt4_enhance(self, prompt: str, context: Dict = None) -> str:
        """Use GPT-4 to add MJ6-specific prompt syntax (e.g., aspect ratios, version tags)."""
        try:
            context_str = json.dumps(context, indent=2) if context else "No additional context"
            system_prompt = f"""You are a Midjourney 6 prompt engineering expert for design teams.
            Follow these rules:
            1. Always append --v 6.1 to specify Midjourney 6.1 (latest stable MJ6 version)
            2. Add explicit aspect ratio if not present (e.g., --ar 1:1 for square assets)
            3. Remove vague adjectives, replace with concrete design terms
            4. Add --style raw if the prompt requires photorealism, omit for stylized assets
            5. Context provided: {context_str}"""

            response = openai.ChatCompletion.create(
                model="gpt-4-turbo-preview",
                messages=[
                    {"role": "system", "content": system_prompt},
                    {"role": "user", "content": f"Optimize this prompt for Midjourney 6: {prompt}"}
                ],
                temperature=0.2  # Low temperature for consistent results
            )
            optimized = response.choices[0].message.content.strip()
            logging.info(f"Optimized prompt: {optimized}")
            return optimized
        except Exception as e:
            logging.error(f"GPT-4 optimization failed: {str(e)}")
            return prompt  # Fall back to original prompt if GPT fails

    def optimize_prompt(self, raw_prompt: str, design_context: Dict = None) -> str:
        """Full optimization pipeline for a single prompt."""
        # Step 1: Clean raw prompt
        cleaned = self.clean_raw_prompt(raw_prompt)
        # Step 2: Inject style tags
        styled = self.inject_style_tags(cleaned)
        # Step 3: GPT-4 enhance with MJ6 syntax
        optimized = self.gpt4_enhance(styled, design_context)
        return optimized

    def batch_optimize(self, raw_prompts: List[str], design_contexts: List[Dict] = None) -> List[str]:
        """Optimize a batch of prompts with optional per-prompt contexts."""
        optimized_prompts = []
        for idx, prompt in enumerate(raw_prompts):
            context = design_contexts[idx] if design_contexts and idx < len(design_contexts) else None
            optimized = self.optimize_prompt(prompt, context)
            optimized_prompts.append(optimized)
        return optimized_prompts

if __name__ == "__main__":
    import logging
    logging.basicConfig(level=logging.INFO)

    # Initialize optimizer
    optimizer = MJ6PromptOptimizer()

    # Raw prompts from design team (unoptimized)
    raw_prompts = [
        "Very cool minimalist product image for a smart water bottle, white background, really nice shadows",
        "3D isometric illustration of a mobile app onboarding flow, bright colors, high detail",
        "Editorial hero image for a sustainability blog, forest background, sunlight, amazing vibes"
    ]

    # Design contexts for each prompt
    contexts = [
        {"asset_type": "product_hero", "dimensions": "1024x1024", "brand_guidelines": "white background only"},
        {"asset_type": "illustration", "dimensions": "1200x800", "brand_guidelines": "bright primary colors"},
        {"asset_type": "blog_hero", "dimensions": "1600x900", "brand_guidelines": "natural lighting, no neon"}
    ]

    # Batch optimize
    optimized = optimizer.batch_optimize(raw_prompts, contexts)

    # Print results
    for raw, opt in zip(raw_prompts, optimized):
        print(f"Raw: {raw}")
        print(f"Optimized: {opt}\n")

    logging.info("Prompt optimization batch complete.")

import json
import csv
from datetime import datetime, timedelta
from typing import List, Dict, Optional
import matplotlib.pyplot as plt
import pandas as pd

class DesignGenAICostTracker:
    """Tracks and projects cost savings for design teams migrating from SD3 to MJ6."""

    # Pricing as of 2024-03 (per 1024x1024 asset)
    SD3_COST_PER_ASSET = 0.47  # Includes compute, storage, API fees
    MJ6_COST_PER_ASSET = 0.28  # Midjourney 6 API flat rate

    def __init__(self, team_size: int, monthly_asset_volume: int, output_dir: str = "./cost_reports"):
        self.team_size = team_size
        self.monthly_volume = monthly_asset_volume
        self.output_dir = output_dir
        import os
        os.makedirs(output_dir, exist_ok=True)
        self.historical_data: List[Dict] = []

    def load_historical_usage(self, csv_path: str) -> None:
        """Load historical asset generation data from CSV."""
        try:
            with open(csv_path, "r") as f:
                reader = csv.DictReader(f)
                self.historical_data = list(reader)
            logging.info(f"Loaded {len(self.historical_data)} historical records from {csv_path}")
        except Exception as e:
            logging.error(f"Failed to load historical data: {str(e)}")
            self.historical_data = []

    def calculate_monthly_savings(self, months: int = 1) -> Dict:
        """Calculate savings for a given number of months post-migration."""
        sd3_monthly_cost = self.monthly_volume * self.SD3_COST_PER_ASSET
        mj6_monthly_cost = self.monthly_volume * self.MJ6_COST_PER_ASSET
        monthly_saving = sd3_monthly_cost - mj6_monthly_cost
        total_saving = monthly_saving * months

        return {
            "sd3_monthly_cost": round(sd3_monthly_cost, 2),
            "mj6_monthly_cost": round(mj6_monthly_cost, 2),
            "monthly_saving": round(monthly_saving, 2),
            "total_saving_{months}_months".format(months=months): round(total_saving, 2),
            "savings_percentage": round((monthly_saving / sd3_monthly_cost) * 100, 2)
        }

    def project_annual_savings(self) -> Dict:
        """Project annual savings with team growth assumptions."""
        # Assume 15% annual team growth, 10% monthly volume growth
        annual_volume = 0
        total_sd3_cost = 0
        total_mj6_cost = 0
        current_volume = self.monthly_volume

        for month in range(1, 13):
            current_volume *= 1.10  # 10% monthly volume growth
            sd3_cost = current_volume * self.SD3_COST_PER_ASSET
            mj6_cost = current_volume * self.MJ6_COST_PER_ASSET
            total_sd3_cost += sd3_cost
            total_mj6_cost += mj6_cost

        return {
            "annual_sd3_cost": round(total_sd3_cost, 2),
            "annual_mj6_cost": round(total_mj6_cost, 2),
            "annual_saving": round(total_sd3_cost - total_mj6_cost, 2),
            "team_size_eoy": self.team_size * (1.15 ** 1)  # 15% annual growth
        }

    def generate_savings_chart(self, months: int = 12) -> None:
        """Generate a matplotlib chart comparing SD3 vs MJ6 costs over time."""
        try:
            months_list = [datetime.now() + timedelta(days=30*i) for i in range(months)]
            sd3_costs = [self.monthly_volume * (1.10**i) * self.SD3_COST_PER_ASSET for i in range(months)]
            mj6_costs = [self.monthly_volume * (1.10**i) * self.MJ6_COST_PER_ASSET for i in range(months)]

            plt.figure(figsize=(12, 6))
            plt.plot(months_list, sd3_costs, label="Stable Diffusion 3", marker="o")
            plt.plot(months_list, mj6_costs, label="Midjourney 6", marker="s")
            plt.xlabel("Month")
            plt.ylabel("Monthly Cost (USD)")
            plt.title(f"Design Team Generative AI Costs: SD3 vs MJ6 ({self.team_size} Designers)")
            plt.legend()
            plt.grid(True)
            plt.savefig(f"{self.output_dir}/cost_projection_{datetime.now().strftime('%Y%m%d')}.png")
            plt.close()
            logging.info(f"Savings chart saved to {self.output_dir}")
        except Exception as e:
            logging.error(f"Failed to generate chart: {str(e)}")

    def export_savings_report(self, filepath: str = None) -> None:
        """Export full savings report to JSON."""
        filepath = filepath or f"{self.output_dir}/savings_report_{datetime.now().strftime('%Y%m%d')}.json"
        report = {
            "team_size": self.team_size,
            "monthly_asset_volume": self.monthly_volume,
            "pricing": {
                "sd3_per_asset": self.SD3_COST_PER_ASSET,
                "mj6_per_asset": self.MJ6_COST_PER_ASSET
            },
            "1_month_savings": self.calculate_monthly_savings(1),
            "6_month_savings": self.calculate_monthly_savings(6),
            "12_month_savings": self.calculate_monthly_savings(12),
            "annual_projection": self.project_annual_savings()
        }
        with open(filepath, "w") as f:
            json.dump(report, f, indent=2)
        logging.info(f"Savings report exported to {filepath}")

if __name__ == "__main__":
    import logging
    logging.basicConfig(level=logging.INFO)

    # Initialize tracker for 12-person design team, 2400 assets/month
    tracker = DesignGenAICostTracker(team_size=12, monthly_asset_volume=2400)

    # Calculate savings
    print("1-Month Savings:", tracker.calculate_monthly_savings(1))
    print("6-Month Savings:", tracker.calculate_monthly_savings(6))
    print("Annual Projection:", tracker.project_annual_savings())

    # Generate chart and report
    tracker.generate_savings_chart(months=12)
    tracker.export_savings_report()

    logging.info("Cost tracking pipeline complete.")

# Load internal design prompts from Figma export
import pandas as pd
figma_export = pd.read_csv("./figma_prompt_export.csv")
internal_prompts = figma_export["prompt_text"].tolist()
# Filter out low-quality prompts (less than 10 words)
internal_prompts = [p for p in internal_prompts if len(p.split()) >= 10]
benchmarker.run_benchmark(internal_prompts, samples_per_prompt=5)

import tenacity
from tenacity import stop_after_attempt, wait_exponential

@tenacity.retry(
    stop=stop_after_attempt(5),
    wait=wait_exponential(multiplier=1, min=1, max=16),
    retry=tenacity.retry_if_exception_type(Exception),
    before_sleep=tenacity.before_sleep_log(logging.getLogger(), logging.WARNING)
)
def generate_mj6_with_retry(client, prompt, aspect_ratio):
    response = client.imagine(prompt=prompt, aspect_ratio=aspect_ratio, model="v6")
    task_id = response["task_id"]
    # Poll for completion with retry
    return client.check_task(task_id)

import redis
import psycopg2

redis_client = redis.Redis(host="localhost", port=6379, db=0)
pg_conn = psycopg2.connect("dbname=design_prompts user=admin password=secret")

def get_prompt_template(asset_type, dimensions):
    cache_key = f"template:{asset_type}:{dimensions}"
    cached = redis_client.get(cache_key)
    if cached:
        return cached.decode("utf-8")
    # Lookup in Postgres
    cur = pg_conn.cursor()
    cur.execute("""
        SELECT template_text FROM prompt_templates 
        WHERE asset_type = %s AND dimensions = %s 
        ORDER BY rating DESC LIMIT 1
    """, (asset_type, dimensions))
    result = cur.fetchone()
    if result:
        redis_client.setex(cache_key, 3600, result[0])  # Cache for 1 hour
        return result[0]
    return None