Create app.py

app.py

@@ -0,0 +1,571 @@
+"""
+Streamlit app for testing machine-generated code detection model with explainability.
+
+This app allows users to:
+1. Input code snippets
+2. Get predictions on whether the code is human-written or machine-generated
+3. View feature importance and explanations for the prediction
+"""
+
+import streamlit as st
+import pandas as pd
+import numpy as np
+import json
+import joblib
+import os
+from typing import Dict, List, Any
+import matplotlib.pyplot as plt
+import seaborn as sns
+import plotly.express as px
+import plotly.graph_objects as go
+from plotly.subplots import make_subplots
+
+# Import local modules
+from code_analytics import extract_all_code_analytics, get_analytics_feature_names
+from entropy_weighted_perplexity import EntropyWeightedPerplexity
+
+# Try to import SHAP for advanced explainability
+try:
+    import shap
+    SHAP_AVAILABLE = True
+except ImportError:
+    SHAP_AVAILABLE = False
+
+st.set_page_config(
+    page_title="AI Code Detection Tool",
+    page_icon="🤖",
+    layout="wide",
+    initial_sidebar_state="expanded"
+)
+
+class ModelLoader:
+    """Handle loading of trained models and metadata."""
+    
+    def __init__(self, model_dir: str = "results"):
+        self.model_dir = model_dir
+        self.model = None
+        self.metadata = None
+        
+    def load_model(self) -> bool:
+        """Load the trained model and metadata."""
+        try:
+            model_path = os.path.join(self.model_dir, "trained_model.pkl")
+            metadata_path = os.path.join(self.model_dir, "model_metadata.json")
+            
+            if not os.path.exists(model_path) or not os.path.exists(metadata_path):
+                return False
+                
+            self.model = joblib.load(model_path)
+            
+            with open(metadata_path, 'r') as f:
+                self.metadata = json.load(f)
+                
+            return True
+        except Exception as e:
+            st.error(f"Error loading model: {e}")
+            return False
+    
+    def get_model_info(self) -> Dict[str, Any]:
+        """Get model information for display."""
+        if self.metadata is None:
+            return {}
+            
+        return {
+            "Model Type": self.metadata.get("model_type", "Unknown"),
+            "Number of Features": len(self.metadata.get("feature_names", [])),
+            "F1 Score": f"{self.metadata.get('metrics', {}).get('f1_macro', 0):.4f}",
+            "Accuracy": f"{self.metadata.get('metrics', {}).get('accuracy', 0):.4f}",
+            "Features Used": "Code Analytics" if not self.metadata.get('config', {}).get('features', {}).get('use_llm_features', False) else "Code Analytics + LLM"
+        }
+
+class CodeAnalyzer:
+    """Handle code analysis and feature extraction."""
+    
+    def __init__(self, model_loader: ModelLoader):
+        self.model_loader = model_loader
+        
+    def extract_features(self, code: str) -> np.ndarray:
+        """
+        Extract features from code using the same pipeline as training.
+        """
+        try:
+            if self.model_loader.metadata is None:
+                raise ValueError("Model metadata not loaded")
+                
+            config = self.model_loader.metadata.get("config", {})
+            features_config = config.get("features", {})
+            
+            all_features = []
+            
+            # Extract LLM features if enabled (disabled in fast_config.yaml)
+            if features_config.get("use_llm_features", False):
+                # Initialize EWP calculator if needed
+                ewp_calculator = EntropyWeightedPerplexity(
+                    model_name=config["model"]["name"],
+                    entropy_window_size=config["model"]["entropy_window_size"],
+                    entropy_weight=config["model"]["entropy_weight"],
+                    perplexity_weight=config["model"]["perplexity_weight"],
+                )
+                
+                # Extract LLM features
+                llm_features = ewp_calculator.calculate_entropy_weighted_score(code)
+                all_features.extend([
+                    llm_features["entropy_weighted_score"],
+                    llm_features["mean_entropy"],
+                    llm_features["mean_windowed_entropy"], 
+                    llm_features["mean_cross_entropy"],
+                    llm_features["sequence_length"],
+                    llm_features["entropy_cross_entropy_ratio"],
+                    llm_features["windowed_raw_entropy_ratio"],
+                ])
+            
+            # Extract code analytics features
+            if features_config.get("use_code_analytics", True):
+                analytics_features = extract_all_code_analytics(code)
+                # Get features in the same order as training
+                analytics_feature_names = get_analytics_feature_names()
+                for feature_name in analytics_feature_names:
+                    all_features.append(analytics_features.get(feature_name, 0.0))
+            
+            return np.array(all_features)
+            
+        except Exception as e:
+            st.error(f"Feature extraction failed: {e}")
+            # Return zeros if extraction fails
+            n_features = len(self.model_loader.metadata.get("feature_names", []))
+            return np.zeros(n_features)
+    
+    def predict(self, code: str) -> Dict[str, Any]:
+        """Make prediction and return results with explanations."""
+        if self.model_loader.model is None:
+            return {"error": "Model not loaded"}
+            
+        try:
+            # Extract features
+            features = self.extract_features(code)
+            features = features.reshape(1, -1)
+            
+            # Make prediction
+            prediction = self.model_loader.model.predict(features)[0]
+            probability = self.model_loader.model.predict_proba(features)[0]
+            
+            # Get feature importance if available
+            feature_importance = self.get_feature_importance(features)
+            
+            return {
+                "prediction": prediction,
+                "probability": probability,
+                "features": features[0],
+                "feature_importance": feature_importance,
+                "label": self.model_loader.metadata["label_mapping"][str(prediction)]
+            }
+            
+        except Exception as e:
+            return {"error": f"Prediction failed: {e}"}
+    
+    def get_feature_importance(self, features: np.ndarray) -> Dict[str, float]:
+        """Get feature importance for the current prediction."""
+        try:
+            if hasattr(self.model_loader.model, 'feature_importances_'):
+                # For tree-based models
+                importances = self.model_loader.model.feature_importances_
+            elif hasattr(self.model_loader.model, 'coef_'):
+                # For linear models
+                importances = np.abs(self.model_loader.model.coef_[0])
+            else:
+                # For ensemble models, try to get feature importance from base estimators
+                if hasattr(self.model_loader.model, 'estimators_'):
+                    importances = []
+                    for estimator in self.model_loader.model.estimators_:
+                        if hasattr(estimator, 'feature_importances_'):
+                            importances.append(estimator.feature_importances_)
+                    if importances:
+                        importances = np.mean(importances, axis=0)
+                    else:
+                        importances = np.ones(len(features[0])) / len(features[0])
+                else:
+                    importances = np.ones(len(features[0])) / len(features[0])
+            
+            feature_names = self.model_loader.metadata.get("feature_names", 
+                                                         [f"Feature_{i}" for i in range(len(features[0]))])
+            
+            return dict(zip(feature_names, importances))
+            
+        except Exception as e:
+            st.warning(f"Could not get feature importance: {e}")
+            return {}
+    
+    def get_shap_explanation(self, code: str) -> Dict[str, Any]:
+        """Get SHAP explanations for the prediction."""
+        if not SHAP_AVAILABLE:
+            return {"error": "SHAP not available"}
+            
+        try:
+            # Extract features for the current code
+            features = self.extract_features(code).reshape(1, -1)
+            
+            # Create a SHAP explainer
+            if hasattr(self.model_loader.model, 'feature_importances_'):
+                # Tree-based model
+                explainer = shap.TreeExplainer(self.model_loader.model)
+            else:
+                # For other models, use KernelExplainer with a background dataset
+                # Use a small random background for efficiency
+                background_size = min(100, 10)  # Small background for speed
+                background_features = np.random.normal(
+                    features.mean(), features.std(), (background_size, features.shape[1])
+                )
+                explainer = shap.KernelExplainer(
+                    self.model_loader.model.predict_proba, background_features
+                )
+            
+            # Get SHAP values
+            shap_values = explainer.shap_values(features)
+            
+            # For binary classification, SHAP returns values for both classes
+            if isinstance(shap_values, list):
+                shap_values = shap_values[1]  # Use positive class
+            
+            feature_names = self.model_loader.metadata.get("feature_names", 
+                                                         [f"Feature_{i}" for i in range(features.shape[1])])
+            
+            return {
+                "shap_values": shap_values[0],
+                "feature_names": feature_names,
+                "base_value": explainer.expected_value if hasattr(explainer, 'expected_value') else 0.5,
+                "feature_values": features[0]
+            }
+            
+        except Exception as e:
+            return {"error": f"SHAP explanation failed: {e}"}
+
+def create_shap_waterfall_plot(shap_explanation: Dict[str, Any], top_n: int = 15):
+    """Create a SHAP waterfall-style plot showing feature contributions."""
+    if "error" in shap_explanation:
+        return None
+        
+    shap_values = shap_explanation["shap_values"]
+    feature_names = shap_explanation["feature_names"]
+    feature_values = shap_explanation["feature_values"]
+    base_value = shap_explanation.get("base_value", 0.5)
+    
+    # Get top contributing features (positive and negative)
+    feature_contributions = list(zip(feature_names, shap_values, feature_values))
+    feature_contributions.sort(key=lambda x: abs(x[1]), reverse=True)
+    top_features = feature_contributions[:top_n]
+    
+    # Create waterfall-style data
+    names = [f[0] for f in top_features]
+    values = [f[1] for f in top_features]
+    colors = ['green' if v > 0 else 'red' for v in values]
+    
+    fig = go.Figure(go.Bar(
+        x=values,
+        y=names,
+        orientation='h',
+        marker_color=colors,
+        text=[f"{v:.4f}" for v in values],
+        textposition="outside"
+    ))
+    
+    fig.update_layout(
+        title=f"SHAP Feature Contributions (Top {top_n})",
+        xaxis_title="SHAP Value (contribution to prediction)",
+        yaxis_title="Features",
+        height=600,
+        yaxis={'categoryorder': 'total ascending'},
+        showlegend=False
+    )
+    
+    # Add vertical line at 0
+    fig.add_vline(x=0, line_dash="dash", line_color="black", opacity=0.5)
+    
+    return fig
+
+def create_feature_importance_plot(feature_importance: Dict[str, float], top_n: int = 20):
+    """Create feature importance visualization."""
+    if not feature_importance:
+        return None
+        
+    # Sort by importance
+    sorted_features = sorted(feature_importance.items(), key=lambda x: abs(x[1]), reverse=True)
+    top_features = sorted_features[:top_n]
+    
+    feature_names = [f[0] for f in top_features]
+    importance_values = [f[1] for f in top_features]
+    
+    # Create horizontal bar plot
+    fig = go.Figure(go.Bar(
+        x=importance_values,
+        y=feature_names,
+        orientation='h',
+        marker_color=px.colors.qualitative.Set3
+    ))
+    
+    fig.update_layout(
+        title=f"Top {top_n} Most Important Features",
+        xaxis_title="Feature Importance",
+        yaxis_title="Features",
+        height=600,
+        yaxis={'categoryorder': 'total ascending'}
+    )
+    
+    return fig
+
+def create_prediction_gauge(probability: np.ndarray, prediction: int):
+    """Create a gauge chart showing prediction confidence."""
+    confidence = max(probability)
+    
+    fig = go.Figure(go.Indicator(
+        mode="gauge+number+delta",
+        value=confidence * 100,
+        domain={'x': [0, 1], 'y': [0, 1]},
+        title={'text': "Prediction Confidence (%)"},
+        gauge={
+            'axis': {'range': [None, 100]},
+            'bar': {'color': "lightgreen" if prediction == 0 else "lightcoral"},
+            'steps': [
+                {'range': [0, 50], 'color': "lightgray"},
+                {'range': [50, 80], 'color': "yellow"},
+                {'range': [80, 100], 'color': "lightgreen"}
+            ],
+            'threshold': {
+                'line': {'color': "red", 'width': 4},
+                'thickness': 0.75,
+                'value': 90
+            }
+        }
+    ))
+    
+    fig.update_layout(height=300)
+    return fig
+
+def main():
+    st.title("🤖 AI Code Detection Tool")
+    st.markdown("### Detect whether code is human-written or machine-generated with explainable AI")
+
+    # Initialize session state
+    if 'model_loader' not in st.session_state:
+        st.session_state.model_loader = ModelLoader()
+        st.session_state.model_loaded = False
+
+    # Sidebar
+    with st.sidebar:
+        st.header("Model Information")
+
+        # Try to load model if not already loaded
+        if not st.session_state.model_loaded:
+            if st.button("Load Model"):
+                with st.spinner("Loading model..."):
+                    if st.session_state.model_loader.load_model():
+                        st.session_state.model_loaded = True
+                        st.success("Model loaded successfully!")
+                    else:
+                        st.error("Failed to load model. Please ensure the model files exist in the 'results' directory.")
+
+        if st.session_state.model_loaded:
+            model_info = st.session_state.model_loader.get_model_info()
+            for key, value in model_info.items():
+                st.metric(key, value)
+
+    # Main content
+    if not st.session_state.model_loaded:
+        st.warning("⚠️ Please load the model first using the sidebar.")
+        st.info("Make sure you have trained a model using the main script with `save_model: true` in the config.")
+        return
+
+    # Initialize code analyzer
+    analyzer = CodeAnalyzer(st.session_state.model_loader)
+
+    # Code input
+    st.header("📝 Enter Code to Analyze")
+
+    # Sample code examples
+    examples = {
+        "Python Function": '''def fibonacci(n):
+    if n <= 1:
+        return n
+    return fibonacci(n-1) + fibonacci(n-2)''',
+        
+        "Simple Loop": '''for i in range(10):
+    print(f"Number: {i}")
+    if i % 2 == 0:
+        print("Even")''',
+        
+        "Class Definition": '''class Calculator:
+    def __init__(self):
+        self.history = []
+    
+    def add(self, a, b):
+        result = a + b
+        self.history.append(f"{a} + {b} = {result}")
+        return result'''
+    }
+
+    # Example selector
+    col1, col2 = st.columns([1, 3])
+    with col1:
+        selected_example = st.selectbox("Load Example:", [""] + list(examples.keys()))
+
+    # Code input area with syntax highlighting
+    if selected_example:
+        code_input = st.text_area("Code:", examples[selected_example], height=200, key="code_input")
+    else:
+        code_input = st.text_area("Code:", height=200, placeholder="Enter your code here...", key="code_input")
+
+    # Code validation
+    if code_input.strip():
+        try:
+            import ast
+            ast.parse(code_input)
+            st.success("✅ Valid Python syntax")
+        except SyntaxError as e:
+            st.warning(f"⚠️ Syntax error detected: {e}")
+            st.info("Note: The model can still analyze syntactically incorrect code, but results may be less reliable.")
+        except Exception:
+            st.info("Code validation skipped (not standard Python)")
+
+    # Analysis options
+    with st.expander("Analysis Options"):
+        show_all_features = st.checkbox("Show all features in results", value=False)
+        use_shap = st.checkbox("Enable SHAP explanations", value=SHAP_AVAILABLE, disabled=not SHAP_AVAILABLE)
+        if not SHAP_AVAILABLE:
+            st.info("Install SHAP (`pip install shap`) for advanced explanations")
+
+    # Analysis button
+    if st.button("🔍 Analyze Code", type="primary"):
+        if not code_input.strip():
+            st.warning("Please enter some code to analyze.")
+            return
+
+        with st.spinner("Analyzing code..."):
+            result = analyzer.predict(code_input)
+
+            if "error" in result:
+                st.error(result["error"])
+                return
+
+            # Display results
+            st.header("📊 Analysis Results")
+
+            col1, col2, col3 = st.columns(3)
+
+            with col1:
+                st.metric(
+                    "Prediction", 
+                    result["label"],
+                    delta=f"{max(result['probability']):.1%} confidence"
+                )
+
+            with col2:
+                human_prob = result["probability"][0]
+                machine_prob = result["probability"][1]
+                st.metric("Human-written", f"{human_prob:.1%}")
+                st.metric("Machine-generated", f"{machine_prob:.1%}")
+
+            with col3:
+                # Confidence gauge
+                gauge_fig = create_prediction_gauge(result["probability"], result["prediction"])
+                st.plotly_chart(gauge_fig, use_container_width=True)
+
+            # Feature importance and SHAP explanations
+            if result["feature_importance"]:
+                st.header("🔍 Feature Importance & Explanations")
+
+                # Create tabs for different explanations
+                tabs = ["Global Importance", "Feature Values"]
+                if SHAP_AVAILABLE:
+                    tabs.append("SHAP Explanations")
+
+                tab_objects = st.tabs(tabs)
+
+                with tab_objects[0]:  # Global Importance
+                    st.subheader("Model's Overall Feature Importance")
+                    fig = create_feature_importance_plot(result["feature_importance"], top_n=20)
+                    if fig:
+                        st.plotly_chart(fig, use_container_width=True)
+
+                    # Show top features in text
+                    st.subheader("Top Contributing Features:")
+                    sorted_features = sorted(result["feature_importance"].items(), 
+                                           key=lambda x: abs(x[1]), reverse=True)
+
+                    col1, col2 = st.columns(2)
+                    with col1:
+                        st.write("**Most Important:**")
+                        for i, (feature, importance) in enumerate(sorted_features[:10], 1):
+                            st.write(f"{i}. **{feature}**: {importance:.4f}")
+
+                    with col2:
+                        st.write("**Feature Description:**")
+                        st.info("These are the features the model finds most important globally across all predictions.")
+
+                with tab_objects[1]:  # Feature Values
+                    st.subheader("Current Code's Feature Values")
+
+                    # Show all features in a dataframe
+                    feature_names = st.session_state.model_loader.metadata.get("feature_names", [])
+                    feature_values = result["features"]
+
+                    if len(feature_names) == len(feature_values):
+                        feature_df = pd.DataFrame({
+                            "Feature": feature_names,
+                            "Value": feature_values,
+                            "Global_Importance": [result["feature_importance"].get(name, 0) for name in feature_names]
+                        }).sort_values("Global_Importance", ascending=False)
+
+                        st.dataframe(feature_df, height=400)
+
+                if SHAP_AVAILABLE and len(tab_objects) > 2:
+                    with tab_objects[2]:  # SHAP Explanations
+                        st.subheader("SHAP Analysis: Why This Prediction?")
+
+                        with st.spinner("Computing SHAP explanations..."):
+                            shap_result = analyzer.get_shap_explanation(code_input)
+
+                        if "error" not in shap_result:
+                            shap_fig = create_shap_waterfall_plot(shap_result, top_n=15)
+                            if shap_fig:
+                                st.plotly_chart(shap_fig, use_container_width=True)
+
+                            st.info("""
+                            **How to read SHAP values:**
+                            - Green bars push the prediction toward "Machine-generated"
+                            - Red bars push the prediction toward "Human-written"  
+                            - Longer bars = stronger influence on this specific prediction
+                            - Values show how much each feature contributed to moving the prediction from the baseline
+                            """)
+                        else:
+                            st.warning(f"SHAP analysis failed: {shap_result['error']}")
+                            st.info("Falling back to global feature importance above.")
+            else:
+                st.warning("Feature importance not available for this model.")
+
+    # Footer
+    st.markdown("---")
+    st.markdown("### About This Tool")
+    col1, col2 = st.columns(2)
+
+    with col1:
+        st.info("""
+        **Purpose**: This tool helps detect whether code was written by humans or generated by AI.
+        
+        **Method**: Uses static code analysis with machine learning, focusing on patterns in:
+        - Code structure and complexity
+        - Naming conventions and style
+        - Syntactic patterns and AST features
+        - Error handling and control flow
+        """)
+
+    with col2:
+        st.warning(
+            """
+        **Limitations**: 
+        - Works with Python code only
+        - Accuracy depends on code length and complexity
+        - Results are probabilistic, not definitive
+        """
+        )
+
+if __name__ == "__main__":
+    main()