app.py

import numpy as np
import pandas as pd
import statsmodels.formula.api as smf
import statsmodels.api as sm
import plotly.graph_objects as go
from plotly.subplots import make_subplots
from scipy.optimize import minimize
import plotly.express as px
from scipy.stats import t, f
import gradio as gr
import io
import os
from zipfile import ZipFile
import warnings

# Suppress specific warnings
warnings.filterwarnings('ignore', category=UserWarning)
warnings.filterwarnings('ignore', category=RuntimeWarning)

class RSM_BoxBehnken:
    def __init__(self, data, x1_name, x2_name, x3_name, y_name, x1_levels, x2_levels, x3_levels):
        """
        Initialize the Response Surface Methodology Box-Behnken Design class
        
        Parameters:
        -----------
        data : pandas.DataFrame
            Experimental design data
        x1_name, x2_name, x3_name : str
            Names of independent variables
        y_name : str
            Name of dependent variable
        x1_levels, x2_levels, x3_levels : list
            Levels of each independent variable
        """
        self.data = data.copy()
        self.model = None
        self.model_simplified = None
        self.optimized_results = None
        self.optimal_levels = None

        # Variable names
        self.x1_name = x1_name
        self.x2_name = x2_name
        self.x3_name = x3_name
        self.y_name = y_name

        # Original levels of variables
        self.x1_levels = x1_levels
        self.x2_levels = x2_levels
        self.x3_levels = x3_levels

    def _get_levels(self, variable_name):
        """
        Get levels for a specific variable
        
        Parameters:
        -----------
        variable_name : str
            Name of the variable
        
        Returns:
        --------
        list
            Levels of the variable
        """
        level_map = {
            self.x1_name: self.x1_levels,
            self.x2_name: self.x2_levels,
            self.x3_name: self.x3_levels
        }
        
        if variable_name not in level_map:
            raise ValueError(f"Unknown variable: {variable_name}")
        
        return level_map[variable_name]

    def fit_model(self, simplified=False):
        """
        Fit the response surface model
        
        Parameters:
        -----------
        simplified : bool, optional
            Whether to fit a simplified model, by default False
        
        Returns:
        --------
        tuple
            Fitted model and Pareto chart
        """
        if simplified:
            formula = f'{self.y_name} ~ {self.x1_name} + {self.x2_name} + {self.x3_name} + ' \
                      f'I({self.x1_name}**2) + I({self.x2_name}**2) + I({self.x3_name}**2)'
            self.model_simplified = smf.ols(formula, data=self.data).fit()
            print("\nSimplified Model:")
            print(self.model_simplified.summary())
            return self.model_simplified, self.pareto_chart(self.model_simplified, "Pareto - Simplified Model")
        else:
            formula = f'{self.y_name} ~ {self.x1_name} + {self.x2_name} + {self.x3_name} + ' \
                      f'I({self.x1_name}**2) + I({self.x2_name}**2) + I({self.x3_name}**2) + ' \
                      f'{self.x1_name}:{self.x2_name} + {self.x1_name}:{self.x3_name} + {self.x2_name}:{self.x3_name}'
            self.model = smf.ols(formula, data=self.data).fit()
            print("Full Model:")
            print(self.model.summary())
            return self.model, self.pareto_chart(self.model, "Pareto - Full Model")

    def optimize(self, method='Nelder-Mead'):
        """
        Optimize the response surface model
        
        Parameters:
        -----------
        method : str, optional
            Optimization method, by default 'Nelder-Mead'
        
        Returns:
        --------
        pandas.DataFrame
            Optimization results table
        """
        if self.model_simplified is None:
            raise ValueError("Fit the simplified model first.")

        def objective_function(x):
            """Objective function for optimization"""
            return -self.model_simplified.predict(pd.DataFrame({
                self.x1_name: [x[0]], 
                self.x2_name: [x[1]], 
                self.x3_name: [x[2]]
            }))

        bounds = [(-1, 1), (-1, 1), (-1, 1)]
        x0 = [0, 0, 0]

        self.optimized_results = minimize(
            objective_function, 
            x0, 
            method=method, 
            bounds=bounds
        )
        self.optimal_levels = self.optimized_results.x
        
        # Convert to natural levels
        optimal_levels_natural = [
            round(self.coded_to_natural(self.optimal_levels[i], var), 3) 
            for i, var in enumerate([self.x1_name, self.x2_name, self.x3_name])
        ]
        
        optimization_table = pd.DataFrame({
            'Variable': [self.x1_name, self.x2_name, self.x3_name],
            'Optimal Level (Natural)': optimal_levels_natural,
            'Optimal Level (Coded)': [round(x, 3) for x in self.optimal_levels]
        })

        return optimization_table

    def coded_to_natural(self, coded_value, variable_name):
        """
        Convert coded value to natural level
        
        Parameters:
        -----------
        coded_value : float
            Coded value of the variable
        variable_name : str
            Name of the variable
        
        Returns:
        --------
        float
            Natural level of the variable
        """
        levels = self._get_levels(variable_name)
        return levels[0] + (coded_value + 1) * (levels[-1] - levels[0]) / 2

    def natural_to_coded(self, natural_value, variable_name):
        """
        Convert natural level to coded value
        
        Parameters:
        -----------
        natural_value : float
            Natural level of the variable
        variable_name : str
            Name of the variable
        
        Returns:
        --------
        float
            Coded value of the variable
        """
        levels = self._get_levels(variable_name)
        return -1 + 2 * (natural_value - levels[0]) / (levels[-1] - levels[0])

    def pareto_chart(self, model, title):
        """
        Create Pareto chart of standardized effects
        
        Parameters:
        -----------
        model : statsmodels.regression.linear_model.RegressionResultsWrapper
            Fitted regression model
        title : str
            Title of the Pareto chart
        
        Returns:
        --------
        plotly.graph_objects.Figure
            Pareto chart
        """
        tvalues = model.tvalues[1:]
        abs_tvalues = np.abs(tvalues)
        sorted_idx = np.argsort(abs_tvalues)[::-1]
        sorted_tvalues = abs_tvalues[sorted_idx]
        sorted_names = tvalues.index[sorted_idx]

        alpha = 0.05
        dof = model.df_resid
        t_critical = t.ppf(1 - alpha / 2, dof)

        fig = px.bar(
            x=sorted_tvalues,
            y=sorted_names,
            orientation='h',
            labels={'x': 'Standardized Effect', 'y': 'Term'},
            title=title
        )
        fig.update_yaxes(autorange="reversed")
        fig.add_vline(x=t_critical, line_dash="dot",
                      annotation_text=f"Critical t = {t_critical:.2f}",
                      annotation_position="bottom right")

        return fig

    def generate_prediction_table(self):
        """
        Generate prediction table with predicted and residual values
        
        Returns:
        --------
        pandas.DataFrame
            Prediction table
        """
        if self.model_simplified is None:
            raise ValueError("Fit the simplified model first.")

        predictions = self.model_simplified.predict(self.data)
        residuals = self.data[self.y_name] - predictions

        prediction_table = self.data.copy()
        prediction_table['Predicted'] = predictions.round(3)
        prediction_table['Residual'] = residuals.round(3)

        return prediction_table[[self.y_name, 'Predicted', 'Residual']]

    def calculate_contribution_percentage(self):
        """
        Calculate percentage contribution of model terms
        
        Returns:
        --------
        pandas.DataFrame
            Contribution percentage table
        """
        if self.model_simplified is None:
            raise ValueError("Fit the simplified model first.")

        anova_table = sm.stats.anova_lm(self.model_simplified, typ=2)
        ss_total = anova_table['sum_sq'].sum()

        contribution_table = []
        
        for index, row in anova_table.iterrows():
            if index != 'Residual':
                factor_name = index.replace('I(', '').replace('**2)', '^2')
                ss_factor = row['sum_sq']
                contribution_percentage = (ss_factor / ss_total) * 100

                contribution_table.append({
                    'Factor': factor_name,
                    'Sum of Squares': round(ss_factor, 3),
                    '% Contribution': round(contribution_percentage, 3)
                })

        return pd.DataFrame(contribution_table)

    def calculate_detailed_anova(self):
        """
        Perform detailed ANOVA analysis
        
        Returns:
        --------
        pandas.DataFrame
            Detailed ANOVA table
        """
        if self.model_simplified is None:
            raise ValueError("Fit the simplified model first.")

        # Preparar datos para ANOVA detallado
        ss_total = np.sum((self.data[self.y_name] - self.data[self.y_name].mean())**2)
        df_total = len(self.data) - 1

        # ANOVA para modelo reducido
        formula_reduced = f'{self.y_name} ~ {self.x1_name} + {self.x2_name} + {self.x3_name} + ' \
                          f'I({self.x1_name}**2) + I({self.x2_name}**2) + I({self.x3_name}**2)'
        model_reduced = smf.ols(formula_reduced, data=self.data).fit()
        anova_reduced = sm.stats.anova_lm(model_reduced, typ=2)

        # Calcular componentes de variación
        ss_regression = anova_reduced['sum_sq'][:-1].sum()
        df_regression = len(anova_reduced) - 1

        ss_residual = self.model_simplified.ssr
        df_residual = self.model_simplified.df_resid

        # Error puro
        replicas = self.data[self.data.duplicated(subset=[self.x1_name, self.x2_name, self.x3_name], keep=False)]
        ss_pure_error = replicas.groupby([self.x1_name, self.x2_name, self.x3_name])[self.y_name].var().sum()
        df_pure_error = len(replicas) - len(replicas.groupby([self.x1_name, self.x2_name, self.x3_name]))

        # Falta de ajuste
        ss_lack_of_fit = ss_residual - ss_pure_error
        df_lack_of_fit = df_residual - df_pure_error

        # Calcular cuadrados medios y estadísticos F
        ms_regression = ss_regression / df_regression
        ms_residual = ss_residual / df_residual
        ms_lack_of_fit = ss_lack_of_fit / df_lack_of_fit
        ms_pure_error = ss_pure_error / df_pure_error

        f_lack_of_fit = ms_lack_of_fit / ms_pure_error
        p_lack_of_fit = 1 - f.cdf(f_lack_of_fit, df_lack_of_fit, df_pure_error)

        # Crear tabla de ANOVA detallada
        detailed_anova_table = pd.DataFrame({
            'Source of Variation': ['Regression', 'Residual', 'Lack of Fit', 'Pure Error', 'Total'],
            'Sum of Squares': [
                round(ss_regression, 3), 
                round(ss_residual, 3), 
                round(ss_lack_of_fit, 3), 
                round(ss_pure_error, 3), 
                round(ss_total, 3)
            ],
            'Degrees of Freedom': [df_regression, df_residual, df_lack_of_fit, df_pure_error, df_total],
            'Mean Square': [
                round(ms_regression, 3), 
                round(ms_residual, 3), 
                round(ms_lack_of_fit, 3), 
                round(ms_pure_error, 3), 
                np.nan
            ],
            'F': [np.nan, np.nan, round(f_lack_of_fit, 3), np.nan, np.nan],
            'p-value': [np.nan, np.nan, round(p_lack_of_fit, 3), np.nan, np.nan]
        })

        return detailed_anova_table
# --- Funciones para la interfaz de Gradio ---

def load_data(x1_name, x2_name, x3_name, y_name, x1_levels_str, x2_levels_str, x3_levels_str, data_str):
    try:
        x1_levels = [float(x.strip()) for x in x1_levels_str.split(',')]
        x2_levels = [float(x.strip()) for x in x2_levels_str.split(',')]
        x3_levels = [float(x.strip()) for x in x3_levels_str.split(',')]

        data_list = [row.split(',') for row in data_str.strip().split('\n')]
        column_names = ['Exp.', x1_name, x2_name, x3_name, y_name]
        data = pd.DataFrame(data_list, columns=column_names)
        data = data.apply(pd.to_numeric, errors='coerce')

        if not all(col in data.columns for col in column_names):
            raise ValueError("El formato de los datos no es correcto.")

        global rsm
        rsm = RSM_BoxBehnken(data, x1_name, x2_name, x3_name, y_name, x1_levels, x2_levels, x3_levels)

        return data, x1_name, x2_name, x3_name, y_name, x1_levels, x2_levels, x3_levels, gr.update(visible=True)

    except Exception as e:
        return None, "", "", "", "", [], [], [], gr.update(visible=False), f"Error: {e}"

def fit_and_optimize_model():
    if 'rsm' not in globals():
        return None, None, None, None, None, None, "Error: Carga los datos primero."
    
    model_completo, pareto_completo = rsm.fit_model()
    model_simplificado, pareto_simplificado = rsm.fit_simplified_model()
    optimization_table = rsm.optimize()
    equation = rsm.get_simplified_equation()
    prediction_table = rsm.generate_prediction_table()
    contribution_table = rsm.calculate_contribution_percentage()
    anova_table = rsm.calculate_detailed_anova()

    equation_formatted = equation.replace(" + ", "<br>+ ").replace(" ** ", "^").replace("*", " × ")
    equation_formatted = f"### Ecuación del Modelo Simplificado:<br>{equation_formatted}"

    
    return model_completo.summary().as_html(), pareto_completo, model_simplificado.summary().as_html(), pareto_simplificado, equation_formatted, optimization_table, prediction_table, contribution_table, anova_table

def generate_rsm_plot(fixed_variable, fixed_level):
    if 'rsm' not in globals():
        return None, "Error: Carga los datos primero."
    
    # Generar todas las gráficas
    all_figs = rsm.generate_all_plots()

    # Crear una lista de figuras para la salida
    plot_outputs = []
    for fig in all_figs:
        # Convertir la figura a una imagen en formato PNG
        img_bytes = fig.to_image(format="png")
        plot_outputs.append(img_bytes)

    # Retornar la lista de imágenes
    return plot_outputs

def download_excel():
    if 'rsm' not in globals():
        return None, "Error: Carga los datos y ajusta el modelo primero."

    output = io.BytesIO()
    with pd.ExcelWriter(output, engine='xlsxwriter') as writer:
        rsm.data.to_excel(writer, sheet_name='Datos', index=False)
        rsm.generate_prediction_table().to_excel(writer, sheet_name='Predicciones', index=False)
        rsm.optimize().to_excel(writer, sheet_name='Optimizacion', index=False)
        rsm.calculate_contribution_percentage().to_excel(writer, sheet_name='Contribucion', index=False)
        rsm.calculate_detailed_anova().to_excel(writer, sheet_name='ANOVA', index=False)

    output.seek(0)
    return gr.File.update(value=output, visible=True, filename="resultados_rsm.xlsx")

def download_images():
    if 'rsm' not in globals():
        return None, "Error: Carga los datos y ajusta el modelo primero."

    # Crear un directorio temporal para guardar las imágenes
    temp_dir = "temp_images"
    os.makedirs(temp_dir, exist_ok=True)

    # Generar todas las gráficas y guardarlas como imágenes PNG
    all_figs = rsm.generate_all_plots()
    for i, fig in enumerate(all_figs):
        img_path = os.path.join(temp_dir, f"plot_{i}.png")
        fig.write_image(img_path)

    # Comprimir las imágenes en un archivo ZIP
    zip_buffer = io.BytesIO()
    with ZipFile(zip_buffer, "w") as zip_file:
        for filename in os.listdir(temp_dir):
            file_path = os.path.join(temp_dir, filename)
            zip_file.write(file_path, arcname=filename)

    # Eliminar el directorio temporal
    for filename in os.listdir(temp_dir):
        file_path = os.path.join(temp_dir, filename)
        os.remove(file_path)
    os.rmdir(temp_dir)

    zip_buffer.seek(0)
    return gr.File.update(value=zip_buffer, visible=True, filename="graficos_rsm.zip")

# --- Crear la interfaz de Gradio ---

with gr.Blocks() as demo:
    gr.Markdown("# Optimización de la producción de AIA usando RSM Box-Behnken")

    with gr.Row():
        with gr.Column():
            gr.Markdown("## Configuración del Diseño")
            x1_name_input = gr.Textbox(label="Nombre de la Variable X1 (ej. Glucosa)", value="Glucosa")
            x2_name_input = gr.Textbox(label="Nombre de la Variable X2 (ej. Extracto de Levadura)", value="Extracto_de_Levadura")
            x3_name_input = gr.Textbox(label="Nombre de la Variable X3 (ej. Triptófano)", value="Triptofano")
            y_name_input = gr.Textbox(label="Nombre de la Variable Dependiente (ej. AIA (ppm))", value="AIA_ppm")
            x1_levels_input = gr.Textbox(label="Niveles de X1 (separados por comas)", value="1, 3.5, 5.5")
            x2_levels_input = gr.Textbox(label="Niveles de X2 (separados por comas)", value="0.03, 0.2, 0.3")
            x3_levels_input = gr.Textbox(label="Niveles de X3 (separados por comas)", value="0.4, 0.65, 0.9")
            data_input = gr.Textbox(label="Datos del Experimento (formato CSV)", lines=5, value="""1,-1,-1,0,166.594
2,1,-1,0,177.557
3,-1,1,0,127.261
4,1,1,0,147.573
5,-1,0,-1,188.883
6,1,0,-1,224.527
7,-1,0,1,190.238
8,1,0,1,226.483
9,0,-1,-1,195.550
10,0,1,-1,149.493
11,0,-1,1,187.683
12,0,1,1,148.621
13,0,0,0,278.951
14,0,0,0,297.238
15,0,0,0,280.896""")
            load_button = gr.Button("Cargar Datos")
            

        with gr.Column():
            gr.Markdown("## Datos Cargados")
            data_output = gr.Dataframe(label="Tabla de Datos")

    # Hacer que la sección de análisis sea visible solo después de cargar los datos
    with gr.Row(visible=False) as analysis_row:
        with gr.Column():
            fit_button = gr.Button("Ajustar Modelo y Optimizar")
            download_excel_button = gr.Button("Descargar Tablas en Excel")
            download_images_button = gr.Button("Descargar Gráficos en ZIP")
            gr.Markdown("**Modelo Completo**")
            model_completo_output = gr.HTML()
            pareto_completo_output = gr.Plot()
            gr.Markdown("**Modelo Simplificado**")
            model_simplificado_output = gr.HTML()
            pareto_simplificado_output = gr.Plot()
            equation_output = gr.HTML()
            optimization_table_output = gr.Dataframe(label="Tabla de Optimización")
            prediction_table_output = gr.Dataframe(label="Tabla de Predicciones")
            contribution_table_output = gr.Dataframe(label="Tabla de % de Contribución")
            anova_table_output = gr.Dataframe(label="Tabla ANOVA Detallada")
        with gr.Column():
            gr.Markdown("## Generar Gráficos de Superficie de Respuesta")
            fixed_variable_input = gr.Dropdown(label="Variable Fija", choices=["Glucosa", "Extracto_de_Levadura", "Triptofano"], value="Glucosa")
            fixed_level_input = gr.Slider(label="Nivel de Variable Fija", minimum=0, maximum=1, step=0.01, value=0.5)
            plot_button = gr.Button("Generar Gráfico")
            rsm_plot_output = gr.Gallery(label="Gráficos RSM", columns=3, preview=True, height="auto")

    load_button.click(
        load_data,
        inputs=[x1_name_input, x2_name_input, x3_name_input, y_name_input, x1_levels_input, x2_levels_input, x3_levels_input, data_input],
        outputs=[data_output, x1_name_input, x2_name_input, x3_name_input, y_name_input, x1_levels_input, x2_levels_input, x3_levels_input, analysis_row]
    )

    fit_button.click(fit_and_optimize_model, outputs=[model_completo_output, pareto_completo_output, model_simplificado_output, pareto_simplificado_output, equation_output, optimization_table_output, prediction_table_output, contribution_table_output, anova_table_output])
    
    plot_button.click(generate_rsm_plot, inputs=[fixed_variable_input, fixed_level_input], outputs=[rsm_plot_output])

    download_excel_button.click(download_excel, outputs=[gr.File()])
    download_images_button.click(download_images, outputs=[gr.File()])

    # Ejemplo de uso
    gr.Markdown("## Ejemplo de uso")
    gr.Markdown("1. Introduce los nombres de las variables y sus niveles en las cajas de texto correspondientes.")
    gr.Markdown("2. Copia y pega los datos del experimento en la caja de texto 'Datos del Experimento'.")
    gr.Markdown("3. Haz clic en 'Cargar Datos' para cargar los datos en la tabla.")
    gr.Markdown("4. Haz clic en 'Ajustar Modelo y Optimizar' para ajustar el modelo y encontrar los niveles óptimos de los factores.")
    gr.Markdown("5. Selecciona una variable fija y su nivel en los controles deslizantes.")
    gr.Markdown("6. Haz clic en 'Generar Gráfico' para generar un gráfico de superficie de respuesta.")
    gr.Markdown("7. Haz clic en 'Descargar Tablas en Excel' para obtener un archivo Excel con todas las tablas generadas.")
    gr.Markdown("8. Haz clic en 'Descargar Gráficos en ZIP' para obtener un archivo ZIP con todos los gráficos generados.")

demo.launch()