Python ile Stable Diffusion API Kullanımı

Görüntü üretimi artık sadece tasarımcıların işi değil. Sistem yöneticileri olarak otomasyona olan aşinalığımız, Stable Diffusion’ı Python ile programatik olarak kullanmayı son derece cazip kılıyor. Bir pipeline kurmak, toplu görüntü üretmek ya da web uygulamanıza görüntü üretim kapasitesi eklemek istiyorsanız, bu yazı tam size göre.

Ortam Hazırlığı ve Kurulum

Önce temiz bir Python sanal ortamı oluşturalım. Bu adımı atlamayın, bağımlılık çatışmaları gerçek bir baş ağrısı olabilir.

python3 -m venv sd-env
source sd-env/bin/activate  # Linux/macOS
# Windows için: sd-envScriptsactivate

pip install --upgrade pip
pip install requests pillow python-dotenv torch torchvision
pip install diffusers transformers accelerate

Stable Diffusion’ı iki farklı şekilde kullanabilirsiniz: ya yerel olarak kendi makinenizde çalıştırırsınız ya da bir API servisi üzerinden erişirsiniz. Her iki senaryoyu da ele alacağız.

Yerel kurulum için AUTOMATIC1111’in WebUI’sini kullanacağız. Bu araç bir API sunucusu da barındırıyor ve üretim ortamlarında oldukça yaygın kullanılıyor.

git clone https://github.com/AUTOMATIC1111/stable-diffusion-webui
cd stable-diffusion-webui
# API modunu aktif ederek başlatın
./webui.sh --api --listen --port 7860

--api bayrağı kritik. Bu olmadan Python kodumuz sunucuyla konuşamaz. --listen ise sadece localhost değil, tüm arayüzlerden erişime izin veriyor. Bunu dikkatli kullanın, üretimde mutlaka bir güvenlik duvarı kuralı ekleyin.

Temel API İstek Yapısı

AUTOMATIC1111 API’si REST tabanlı çalışıyor ve endpointleri oldukça temiz. Temel bir text-to-image isteğini şöyle gönderiyoruz:

import requests
import base64
import json
from pathlib import Path

API_URL = "http://localhost:7860"

def generate_image(prompt, negative_prompt="", steps=20, width=512, height=512):
    """Temel görüntü üretim fonksiyonu"""
    
    payload = {
        "prompt": prompt,
        "negative_prompt": negative_prompt,
        "steps": steps,
        "width": width,
        "height": height,
        "cfg_scale": 7,
        "sampler_name": "DPM++ 2M Karras",
    }
    
    response = requests.post(
        f"{API_URL}/sdapi/v1/txt2img",
        json=payload,
        timeout=120
    )
    
    if response.status_code != 200:
        raise Exception(f"API hatasi: {response.status_code} - {response.text}")
    
    result = response.json()
    image_data = base64.b64decode(result["images"][0])
    
    return image_data

# Kullanım
image_bytes = generate_image(
    prompt="a serene mountain landscape, photorealistic, 8k",
    negative_prompt="blurry, low quality, distorted"
)

with open("output.png", "wb") as f:
    f.write(image_bytes)

print("Görüntü kaydedildi: output.png")

Bu basit yapı çoğu senaryo için yeterli, ama gerçek dünyada çok daha fazlasına ihtiyaç duyuyorsunuz.

Yapılandırma Yönetimi

Sabit değerleri koda gömmek yerine bir config sistemi kuralım. Özellikle birden fazla sunucunuz varsa ya da farklı model profilleriniz olduğunda bu yaklaşım hayat kurtarıyor.

import os
from dotenv import load_dotenv
from dataclasses import dataclass, field
from typing import Optional

load_dotenv()

@dataclass
class SDConfig:
    api_url: str = field(default_factory=lambda: os.getenv("SD_API_URL", "http://localhost:7860"))
    api_key: Optional[str] = field(default_factory=lambda: os.getenv("SD_API_KEY"))
    default_steps: int = int(os.getenv("SD_DEFAULT_STEPS", "20"))
    default_width: int = int(os.getenv("SD_DEFAULT_WIDTH", "512"))
    default_height: int = int(os.getenv("SD_DEFAULT_HEIGHT", "512"))
    default_cfg_scale: float = float(os.getenv("SD_CFG_SCALE", "7.0"))
    output_dir: str = os.getenv("SD_OUTPUT_DIR", "./outputs")
    timeout: int = int(os.getenv("SD_TIMEOUT", "120"))
    
    def __post_init__(self):
        Path(self.output_dir).mkdir(parents=True, exist_ok=True)

config = SDConfig()

.env dosyanız şu şekilde görünmeli:

SD_API_URL=http://192.168.1.100:7860
SD_API_KEY=your-optional-api-key
SD_DEFAULT_STEPS=25
SD_OUTPUT_DIR=/var/www/generated-images
SD_TIMEOUT=180

Gelişmiş İstemci Sınıfı

Gerçek projeler için bir istemci sınıfı oluşturmak en doğru yaklaşım. Hata yönetimi, yeniden deneme mekanizması ve loglama ekleyelim:

import requests
import base64
import logging
import time
from pathlib import Path
from datetime import datetime
from typing import Optional, List, Dict, Any

logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger("SDClient")

class StableDiffusionClient:
    def __init__(self, config: SDConfig):
        self.config = config
        self.session = requests.Session()
        
        if config.api_key:
            self.session.headers.update({"Authorization": f"Bearer {config.api_key}"})
        
        self.session.headers.update({"Content-Type": "application/json"})
    
    def _make_request(self, endpoint: str, payload: Dict[str, Any], retries: int = 3) -> Dict:
        """Yeniden deneme mekanizmasıyla HTTP isteği gönderir"""
        url = f"{self.config.api_url}{endpoint}"
        
        for attempt in range(retries):
            try:
                logger.info(f"İstek gönderiliyor: {endpoint} (deneme {attempt + 1}/{retries})")
                response = self.session.post(url, json=payload, timeout=self.config.timeout)
                response.raise_for_status()
                return response.json()
                
            except requests.exceptions.Timeout:
                logger.warning(f"Zaman aşımı (deneme {attempt + 1})")
                if attempt == retries - 1:
                    raise
                time.sleep(2 ** attempt)  # Üstel geri çekilme
                
            except requests.exceptions.HTTPError as e:
                logger.error(f"HTTP hatası: {e.response.status_code}")
                raise
                
            except requests.exceptions.ConnectionError:
                logger.error("Bağlantı hatası - API çalışıyor mu?")
                if attempt == retries - 1:
                    raise
                time.sleep(5)
        
    def txt2img(self, 
                prompt: str,
                negative_prompt: str = "",
                steps: Optional[int] = None,
                width: Optional[int] = None,
                height: Optional[int] = None,
                cfg_scale: Optional[float] = None,
                seed: int = -1,
                sampler: str = "DPM++ 2M Karras",
                batch_size: int = 1) -> List[bytes]:
        """Text-to-image üretimi"""
        
        payload = {
            "prompt": prompt,
            "negative_prompt": negative_prompt,
            "steps": steps or self.config.default_steps,
            "width": width or self.config.default_width,
            "height": height or self.config.default_height,
            "cfg_scale": cfg_scale or self.config.default_cfg_scale,
            "seed": seed,
            "sampler_name": sampler,
            "batch_size": batch_size,
            "save_images": False,
        }
        
        result = self._make_request("/sdapi/v1/txt2img", payload)
        
        images = []
        for img_data in result["images"]:
            images.append(base64.b64decode(img_data))
        
        logger.info(f"{len(images)} görüntü üretildi")
        return images
    
    def save_images(self, images: List[bytes], prefix: str = "output") -> List[Path]:
        """Görüntüleri diske kaydeder"""
        saved_paths = []
        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
        
        for i, img_data in enumerate(images):
            filename = f"{prefix}_{timestamp}_{i:03d}.png"
            filepath = Path(self.config.output_dir) / filename
            
            with open(filepath, "wb") as f:
                f.write(img_data)
            
            saved_paths.append(filepath)
            logger.info(f"Kaydedildi: {filepath}")
        
        return saved_paths
    
    def get_models(self) -> List[Dict]:
        """Mevcut modellerin listesini getirir"""
        response = self.session.get(f"{self.config.api_url}/sdapi/v1/sd-models")
        response.raise_for_status()
        return response.json()
    
    def get_samplers(self) -> List[Dict]:
        """Mevcut sampler listesini getirir"""
        response = self.session.get(f"{self.config.api_url}/sdapi/v1/samplers")
        response.raise_for_status()
        return response.json()
    
    def check_health(self) -> bool:
        """API sağlık kontrolü"""
        try:
            response = self.session.get(f"{self.config.api_url}/sdapi/v1/memory", timeout=10)
            return response.status_code == 200
        except Exception:
            return False

Toplu Görüntü Üretimi

Sysadmin’lerin en sevdiği senaryo: bir liste alıp hepsini otomatik olarak işlemek. Diyelim ki bir e-ticaret sitesi için ürün görselleri üretmeniz gerekiyor.

import csv
import concurrent.futures
from pathlib import Path

def batch_generate_from_csv(csv_file: str, client: StableDiffusionClient, max_workers: int = 2):
    """CSV dosyasından toplu görüntü üretir"""
    
    results = {"success": 0, "failed": 0, "errors": []}
    
    with open(csv_file, "r", encoding="utf-8") as f:
        reader = csv.DictReader(f)
        tasks = list(reader)
    
    logger.info(f"Toplam {len(tasks)} görüntü üretilecek")
    
    def process_task(task):
        try:
            product_name = task.get("product_name", "item")
            prompt = task.get("prompt", "")
            negative_prompt = task.get("negative_prompt", "low quality, blurry")
            
            if not prompt:
                raise ValueError(f"Prompt eksik: {product_name}")
            
            images = client.txt2img(
                prompt=prompt,
                negative_prompt=negative_prompt,
                width=int(task.get("width", 512)),
                height=int(task.get("height", 512))
            )
            
            saved = client.save_images(images, prefix=product_name.replace(" ", "_"))
            return {"status": "success", "product": product_name, "files": saved}
            
        except Exception as e:
            logger.error(f"Hata ({task.get('product_name', 'unknown')}): {e}")
            return {"status": "failed", "product": task.get("product_name"), "error": str(e)}
    
    # max_workers=2 ile paralel ama GPU'yu patlatmadan çalıştır
    with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor:
        futures = {executor.submit(process_task, task): task for task in tasks}
        
        for future in concurrent.futures.as_completed(futures):
            result = future.result()
            if result["status"] == "success":
                results["success"] += 1
            else:
                results["failed"] += 1
                results["errors"].append(result)
    
    logger.info(f"Tamamlandi - Basarili: {results['success']}, Basarisiz: {results['failed']}")
    return results

# Kullanım
config = SDConfig()
client = StableDiffusionClient(config)

if client.check_health():
    batch_results = batch_generate_from_csv("products.csv", client)
else:
    logger.error("API erişilemiyor!")

CSV dosyanız şu formatta olabilir:

# products.csv örneği
# product_name,prompt,negative_prompt,width,height
echo "product_name,prompt,negative_prompt,width,height
leather_bag,professional product photo of a brown leather bag white background,blurry low quality,512,512
wooden_table,minimalist wooden dining table studio lighting,distorted cartoonish,768,512" > products.csv

Image-to-Image İşlemi

Bazen sıfırdan üretmek yerine mevcut bir görüntüyü dönüştürmek istersiniz. Bu özellikle tasarım revizyonlarında çok işe yarıyor.

from PIL import Image
import io

def img2img(client: StableDiffusionClient, 
            source_image_path: str,
            prompt: str,
            strength: float = 0.75,
            negative_prompt: str = "") -> List[bytes]:
    """Mevcut görüntüyü dönüştürür"""
    
    # Görüntüyü base64'e dönüştür
    with Image.open(source_image_path) as img:
        # Boyutu 512'nin katlarına yuvarla
        width = (img.width // 64) * 64
        height = (img.height // 64) * 64
        img = img.resize((width, height), Image.LANCZOS)
        
        buffer = io.BytesIO()
        img.save(buffer, format="PNG")
        img_base64 = base64.b64encode(buffer.getvalue()).decode("utf-8")
    
    payload = {
        "init_images": [img_base64],
        "prompt": prompt,
        "negative_prompt": negative_prompt,
        "denoising_strength": strength,  # 0.0 = değişiklik yok, 1.0 = tamamen yeni
        "steps": 30,
        "cfg_scale": 7,
        "sampler_name": "DPM++ 2M Karras",
    }
    
    result = client._make_request("/sdapi/v1/img2img", payload)
    
    images = [base64.b64decode(img) for img in result["images"]]
    return images

# Kullanım
transformed = img2img(
    client=client,
    source_image_path="original_photo.jpg",
    prompt="oil painting style, artistic, vibrant colors",
    strength=0.6,
    negative_prompt="photo, realistic"
)

client.save_images(transformed, prefix="transformed")

strength parametresini iyi ayarlayın. 0.3-0.5 arasında orijinale benzerlik yüksek tutulur, 0.7-0.9 arasında çok daha dramatik dönüşümler olur.

Model Değiştirme ve LoRA Kullanımı

Farklı görevler için farklı modeller kullanmak gerekebilir. API üzerinden bunu programatik yapabilirsiniz:

def switch_model(client: StableDiffusionClient, model_name: str):
    """Çalışma zamanında model değiştirir"""
    
    # Mevcut modelleri listele
    models = client.get_models()
    model_titles = [m["title"] for m in models]
    
    if model_name not in model_titles:
        logger.error(f"Model bulunamadı: {model_name}")
        logger.info(f"Mevcut modeller: {model_titles}")
        return False
    
    payload = {"sd_model_checkpoint": model_name}
    
    try:
        response = client.session.post(
            f"{client.config.api_url}/sdapi/v1/options",
            json=payload,
            timeout=60
        )
        response.raise_for_status()
        logger.info(f"Model değiştirildi: {model_name}")
        return True
    except Exception as e:
        logger.error(f"Model değiştirme hatası: {e}")
        return False

def generate_with_lora(client: StableDiffusionClient, 
                       base_prompt: str,
                       lora_name: str,
                       lora_weight: float = 0.8) -> List[bytes]:
    """LoRA ağırlığı ekleyerek görüntü üretir"""
    
    # LoRA syntax: <lora:lora_name:weight>
    lora_prompt = f"{base_prompt} <lora:{lora_name}:{lora_weight}>"
    
    return client.txt2img(
        prompt=lora_prompt,
        negative_prompt="low quality, blurry, distorted",
        steps=30
    )

# Kullanım
switch_model(client, "realisticVisionV60B1_v51VAE.safetensors")

portrait_images = generate_with_lora(
    client=client,
    base_prompt="professional headshot, business attire, studio lighting",
    lora_name="add_detail",
    lora_weight=0.7
)

İlerleme Takibi ve Asenkron Kullanım

Uzun üretim işlemlerinde ilerlemeyi takip etmek kullanıcı deneyimi açısından önemli. AUTOMATIC1111 bu için bir progress endpoint sunuyor:

import asyncio
import aiohttp
import threading

def monitor_progress(api_url: str, stop_event: threading.Event):
    """Arka planda ilerlemeyi takip eder"""
    
    while not stop_event.is_set():
        try:
            response = requests.get(f"{api_url}/sdapi/v1/progress", timeout=5)
            if response.status_code == 200:
                data = response.json()
                progress = data.get("progress", 0)
                state = data.get("state", {})
                
                if progress > 0:
                    bar_length = 40
                    filled = int(bar_length * progress)
                    bar = "=" * filled + "-" * (bar_length - filled)
                    job = state.get("job", "")
                    print(f"r[{bar}] {progress*100:.1f}% - {job}", end="", flush=True)
        except Exception:
            pass
        
        time.sleep(1)
    
    print()  # Son satır sonu

def generate_with_progress(client: StableDiffusionClient, prompt: str) -> List[bytes]:
    """İlerleme göstergesiyle görüntü üretir"""
    
    stop_event = threading.Event()
    monitor_thread = threading.Thread(
        target=monitor_progress,
        args=(client.config.api_url, stop_event),
        daemon=True
    )
    monitor_thread.start()
    
    try:
        images = client.txt2img(prompt=prompt, steps=30)
        return images
    finally:
        stop_event.set()
        monitor_thread.join(timeout=2)

# Kullanım
images = generate_with_progress(
    client,
    "epic fantasy castle at sunset, detailed, cinematic lighting"
)
client.save_images(images, prefix="fantasy_castle")

Üretim Ortamı İpuçları

Gerçek bir üretim ortamında dikkat etmeniz gereken birkaç önemli nokta var:

GPU bellek yönetimi: Yüksek çözünürlüklü görüntüler GPU belleğini doldurabilir. Hata aldığınızda çözünürlüğü küçültün ya da --medvram bayrağını kullanın.

Kuyruk sistemi: Eş zamanlı çok istek gelirse sistem çöker. Bir Redis tabanlı kuyruk sistemi (Celery + Redis) kurmanızı öneririm.

Timeout değerleri: Uzun işlemler için timeout değerini artırın. 512×512 için 60 saniye, 1024×1024 için 180 saniye makul.

Güvenlik: API’yi dışarıya açıyorsanız:

# Nginx reverse proxy ile temel auth ekleyin
apt install apache2-utils
htpasswd -c /etc/nginx/.htpasswd sduser

Çıktı yönetimi: Disk dolmasını önlemek için eski dosyaları temizleyin:

# Cron ile 7 günden eski dosyaları sil
0 3 * * * find /var/www/generated-images -name "*.png" -mtime +7 -delete

Model cache: Modeller büyük dosyalar. SSD üzerinde tutun ve NFS/NAS kullanıyorsanız gecikmeyi göz önünde bulundurun.

Log rotasyonu: Python loglama ile çok fazla çıktı üretebilirsiniz. logging.handlers.RotatingFileHandler kullanarak log dosyalarını kontrol altında tutun.

Batch boyutu vs. paralel istek: Aynı anda birden fazla görüntü üretmek için batch_size parametresini artırmak, ayrı istekler göndermekten genellikle daha verimli.

Seed yönetimi: Tekrar üretilebilirlik için seed değerini kaydedin. -1 rastgele seed kullanır ama API response’unda kullanılan seed değeri her zaman dönüyor.

Sonuç

Python ile Stable Diffusion API entegrasyonu, başta karmaşık görünse de temel yapıyı kavradıktan sonra son derece esnek bir araç haline geliyor. Anlattığımız yapı ile toplu görüntü üretiminden model yönetimine, ilerleme takibinden hata toleranslı sistemlere kadar geniş bir yelpazede çalışabilirsiniz.

Sysadmin perspektifinden baktığımızda, bu işin en kritik kısmı güvenilir bir altyapı kurmak. Yeniden deneme mekanizmaları, sağlık kontrolleri ve doğru timeout değerleri olmadan üretim ortamında ciddi sorunlarla karşılaşabilirsiniz. Yazdığımız StableDiffusionClient sınıfı bu ihtiyaçların büyük bölümünü karşılıyor, ancak kendi projenizin gereksinimlerine göre genişletmeniz gerekecek.

Bir sonraki adım olarak ControlNet entegrasyonunu, ControlNet ile pose kontrolünü ya da Stable Diffusion XL modellerinin API üzerinden kullanımını inceleyebilirsiniz. Bunlar biraz daha ileri seviye ama temel yapı aynı kalıyor.