Spaces:
Sleeping
Sleeping
| import logging | |
| import random | |
| import asyncio | |
| from concurrent.futures import ThreadPoolExecutor | |
| from helpercodes.image_functions import generate_image | |
| from helpercodes.ourdata_advanced import get_links, generate_gpt_prompt_advanced_feature, get_dataset_dict | |
| def _generate_single_image_data( | |
| df_subset, | |
| input_keywords, | |
| sentiment, | |
| image_elements=None, | |
| text_elements=None, | |
| design_style=None, aspect_ratio='1:1', input_image=None, is_text=True, non_compliant=None, emotion=None | |
| ): | |
| try: | |
| # 1) Fetch results | |
| results = get_links(df_subset, input_keywords) | |
| if len(results) < 2: | |
| logging.warning(f"Not enough results for keywords: {input_keywords}") | |
| return None | |
| # 2) Generate prompt | |
| prompt = generate_gpt_prompt_advanced_feature(results, sentiment, image_elements, text_elements, input_image, is_text, non_compliant, emotion) | |
| logging.info(f"Generated prompt: {prompt}") | |
| # 3) Generate and encode image | |
| image = generate_image(prompt, aspect_ratio, design_style) | |
| logging.info(f"Image successfully generated for keywords: {input_keywords}") | |
| return image | |
| except Exception as e: | |
| logging.error(f"Error in _generate_single_image_data: {e}", exc_info=True) | |
| return None | |
| async def process_advanced_features(style, sentiment, image_elements=None, text_elements=None, design_style=None, aspect_ratio='1:1', input_image=None, is_text=True, non_compliant=None, emotion=None): | |
| try: | |
| data, style_dict = get_dataset_dict() | |
| # Check if the style exists in the dictionary | |
| if style not in style_dict or not style_dict[style]: | |
| logging.warning(f"No keywords found for style: {style}") | |
| return {"data": []} | |
| # Get the subset of data for the style | |
| df_subset = data[data['styles'] == style].copy() | |
| if df_subset.empty: | |
| logging.warning(f"No data found for style: {style}") | |
| return {"data": []} | |
| # Sample keywords and generate images | |
| sampled_keywords = random.sample(style_dict[style], min(len(style_dict[style]), 5)) | |
| input_keywords = ", ".join(sampled_keywords) | |
| logging.info(f"Selected keywords for style '{style}': {input_keywords}") | |
| # We'll generate two images in parallel: | |
| loop = asyncio.get_running_loop() | |
| # Thread pool for blocking tasks | |
| with ThreadPoolExecutor() as executor: | |
| tasks = [] | |
| for _ in range(2): | |
| tasks.append( | |
| loop.run_in_executor( | |
| executor, | |
| _generate_single_image_data, | |
| df_subset, | |
| input_keywords, | |
| sentiment, | |
| image_elements, | |
| text_elements, | |
| design_style, aspect_ratio, input_image, is_text, | |
| non_compliant, emotion | |
| ) | |
| ) | |
| # Wait for both tasks in parallel | |
| results = await asyncio.gather(*tasks) | |
| # Filter out any None (e.g., if we didn't have enough results) | |
| response = [r for r in results if r is not None] | |
| if not response: | |
| logging.warning("No images generated successfully.") | |
| return response | |
| except Exception as e: | |
| logging.error(f"Error in process_advanced_features: {e}", exc_info=True) | |
| return {"data": []} | |
| def process_advanced_data_sync(style, sentiment, image_elements=None, text_elements=None, design_style=None, aspect_ratio='1:1', input_image=None, is_text=True, non_compliant=None, emotion=None): | |
| try: | |
| logging.info(f"Processing advanced data sync for style: {style}") | |
| return asyncio.run(process_advanced_features(style, sentiment, image_elements, text_elements, design_style, aspect_ratio, input_image, is_text, non_compliant, emotion)) | |
| except RuntimeError as e: | |
| logging.error("RuntimeError in process_advanced_data_sync: Async event loop issue", exc_info=True) | |
| return {"data": []} | |
| except Exception as e: | |
| logging.error(f"Unexpected error in process_advanced_data_sync: {e}", exc_info=True) | |
| return {"data": []} |