akkiii4653/learning-intelligence-tool

1import os2from typing import List, Dict, Any3 4import joblib5import numpy as np6import pandas as pd7 8 9class LearningIntelligence:10    """11    Core AI engine for the Learning Intelligence Tool.12 13    - Loads trained completion model.14    - Validates and preprocesses input data.15    - Generates completion probabilities and risk flags.16    - Computes chapter difficulty scores.17    - Produces summary insights for mentors/admins.18    """19 20    def __init__(21        self,22        model_path: str = "models/completion_model.pkl",23        feature_cols: List[str] = None,24    ) -> None:25        self.features = feature_cols or ["Chapter_Order", "Time_Spent", "Scores"]26 27        if not os.path.exists(model_path):28            raise FileNotFoundError(29                f"Model file not found at '{model_path}'. "30                "Train the model first (scripts/train_completion_model.py)."31            )32 33        self.model = joblib.load(model_path)34 35        if not hasattr(self.model, "predict_proba"):36            raise TypeError(37                "Loaded model does not support 'predict_proba'. "38                "Use a classifier with probability estimates enabled."39            )40 41    # -------------------------42    # Preprocessing43    # -------------------------44    def preprocess(self, df: pd.DataFrame) -> pd.DataFrame:45        if df is None or df.empty:46            raise ValueError("Input DataFrame is empty or None.")47 48        missing_cols = [col for col in self.features if col not in df.columns]49        if missing_cols:50            raise ValueError(f"Input missing required columns: {missing_cols}")51 52        processed = df[self.features].copy()53        for col in self.features:54            processed[col] = pd.to_numeric(processed[col], errors="coerce")55 56        if processed.isna().any().any():57            processed = processed.fillna(processed.median(numeric_only=True))58 59        return processed60 61    # -------------------------62    # Main analysis63    # -------------------------64    def analyze(self, df: pd.DataFrame) -> pd.DataFrame:65        """66        Generate:67        - Completion predictions (0/1)68        - Completion probabilities69        - Risk flags (High / Medium / Low)70        """71        if df is None or df.empty:72            raise ValueError("Input DataFrame is empty or None.")73 74        data = df.copy()75        processed_data = self.preprocess(data)76 77        proba = self.model.predict_proba(processed_data)[:, 1]78        preds = self.model.predict(processed_data)79 80        data["Completion_Probability"] = proba81        data["Predicted_Status"] = preds82 83        # Risk thresholds (tuned to create all 3 buckets)84        data["Risk_Flag"] = np.where(85            proba < 0.30,86            "High Risk",87            np.where(proba < 0.85, "Medium Risk", "Low Risk"),88        )89 90        return data91 92    # -------------------------93    # Chapter difficulty94    # -------------------------95    def calculate_chapter_difficulty(self, df: pd.DataFrame) -> pd.DataFrame:96        """97        Identify difficult chapters using:98        - Low average scores99        - High average time spent100        """101        if df is None or df.empty:102            raise ValueError("Input DataFrame is empty or None.")103 104        required_cols = ["Chapter_Order", "Scores", "Time_Spent", "Student_ID"]105        missing = [c for c in required_cols if c not in df.columns]106        if missing:107            raise ValueError(108                f"Input missing required columns for chapter difficulty: {missing}"109            )110 111        difficulty = (112            df.groupby("Chapter_Order")113            .agg(114                {115                    "Scores": "mean",116                    "Time_Spent": "mean",117                    "Student_ID": "count",118                }119            )120            .reset_index()121            .rename(columns={"Student_ID": "Student_Count"})122        )123 124        max_time = difficulty["Time_Spent"].max()125        if max_time == 0 or np.isnan(max_time):126            time_norm = np.zeros(len(difficulty))127        else:128            time_norm = difficulty["Time_Spent"] / max_time129 130        difficulty["Difficulty_Score"] = (131            (100 - difficulty["Scores"]) * 0.7 + time_norm * 30132        )133 134        difficulty = difficulty.sort_values(135            "Difficulty_Score", ascending=False136        ).reset_index(drop=True)137 138        return difficulty139 140    # -------------------------141    # Summary insights142    # -------------------------143    def generate_insights(144        self, results: pd.DataFrame, chapter_stats: pd.DataFrame145    ) -> Dict[str, Any]:146        """147        Generate human-readable summary insights:148        - Overall completion & risk distribution149        - Top high-risk students150        - Hardest chapters151        - Short text summary152        """153        insights: Dict[str, Any] = {}154 155        # Completion rate156        if "Predicted_Status" in results.columns:157            completion_rate = (results["Predicted_Status"] == 1).mean()158            insights["completion_rate"] = round(completion_rate * 100, 1)159        else:160            insights["completion_rate"] = None161 162        # Risk distribution163        if "Risk_Flag" in results.columns:164            risk_counts = results["Risk_Flag"].value_counts().to_dict()165        else:166            risk_counts = {}167        insights["risk_distribution"] = risk_counts168 169        # Top high-risk students170        if {"Student_ID", "Risk_Flag", "Completion_Probability"}.issubset(results.columns):171            high_risk_students = (172                results[results["Risk_Flag"] == "High Risk"]173                .sort_values("Completion_Probability")174                .head(10)[["Student_ID", "Scores", "Time_Spent", "Completion_Probability"]]175                .to_dict(orient="records")176            )177        else:178            high_risk_students = []179        insights["high_risk_students"] = high_risk_students180 181        # Hardest chapters182        if {"Chapter_Order", "Difficulty_Score"}.issubset(chapter_stats.columns):183            hardest_chapters = (184                chapter_stats.sort_values("Difficulty_Score", ascending=False)185                .head(5)[["Chapter_Order", "Difficulty_Score", "Scores", "Time_Spent"]]186                .to_dict(orient="records")187            )188        else:189            hardest_chapters = []190        insights["hardest_chapters"] = hardest_chapters191 192        # Short natural language summary193        summary_lines = []194 195        if insights["completion_rate"] is not None:196            summary_lines.append(197                f"Overall predicted completion rate is {insights['completion_rate']}%."198            )199 200        if risk_counts:201            hr = risk_counts.get("High Risk", 0)202            mr = risk_counts.get("Medium Risk", 0)203            lr = risk_counts.get("Low Risk", 0)204            summary_lines.append(205                f"Risk distribution: {hr} high-risk, {mr} medium-risk, {lr} low-risk learners."206            )207 208        if hardest_chapters:209            hardest = hardest_chapters[0]210            summary_lines.append(211                f"Chapter {hardest['Chapter_Order']} appears most difficult with a difficulty score of "212                f"{round(hardest['Difficulty_Score'], 1)}."213            )214 215        insights["summary_text"] = " ".join(summary_lines) if summary_lines else ""216 217        return insights218