File size: 1,958 Bytes
348a0a6
 
127d226
348a0a6
 
 
8d513ea
5fa8a0c
503daa9
 
5fa8a0c
d8aedbb
348a0a6
 
 
 
 
 
 
 
 
503daa9
348a0a6
 
 
 
 
 
 
 
 
 
 
127d226
348a0a6
 
 
 
 
 
 
503daa9
348a0a6
 
 
 
 
 
 
503daa9
348a0a6
 
 
 
 
5fa8a0c
 
 
348a0a6
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
from fastapi import FastAPI, HTTPException
from fastapi.responses import StreamingResponse
import requests
from urllib.parse import quote
import asyncio
from io import BytesIO
import os

url = os.environ.get("url")

app = FastAPI()

def create_job(prompt, model, sampler, seed, neg):
    if model is None:
        model = 'Realistic_Vision_V5.0.safetensors [614d1063]'
    if sampler is None:
        sampler = 'DPM++ 2M Karras'
    if seed is None:
        seed = '-1'
    if neg is None:
        neg = "(long list of negative prompts removed for brevity)"
    url = f'https://api.{url}.com/generate'
    params = {
        'new': 'true',
        'prompt': quote(prompt),
        'model': model,
        'negative_prompt': neg,
        'steps': '100',
        'cfg': '9.5',
        'seed': seed,
        'sampler': sampler,
        'upscale': 'True',
        'aspect_ratio': 'square'
    }
    response = requests.get(url, params=params)
    response.raise_for_status()
    return response.json()['job']

@app.get("/generate_image")
async def generate_image(prompt: str, model: str = None, sampler: str = None, seed: str = None, neg: str = None):
    job_id = create_job(prompt, model, sampler, seed, neg)
    url = f'https://api.{url}.com/job/{job_id}'
    headers = {'accept': '*/*'}
    
    while True:
        response = requests.get(url=url, headers=headers)
        response.raise_for_status()
        job_response = response.json()
        if job_response['status'] == 'succeeded':
            image_url = f'https://images.{url}.xyz/{job_id}.png'
            image_response = requests.get(image_url)
            image_response.raise_for_status()
            image = BytesIO(image_response.content)
            return StreamingResponse(image, media_type='image/png')
        await asyncio.sleep(2)  # Add a delay to prevent excessive requests

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000, debug=True)