Delete execution_programs.py

execution_programs.py

@@ -1,224 +0,0 @@
-import io
-import contextlib
-import streamlit as st
-import pandas as pd
-import matplotlib.pyplot as plt
-import seaborn as sns
-from datetime import datetime
-import re
-from io import StringIO
-import plotly.graph_objects as go
-from llm_models import text_llm
-from prompt_models import prompt_analyze
-
-
-def analyze_code_execution(code: str, execution_result: dict, df: pd.DataFrame) -> str:
-    """
-    Analyze only the execution results (output and variables) to provide focused insights.
-    
-    Args:
-        code: The Python code that was executed
-        execution_result: The dictionary containing execution results
-        df: The DataFrame being analyzed (for context only)
-        
-    Returns:
-        str: Natural language analysis focused on the execution results
-    """
-    # Extract relevant information from execution_result
-    output = execution_result.get('stdout', 'Tidak ada output teks')
-    variables = execution_result.get('variables', {})
-    error = execution_result.get('error', None)
-    
-    # Prepare variable summaries
-    var_summaries = []
-    for var_name, var_value in variables.items():
-        if isinstance(var_value, (pd.DataFrame, pd.Series)):
-            var_summaries.append(f"- {var_name}: {type(var_value).__name__} dengan bentuk {var_value.shape}")
-        elif isinstance(var_value, (plt.Figure, sns.axisgrid.Grid)):
-            var_summaries.append(f"- {var_name}: Visualisasi plot")
-        else:
-            var_summaries.append(f"- {var_name}: {type(var_value).__name__}")
-
-
-    prompt = prompt_analyze(output, var_summaries, df, error)
-    # Get analysis from the text model
-    response = text_llm(prompt)
-
-    # Clear thinking output
-    clean_response = re.sub(r'<\s*think\s*>.*?<\s*/\s*think\s*>', '', response, flags=re.DOTALL | re.IGNORECASE)
-    
-    return clean_response
-
-# =======================
-def execute_code(code: str, df: pd.DataFrame):
-    """Execute Python code safely and return outputs"""
-    # Extract Python code from markdown blocks
-    if "```python" in code:
-        clean_code = code.split("```python")[1].split("```")[0].strip()
-    else:
-        clean_code = code.strip()
-    
-    # Add required imports if missing
-    required_imports = [
-        "import pandas as pd",
-        "import matplotlib.pyplot as plt",
-        "import seaborn as sns",
-        "import plotly.express as px",
-        "import plotly.graph_objects as go"
-    ]
-    
-    for imp in required_imports:
-        if imp not in clean_code:
-            clean_code = imp + "\n" + clean_code
-    
-    # Prepare execution environment
-    local_vars = {
-        'df': df,
-        'plt': plt,
-        'sns': sns,
-        'pd': pd
-    }
-    
-    # Capture outputs
-    stdout = io.StringIO()
-    stderr = io.StringIO()
-    
-    try:
-        with contextlib.redirect_stdout(stdout), contextlib.redirect_stderr(stderr):
-            exec(clean_code, local_vars)
-            
-                # Deteksi plot Plotly
-            plotly_figs = []
-            for var_name, var_value in local_vars.items():
-                if isinstance(var_value, go.Figure):
-                    plotly_figs.append(var_value)
-            
-            return {
-                'code': clean_code,
-                'stdout': stdout.getvalue(),
-                'stderr': stderr.getvalue(),
-                'figure': plotly_figs,
-                'success': True,
-                'variables': {k: v for k, v in local_vars.items() 
-                              if not k.startswith('_') and k not in ['df', 'plt', 'sns', 'pd']}
-            }
-
-    except Exception as e:
-        return {
-            'code': clean_code,
-            'error': str(e),
-            'stderr': stderr.getvalue(),
-            'success': False
-        }
-
-def add_to_history(execution_result):
-    """Add execution result to history with timestamp"""
-    history_item = {
-        **execution_result,
-        'timestamp': datetime.now().strftime("%Y-%m-%d %H:%M:%S")
-    }
-    st.session_state.execution_history.append(history_item)
-
-    # Generate explanation for successful executions
-    if execution_result['success']:
-        with st.spinner("Analyzing results..."):
-            df_to_analyze = st.session_state.df_cleaned if 'df_cleaned' in st.session_state else st.session_state.df
-            
-            analysis = analyze_code_execution(
-                code=execution_result['code'],
-                execution_result=execution_result,
-                df=df_to_analyze  # Only used for context, not analyzed
-            )
-            
-            history_item['explanation'] = f"""
-            **Analisis Hasil Eksekusi:**
-            {analysis}
-            """
-            
-
-
-# =======================
-# Display Functions
-# =======================
-
-def display_code_with_highlight(code: str):
-    """Display formatted Python code using Streamlit's built-in code block"""
-    st.code(code, language='python')
-
-def display_history():
-    """Display all execution history items"""
-    st.markdown("### Execution History")
-    
-    if not st.session_state.execution_history:
-        st.info("No executions yet. Run some code to see results here.")
-        return
-    
-    for i, item in enumerate(reversed(st.session_state.execution_history)):
-        with st.container():
-            st.markdown(f"### Execution #{len(st.session_state.execution_history)-i}")
-            st.caption(f"Executed at {item['timestamp']}")
-            
-            # Display code
-            with st.expander("View Code", expanded=False):
-                display_code_with_highlight(item['code'])
-            
-            # Display outputs
-            if item['success']:
-                if 'explanation' in item:
-                    with st.expander("See Explanation"):
-                        st.markdown("**Explanation:**")
-                        st.info(item['explanation'])
-
-                if item['stdout']:
-                    with st.expander("View Output", expanded=False):
-                        st.markdown("**Output:**")
-                        lines = item['stdout'].split('\n')
-                        text_content = []
-                        table_data = []
-                        header_detected = False
-
-                        for idx, line in enumerate(lines):
-                            cleaned_line = line.strip()
-                            if not cleaned_line:
-                                continue
-                            
-                            if not header_detected:
-                                if re.match(r"^[\w\s_]+$", cleaned_line) and len(cleaned_line.split()) > 1:
-                                    if (idx + 1 < len(lines)) and re.match(r"^\d+\s+[\d\.e+-]+", lines[idx+1].strip()):
-                                        header_detected = True
-                                        table_data = lines[idx:]
-                                        break
-                                    else:
-                                        text_content.append(cleaned_line)
-                                else:
-                                    text_content.append(cleaned_line)
-                        
-                        if text_content:
-                            st.write("\n".join(text_content))
-                            
-                        if table_data:
-                            try:
-                                clean_table = [line.strip() for line in table_data if line.strip()]
-                                df = pd.read_csv(StringIO("\n".join(clean_table)), sep=r"\s+", engine="python")
-                                st.dataframe(df)
-                            except Exception as e:
-                                st.write("\n".join(table_data))
-
-                
-                if item['figure']:
-                    st.markdown("**Visualisasi Interaktif:**")
-                    for fig in item['figure']:
-                        st.plotly_chart(fig, use_container_width=True)
-
-                        
-                if item['variables']:
-                    with st.expander("Created Variables"):
-                        st.json({k: str(type(v)) for k, v in item['variables'].items()})
-            else:
-                st.error("Execution failed")
-                st.error(item['error'])
-                if item['stderr']:
-                    st.text("Error details:")
-                    st.text(item['stderr'])
-            
-            st.markdown("---")