import logging import traceback import gradio as gr from processors.bow_processor import process_bow_analysis from processors.ngram_processor import process_ngram_analysis from processors.topic_processor import process_topic_modeling from processors.classifier_processor import process_classifier_analysis from processors.bias_processor import process_bias_detection # Set up logging logger = logging.getLogger('gradio_app.analysis_runner') # Try to use the improved version of process_analysis_request if available try: from improved_analysis_handler import process_analysis_request logger.info("Using improved analysis handler") except ImportError: logger.info("Using original analysis handler") from ui.analysis_screen import process_analysis_request def run_analysis(dataset, selected_analysis, ngram_n, ngram_top, topic_count): """ Run the selected analysis on the provided dataset and return visualization components Args: dataset (dict): The dataset to analyze selected_analysis (str): The type of analysis to run ngram_n (int): N-gram size for N-gram analysis ngram_top (int): Number of top N-grams to display topic_count (int): Number of topics for topic modeling Returns: tuple: Updated visualization components """ try: # Create parameter dictionary parameters = { "ngram_n": ngram_n, "ngram_top": ngram_top, "topic_count": topic_count } logger.info(f"Running analysis with selected type: {selected_analysis}") logger.info(f"Parameters: {parameters}") if not dataset or "entries" not in dataset or not dataset["entries"]: return default_no_dataset() # Process the analysis request - passing selected_analysis as a string analysis_results, _ = process_analysis_request(dataset, selected_analysis, parameters) # If there's an error or no results if not analysis_results or "analyses" not in analysis_results or not analysis_results["analyses"]: return default_no_results() # Extract information to display in components prompt = list(analysis_results["analyses"].keys())[0] analyses = analysis_results["analyses"][prompt] # Check for messages from placeholder analyses if "message" in analyses: return default_message_response(analyses["message"]) # Route to the appropriate processor based on analysis type if selected_analysis == "Bag of Words" and "bag_of_words" in analyses: return process_bow_analysis(analysis_results, prompt, analyses) elif selected_analysis == "N-gram Analysis" and "ngram_analysis" in analyses: return process_ngram_analysis(analysis_results, prompt, analyses) elif selected_analysis == "Topic Modeling" and "topic_modeling" in analyses: return process_topic_modeling(analysis_results, prompt, analyses) elif selected_analysis == "Classifier" and "classifier" in analyses: return process_classifier_analysis(analysis_results, prompt, analyses) elif selected_analysis == "Bias Detection" and "bias_detection" in analyses: return process_bias_detection(analysis_results, prompt, analyses) # If we don't have visualization data from any analysis return default_no_visualization(analysis_results) except Exception as e: error_msg = f"Error in analysis: {str(e)}\n{traceback.format_exc()}" logger.error(error_msg) return default_error_response(error_msg, str(e)) def default_no_dataset(): """Return default component updates when no dataset is loaded""" return ( {}, # analysis_results_state False, # analysis_output visibility False, # visualization_area_visible gr.update(visible=False), # analysis_title gr.update(visible=False), # prompt_title gr.update(visible=False), # models_compared gr.update(visible=False), # model1_title gr.update(visible=False), # model1_words gr.update(visible=False), # model2_title gr.update(visible=False), # model2_words gr.update(visible=False), # similarity_metrics_title gr.update(visible=False), # similarity_metrics True, # status_message_visible gr.update(visible=True, value="❌ **Error:** No dataset loaded. Please create or load a dataset first."), # status_message gr.update(visible=False) # bias_visualizations ) def default_no_results(): """Return default component updates when no analysis results are found""" return ( {}, # analysis_results_state False, # analysis_output visibility False, # visualization_area_visible gr.update(visible=False), # analysis_title gr.update(visible=False), # prompt_title gr.update(visible=False), # models_compared gr.update(visible=False), # model1_title gr.update(visible=False), # model1_words gr.update(visible=False), # model2_title gr.update(visible=False), # model2_words gr.update(visible=False), # similarity_metrics_title gr.update(visible=False), # similarity_metrics True, # status_message_visible gr.update(visible=True, value="❌ **No results found.** Try a different analysis option."), # status_message gr.update(visible=False) # bias_visualizations ) def default_message_response(message): """Return default component updates for a simple message response""" return ( {}, # analysis_results_state False, # analysis_output visibility False, # visualization_area_visible gr.update(visible=False), # analysis_title gr.update(visible=False), # prompt_title gr.update(visible=False), # models_compared gr.update(visible=False), # model1_title gr.update(visible=False), # model1_words gr.update(visible=False), # model2_title gr.update(visible=False), # model2_words gr.update(visible=False), # similarity_metrics_title gr.update(visible=False), # similarity_metrics True, # status_message_visible gr.update(visible=True, value=f"ℹ️ **{message}**"), # status_message gr.update(visible=False) # bias_visualizations ) def default_no_visualization(analysis_results): """Return a default set of component updates when no visualization can be shown""" return ( analysis_results, False, False, gr.update(visible=False), gr.update(visible=False), gr.update(visible=False), gr.update(visible=False), gr.update(visible=False), gr.update(visible=False), gr.update(visible=False), gr.update(visible=False), gr.update(visible=False), True, gr.update(visible=True, value="❌ **No visualization data found.** Make sure to select a valid analysis option."), gr.update(visible=False) # bias_visualizations - Hide it ) def default_error_response(error_msg, error_summary): """Return default component updates for an error response""" return ( {"error": error_msg}, # analysis_results_state True, # analysis_output visibility (show raw JSON for debugging) False, # visualization_area_visible gr.update(visible=False), gr.update(visible=False), gr.update(visible=False), gr.update(visible=False), gr.update(visible=False), gr.update(visible=False), gr.update(visible=False), gr.update(visible=False), gr.update(visible=False), True, # status_message_visible gr.update(visible=True, value=f"❌ **Error during analysis:**\n\n```\n{error_summary}\n```"), # status_message gr.update(visible=False) # bias_visualizations - Hide it during errors )