analysis_runner.py

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
    )