""" title: Generate Image with Fal.AI author: Tong Vuu author_url: https://github.com/silverwolfceh requirements: requests,pydantic version: 0.1 """ from pydantic import BaseModel, Field import requests import time import base64 from requests.exceptions import Timeout, RequestException from urllib.parse import quote class Tools: class Valves(BaseModel): FAL_URL: str = Field( default="https://queue.fal.run/fal-ai/", description="FAL API URL for text to image generation", ) FAL_KEY: str = Field( default="", description="API key for Fal API", ) FAL_MODEL: str = Field( default="imagen3", description="Fal.AI model for image gen" ) PROXY_ENABLED: bool = Field( default=False, description="Whether to use a proxy for API requests", ) PROXIES_ADDR: str = Field( default="http://127.0.0.1:3128", description="Proxy settings for API requests", ) def __init__(self): self.valves = self.Valves() async def generate_images( self, prompt_in_english: str, seed: float, randomize_seed: bool, width: float, height: float, num_inference_steps: float, __event_emitter__=None, ): """ Generate an image by providing arguments. :param prompt_in_english: prompt to use for image generation in English. :param seed: seed for random number generation. 0 if randomize_seed is True. :param randomize_seed: whether to randomize the seed. Default is True. :param width: width of the generated image. Default is 1280. :param height: height of the generated image. Default is 720. :param num_inference_steps: number of inference steps for the image generation. Default is 4. """ try: # print( # f"Received input: {prompt_in_english}, {seed}, {randomize_seed}, {width}, {height}, {num_inference_steps}" # ) # Notify start of generation await __event_emitter__( { "type": "status", "data": {"description": "Start generating image", "done": False}, } ) # Build the URL with parameters encoded_prompt = prompt_in_english url = f"{self.valves.FAL_URL}{self.valves.FAL_MODEL}" headers = { "Content-Type": "application/json", "Authorization": f"Key {self.valves.FAL_KEY}", } data = {"prompt": encoded_prompt, "seed": 0 if seed is None else int(seed)} proxies = None if self.valves.PROXY_ENABLED: proxies = { "http": self.valves.PROXIES_ADDR, "https": self.valves.PROXIES_ADDR, } sess = requests.session() sess.proxies = proxies sess.headers = headers sess.verify = False response = sess.post(url, json=data, timeout=30) if response.status_code != 200: print(f"API request failed 0x1 {response.status_code}") await __event_emitter__( { "type": "message", "data": {"content": f"Failed to generate image [error: 0x1]"}, } ) return "Failed" else: requestid = response.json().get("request_id") status = response.json().get("status") cnt = 1 while status.lower() != "completed": await __event_emitter__( { "type": "status", "data": { "description": f"Still wating for queue {status}", "done": False, }, } ) time.sleep(1) url = f"{self.valves.FAL_URL}{self.valves.FAL_MODEL}/requests/{requestid}/status" response = sess.get(url) if response.status_code != 200 and response.status_code != 202: print( f"API request failed 0x2 {response.status_code} [Tried: {cnt}]" ) await __event_emitter__( { "type": "message", "data": { "content": f"Failed to generate image [error: 0x2] Tried: {cnt}" }, } ) return "Failed" else: status = response.json().get("status") cnt = cnt + 1 await __event_emitter__( { "type": "status", "data": {"description": f"Getting the result", "done": False}, } ) url = ( f"{self.valves.FAL_URL}{self.valves.FAL_MODEL}/requests/{requestid}" ) response = sess.get(url) if response.status_code != 200: print(f"API request failed 0x3 {response.status_code}") await __event_emitter__( { "type": "message", "data": { "content": f"Failed to generate image [error: 0x3]" }, } ) return "Failed" else: imageurl = response.json()["images"][0]["url"] seed = response.json()["seed"] response = sess.get(imageurl) if response.status_code != 200: print(f"API request failed 0x4 {response.status_code}") await __event_emitter__( { "type": "message", "data": { "content": f"Failed to generate image [error: 0x4]" }, } ) return "Failed" else: # Convert image to base64 image_content = response.content image_url = f"data:image/jpeg;base64,{base64.b64encode(image_content).decode('utf-8')}" # Send completion status await __event_emitter__( { "type": "status", "data": { "description": "Image generated", "done": True, }, } ) # Send the image await __event_emitter__( { "type": "message", "data": { "content": f"Generated image for prompt: '{prompt_in_english}'\n\n [Seed: {seed}]" }, } ) return f"Image successfully generated for prompt: '{prompt_in_english}'" except Timeout: error_msg = ( "Request timed out while generating the image. Please try again later." ) print(error_msg) await __event_emitter__( { "type": "status", "data": {"description": error_msg, "done": True}, } ) return error_msg except RequestException as e: error_msg = f"Network error occurred: {str(e)}" print(error_msg) await __event_emitter__( { "type": "status", "data": {"description": error_msg, "done": True}, } ) return error_msg except Exception as e: error_msg = f"An error occurred: {str(e)}" print(f"Unexpected error: {str(e)}") await __event_emitter__( { "type": "status", "data": {"description": error_msg, "done": True}, } ) return error_msg