# ============================================================================= # GRAPH VISUALIZATIONS v2.0 - Pro-Level Mindmaps, Knowledge Graphs # ============================================================================= # # PRO FEATURES: # - Advanced Radial Mindmap with hierarchical tree layout # - Force-directed Knowledge Graph with community detection # - Entity Clustering Visualization # - Sankey Flow Diagrams # - Interactive zoom/pan with detailed tooltips # - Premium styling with gradients and animations # # All dynamically generated from actual graph data - NO hardcoding! # import json import math from typing import Dict, List, Optional, Tuple, Any import pandas as pd import networkx as nx from collections import defaultdict def generate_mindmap( graph: nx.Graph, query: str, focus_entity: Optional[str] = None, max_depth: int = 3, max_children: int = 5 ) -> Tuple[Optional[Dict], str]: """ 🧠 Generate a proper MINDMAP with nodes and branches (tree-style diagram). Uses improved BFS tree layout to show true hierarchy: Root -> Level 1 (e.g. Categories) -> Level 2 (e.g. Products) """ print(f"[MINDMAP] Starting mindmap generation for query: {query}") if not graph or graph.number_of_nodes() == 0: return None, "No graph data available for mindmap" def clean_name(name: str) -> str: name_str = str(name) if ':' in name_str: name_str = name_str.split(':')[-1].strip() return name_str[:20] # 1. Determine Root Node # Priority: Explicit focus -> Query match -> Centrality -> Degree root_node = focus_entity if not root_node: # Try finding entity mentioned in query query_lower = query.lower() for node in graph.nodes(): if str(node).lower() in query_lower: root_node = node break if not root_node: # Fallback to Degree Centrality (most connected node) degrees = sorted(graph.degree(), key=lambda x: x[1], reverse=True) if degrees: root_node = degrees[0][0] if not root_node: return None, "Could not determine root node for mindmap" print(f"[MINDMAP] Selected root: {root_node}") # 2. Build BFS Tree from Root # This creates a hierarchical structure ensuring no cycles try: # Limit graph size for performance before building tree # Get egocentric subgraph radius max_depth subgraph = nx.ego_graph(graph, root_node, radius=max_depth) tree = nx.bfs_tree(subgraph, root_node, depth_limit=max_depth) except Exception as e: print(f"[MINDMAP] Error building tree: {e}") # Fallback shallow tree tree = nx.Graph() tree.add_node(root_node) for n in list(graph.neighbors(root_node))[:max_children]: tree.add_edge(root_node, n) # 3. Calculate Layout Positions (Reingold-Tilford / Tree Layout) # Custom implementation for Plotly since NX doesn't have a great tree layout built-in for all vers node_data = [] # {name, x, y, size, color, depth} edges = [] # (x1, y1, x2, y2) # Identify levels levels = {} # node -> depth levels[root_node] = 0 bfs_layers = list(nx.bfs_layers(tree, [root_node])) # Process layers to determining positions # Simple strategy: Root at 0,0. Children spaced out vertically at x=1, etc. pos = {} max_y_at_level = {} for depth, layer in enumerate(bfs_layers): # Limit children per layer to keep map readable if depth > 0: # Sort by degree in original graph to show important nodes first layer = sorted(layer, key=lambda n: graph.degree(n), reverse=True) layer = layer[:max_children * (depth+1)] # Allow more nodes at deeper levels layer_height = len(layer) y_start = -(layer_height - 1) * 0.8 / 2 for i, node in enumerate(layer): levels[node] = depth x = depth * 2 # Horizontal spacing y = y_start + i * 0.8 # Vertical spacing pos[node] = (x, y) # Store edge info (except for root) if depth > 0: # Find parent in previous layer parents = list(tree.predecessors(node)) if parents: parent = parents[0] if parent in pos: px, py = pos[parent] edges.append((px, py, x, y)) # 4. Generate Visualization Data colors = ['#6366f1', '#10b981', '#f59e0b', '#ef4444', '#8b5cf6', '#ec4899'] for node, (x, y) in pos.items(): depth = levels.get(node, 0) color = colors[depth % len(colors)] size = max(15, 40 - depth * 8) node_data.append({ 'name': clean_name(node), 'x': x, 'y': y, 'size': size, 'color': color, 'hover': f"{node}
Level {depth}" }) # 5. Build Chart JSON edge_x = [] edge_y = [] for x1, y1, x2, y2 in edges: # Curved lines (simple bezier approximation via None point) # For simplicity in scatter plot, we use straight lines or segmented edge_x.extend([x1, x2, None]) edge_y.extend([y1, y2, None]) chart = { "data": [ # Edges { "type": "scatter", "x": edge_x, "y": edge_y, "mode": "lines", "line": { "width": 1.5, "color": "#cbd5e1", "shape": "spline" # Attempt curved lines }, "hoverinfo": "none" }, # Nodes { "type": "scatter", "x": [n['x'] for n in node_data], "y": [n['y'] for n in node_data], "mode": "markers+text", "marker": { "size": [n['size'] for n in node_data], "color": [n['color'] for n in node_data], "line": {"width": 2, "color": "white"} }, "text": [n['name'] for n in node_data], "textposition": "bottom center" if len(node_data) < 20 else "middle right", "hovertext": [n['hover'] for n in node_data], "hoverinfo": "text" } ], "layout": { "title": { "text": f"Mindmap: {clean_name(root_node)}", "font": {"size": 20, "color": "#1e293b"} }, "showlegend": False, "xaxis": {"visible": False, "showgrid": False, "zeroline": False}, "yaxis": {"visible": False, "showgrid": False, "zeroline": False}, "plot_bgcolor": "rgba(0,0,0,0)", "paper_bgcolor": "rgba(0,0,0,0)", "height": 600, "margin": {"l": 50, "r": 50, "t": 80, "b": 50} } } explanation = f"""**Hierarchical Mindmap** Starts from **{clean_name(root_node)}** and branches out to {len(node_data)-1} connected entities. • **Level 1**: Main Categories/Connections • **Level 2**: Detailed Items """ return chart, explanation def generate_knowledge_graph( graph: nx.Graph, query: str, max_nodes: int = 40, currency_symbol: str = "₹" ) -> Tuple[Optional[Dict], str]: """ 🕸️ Generate an Advanced Knowledge Graph with COMMUNITY DETECTION. Features: - Nodes sized by PageRank (influence) - Colors by Community (Louvain/Modularity) - Filtering to show most relevant nodes """ if not graph or graph.number_of_nodes() == 0: return None, "No graph data available" # 1. Select Important Nodes (PageRank or Degree) # PageRank is better for finding "influential" nodes not just hubs try: importance = nx.pagerank(graph) except: importance = dict(nx.degree(graph)) # Normalize m = max(importance.values()) or 1 importance = {k: v/m for k,v in importance.items()} # Sort nodes by importance top_nodes_with_score = sorted(importance.items(), key=lambda x: x[1], reverse=True)[:max_nodes] top_nodes = [n[0] for n in top_nodes_with_score] subgraph = graph.subgraph(top_nodes) # 2. Community Detection (Greedy Modularity) # Groups nodes that are densely connected try: from networkx.algorithms.community import greedy_modularity_communities communities = list(greedy_modularity_communities(subgraph)) # Map node -> community_id partition = {} for idx, comm in enumerate(communities): for node in comm: partition[node] = idx except ImportError: # Fallback if algo missing partition = {n: 0 for n in top_nodes} except Exception as e: print(f"Community detection failed: {e}") partition = {n: 0 for n in top_nodes} # 3. Layout (Spring / Fruchterman-Reingold) pos = nx.spring_layout(subgraph, k=0.5, iterations=50, seed=42) # 4. Build Traces node_x, node_y = [], [] node_text, node_size, node_color = [], [], [] # Palette for communities community_colors = [ '#3b82f6', '#ef4444', '#10b981', '#f59e0b', '#8b5cf6', '#ec4899', '#06b6d4', '#84cc16', '#6366f1', '#d946ef' ] for node in top_nodes: x, y = pos[node] node_x.append(x) node_y.append(y) # Metadata comm_id = partition.get(node, 0) score = importance.get(node, 0) # Size based on importance (min 10, max 50) size = 10 + (math.sqrt(score) * 40) node_size.append(size) # Color by community node_color.append(community_colors[comm_id % len(community_colors)]) # Label node_type = graph.nodes[node].get('type', 'Unknown') node_text.append(f"{node}
Type: {node_type}
Group: {comm_id+1}") # Edges edge_x, edge_y = [], [] for u, v in subgraph.edges(): x0, y0 = pos[u] x1, y1 = pos[v] edge_x.extend([x0, x1, None]) edge_y.extend([y0, y1, None]) chart = { "data": [ { "type": "scatter", "x": edge_x, "y": edge_y, "mode": "lines", "line": {"width": 1, "color": "rgba(100, 116, 139, 0.4)"}, "hoverinfo": "none" }, { "type": "scatter", "x": node_x, "y": node_y, "mode": "markers+text", "marker": { "size": node_size, "color": node_color, "line": {"width": 1.5, "color": "white"} }, "text": [str(n)[:15] for n in top_nodes], "textposition": "top center", "hoverinfo": "text", "hovertext": node_text } ], "layout": { "title": {"text": "Network Analysis (Community Detected)", "font": {"size": 20}}, "showlegend": False, "hovermode": "closest", "plot_bgcolor": "rgba(0,0,0,0)", "paper_bgcolor": "rgba(0,0,0,0)", "xaxis": {"visible": False}, "yaxis": {"visible": False}, "height": 600, "margin": {"l": 20, "r": 20, "t": 60, "b": 20} } } explanation = f"""**Smart Knowledge Graph** Visualizes the top {len(top_nodes)} most influential entities, grouped by behavior. • **Colors**: Represent {len(set(partition.values()))} detected communities (closely related groups). • **Node Size**: Indicates network influence (centrality). """ return chart, explanation def generate_relationship_diagram( df: pd.DataFrame, query: str, source_col: str = "customer", target_col: str = "product", value_col: str = "amount", top_n: int = 20, currency_symbol: str = "₹" ) -> Tuple[Optional[Dict], str]: """ 🔗 Generate a Sankey diagram showing relationships between entities. Shows flow/connections between two entity types (e.g., customers → products). """ from core.llm import chat if df is None or df.empty: return None, "No data available" # LLM determines best columns if not obvious if source_col not in df.columns or target_col not in df.columns: cols = list(df.columns) # Simple heuristic fallback cat_cols = [c for c in df.columns if df[c].dtype == 'object'] num_cols = [c for c in df.columns if df[c].dtype != 'object'] if len(cat_cols) >= 2: source_col = cat_cols[0] target_col = cat_cols[1] if num_cols: value_col = num_cols[0] # Aggregate data try: grouped = df.groupby([source_col, target_col])[value_col].sum().reset_index() grouped = grouped.nlargest(top_n, value_col) except: return None, f"Could not aggregate data with columns: {source_col}, {target_col}, {value_col}" # Create node list sources = grouped[source_col].unique().tolist() targets = grouped[target_col].unique().tolist() all_nodes = sources + targets node_map = {n: i for i, n in enumerate(all_nodes)} # Build Sankey data source_indices = [node_map[s] for s in grouped[source_col]] target_indices = [node_map[t] for t in grouped[target_col]] values = [float(v) for v in grouped[value_col]] chart = { "data": [{ "type": "sankey", "node": { "pad": 15, "thickness": 20, "line": {"color": "black", "width": 0.5}, "label": [str(n)[:20] for n in all_nodes], "color": ["#6366f1"] * len(sources) + ["#ec4899"] * len(targets) }, "link": { "source": source_indices, "target": target_indices, "value": values, "color": "rgba(99, 102, 241, 0.2)" } }], "layout": { "title": {"text": f"{source_col.title()} → {target_col.title()} Flow", "font": {"size": 20}}, "height": 500, "font": {"size": 12}, "plot_bgcolor": "rgba(0,0,0,0)", "paper_bgcolor": "rgba(0,0,0,0)", } } total_flow = sum(values) explanation = f"""**Relationship Flow** Mapping the flow of value from **{source_col}** to **{target_col}**. • **Top connections**: {len(values)} pathways shown • **Total Value**: {currency_symbol}{total_flow:,.0f} """ return chart, explanation def detect_graph_visualization_type(query: str) -> str: """ Detect graph visualization type from query using keywords. Returns: 'mindmap', 'knowledge_graph', 'relationship', 'radial', 'cluster', or 'none' """ q = query.lower() # Direct keyword matching (most reliable) if any(kw in q for kw in ['mindmap', 'mind map', 'mind-map', 'hierarchy', 'tree structure', 'tree diagram']): return 'mindmap' if any(kw in q for kw in ['radial', 'radial graph', 'radial layout', 'circular layout']): return 'radial' if any(kw in q for kw in ['knowledge graph', 'network', 'network graph', 'entity network']): return 'knowledge_graph' if any(kw in q for kw in ['cluster', 'clustering', 'group entities', 'community', 'communities']): return 'cluster' if any(kw in q for kw in ['sankey', 'relationship', 'flow', 'path', 'connection', 'link']): return 'relationship' # Implicit visualization requests if any(kw in q for kw in ['visualize relationships', 'show connections', 'how are they connected']): return 'knowledge_graph' return 'none' def wants_graph_visualization(query: str) -> bool: """Check if user wants a graph-based visualization.""" return detect_graph_visualization_type(query) != 'none' # ============================================================================= # PRO-LEVEL GRAPH VISUALIZATIONS v2.0 # ============================================================================= def generate_radial_mindmap( graph: nx.Graph, query: str, focus_entity: Optional[str] = None, max_depth: int = 3, max_children: int = 8 ) -> Tuple[Optional[Dict], str]: """ 🌟 PRO RADIAL MINDMAP - Circular hierarchical layout. Creates a stunning radial tree with the root at center, branches spreading outward in concentric circles. """ print(f"[RADIAL MINDMAP] Generating for query: {query}") if not graph or graph.number_of_nodes() == 0: return None, "No graph data available for radial mindmap" def clean_name(name: str) -> str: name_str = str(name) if ':' in name_str: name_str = name_str.split(':')[-1].strip() return name_str[:25] # Determine root node root_node = focus_entity if not root_node: degrees = sorted(graph.degree(), key=lambda x: x[1], reverse=True) if degrees: root_node = degrees[0][0] if not root_node: return None, "Could not determine root node" # Build BFS tree try: subgraph = nx.ego_graph(graph, root_node, radius=max_depth) tree = nx.bfs_tree(subgraph, root_node, depth_limit=max_depth) except: tree = nx.Graph() tree.add_node(root_node) for n in list(graph.neighbors(root_node))[:max_children]: tree.add_edge(root_node, n) # Calculate radial positions bfs_layers = list(nx.bfs_layers(tree, [root_node])) node_data = [] edges = [] levels = {root_node: 0} pos = {root_node: (0, 0)} # Premium gradient colors level_colors = [ '#6366F1', # Indigo (root) '#8B5CF6', # Violet (level 1) '#A855F7', # Purple (level 2) '#C084FC', # Light purple (level 3) '#DDD6FE', # Very light purple (level 4) ] for depth, layer in enumerate(bfs_layers): if depth == 0: continue # Root already positioned # Limit nodes per layer layer = sorted(layer, key=lambda n: graph.degree(n), reverse=True)[:max_children * depth] radius = depth * 1.5 angle_step = 2 * math.pi / max(len(layer), 1) for i, node in enumerate(layer): levels[node] = depth angle = i * angle_step - math.pi / 2 # Start from top x = radius * math.cos(angle) y = radius * math.sin(angle) pos[node] = (x, y) # Find parent and add edge parents = list(tree.predecessors(node)) if parents and parents[0] in pos: px, py = pos[parents[0]] edges.append((px, py, x, y)) # Build node data for node, (x, y) in pos.items(): depth = levels.get(node, 0) color = level_colors[min(depth, len(level_colors) - 1)] size = max(20, 50 - depth * 10) node_data.append({ 'name': clean_name(node), 'x': x, 'y': y, 'size': size, 'color': color, 'hover': f"{node}
Level {depth}" }) # Build edge traces with curved lines edge_x, edge_y = [], [] for x1, y1, x2, y2 in edges: # Create smooth bezier curve cx, cy = (x1 + x2) / 2 * 1.1, (y1 + y2) / 2 * 1.1 for t in [0, 0.25, 0.5, 0.75, 1]: bx = (1-t)**2 * x1 + 2*(1-t)*t * cx + t**2 * x2 by = (1-t)**2 * y1 + 2*(1-t)*t * cy + t**2 * y2 edge_x.append(bx) edge_y.append(by) edge_x.append(None) edge_y.append(None) chart = { "data": [ { "type": "scatter", "x": edge_x, "y": edge_y, "mode": "lines", "line": {"width": 1.5, "color": "#CBD5E1"}, "hoverinfo": "none" }, { "type": "scatter", "x": [n['x'] for n in node_data], "y": [n['y'] for n in node_data], "mode": "markers+text", "marker": { "size": [n['size'] for n in node_data], "color": [n['color'] for n in node_data], "line": {"width": 2, "color": "white"} }, "text": [n['name'] for n in node_data], "textposition": "middle center", "textfont": {"size": 10, "color": "white"}, "hovertext": [n['hover'] for n in node_data], "hoverinfo": "text" } ], "layout": { "title": {"text": f"🌟 Radial Mindmap: {clean_name(root_node)}", "font": {"size": 18}}, "showlegend": False, "xaxis": {"visible": False, "showgrid": False, "zeroline": False}, "yaxis": {"visible": False, "showgrid": False, "zeroline": False, "scaleanchor": "x"}, "plot_bgcolor": "rgba(0,0,0,0)", "paper_bgcolor": "rgba(0,0,0,0)", "height": 600, "margin": {"l": 20, "r": 20, "t": 60, "b": 20} } } explanation = f"""**🌟 Radial Mindmap** Central entity: **{clean_name(root_node)}** with {len(node_data)-1} connected entities across {len(bfs_layers)-1} levels. """ return chart, explanation def generate_entity_cluster( df: pd.DataFrame, query: str, entity_col: Optional[str] = None, value_col: Optional[str] = None, top_n: int = 30, currency_symbol: str = "₹" ) -> Tuple[Optional[Dict], str]: """ 🔮 ENTITY CLUSTERING - Group entities by similarity/value. Creates a bubble cluster where: - Bubble size = value/importance - Color = cluster/category - Position = similarity grouping """ if df is None or df.empty: return None, "No data available for clustering" # Auto-detect columns if not specified if not entity_col: cat_cols = [c for c in df.columns if df[c].dtype == 'object'] entity_col = cat_cols[0] if cat_cols else None if not value_col: num_cols = [c for c in df.columns if df[c].dtype in ['int64', 'float64']] value_col = num_cols[0] if num_cols else None if not entity_col or not value_col: return None, "Could not detect suitable columns for clustering" # Aggregate by entity grouped = df.groupby(entity_col)[value_col].sum().sort_values(ascending=False).head(top_n) entities = grouped.index.tolist() values = grouped.values.tolist() # Normalize for sizing (radius 10-50) max_val = max(values) if values else 1 sizes = [max(15, min(60, (v / max_val) * 45 + 15)) for v in values] # Create force-directed-like layout n = len(entities) positions = [] for i in range(n): # Spiral layout angle = i * 0.5 radius = math.sqrt(i + 1) * 0.8 x = radius * math.cos(angle) y = radius * math.sin(angle) positions.append((x, y)) # Cluster colors based on value quartiles colors = [] for v in values: ratio = v / max_val if ratio > 0.75: colors.append('#22C55E') # Green (top performers) elif ratio > 0.5: colors.append('#6366F1') # Indigo (good) elif ratio > 0.25: colors.append('#F59E0B') # Amber (medium) else: colors.append('#EF4444') # Red (low) chart = { "data": [{ "type": "scatter", "mode": "markers+text", "x": [p[0] for p in positions], "y": [p[1] for p in positions], "text": [str(e)[:15] for e in entities], "textposition": "middle center", "textfont": {"size": 9, "color": "white"}, "marker": { "size": sizes, "color": colors, "opacity": 0.85, "line": {"width": 2, "color": "white"} }, "hovertemplate": "%{text}
" + currency_symbol + "%{customdata:,.0f}", "customdata": values }], "layout": { "title": {"text": "🔮 Entity Clusters by Value", "font": {"size": 18}}, "showlegend": False, "xaxis": {"visible": False}, "yaxis": {"visible": False}, "plot_bgcolor": "rgba(0,0,0,0)", "paper_bgcolor": "rgba(0,0,0,0)", "height": 550, "margin": {"l": 20, "r": 20, "t": 60, "b": 20}, "annotations": [ {"text": "🟢 Top Performers", "x": 0.02, "y": 0.98, "xref": "paper", "yref": "paper", "showarrow": False, "font": {"size": 10}}, {"text": "🔴 Low Performers", "x": 0.02, "y": 0.93, "xref": "paper", "yref": "paper", "showarrow": False, "font": {"size": 10}} ] } } total_value = sum(values) explanation = f"""**🔮 Entity Clustering Analysis** Showing {len(entities)} {entity_col}s clustered by {value_col}. • **Total**: {currency_symbol}{total_value:,.0f} • **Top Entity**: {entities[0]} ({currency_symbol}{values[0]:,.0f}) • 🟢 Green = Top quartile, 🔴 Red = Bottom quartile """ return chart, explanation def get_best_graph_visualization( query: str, graph: Optional[nx.Graph] = None, df: Optional[pd.DataFrame] = None, currency_symbol: str = "₹" ) -> Tuple[Optional[Dict], str]: """ 🎯 INTELLIGENT GRAPH VIZ SELECTOR Automatically selects the best graph visualization based on: 1. Query keywords 2. Available data (graph vs DataFrame) 3. Data characteristics """ viz_type = detect_graph_visualization_type(query) if viz_type == 'radial' and graph: return generate_radial_mindmap(graph, query) if viz_type == 'mindmap' and graph: return generate_mindmap(graph, query) if viz_type == 'knowledge_graph' and graph: return generate_knowledge_graph(graph, query, currency_symbol=currency_symbol) if viz_type == 'cluster' and df is not None: return generate_entity_cluster(df, query, currency_symbol=currency_symbol) if viz_type == 'relationship' and df is not None: return generate_relationship_diagram(df, query, currency_symbol=currency_symbol) # Default: try knowledge graph if graph available, else cluster if graph and graph.number_of_nodes() > 0: return generate_knowledge_graph(graph, query, currency_symbol=currency_symbol) if df is not None and not df.empty: return generate_entity_cluster(df, query, currency_symbol=currency_symbol) return None, "No suitable data for graph visualization"