diff --git a/codewiki/src/be/dependency_analyzer/topo_sort.py b/codewiki/src/be/dependency_analyzer/topo_sort.py index da9ae3a0..633b5e38 100644 --- a/codewiki/src/be/dependency_analyzer/topo_sort.py +++ b/codewiki/src/be/dependency_analyzer/topo_sort.py @@ -7,28 +7,28 @@ """ import logging -from typing import Dict, List, Set, Any from collections import deque +from typing import Any -from codewiki.src.be.dependency_analyzer.models.core import Node from codewiki.src.be.dependency_analyzer.leaf_selection import ( LEAF_REDUCTION_THRESHOLD, compute_valid_leaf_types, filter_leaf_nodes, ) +from codewiki.src.be.dependency_analyzer.models.core import Node logger = logging.getLogger(__name__) -def detect_cycles(graph: Dict[str, Set[str]]) -> List[List[str]]: +def detect_cycles(graph: dict[str, set[str]]) -> list[list[str]]: """ Detect cycles in a dependency graph using Tarjan's algorithm to find strongly connected components. - + Args: graph: A dependency graph represented as adjacency lists (node -> set of dependencies) - + Returns: A list of lists, where each inner list contains the nodes in a cycle """ @@ -39,7 +39,7 @@ def detect_cycles(graph: Dict[str, Set[str]]) -> List[List[str]]: onstack = set() # nodes currently on the stack stack = [] # stack of nodes result = [] # list of cycles (strongly connected components) - + def strongconnect(node): # Set the depth index for node index[node] = index_counter[0] @@ -47,7 +47,7 @@ def strongconnect(node): index_counter[0] += 1 stack.append(node) onstack.add(node) - + # Consider successors for successor in graph.get(node, set()): if successor not in index: @@ -57,7 +57,7 @@ def strongconnect(node): elif successor in onstack: # Successor is on the stack and hence in the current SCC lowlink[node] = min(lowlink[node], index[successor]) - + # If node is a root node, pop the stack and generate an SCC if lowlink[node] == index[node]: # Start a new strongly connected component @@ -68,46 +68,47 @@ def strongconnect(node): scc.append(successor) if successor == node: break - + # Only include SCCs with more than one node (actual cycles) if len(scc) > 1: result.append(scc) - + # Visit each node for node in graph: if node not in index: strongconnect(node) - + return result -def resolve_cycles(graph: Dict[str, Set[str]]) -> Dict[str, Set[str]]: + +def resolve_cycles(graph: dict[str, set[str]]) -> dict[str, set[str]]: """ Resolve cycles in a dependency graph by identifying strongly connected components and breaking cycles. - + Args: graph: A dependency graph represented as adjacency lists (node -> set of dependencies) - + Returns: A new acyclic graph with the same nodes but with cycles broken """ # Detect cycles (SCCs) cycles = detect_cycles(graph) - + if not cycles: logger.debug("No cycles detected in the dependency graph") return graph - + logger.debug(f"Detected {len(cycles)} cycles in the dependency graph") - + # Create a copy of the graph to modify new_graph = {node: deps.copy() for node, deps in graph.items()} - + # Process each cycle for i, cycle in enumerate(cycles): - logger.debug(f"Cycle {i+1}: {' -> '.join(cycle)}") - + logger.debug(f"Cycle {i + 1}: {' -> '.join(cycle)}") + # Strategy: Break the cycle by removing the "weakest" dependency # Here, we just arbitrarily remove the last edge to make the graph acyclic # In a real-world scenario, you might use heuristics to determine which edge to break @@ -115,189 +116,192 @@ def resolve_cycles(graph: Dict[str, Set[str]]) -> Dict[str, Set[str]]: for j in range(len(cycle) - 1): current = cycle[j] next_node = cycle[j + 1] - + if next_node in new_graph[current]: logger.debug(f"Breaking cycle by removing dependency: {current} -> {next_node}") new_graph[current].remove(next_node) break - + return new_graph -def topological_sort(graph: Dict[str, Set[str]]) -> List[str]: + +def topological_sort(graph: dict[str, set[str]]) -> list[str]: """ Perform a topological sort on a dependency graph. - + Args: graph: A dependency graph represented as adjacency lists (node -> set of dependencies) - + Returns: A list of nodes in topological order (dependencies first) """ # First, check for and resolve cycles acyclic_graph = resolve_cycles(graph) - + # Initialize in-degree counter for all nodes in_degree = {node: 0 for node in acyclic_graph} - + # Count in-degrees for node, dependencies in acyclic_graph.items(): for dep in dependencies: if dep in in_degree: in_degree[dep] += 1 - + # Queue of nodes with no dependencies (in-degree of 0) queue = deque([node for node, degree in in_degree.items() if degree == 0]) - + # Result list to store the topological order result = [] - + # Process nodes in topological order while queue: node = queue.popleft() result.append(node) - + # Reduce in-degree for each node that depends on the current node for dependent, deps in acyclic_graph.items(): if node in deps: in_degree[dependent] -= 1 if in_degree[dependent] == 0: queue.append(dependent) - + # Check if the sort was successful (all nodes included) if len(result) != len(acyclic_graph): logger.warning("Topological sort failed: graph has cycles that weren't resolved") # Return all nodes in some order to avoid breaking the process return list(acyclic_graph.keys()) - + # Reverse the result to get dependencies first return result[::-1] -def dependency_first_dfs(graph: Dict[str, Set[str]]) -> List[str]: + +def dependency_first_dfs(graph: dict[str, set[str]]) -> list[str]: """ Perform a depth-first traversal of the dependency graph, starting from root nodes that have no dependencies. - + The graph uses natural dependency direction: - If A depends on B, the graph has an edge A → B - This means an edge from X to Y represents "X depends on Y" - Root nodes (nodes with no incoming edges/dependencies) are processed first, followed by nodes that depend on them - + Args: graph: A dependency graph with natural direction (A→B if A depends on B) - + Returns: A list of nodes in an order where dependencies come before their dependents """ # First, resolve cycles to ensure we have a DAG acyclic_graph = resolve_cycles(graph) - + # Find root nodes (nodes with no dependencies) root_nodes = [] # Create a reverse graph to easily check if a node has incoming edges has_incoming_edge = {node: False for node in acyclic_graph} - + for node, deps in acyclic_graph.items(): for dep in deps: has_incoming_edge[dep] = True - + # Nodes with no incoming edges are root nodes for node in acyclic_graph: if not has_incoming_edge.get(node, False) and node in acyclic_graph: root_nodes.append(node) - + if not root_nodes: logger.warning("No root nodes found in the graph, using arbitrary starting point") root_nodes = list(acyclic_graph.keys())[:1] # Use the first node as starting point - + # Track visited nodes visited = set() result = [] - + # DFS function that processes dependencies first def dfs(node): if node in visited: return visited.add(node) - + # Visit all dependencies first for dep in sorted(acyclic_graph.get(node, set())): dfs(dep) - + # Add this node to the result after all its dependencies result.append(node) - + # Start DFS from each root node for root in sorted(root_nodes): dfs(root) - + # Check if all nodes were visited if len(result) != len(acyclic_graph): # Some nodes weren't visited - try to visit remaining nodes for node in sorted(acyclic_graph.keys()): if node not in visited: dfs(node) - + return result -def build_graph_from_components(components: Dict[str, Any]) -> Dict[str, Set[str]]: + +def build_graph_from_components(components: dict[str, Any]) -> dict[str, set[str]]: """ Build a dependency graph from a collection of code components. - + The graph uses the natural dependency direction: - If A depends on B, we create an edge A → B - This means an edge from node X to node Y represents "X depends on Y" - Root nodes (nodes with no dependencies) are components that don't depend on anything - + Args: components: A dictionary of code components, where each component has a 'depends_on' attribute - + Returns: A dependency graph with natural dependency direction """ graph = {} - + for comp_id, component in components.items(): # Initialize the node's adjacency list if comp_id not in graph: graph[comp_id] = set() - + # Add dependencies for dep_id in component.depends_on: # Only include dependencies that are actual components in our repository if dep_id in components: graph[comp_id].add(dep_id) - - return graph + + return graph -def get_leaf_nodes(graph: Dict[str, Set[str]], components: Dict[str, Node]) -> List[str]: +def get_leaf_nodes(graph: dict[str, set[str]], components: dict[str, Node]) -> list[str]: """ Find leaf nodes (nodes that no other nodes depend on) and build dependency trees showing the full dependency chain from each leaf back to the ultimate dependencies. - + The graph uses natural dependency direction: - If A depends on B, the graph has an edge A → B - Leaf nodes are nodes that appear in no other node's dependency set - Each tree shows the dependency chain: leaf → its dependencies → their dependencies, etc. - + Args: graph: A dependency graph with natural direction (A→B if A depends on B) - + Returns: A list of leaf nodes """ # First, resolve cycles to ensure we have a DAG acyclic_graph = resolve_cycles(graph) - + # Find leaf nodes (nodes that no other nodes depend on) leaf_nodes = set(acyclic_graph.keys()) valid_types = compute_valid_leaf_types(components) - def concise_node(leaf_nodes: Set[str]) -> Set[str]: + def concise_node(leaf_nodes: set[str]) -> set[str]: concise_leaf_nodes = set() for node in leaf_nodes: if node.endswith("__init__"): @@ -310,16 +314,23 @@ def concise_node(leaf_nodes: Set[str]) -> Set[str]: concise_leaf_nodes = concise_node(leaf_nodes) if len(concise_leaf_nodes) >= LEAF_REDUCTION_THRESHOLD: - logger.debug(f"Leaf nodes are too many ({len(concise_leaf_nodes)}), removing dependencies of other nodes") + count_before = len(concise_leaf_nodes) + logger.info( + "Leaf nodes are too many (%d >= %d); reducing to components " + "that nothing else depends on.", + count_before, + LEAF_REDUCTION_THRESHOLD, + ) # Remove nodes that are dependencies of other nodes - for node, deps in acyclic_graph.items(): + for deps in acyclic_graph.values(): for dep in deps: leaf_nodes.discard(dep) - + concise_leaf_nodes = concise_node(leaf_nodes) - + logger.info("Reduced from %d to %d leaf nodes.", count_before, len(concise_leaf_nodes)) + if not leaf_nodes: logger.warning("No leaf nodes found in the graph") return [] - - return concise_leaf_nodes \ No newline at end of file + + return concise_leaf_nodes