Spaces:
Sleeping
Sleeping
File size: 4,006 Bytes
17129b4 c4972c9 17129b4 d080bf0 17129b4 c4972c9 17129b4 c4972c9 17129b4 c4972c9 17129b4 c4972c9 17129b4 d080bf0 17129b4 c4972c9 17129b4 c4972c9 17129b4 c4972c9 17129b4 c4972c9 17129b4 d080bf0 17129b4 c4972c9 |
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 101 102 103 104 |
import logging
import asyncio
from concurrent.futures import ThreadPoolExecutor
from helpercodes.image_functions import generate_image
from helpercodes.user_data import generate_gpt_prompt, _encode_image_to_base64
# Configure Logging
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s - %(levelname)s - %(message)s"
)
def _generate_prompt_and_image(pos: str, neg: str, category: str, story_instruction: str,
sentiment, image_elements, is_text, text_elements, non_compliant, emotion,
aspect_ratio, design):
try:
logging.info(f"Generating GPT prompt for category: {category}, instruction: {story_instruction}")
# Generate the GPT prompt
prompt = generate_gpt_prompt(
pos,
neg,
category,
story_instruction,
sentiment, image_elements, is_text, text_elements, non_compliant, emotion
)
logging.info(f"Generated GPT prompt: {prompt}")
# Generate the image from the prompt
image = generate_image(prompt, aspect_ratio, design)
if image:
logging.info("Successfully generated image.")
else:
logging.warning("Image generation failed.")
return image
except Exception as e:
logging.error(f"Error in _generate_prompt_and_image: {e}", exc_info=True)
return None
async def process_custom_data(pos, category, neg, sentiment, image_elements, is_text, text_elements,
non_compliant, emotion, aspect_ratio, design):
try:
logging.info("Processing custom data...")
# The two different instructions
instruction_var = (
"which should have new offers, revised text and graphics but style should remain same"
)
instruction_same = (
"which should have new offers and revised text but graphics and style should remain same"
)
# We'll use asyncio's event loop to run blocking code in a ThreadPoolExecutor
loop = asyncio.get_running_loop()
# Create the thread pool
with ThreadPoolExecutor() as executor:
# Schedule both tasks in parallel
future_var = loop.run_in_executor(
executor,
_generate_prompt_and_image,
pos,
neg,
category,
instruction_var,
sentiment, image_elements, is_text, text_elements,
non_compliant, emotion, aspect_ratio, design
)
future_same = loop.run_in_executor(
executor,
_generate_prompt_and_image,
pos,
neg,
category,
instruction_same,
sentiment, image_elements, is_text, text_elements,
non_compliant, emotion, aspect_ratio, design
)
# Wait for both images to finish
image_var, image_same = await asyncio.gather(future_var, future_same)
return image_var, image_same
except Exception as e:
logging.error(f"Unexpected error in process_custom_data: {e}", exc_info=True)
return None, None
def process_user_data_sync(pos, category, neg, sentiment, image_elements, is_text, text_elements,
non_compliant, emotion, aspect_ratio, design):
try:
logging.info("Executing process_user_data_sync...")
return asyncio.run(process_custom_data(pos, category, neg, sentiment, image_elements, is_text, text_elements,
non_compliant, emotion, aspect_ratio, design))
except RuntimeError as e:
logging.error("RuntimeError in process_user_data_sync: Async event loop issue", exc_info=True)
return None, None
except Exception as e:
logging.error(f"Unexpected error in process_user_data_sync: {e}", exc_info=True)
return None, None |