Spaces:
Sleeping
Sleeping
File size: 8,053 Bytes
e66f533 |
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 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 |
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
) |