"""Adaptive learning path service. Generates personalized learning sequences for students based on their current mastery profile, target learning outcomes, and knowledge graph dependencies. Uses rule-based path optimization with prerequisite analysis. """ import logging from datetime import datetime, timezone from typing import Dict, List, Tuple import pandas as pd from app.core.exceptions import EntityNotFoundError, DatasetError from app.data.loader import DatasetLoader from app.services.knowledge_graph_service import KnowledgeGraphService from app.schemas.learning_path import ( LearningPathRequest, LearningPathResponse, LearningPathStep ) logger = logging.getLogger(__name__) class AdaptiveLearningPathService: """Generates adaptive learning paths for students. Uses knowledge graph traversal combined with student mastery profiles to create personalized learning sequences. Prioritizes weak prerequisites and respects dependency chains while considering difficulty progression. """ def __init__(self, loader: DatasetLoader, knowledge_graph: KnowledgeGraphService) -> None: self._loader = loader self._knowledge_graph = knowledge_graph # Difficulty to study time mapping (minutes) self._difficulty_time_map = { "easy": 15, "medium": 25, "hard": 40 } # Mastery thresholds self._mastery_thresholds = { "weak": 0.4, "developing": 0.6, "proficient": 0.8 } def generate_learning_path(self, request: LearningPathRequest) -> LearningPathResponse: """Generate an adaptive learning path for a student. Algorithm: 1. Load student's current mastery profile 2. Identify prerequisites for target LO that are weak/missing 3. Build ordered learning sequence respecting dependencies 4. Add difficulty progression and time estimates 5. Generate completion probability and recommendations Args: request: Learning path generation request Returns: LearningPathResponse with ordered learning steps Raises: EntityNotFoundError: If student or target LO not found """ timestamp = datetime.now(timezone.utc).isoformat() # 1. Validate and load student mastery profile student_mastery = self._load_student_mastery(request.student_id) # 2. Validate target LO exists target_lo_info = self._get_lo_info(request.target_lo_id) # 3. Get all prerequisites for target LO all_prerequisites = self._knowledge_graph.get_prerequisites( request.target_lo_id, max_depth=None ) # 4. Identify weak prerequisites that need attention weak_prerequisites = self._identify_weak_prerequisites( all_prerequisites, student_mastery, request.include_mastered ) # 5. Build learning path with dependency ordering learning_steps = self._build_learning_path( weak_prerequisites, request.target_lo_id, student_mastery, request.max_steps, request.difficulty_preference ) # 6. Calculate path metadata total_time = sum(step.estimated_study_time for step in learning_steps) overall_mastery = self._calculate_overall_mastery(student_mastery) path_difficulty = self._determine_path_difficulty(learning_steps) completion_probability = self._estimate_completion_probability( learning_steps, overall_mastery ) # 7. Generate recommendations next_action, teacher_notes = self._generate_recommendations( learning_steps, weak_prerequisites, overall_mastery ) return LearningPathResponse( student_id=request.student_id, target_lo_id=request.target_lo_id, timestamp=timestamp, learning_path=learning_steps, total_steps=len(learning_steps), estimated_total_time=total_time, current_overall_mastery=overall_mastery, weak_prerequisites=weak_prerequisites, path_difficulty=path_difficulty, completion_probability=completion_probability, next_action=next_action, teacher_notes=teacher_notes ) def _load_student_mastery(self, student_id: str) -> Dict[str, float]: """Load student's mastery profile as a dictionary. Returns: Dict mapping lo_id to mastery_score """ try: # Validate student exists student_profiles = self._loader.load_table("student_profiles") if student_id not in student_profiles["student_id"].values: raise EntityNotFoundError(f"Student '{student_id}' not found") # Load mastery profiles mastery_profiles = self._loader.load_table("mastery_profiles") student_mastery = mastery_profiles[ mastery_profiles["student_id"] == student_id ] # Convert to dictionary mastery_dict = {} for _, row in student_mastery.iterrows(): lo_id = str(row["lo_id"]) mastery_score = float(row.get("mastery_score", 0.0)) mastery_dict[lo_id] = max(0.0, min(1.0, mastery_score)) return mastery_dict except Exception as exc: logger.error("Failed to load student mastery for %s: %s", student_id, exc) raise EntityNotFoundError(f"Unable to load mastery profile for student '{student_id}'") from exc def _get_lo_info(self, lo_id: str) -> Dict: """Get learning outcome metadata.""" try: learning_outcomes = self._loader.load_table("learning_outcomes") lo_row = learning_outcomes[learning_outcomes["lo_id"] == lo_id] if lo_row.empty: raise EntityNotFoundError(f"Learning outcome '{lo_id}' not found") return lo_row.iloc[0].to_dict() except Exception as exc: logger.error("Failed to get LO info for %s: %s", lo_id, exc) raise EntityNotFoundError(f"Learning outcome '{lo_id}' not found") from exc def _identify_weak_prerequisites( self, prerequisites: List[str], student_mastery: Dict[str, float], include_mastered: bool ) -> List[str]: """Identify prerequisites that need attention based on mastery scores.""" weak_prerequisites = [] for lo_id in prerequisites: mastery_score = student_mastery.get(lo_id, 0.0) # Include if weak/developing or if explicitly requested if mastery_score < self._mastery_thresholds["proficient"]: weak_prerequisites.append(lo_id) elif include_mastered and mastery_score >= self._mastery_thresholds["proficient"]: weak_prerequisites.append(lo_id) return weak_prerequisites def _build_learning_path( self, weak_prerequisites: List[str], target_lo_id: str, student_mastery: Dict[str, float], max_steps: int, difficulty_preference: str ) -> List[LearningPathStep]: """Build ordered learning path respecting dependencies.""" learning_steps = [] # Add weak prerequisites in dependency order ordered_prerequisites = self._order_by_dependencies(weak_prerequisites) step_number = 1 for lo_id in ordered_prerequisites[:max_steps-1]: # Reserve one step for target step = self._create_learning_step( step_number, lo_id, student_mastery, is_prerequisite=True, difficulty_preference=difficulty_preference ) learning_steps.append(step) step_number += 1 # Add target LO as final step if there's room if step_number <= max_steps: target_step = self._create_learning_step( step_number, target_lo_id, student_mastery, is_prerequisite=False, difficulty_preference=difficulty_preference ) learning_steps.append(target_step) return learning_steps def _order_by_dependencies(self, lo_ids: List[str]) -> List[str]: """Order LOs by dependency chain (prerequisites first).""" ordered = [] remaining = set(lo_ids) while remaining: # Find LOs with no remaining prerequisites ready_los = [] for lo_id in remaining: prerequisites = self._knowledge_graph.get_prerequisites(lo_id, max_depth=1) if not any(prereq in remaining for prereq in prerequisites): ready_los.append(lo_id) if not ready_los: # Break cycles by taking the first remaining LO ready_los = [next(iter(remaining))] # Sort by difficulty (easier first) and add to ordered list ready_los.sort(key=lambda lo: self._get_difficulty_score(lo)) ordered.extend(ready_los) remaining -= set(ready_los) return ordered def _create_learning_step( self, step_number: int, lo_id: str, student_mastery: Dict[str, float], is_prerequisite: bool, difficulty_preference: str ) -> LearningPathStep: """Create a learning path step with metadata.""" lo_info = self._get_lo_info(lo_id) mastery_score = student_mastery.get(lo_id, 0.0) mastery_label = self._get_mastery_label(mastery_score) # Estimate study time based on difficulty and current mastery difficulty = str(lo_info.get("difficulty", "medium")).lower() base_time = self._difficulty_time_map.get(difficulty, 25) # Adjust time based on mastery (less time if already partially mastered) time_multiplier = max(0.3, 1.0 - mastery_score) estimated_time = int(base_time * time_multiplier) # Generate reason for inclusion if is_prerequisite: if mastery_score < self._mastery_thresholds["weak"]: reason = f"Weak prerequisite (mastery: {mastery_score:.1%}) - needs foundational work" elif mastery_score < self._mastery_thresholds["developing"]: reason = f"Developing prerequisite (mastery: {mastery_score:.1%}) - needs reinforcement" else: reason = f"Prerequisite review (mastery: {mastery_score:.1%}) - ensure solid foundation" else: reason = f"Target learning outcome - current mastery: {mastery_score:.1%}" return LearningPathStep( step_number=step_number, lo_id=lo_id, title=str(lo_info.get("title", "")), grade=int(lo_info.get("grade", 6)), subject=str(lo_info.get("subject", "")), chapter=str(lo_info.get("chapter", "")), difficulty=difficulty, bloom_level=str(lo_info.get("bloom_level", "Understand")), current_mastery=mastery_score, mastery_label=mastery_label, is_prerequisite=is_prerequisite, estimated_study_time=estimated_time, reason=reason ) def _get_difficulty_score(self, lo_id: str) -> int: """Get numeric difficulty score for sorting (easier first).""" try: lo_info = self._get_lo_info(lo_id) difficulty = str(lo_info.get("difficulty", "medium")).lower() return {"easy": 1, "medium": 2, "hard": 3}.get(difficulty, 2) except Exception: return 2 def _get_mastery_label(self, mastery_score: float) -> str: """Convert mastery score to label.""" if mastery_score < self._mastery_thresholds["weak"]: return "weak" elif mastery_score < self._mastery_thresholds["developing"]: return "developing" elif mastery_score < self._mastery_thresholds["proficient"]: return "proficient" else: return "mastered" def _calculate_overall_mastery(self, student_mastery: Dict[str, float]) -> float: """Calculate student's overall mastery score.""" if not student_mastery: return 0.0 return sum(student_mastery.values()) / len(student_mastery) def _determine_path_difficulty(self, learning_steps: List[LearningPathStep]) -> str: """Determine overall path difficulty.""" if not learning_steps: return "easy" difficulty_counts = {"easy": 0, "medium": 0, "hard": 0} for step in learning_steps: difficulty_counts[step.difficulty] += 1 # Return the most common difficulty return max(difficulty_counts, key=difficulty_counts.get) def _estimate_completion_probability( self, learning_steps: List[LearningPathStep], overall_mastery: float ) -> float: """Estimate probability of path completion.""" if not learning_steps: return 1.0 # Base probability from overall mastery base_prob = 0.3 + (overall_mastery * 0.5) # Adjust for path length (longer paths are harder to complete) length_penalty = max(0.1, 1.0 - (len(learning_steps) * 0.05)) # Adjust for difficulty distribution hard_steps = sum(1 for step in learning_steps if step.difficulty == "hard") difficulty_penalty = max(0.1, 1.0 - (hard_steps * 0.1)) completion_prob = base_prob * length_penalty * difficulty_penalty return max(0.1, min(0.95, completion_prob)) def _generate_recommendations( self, learning_steps: List[LearningPathStep], weak_prerequisites: List[str], overall_mastery: float ) -> Tuple[str, str]: """Generate next action and teacher notes.""" if not learning_steps: next_action = "No learning path needed - target already mastered" teacher_notes = "Student has strong mastery of target and prerequisites" return next_action, teacher_notes first_step = learning_steps[0] # Next action if first_step.current_mastery < self._mastery_thresholds["weak"]: next_action = f"Start with foundational work on {first_step.title}" else: next_action = f"Begin with {first_step.title} ({first_step.estimated_study_time} min)" # Teacher notes weak_count = len(weak_prerequisites) if weak_count == 0: teacher_notes = "Student is ready for target LO with minimal prerequisite work" elif weak_count <= 3: teacher_notes = f"Student needs work on {weak_count} prerequisites before target LO" else: teacher_notes = f"Student has {weak_count} weak prerequisites - consider breaking into smaller goals" if overall_mastery < 0.4: teacher_notes += ". Consider additional support or scaffolding." return next_action, teacher_notes