app.py

import gradio as gr
import pandas as pd
import networkx as nx
from pyvis.network import Network
import tempfile
import os

def calculate_centralities(df, is_directed):
    # Build NetworkX graph
    G = nx.from_pandas_edgelist(df, 'Source', 'Target', create_using=nx.DiGraph() if is_directed else nx.Graph())
    
    # Calculate centralities
    deg_cent = nx.degree_centrality(G)
    bet_cent = nx.betweenness_centrality(G)
    
    # Eigenvector has a fallback for non-convergence or directed graphs
    try:
        eig_cent = nx.eigenvector_centrality(G, max_iter=1000)
    except:
        try:
            eig_cent = nx.eigenvector_centrality_numpy(G)
        except:
            eig_cent = {node: 0.0 for node in G.nodes()}
            
    clo_cent = nx.closeness_centrality(G)
    
    # Build table
    records = []
    for node in G.nodes():
        records.append({
            "Node": node,
            "Degree Centrality": deg_cent.get(node, 0.0),
            "Betweenness Centrality": bet_cent.get(node, 0.0),
            "Eigenvector Centrality": eig_cent.get(node, 0.0),
            "Closeness Centrality": clo_cent.get(node, 0.0)
        })
        
    df_cent = pd.DataFrame(records).sort_values("Degree Centrality", ascending=False)
    return G, df_cent

def get_color_gradient(value, max_val):
    # Maps centrality value to an aesthetic gradient: low = muted brown, high = hot orange/white
    if max_val <= 0:
        return "#ff7043"
    ratio = min(value / max_val, 1.0)
    # Interpolate colors between #3d281c (wash) and #ff7043 (accent) or #ffffff
    r = int(61 + (255 - 61) * ratio)
    g = int(40 + (112 - 40) * ratio)
    b = int(28 + (67 - 28) * ratio)
    return f"#{r:02x}{g:02x}{b:02x}"

def generate_vis_html(G, df_cent, active_metric):
    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, 112, 67, 0.25)",
          "highlight": "#ff7043"
        },
        "smooth": {
          "type": "continuous"
        }
      },
      "physics": {
        "barnesHut": {
          "gravitationalConstant": -12000,
          "centralGravity": 0.3,
          "springLength": 120,
          "springConstant": 0.04
        }
      }
    }
    """)
    
    # Score dictionary
    scores = dict(zip(df_cent['Node'], df_cent[active_metric]))
    max_score = max(scores.values()) if scores else 1.0
    
    for node in G.nodes():
        score = scores.get(node, 0.0)
        # Sizing logic: baseline = 10, scaled up to max 45
        size = 10 + (35 * (score / max_score if max_score > 0 else 0))
        color = get_color_gradient(score, max_score)
        
        net.add_node(
            node, 
            label=node, 
            size=size, 
            color=color, 
            title=f"Centrality Score: {score:.5f}"
        )
        
    # Add edges
    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('"', '&quot;')
    iframe_code = f'<iframe srcdoc="{escaped_html}" style="width: 100%; height: 530px; border: 1px solid rgba(255, 255, 255, 0.08); border-radius: 8px;"></iframe>'
    return iframe_code

def analyze_centrality(file_obj, is_directed, active_metric):
    if file_obj is None:
        return "Please upload a CSV or Excel network dataset.", "", 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
        
    # 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
        
    # Calculate scores
    G, df_cent = calculate_centralities(df, is_directed)
    
    # General stats
    stats_html = f"""
    <div style='display: grid; grid-template-columns: repeat(2, 1fr); gap: 1rem; margin-bottom: 1rem;'>
        <div style='background: rgba(255, 255, 255, 0.03); border: 1px solid rgba(255, 255, 255, 0.08); border-radius: 8px; padding: 1rem; text-align: center;'>
            <div style='font-size: 0.75rem; text-transform: uppercase; color: #ff7043; letter-spacing: 0.1em;'>Network Nodes</div>
            <div style='font-size: 2rem; font-weight: bold; margin-top: 0.5rem;'>{G.number_of_nodes()}</div>
        </div>
        <div style='background: rgba(255, 255, 255, 0.03); border: 1px solid rgba(255, 255, 255, 0.08); border-radius: 8px; padding: 1rem; text-align: center;'>
            <div style='font-size: 0.75rem; text-transform: uppercase; color: #ff7043; letter-spacing: 0.1em;'>Network Edges</div>
            <div style='font-size: 2rem; font-weight: bold; margin-top: 0.5rem;'>{G.number_of_edges()}</div>
        </div>
    </div>
    """
    
    # Generate PyVis HTML
    vis_html = generate_vis_html(G, df_cent, active_metric)
    
    # Sort for displaying
    display_df = df_cent.sort_values(active_metric, ascending=False)
    
    # Download scores CSV
    out_csv = tempfile.mktemp(suffix=".csv")
    df_cent.to_csv(out_csv, index=False)
    
    return "", stats_html, vis_html, display_df, 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="Centrality Analysis") as demo:
    gr.Markdown(
        """
        # 👑 Network Centrality Analysis Suite
        ### Quantify node influence and structural power inside complex networks using four classical centrality algorithms. Drag, zoom, and visualize node importance dynamically!
        """
    )
    
    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)
            
            active_metric = gr.Radio(
                choices=["Degree Centrality", "Betweenness Centrality", "Eigenvector Centrality", "Closeness Centrality"],
                value="Degree Centrality",
                label="Centrality Measure",
                info="Degree (total links), Betweenness (brokerage), Eigenvector (influence of connections), Closeness (distance)."
            )
            
            btn = gr.Button("Calculate Centrality Rankings", variant="primary")
            
        with gr.Column(scale=2):
            stats_box = gr.HTML()
            
            with gr.Tabs():
                with gr.TabItem("Interactive Graph Scaling"):
                    vis_box = gr.HTML()
                with gr.TabItem("Rankings Table"):
                    table_box = gr.Dataframe(headers=["Node", "Degree Centrality", "Betweenness Centrality", "Eigenvector Centrality", "Closeness Centrality"])
                    download_btn = gr.File(label="Download Calculated Rankings CSV", visible=False)

    def process(file_obj, is_directed, metric):
        err, stats, vis, table, csv_path = analyze_centrality(file_obj, is_directed, metric)
        if err:
            return gr.update(value=err, visible=True), "", "", None, gr.update(visible=False)
        return gr.update(visible=False), stats, vis, table, csv_path

    btn.click(
        process,
        inputs=[file_obj, is_directed, active_metric],
        outputs=[error_msg, stats_box, vis_box, table_box, download_btn]
    )

if __name__ == "__main__":
    demo.launch()