File size: 3,939 Bytes
d091da8 39c89a5 b4fbe2e d091da8 dd2e4cf 39c89a5 d091da8 39c89a5 d091da8 78b81a5 b4fbe2e d091da8 39c89a5 b4fbe2e 39c89a5 d091da8 eca7f7a d091da8 39c89a5 d091da8 39c89a5 b4fbe2e 39c89a5 b4fbe2e d091da8 39c89a5 b4fbe2e d091da8 b4fbe2e 39c89a5 d9ce58c b4fbe2e d9ce58c b4fbe2e d9ce58c b4fbe2e d9ce58c b4fbe2e d9ce58c b4fbe2e |
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 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 |
import os
import aiohttp
from typing import Optional
class FluxClient:
def __init__(self, api_key: str):
self.api_key = api_key
self.endpoint = os.getenv("FLUX_ENDPOINT")
self._session = None
async def _get_session(self):
if self._session is None:
self._session = aiohttp.ClientSession()
return self._session
async def generate_image(self,
prompt: str,
width: int,
height: int,
num_inference_steps: int = 5,
guidance_scale: float = 9.0) -> Optional[bytes]:
"""Génère une image à partir d'un prompt."""
try:
# Ensure dimensions are multiples of 8
width = (width // 8) * 8
height = (height // 8) * 8
print(f"Sending request to Hugging Face API: {self.endpoint}")
print(f"Headers: Authorization: Bearer {self.api_key[:4]}...")
print(f"Request body: {prompt[:100]}...")
session = await self._get_session()
async with session.post(
self.endpoint,
headers={
"Authorization": f"Bearer {self.api_key}",
"Accept": "image/jpeg"
},
json={
"inputs": prompt,
"parameters": {
"num_inference_steps": num_inference_steps,
"guidance_scale": guidance_scale,
"width": width,
"height": height,
"negative_prompt": "Bubbles, text, caption. Do not include bright or clean clothing."
}
}
) as response:
print(f"Response status code: {response.status}")
print(f"Response headers: {response.headers}")
print(f"Response content type: {response.headers.get('content-type', 'unknown')}")
if response.status == 200:
content = await response.read()
content_length = len(content)
print(f"Received successful response with content length: {content_length}")
if isinstance(content, bytes):
print("Response content is bytes (correct)")
else:
print(f"Warning: Response content is {type(content)}")
return content
else:
error_content = await response.text()
print(f"Error from Flux API: {response.status}")
print(f"Response content: {error_content}")
return None
except Exception as e:
print(f"Error in FluxClient.generate_image: {str(e)}")
import traceback
print(f"Traceback: {traceback.format_exc()}")
return None
async def close(self):
if self._session:
await self._session.close()
self._session = None
async def check_health(self) -> bool:
"""
Vérifie la disponibilité du service Flux en tentant de générer une petite image.
Returns:
bool: True si le service est disponible, False sinon
"""
try:
# Test simple prompt pour générer une petite image
test_image = await self.generate_image(
prompt="test image, simple circle",
width=64, # Petite image pour le test
height=64,
num_inference_steps=1 # Minimum d'étapes pour être rapide
)
return test_image is not None
except Exception as e:
print(f"Health check failed: {str(e)}")
raise |