import gradio as gr
import pandas as pd
import networkx as nx
import networkx.community as nx_comm
from pyvis.network import Network
import tempfile
import os
# Premium color palette for nodes grouped by community
COMMUNITY_COLORS = [
"#ff7043", # Hot Orange
"#4db6ac", # Teal Accent
"#9575cd", # Deep Purple
"#ffd54f", # Warm Yellow
"#64b5f6", # Ocean Blue
"#f06292", # Vivid Pink
"#81c784", # Forest Green
"#ffffff", # Pure White
"#a1887f", # Warm Clay
"#ba68c8" # Violet
]
def run_louvain_communities(df, is_directed, seed_val=42):
G = nx.from_pandas_edgelist(df, 'Source', 'Target', create_using=nx.DiGraph() if is_directed else nx.Graph())
# Louvain communities require undirected graph for computation in standard networkx
G_undirected = G.to_undirected() if is_directed else G
try:
communities = nx_comm.louvain_communities(G_undirected, seed=seed_val)
# Calculate modularity
modularity = nx_comm.modularity(G_undirected, communities)
except Exception as e:
# Fallback to label propagation if Louvain fails for any reason
communities = list(nx_comm.label_propagation_communities(G_undirected))
modularity = nx_comm.modularity(G_undirected, communities)
# Build results table
node_to_community = {}
community_records = []
for comm_idx, comm in enumerate(communities):
comm_color = COMMUNITY_COLORS[comm_idx % len(COMMUNITY_COLORS)]
members = sorted(list(comm))
members_str = ", ".join(members)
community_records.append({
"Community ID": comm_idx + 1,
"Node Count": len(comm),
"Members": members_str
})
for node in comm:
node_to_community[node] = {
"Community ID": comm_idx + 1,
"Color": comm_color
}
df_comm_summary = pd.DataFrame(community_records).sort_values("Node Count", ascending=False)
node_records = []
for node, info in node_to_community.items():
node_records.append({
"Node": node,
"Community ID": info["Community ID"]
})
df_nodes = pd.DataFrame(node_records).sort_values("Community ID")
return G, df_comm_summary, df_nodes, node_to_community, modularity
def generate_vis_html(G, node_to_community):
net = Network(
height="500px",
width="100%",
bgcolor="#16100c",
font_color="#f4eee6",
notebook=False
)
net.set_options("""
var options = {
"nodes": {
"borderWidth": 2,
"font": {
"color": "#f4eee6",
"size": 14,
"face": "Inter, sans-serif"
}
},
"edges": {
"color": {
"color": "rgba(255, 255, 255, 0.15)",
"highlight": "#ff7043"
},
"smooth": {
"type": "continuous"
}
},
"physics": {
"barnesHut": {
"gravitationalConstant": -12000,
"centralGravity": 0.3,
"springLength": 120,
"springConstant": 0.04
}
}
}
""")
degrees = dict(G.degree())
for node in G.nodes():
deg = degrees.get(node, 1)
size = 10 + (deg * 1.5)
info = node_to_community.get(node, {"Community ID": 0, "Color": "#ff7043"})
net.add_node(
node,
label=node,
size=size,
color=info["Color"],
title=f"Community ID: {info['Community ID']} | Connections: {deg}"
)
for edge in G.edges():
net.add_edge(edge[0], edge[1])
temp_dir = tempfile.gettempdir()
temp_path = os.path.join(temp_dir, next(tempfile._get_candidate_names()) + ".html")
net.save_graph(temp_path)
with open(temp_path, "r", encoding="utf-8") as f:
html_content = f.read()
try:
os.remove(temp_path)
except:
pass
escaped_html = html_content.replace('"', '"')
iframe_code = f''
return iframe_code
def analyze_communities(file_obj, is_directed):
if file_obj is None:
return "Please upload a CSV or Excel network dataset.", "", None, None, None, None
try:
if file_obj.name.endswith('.csv'):
df = pd.read_csv(file_obj.name)
else:
df = pd.read_excel(file_obj.name)
except Exception as e:
return f"Error reading file: {str(e)}", "", None, None, None, None
# Standardize column headers
rename_map = {}
for col in df.columns:
if col.lower() in ['source', 'from', 'node1']:
rename_map[col] = 'Source'
elif col.lower() in ['target', 'to', 'node2']:
rename_map[col] = 'Target'
df = df.rename(columns=rename_map)
if 'Source' not in df.columns or 'Target' not in df.columns:
return "CSV/Excel must contain at least 'Source' and 'Target' columns representing network edges.", "", None, None, None, None
# Calculate
G, df_comm_summary, df_nodes, node_to_community, modularity = run_louvain_communities(df, is_directed)
# Stats
stats_html = f"""
Communities Detected
{len(df_comm_summary)}
Modularity Score
{modularity:.4f}
Total Nodes
{G.number_of_nodes()}
"""
# Generate PyVis HTML
vis_html = generate_vis_html(G, node_to_community)
# Download nodes CSV
out_csv = tempfile.mktemp(suffix=".csv")
df_nodes.to_csv(out_csv, index=False)
return "", stats_html, vis_html, df_comm_summary, df_nodes, gr.update(value=out_csv, visible=True)
theme = gr.themes.Default(
primary_hue="orange",
neutral_hue="stone"
).set(
body_background_fill="#0d0907",
body_text_color="#c4bbae",
block_background_fill="#16100c",
block_border_width="1px",
block_label_text_color="#f4eee6"
)
with gr.Blocks(theme=theme, title="Community Detection") as demo:
gr.Markdown(
"""
# 👥 Network Community & Subgroup Detector
### Run structural community partitioning (Louvain Modularity) on network datasets. Automatically group nodes into clusters and color-code subgroups!
"""
)
error_msg = gr.Markdown("", visible=False)
with gr.Row():
with gr.Column(scale=1):
file_obj = gr.File(label="Upload CSV or Excel Network File", file_types=[".csv", ".xlsx"])
is_directed = gr.Checkbox(label="Is Directed Network", value=False)
btn = gr.Button("Detect Network Communities", variant="primary")
gr.Markdown(
"""
💡 **How Modularity Works**: Modularity ranges between `-0.5` and `1.0`.
Scores above `0.3` suggest a very strong, well-separated community structure where nodes cluster in tight cliques.
"""
)
with gr.Column(scale=2):
stats_box = gr.HTML()
with gr.Tabs():
with gr.TabItem("Interactive Subgroups Map"):
vis_box = gr.HTML()
with gr.TabItem("Communities List"):
table_summary = gr.Dataframe(headers=["Community ID", "Node Count", "Members"])
with gr.TabItem("Node Grouping Table"):
table_nodes = gr.Dataframe(headers=["Node", "Community ID"])
download_btn = gr.File(label="Download Community Assignments CSV", visible=False)
def process(file_obj, is_directed):
err, stats, vis, summary, nodes, csv_path = analyze_communities(file_obj, is_directed)
if err:
return gr.update(value=err, visible=True), "", "", None, None, gr.update(visible=False)
return gr.update(visible=False), stats, vis, summary, nodes, csv_path
btn.click(
process,
inputs=[file_obj, is_directed],
outputs=[error_msg, stats_box, vis_box, table_summary, table_nodes, download_btn]
)
if __name__ == "__main__":
demo.launch()