"""
Interactive Curriculum Visualizer - FIXED VERSION
Creates CurricularAnalytics-style network graphs
"""
import streamlit as st
import networkx as nx
import plotly.graph_objects as go
import pickle
import json
from typing import Dict, List, Tuple
import numpy as np
class CurriculumVisualizer:
"""
Creates interactive curriculum dependency graphs
Similar to CurricularAnalytics.org
"""
def __init__(self, graph: nx.DiGraph):
self.graph = graph
self.courses = dict(graph.nodes(data=True))
self.positions = None
self.layers = None
def calculate_metrics(self, course_id: str) -> Dict:
"""Calculate blocking factor, delay factor, centrality"""
# Blocking Factor: courses this blocks
blocking = len(list(nx.descendants(self.graph, course_id)))
# Correctly calculate the delay factor by finding the longest path
delay = 0
sinks = [n for n, d in self.graph.out_degree() if d == 0]
max_len = 0
for sink in sinks:
try:
paths = list(nx.all_simple_paths(self.graph, source=course_id, target=sink))
if paths:
current_max = max(len(p) for p in paths)
if current_max > max_len:
max_len = current_max
except (nx.NetworkXNoPath, nx.NodeNotFound):
continue
delay = max_len
# Centrality: betweenness centrality
centrality_dict = nx.betweenness_centrality(self.graph)
centrality = centrality_dict.get(course_id, 0) * 100
# Complexity (from your analyzer)
complexity = self.courses[course_id].get('complexity', 0)
return {
'blocking': blocking,
'delay': delay,
'centrality': round(centrality, 1),
'complexity': complexity
}
def create_hierarchical_layout(self) -> Dict:
"""Create semester-based layout like CurricularAnalytics"""
# Topological sort to get course ordering
try:
topo_order = list(nx.topological_sort(self.graph))
except nx.NetworkXError:
# Has cycles, use DFS order
topo_order = list(nx.dfs_preorder_nodes(self.graph))
# Calculate depth for each node (semester level)
depths = {}
for node in topo_order:
predecessors = list(self.graph.predecessors(node))
if not predecessors:
depths[node] = 0
else:
depths[node] = max(depths.get(p, 0) for p in predecessors) + 1
# Group by depth (semester)
layers = {}
for node, depth in depths.items():
if depth not in layers:
layers[depth] = []
layers[depth].append(node)
# Create positions
positions = {}
max_width = max(len(nodes) for nodes in layers.values()) if layers else 1
for depth, nodes in layers.items():
width = len(nodes)
spacing = 2.0 / (width + 1) if width > 0 else 1
for i, node in enumerate(nodes):
x = (i + 1) * spacing - 1 # Center around 0
y = -depth * 2 # Vertical spacing
positions[node] = (x, y)
self.positions = positions
self.layers = layers
return positions
def create_interactive_plot(self, highlight_path: List[str] = None) -> go.Figure:
"""Create Plotly interactive network graph"""
if not self.positions:
self.create_hierarchical_layout()
# Create edge traces
edge_traces = []
for edge in self.graph.edges():
if edge[0] not in self.positions or edge[1] not in self.positions:
continue
x0, y0 = self.positions[edge[0]]
x1, y1 = self.positions[edge[1]]
# Check if edge is on critical path
is_critical = False
if highlight_path and edge[0] in highlight_path and edge[1] in highlight_path:
try:
idx0 = highlight_path.index(edge[0])
idx1 = highlight_path.index(edge[1])
is_critical = idx1 == idx0 + 1
except ValueError:
is_critical = False
edge_trace = go.Scatter(
x=[x0, x1, None],
y=[y0, y1, None],
mode='lines',
line=dict(
width=3 if is_critical else 1,
color='red' if is_critical else '#888'
),
hoverinfo='none',
showlegend=False
)
edge_traces.append(edge_trace)
# Create node trace
node_x = []
node_y = []
node_text = []
node_color = []
node_size = []
for node in self.graph.nodes():
if node not in self.positions:
continue
x, y = self.positions[node]
node_x.append(x)
node_y.append(y)
# Get course info
course_data = self.courses.get(node, {})
metrics = self.calculate_metrics(node)
# Create hover text
hover_text = f"""
{node}: {course_data.get('name', 'Unknown')}
Credits: {course_data.get('credits', 4)}
Metrics:
Complexity: {metrics['complexity']}
Blocking Factor: {metrics['blocking']}
Delay Factor: {metrics['delay']}
Centrality: {metrics['centrality']}
Prerequisites: {', '.join(self.graph.predecessors(node)) or 'None'}
"""
node_text.append(hover_text)
# Color by complexity
node_color.append(metrics['complexity'])
# Size by blocking factor
node_size.append(15 + metrics['blocking'] * 2)
node_trace = go.Scatter(
x=node_x,
y=node_y,
mode='markers+text',
text=[node for node in self.graph.nodes() if node in self.positions],
textposition="top center",
textfont=dict(size=10),
hovertext=node_text,
hoverinfo='text',
marker=dict(
showscale=True,
colorscale='Viridis',
size=node_size,
color=node_color,
colorbar=dict(
thickness=15,
title=dict(text="Complexity", side="right"),
xanchor="left"
),
line=dict(width=2, color='white')
)
)
# Create figure
fig = go.Figure(data=edge_traces + [node_trace])
# FIXED: Updated layout with proper title syntax (no more titlefont_size)
fig.update_layout(
title=dict(
text="Interactive Curriculum Map",
font=dict(size=20)
),
showlegend=False,
hovermode='closest',
margin=dict(b=0, l=0, r=0, t=40),
xaxis=dict(showgrid=False, zeroline=False, showticklabels=False),
yaxis=dict(showgrid=False, zeroline=False, showticklabels=False),
plot_bgcolor='white',
height=800
)
return fig
def find_critical_path(self) -> List[str]:
"""Find the longest path (critical path) in curriculum"""
if not nx.is_directed_acyclic_graph(self.graph):
return []
# Find all paths from sources to sinks
sources = [n for n in self.graph.nodes() if self.graph.in_degree(n) == 0]
sinks = [n for n in self.graph.nodes() if self.graph.out_degree(n) == 0]
longest_path = []
max_length = 0
for source in sources:
for sink in sinks:
try:
paths = list(nx.all_simple_paths(self.graph, source, sink))
for path in paths:
if len(path) > max_length:
max_length = len(path)
longest_path = path
except nx.NetworkXNoPath:
continue
return longest_path
def export_to_curricular_analytics_format(self, plan: Dict) -> Dict:
"""Export plan in CurricularAnalytics JSON format"""
ca_format = {
"curriculum": {
"name": "Generated Curriculum",
"courses": [],
"dependencies": []
},
"metrics": {}
}
# Add courses
for course_id in self.graph.nodes():
course_data = self.courses.get(course_id, {})
metrics = self.calculate_metrics(course_id)
ca_format["curriculum"]["courses"].append({
"id": course_id,
"name": course_data.get('name', ''),
"credits": course_data.get('credits', 4),
"complexity": metrics['complexity'],
"blocking_factor": metrics['blocking'],
"delay_factor": metrics['delay'],
"centrality": metrics['centrality']
})
# Add dependencies
for edge in self.graph.edges():
ca_format["curriculum"]["dependencies"].append({
"source": edge[0],
"target": edge[1],
"type": "prerequisite"
})
return ca_format
def run_visualizer():
"""Streamlit app for visualization"""
st.set_page_config(page_title="Curriculum Visualizer", layout="wide")
st.title("πΊοΈ Interactive Curriculum Visualizer")
# Sidebar
with st.sidebar:
st.header("Controls")
# File upload
uploaded_file = st.file_uploader("Upload curriculum graph", type=['pkl'])
# Display options
show_critical = st.checkbox("Highlight Critical Path", value=True)
show_metrics = st.checkbox("Show Metrics Panel", value=True)
# Filter options
st.subheader("Filter Courses")
min_complexity = st.slider("Min Complexity", 0, 200, 0)
subjects = st.multiselect("Subjects", ["CS", "DS", "MATH", "IS", "CY"])
# Main content
if uploaded_file:
# Load graph
graph = pickle.load(uploaded_file)
visualizer = CurriculumVisualizer(graph)
# Apply filters
if subjects:
nodes_to_keep = [
n for n in graph.nodes()
if graph.nodes[n].get('subject') in subjects
]
filtered_graph = graph.subgraph(nodes_to_keep).copy()
visualizer = CurriculumVisualizer(filtered_graph)
# Create visualization
col1, col2 = st.columns([3, 1] if show_metrics else [1])
with col1:
# Find critical path
critical_path = []
if show_critical:
critical_path = visualizer.find_critical_path()
if critical_path:
st.info(f"Critical Path: {' β '.join(critical_path[:5])}...")
# Create and display plot
fig = visualizer.create_interactive_plot(critical_path)
st.plotly_chart(fig, use_container_width=True)
if show_metrics:
with col2:
st.subheader("π Curriculum Metrics")
# Overall metrics
total_courses = visualizer.graph.number_of_nodes()
total_prereqs = visualizer.graph.number_of_edges()
st.metric("Total Courses", total_courses)
st.metric("Total Prerequisites", total_prereqs)
if total_courses > 0:
st.metric("Avg Prerequisites", f"{total_prereqs/total_courses:.1f}")
st.divider()
# Most complex courses
st.subheader("Most Complex Courses")
complexities = []
for node in visualizer.graph.nodes():
metrics = visualizer.calculate_metrics(node)
complexities.append((node, metrics['complexity']))
complexities.sort(key=lambda x: x[1], reverse=True)
for course, complexity in complexities[:5]:
name = visualizer.courses.get(course, {}).get('name', course)
st.write(f"**{course}**: {name}")
st.progress(min(complexity/200, 1.0) if complexity else 0.0)
# Export button
st.divider()
if st.button("Export to CA Format"):
ca_json = visualizer.export_to_curricular_analytics_format({})
st.download_button(
"Download JSON",
json.dumps(ca_json, indent=2),
"curriculum_analytics.json",
"application/json"
)
else:
# Demo/instruction
st.info("Upload a curriculum graph file to visualize")
with st.expander("About this Visualizer"):
st.write("""
This tool creates interactive curriculum dependency graphs similar to CurricularAnalytics.org.
**Features:**
- Hierarchical layout by semester level
- Color coding by complexity
- Node size by blocking factor
- Critical path highlighting
- Interactive hover details
- Export to CurricularAnalytics format
**Metrics Calculated:**
- **Blocking Factor**: Number of courses this prerequisite blocks
- **Delay Factor**: Length of longest path through this course
- **Centrality**: Importance in the curriculum network
- **Complexity**: Combined metric from all factors
""")
if __name__ == "__main__":
run_visualizer()