"""Knowledge graph service for learning outcome dependencies. Builds and maintains a directed acyclic graph (DAG) of learning outcome dependencies from lo_dependencies.csv. Provides graph traversal methods for prerequisite chains, successor paths, and dependency analysis. """ import logging from datetime import datetime, timezone from typing import Dict, List import networkx as nx import pandas as pd from app.core.exceptions import EntityNotFoundError, DatasetError from app.data.loader import DatasetLoader from app.schemas.knowledge_graph import ( KnowledgeGraphResponse, LONode, LORelationship ) logger = logging.getLogger(__name__) class KnowledgeGraphService: """Manages learning outcome dependency graph and provides traversal methods. Loads lo_dependencies.csv and learning_outcomes.csv to build a NetworkX directed graph. Provides methods for finding prerequisites, successors, learning paths, and dependency analysis. """ def __init__(self, loader: DatasetLoader) -> None: self._loader = loader self._graph: nx.DiGraph | None = None self._lo_metadata: Dict[str, Dict] = {} self._build_graph() def _build_graph(self) -> None: """Build the knowledge graph from dataset tables.""" try: # Load learning outcomes for metadata learning_outcomes = self._loader.load_table("learning_outcomes") self._lo_metadata = { row["lo_id"]: { "title": str(row.get("title", "")), "grade": int(row.get("grade", 6)), "subject": str(row.get("subject", "")), "chapter": str(row.get("chapter", "")), "difficulty": str(row.get("difficulty", "Medium")).lower(), "bloom_level": str(row.get("bloom_level", "Understand")) } for _, row in learning_outcomes.iterrows() } # Load dependencies and build graph dependencies = self._loader.load_table("lo_dependencies") self._graph = nx.DiGraph() # Add all LO nodes first for lo_id in self._lo_metadata.keys(): self._graph.add_node(lo_id) # Strength column is a string category in this dataset — map to float _strength_map = {"weak": 0.33, "medium": 0.66, "strong": 1.0} # Add dependency edges (prerequisite -> dependent) for _, row in dependencies.iterrows(): prerequisite_lo = str(row["prerequisite_lo_id"]) dependent_lo = str(row["lo_id"]) relationship_type = str(row.get("relationship_type", "prerequisite")) raw_strength = row.get("strength", "medium") # Handle both numeric and string strength values try: strength = float(raw_strength) except (ValueError, TypeError): strength = _strength_map.get(str(raw_strength).lower(), 0.66) if prerequisite_lo in self._lo_metadata and dependent_lo in self._lo_metadata: self._graph.add_edge( prerequisite_lo, dependent_lo, relationship_type=relationship_type, strength=strength ) logger.info( "Built knowledge graph: %d nodes, %d edges", self._graph.number_of_nodes(), self._graph.number_of_edges() ) except Exception as exc: logger.error("Failed to build knowledge graph: %s", exc) raise DatasetError(f"Knowledge graph construction failed: {exc}") from exc def get_knowledge_graph(self, lo_id: str, max_depth: int = 2) -> KnowledgeGraphResponse: """Get knowledge graph information for a learning outcome. Args: lo_id: Learning outcome ID to query max_depth: Maximum depth to traverse for prerequisites/successors Returns: KnowledgeGraphResponse with LO info and relationships Raises: EntityNotFoundError: If lo_id is not found in the graph """ if not self._graph or lo_id not in self._graph: raise EntityNotFoundError(f"Learning outcome '{lo_id}' not found in knowledge graph") timestamp = datetime.now(timezone.utc).isoformat() # Get LO metadata lo_info = self._create_lo_node(lo_id) # Get prerequisites (nodes that point to this LO) prerequisites = self._get_prerequisites(lo_id, max_depth) # Get successors (nodes this LO points to) successors = self._get_successors(lo_id, max_depth) # Calculate depth from root (nodes with no predecessors) depth_from_root = self._calculate_depth_from_root(lo_id) return KnowledgeGraphResponse( lo_id=lo_id, timestamp=timestamp, lo_info=lo_info, prerequisites=prerequisites, successors=successors, prerequisite_count=len(prerequisites), successor_count=len(successors), depth_from_root=depth_from_root ) def get_prerequisites(self, lo_id: str, max_depth: int = None) -> List[str]: """Get all prerequisite LO IDs for a given learning outcome. Args: lo_id: Learning outcome ID max_depth: Maximum depth to traverse (None for all) Returns: List of prerequisite LO IDs in dependency order """ if not self._graph or lo_id not in self._graph: return [] prerequisites = [] visited = set() def _traverse_prerequisites(current_lo: str, depth: int) -> None: if max_depth is not None and depth >= max_depth: return if current_lo in visited: return visited.add(current_lo) # Get direct prerequisites for pred in self._graph.predecessors(current_lo): if pred not in prerequisites: prerequisites.append(pred) _traverse_prerequisites(pred, depth + 1) _traverse_prerequisites(lo_id, 0) return prerequisites def get_successors(self, lo_id: str, max_depth: int = None) -> List[str]: """Get all successor LO IDs for a given learning outcome. Args: lo_id: Learning outcome ID max_depth: Maximum depth to traverse (None for all) Returns: List of successor LO IDs """ if not self._graph or lo_id not in self._graph: return [] successors = [] visited = set() def _traverse_successors(current_lo: str, depth: int) -> None: if max_depth is not None and depth >= max_depth: return if current_lo in visited: return visited.add(current_lo) # Get direct successors for succ in self._graph.successors(current_lo): if succ not in successors: successors.append(succ) _traverse_successors(succ, depth + 1) _traverse_successors(lo_id, 0) return successors def get_learning_path(self, start_lo: str, target_lo: str) -> List[str]: """Find the shortest learning path between two learning outcomes. Args: start_lo: Starting learning outcome ID target_lo: Target learning outcome ID Returns: List of LO IDs representing the shortest path Raises: EntityNotFoundError: If either LO is not found or no path exists """ if not self._graph: raise DatasetError("Knowledge graph not initialized") if start_lo not in self._graph: raise EntityNotFoundError(f"Start LO '{start_lo}' not found in knowledge graph") if target_lo not in self._graph: raise EntityNotFoundError(f"Target LO '{target_lo}' not found in knowledge graph") try: # Find shortest path in the directed graph path = nx.shortest_path(self._graph, start_lo, target_lo) return path except nx.NetworkXNoPath: # No direct path - check if target is a prerequisite of start try: reverse_path = nx.shortest_path(self._graph, target_lo, start_lo) # Return reverse path (target should be learned first) return list(reversed(reverse_path)) except nx.NetworkXNoPath: raise EntityNotFoundError( f"No learning path found between '{start_lo}' and '{target_lo}'" ) def is_prerequisite(self, prerequisite_lo: str, dependent_lo: str) -> bool: """Check if one LO is a prerequisite of another. Args: prerequisite_lo: Potential prerequisite LO ID dependent_lo: Dependent LO ID Returns: True if prerequisite_lo is a prerequisite of dependent_lo """ if not self._graph: return False return nx.has_path(self._graph, prerequisite_lo, dependent_lo) def get_root_los(self) -> List[str]: """Get all root learning outcomes (no prerequisites). Returns: List of LO IDs that have no prerequisites """ if not self._graph: return [] return [node for node in self._graph.nodes() if self._graph.in_degree(node) == 0] def get_leaf_los(self) -> List[str]: """Get all leaf learning outcomes (no successors). Returns: List of LO IDs that have no successors """ if not self._graph: return [] return [node for node in self._graph.nodes() if self._graph.out_degree(node) == 0] def _get_prerequisites(self, lo_id: str, max_depth: int) -> List[LONode]: """Get prerequisite LO nodes with metadata.""" prerequisite_ids = self.get_prerequisites(lo_id, max_depth) return [self._create_lo_node(lo_id) for lo_id in prerequisite_ids] def _get_successors(self, lo_id: str, max_depth: int) -> List[LONode]: """Get successor LO nodes with metadata.""" successor_ids = self.get_successors(lo_id, max_depth) return [self._create_lo_node(lo_id) for lo_id in successor_ids] def _create_lo_node(self, lo_id: str) -> LONode: """Create an LONode from LO metadata.""" metadata = self._lo_metadata.get(lo_id, {}) return LONode( lo_id=lo_id, title=metadata.get("title", ""), grade=metadata.get("grade", 6), subject=metadata.get("subject", ""), chapter=metadata.get("chapter", ""), difficulty=metadata.get("difficulty", "medium"), bloom_level=metadata.get("bloom_level", "Understand") ) def _calculate_depth_from_root(self, lo_id: str) -> int: """Calculate the depth of an LO from root nodes.""" if not self._graph: return 0 # Find shortest path from any root to this LO root_los = self.get_root_los() min_depth = float('inf') for root_lo in root_los: try: path = nx.shortest_path(self._graph, root_lo, lo_id) min_depth = min(min_depth, len(path) - 1) except nx.NetworkXNoPath: continue return int(min_depth) if min_depth != float('inf') else 0