""" VOSviewer-faithful bibliometric network renderer. Implements render_coauthorship_network, render_keyword_network, and render_topic_network using PyVis embedded in Streamlit via st.components.v1.html(). Every visual specification in Phase 4 Section B is implemented here. """ import json import math import re from statistics import median from typing import Any, Dict, List, Optional, Tuple import networkx as nx import hashlib as _hashlib try: with open(__file__, "rb") as _f: _VIZ_VERSION = "v" + _hashlib.md5(_f.read()).hexdigest()[:10] except Exception: _VIZ_VERSION = "v1" from config import ( CANVAS_BG, COMMUNITY_COLORS, NODE_SIZE_MIN, NODE_SIZE_MAX, EDGE_WIDTH_MIN, EDGE_WIDTH_MAX, COLOR_SURFACE_ELEVATED, COLOR_TEXT_SECONDARY, COLOR_PRIMARY, ) from utils.helpers import ( lighten_hex, hex_to_rgba, scale_node_size, scale_edge_width, truncate, format_author_short, percentile, ) # ── Physics options ─────────────────────────────────────────────────────────── def get_physics_options(node_count: int, network_type: str = "default") -> Dict: if node_count < 50: grav = -3000 spring = 200 overlap = 0.8 elif node_count > 500: grav = -8000 spring = 100 overlap = 1.0 else: grav = -5000 spring = 150 overlap = 0.9 # Co-authorship network: use forceAtlas2Based solver instead of barnesHut. # barnesHut applies global pairwise repulsion between ALL nodes — this pushes # intra-cluster nodes into polygon rings regardless of avoidOverlap settings. # forceAtlas2Based uses hub-weighted repulsion: high-degree nodes repel more, # which naturally separates communities while keeping cluster members in # tight organic blobs (not rings). centralGravity=0.01 keeps all clusters # in one compact mass without scattering them across the canvas. if network_type == "coauthorship": return { "layout": { "improvedLayout": False, }, "physics": { "enabled": True, "solver": "forceAtlas2Based", "forceAtlas2Based": { # springLength=1 collapses connected nodes toward the same # point; avoidOverlap=1.0 then stops them at exact touching # distance (sum of radii). This creates a unique BLOB # equilibrium — nodes pack at contact — rather than the ring # equilibrium produced by large springLength where nodes # settle equidistant around a hub. Since node sizes vary # (paper-count scaled 18-85), touching distances differ per # pair, naturally breaking symmetry. "gravitationalConstant": -20, "centralGravity": 0.02, "springLength": 1, "springConstant": 0.08, "damping": 0.4, "avoidOverlap": 1.0, }, "maxVelocity": 80, "minVelocity": 0.3, "stabilization": { "enabled": True, "iterations": 600, "updateInterval": 10, "fit": True, }, "timestep": 0.2, }, "interaction": { "hover": True, "tooltipDelay": 150, "hideEdgesOnDrag": True, "hideEdgesOnZoom": False, "multiselect": True, "navigationButtons": False, "keyboard": {"enabled": False}, "zoomView": True, }, "nodes": { "chosen": True, "physics": True, "shadow": False, "font": { "size": 20, "color": "#000000", "strokeWidth": 2, "strokeColor": "#FFFFFF", "vadjust": 0, }, }, "edges": { "chosen": True, "physics": True, "hoverWidth": 2.5, "selectionWidth": 3.0, "smooth": {"type": "continuous", "roundness": 0.1}, }, } # Keyword networks: zero centralGravity + balanced spring/repulsion. # centralGravity=0 prevents circular ring. Stronger repulsion + longer softer # springs spread clusters wide across canvas without intra-cluster overlap. elif network_type == "keyword": grav = -55000 # strong repulsion separates clusters across canvas central_grav = 0.0 # zero — topology drives placement, no ring force spring = 220 # longer: gives nodes within each cluster room spring_const = 0.05 # soft: cluster structure without squashing nodes damping = 0.10 overlap = 1.0 iterations = 6000 timestep = 0.20 max_vel = 100 min_vel = 0.10 else: central_grav = 0.15 spring_const = 0.04 damping = 0.12 iterations = 2000 timestep = 0.35 max_vel = 60 min_vel = 0.3 return { "physics": { "enabled": True, "barnesHut": { "gravitationalConstant": grav, "centralGravity": central_grav, "springLength": spring, "springConstant": spring_const, "damping": damping, "avoidOverlap": overlap, }, "maxVelocity": max_vel, "minVelocity": min_vel, "stabilization": { "enabled": True, "iterations": iterations, "updateInterval": 25, "fit": True, }, "timestep": timestep, }, "interaction": { "hover": True, "tooltipDelay": 150, "hideEdgesOnDrag": True, "hideEdgesOnZoom": False, "multiselect": True, "navigationButtons": False, "keyboard": {"enabled": False}, "zoomView": True, }, "nodes": { "chosen": True, "physics": True, "shadow": False, "font": { "size": 22 if network_type == "keyword" else 14, "color": "#000000", "strokeWidth": 2, "strokeColor": "#FFFFFF", "vadjust": -40 if network_type == "keyword" else 0, }, }, "edges": { "chosen": True, "physics": True, "hoverWidth": 2.5, "selectionWidth": 3.0, "smooth": {"type": "continuous", "roundness": 0.1}, }, } # ── Tooltip container style ─────────────────────────────────────────────────── _TOOLTIP_STYLE = ( "background:#1C2539;border:1px solid #3B4A6B;border-radius:8px;" "padding:10px 14px;font-family:'Open Sans',Arial,sans-serif;" "font-size:12px;color:#F0F4FF;max-width:260px;" "box-shadow:0 4px 16px rgba(0,0,0,0.5);line-height:1.6;" ) def _wrap_tooltip(content: str) -> str: return f'
{content}
' # ── Node sizing helpers ─────────────────────────────────────────────────────── def _compute_node_sizes( graph: nx.Graph, weight_attr: str = "weight", size_min: Optional[float] = None, size_max: Optional[float] = None, ) -> Dict[Any, float]: if size_min is None: size_min = NODE_SIZE_MIN if size_max is None: size_max = NODE_SIZE_MAX weights = {n: graph.nodes[n].get(weight_attr, 1) for n in graph.nodes()} w_min = min(weights.values(), default=1) w_max = max(weights.values(), default=1) return { n: scale_node_size(w, w_min, w_max, size_min, size_max) for n, w in weights.items() } def _compute_edge_widths( graph: nx.Graph, width_min: Optional[float] = None, width_max: Optional[float] = None, ) -> Dict[Tuple, float]: if width_min is None: width_min = EDGE_WIDTH_MIN if width_max is None: width_max = EDGE_WIDTH_MAX edge_weights = { (u, v): graph[u][v].get("weight", 1) for u, v in graph.edges() } if not edge_weights: return {} w_min = min(edge_weights.values()) w_max = max(edge_weights.values()) return { edge: scale_edge_width(w, w_min, w_max, width_min, width_max) for edge, w in edge_weights.items() } # ── Label visibility ────────────────────────────────────────────────────────── def _label_font(weight: float, p50: float, p75: float) -> Dict: if weight >= p75: return {"size": 16, "color": "#000000", "face": "arial", "strokeWidth": 3, "strokeColor": "#FFFFFF"} if weight >= p50: return {"size": 13, "color": "#000000", "face": "arial", "strokeWidth": 2, "strokeColor": "#FFFFFF"} return {"size": 11, "color": "#000000", "face": "arial", "strokeWidth": 1, "strokeColor": "#FFFFFF"} # ── PyVis HTML post-processing ──────────────────────────────────────────────── _KEYWORD_FIT_JS = """ """ _STABILIZE_JS = """ """ _HIGHLIGHT_JS = """ """ _CONTROLS_JS = """ """ _LABEL_OVERLAP_JS = """ """ def _post_process_html(html: str, node_count: int = 0, network_type: str = "default") -> str: for old in [ 'background-color: #ffffff;', 'background-color:#ffffff;', 'background-color: white;', 'background: white;', 'background:#ffffff;', 'background: #ffffff;', ]: html = html.replace(old, f'background:{CANVAS_BG};') html = html.replace(old.upper(), f'background:{CANVAS_BG};') html = re.sub( r'(#mynetwork\s*\{[^}]*)', rf'\1background:{CANVAS_BG} !important;', html, flags=re.DOTALL, ) # Centre the PyVis loading bar — override hard-coded top:400px with flexbox _loading_bar_css = ( "" ) overlap_js = _LABEL_OVERLAP_JS if network_type == "keyword" else "" keyword_fit_js = _KEYWORD_FIT_JS if network_type == "keyword" else "" html = html.replace( "", _loading_bar_css + _STABILIZE_JS + _HIGHLIGHT_JS + _CONTROLS_JS + overlap_js + keyword_fit_js + "", ) return html def _pyvis_to_html(net, node_count: int = 0, network_type: str = "default") -> str: """Generate PyVis HTML string (no disk write).""" html = net.generate_html(notebook=False) return _post_process_html(html, node_count, network_type) # ── Controls panel ──────────────────────────────────────────────────────────── def _render_controls( graph: nx.Graph, key_prefix: str, session_key_html: str, rebuild_fn, show_search: bool = True, show_density: bool = True, show_freeze: bool = True, show_png: bool = True, ) -> Tuple[str, float, float, Optional[List[int]], bool]: """ Render the controls strip above a network graph. Returns (search_term, min_link, min_size, selected_communities, freeze). show_search / show_density / show_freeze let callers hide specific controls; hidden controls return their default values (empty string / False). """ import streamlit as st # Build column widths dynamically based on which controls are visible. col_widths = [] if show_search: col_widths.append(3) # search col_widths.extend([2, 2, 2]) # min_link, min_size, communities always shown if show_density: col_widths.append(1) if show_freeze: col_widths.append(1) if show_png: col_widths.append(1) # PNG (optional) cols = st.columns(col_widths) ci = 0 # column index cursor if show_search: with cols[ci]: search = st.text_input( "Search nodes…", key=f"{key_prefix}_search", placeholder="Search nodes…", label_visibility="collapsed", ) ci += 1 else: search = "" all_edge_weights = [ graph[u][v].get("weight", 1) for u, v in graph.edges() ] edge_p90 = percentile(all_edge_weights, 90) if all_edge_weights else 1.0 with cols[ci]: min_link = st.slider( "Min Link Strength", min_value=1, max_value=max(int(edge_p90), 2), value=1, key=f"{key_prefix}_min_link", ) ci += 1 all_node_weights = [ graph.nodes[n].get("weight", 1) for n in graph.nodes() ] node_p75 = percentile(all_node_weights, 75) if all_node_weights else 1.0 with cols[ci]: min_size = st.slider( "Min Node Size", min_value=1, max_value=max(int(node_p75), 2), value=1, key=f"{key_prefix}_min_size", ) ci += 1 all_communities = sorted( set( graph.nodes[n].get("community_id", 0) for n in graph.nodes() ) ) with cols[ci]: selected_comms = st.multiselect( "Communities", options=all_communities, default=all_communities, key=f"{key_prefix}_comms", ) ci += 1 if show_density: with cols[ci]: density_on = st.checkbox("Density", key=f"{key_prefix}_density") ci += 1 if show_freeze: with cols[ci]: freeze = st.checkbox("❄ Freeze", key=f"{key_prefix}_freeze") ci += 1 else: freeze = False if show_png: with cols[ci]: export_png = st.button("↗ PNG", key=f"{key_prefix}_export") if export_png: st.info("Right-click the graph → Save image as… to export PNG.") return search, float(min_link), float(min_size), selected_comms, freeze # ── Network rendering core ──────────────────────────────────────────────────── def _build_pyvis_network( graph: nx.Graph, node_sizes: Dict, edge_widths: Dict, node_weights: Dict, label_fn, tooltip_fn, edge_tooltip_fn, directed: bool = False, shape_fn=None, edge_alpha: float = 0.55, edge_roundness: float = 0.10, font_size_boost: int = 0, node_opacity: float = 1.0, network_type: str = "default", ) -> Any: """ Build a PyVis Network object from a NetworkX graph with full VOSviewer-faithful styling applied. """ try: from pyvis.network import Network except ImportError as exc: raise ImportError("pyvis is not installed. Run: pip install pyvis") from exc w_list = list(node_weights.values()) p50 = percentile(w_list, 50) if w_list else 1.0 p75 = percentile(w_list, 75) if w_list else 1.0 net = Network( height="850px", width="100%", directed=directed, bgcolor=CANVAS_BG, font_color="#000000", ) net.toggle_physics(True) for node in graph.nodes(): data = graph.nodes[node] fill_hex = data.get("color_hex", COMMUNITY_COLORS[0]) size = node_sizes.get(node, NODE_SIZE_MIN) weight = node_weights.get(node, 1) label = label_fn(node, data) # Note: keyword network label overlap is handled by _LABEL_OVERLAP_JS # injected into the HTML — no static hiding here. tooltip = tooltip_fn(node, data, graph) shape = shape_fn(data) if shape_fn else "dot" if network_type == "keyword": # VOSviewer-style: font size scales with visual node size. # Hub nodes keep large prominent labels; smaller nodes get # proportionally smaller fonts that don't bleed onto neighbours. node_size_val = node_sizes.get(node, NODE_SIZE_MIN) font_px = max(10, min(30, int(node_size_val * 0.33))) stroke_w = 3 if font_px >= 20 else (2 if font_px >= 14 else 1) font = { "size": font_px, "color": "#000000", "face": "arial", "strokeWidth": stroke_w, "strokeColor": "#FFFFFF", } else: font = _label_font(weight, p50, p75) if font_size_boost: font = {**font, "size": font["size"] + font_size_boost} bg = hex_to_rgba(fill_hex, node_opacity) if node_opacity < 1.0 else fill_hex node_color = { "background": bg, "border": bg, "highlight": {"background": bg, "border": "#000000"}, "hover": {"background": bg, "border": "#333333"}, } net.add_node( node, label=label, title=tooltip, size=size, shape=shape, color=node_color, borderWidth=0, borderWidthSelected=0, font=font, ) for u, v in graph.edges(): src_community = graph.nodes[u].get("community_id", 0) src_color_hex = graph.nodes[u].get("color_hex", COMMUNITY_COLORS[0]) edge_color = { "color": hex_to_rgba(src_color_hex, edge_alpha), "highlight": hex_to_rgba(src_color_hex, min(1.0, edge_alpha + 0.45)), "hover": hex_to_rgba(src_color_hex, min(1.0, edge_alpha + 0.30)), "inherit": False, } width = edge_widths.get((u, v), EDGE_WIDTH_MIN) edge_data = graph[u][v] if isinstance(graph, nx.Graph) else {} tooltip = edge_tooltip_fn(u, v, edge_data) net.add_edge( u, v, width=width, color=edge_color, title=tooltip, arrows="" if not directed else "to", smooth={"type": "continuous", "roundness": edge_roundness} if not directed else {"type": "curvedCW", "roundness": edge_roundness}, ) physics_opts = get_physics_options(graph.number_of_nodes(), network_type) net.set_options(json.dumps(physics_opts)) return net def _default_edge_tooltip(u, v, data) -> str: weight = data.get("weight", 1) pmids = data.get("evidence_pmids", []) pmid_str = ", ".join(str(p) for p in pmids[:3]) if pmids else "N/A" content = ( f'
{u} ↔ {v}
' f'
Co-occurrence: {weight} papers
' f'
' f'PMIDs: {pmid_str}
' ) return _wrap_tooltip(content) # ── A) Co-authorship network ────────────────────────────────────────────────── def render_coauthorship_network( graph: nx.Graph, papers_df=None, key_prefix: str = "coauth", ): """ Render the Author Collaboration Network. Section title + subtitle + controls panel + PyVis graph + stats panel. """ import streamlit as st st.markdown( "### 01 — Author Collaboration Network\n" "" "Node size = publication count  |  " "Color = research cluster  |  " "Edge thickness = collaboration strength" "", unsafe_allow_html=True, ) st.markdown("
", unsafe_allow_html=True) if graph is None or graph.number_of_nodes() == 0: st.info("No co-authorship data available. Run the pipeline first.") return search, min_link, min_size, sel_comms, freeze = _render_controls( graph, key_prefix, f"_coauth_html", lambda g: g, show_search=False, show_density=False, show_freeze=False, show_png=False, ) # Apply filters filtered = _filter_graph(graph, min_link, min_size, sel_comms, search) cache_key = ( f"_coauth_html_{_VIZ_VERSION}_{key_prefix}_{min_link}_{min_size}_" f"{','.join(map(str, sel_comms))}_{search}_{freeze}_coauth" ) if cache_key not in st.session_state or st.session_state[cache_key] is None: node_sizes = _compute_node_sizes(filtered, size_min=18, size_max=85) edge_widths = _compute_edge_widths(filtered) node_weights = {n: filtered.nodes[n].get("weight", 1) for n in filtered.nodes()} def label_fn(node, data): return format_author_short(str(node)) def tooltip_fn(node, data, g): deg = g.degree(node) paper_count = data.get("weight", 0) cid = data.get("community_id", 0) neighbors = sorted( g.neighbors(node), key=lambda nb: g[node][nb].get("weight", 0), reverse=True, )[:5] collab_list = "".join( f"
• {format_author_short(nb)}
" for nb in neighbors ) content = ( f'
' f'{node}
' f'
' f'Research Cluster {cid}
' f'
' f'Papers: ' f'{paper_count}
' f'
' f'Collaborations: ' f'{deg}
' f'
' f'Top collaborators:
{collab_list}
' ) return _wrap_tooltip(content) net = _build_pyvis_network( filtered, node_sizes, edge_widths, node_weights, label_fn, tooltip_fn, lambda u, v, d: "", network_type="coauthorship", node_opacity=0.78, font_size_boost=8, ) if freeze: net.toggle_physics(False) # Embed random x,y directly into each PyVis node dict BEFORE # generate_html() serialises them into the JS DataSet. # vis.js reads x/y from the DataSet at construction time — when # coordinates are already present it skips its circular seeding # entirely and starts physics from these positions. # This is more reliable than post-hoc JS injection because it # happens before `new vis.Network()` is even called. import random as _rnd _spread = 600 for _node in net.nodes: _node['x'] = _rnd.uniform(-_spread, _spread) _node['y'] = _rnd.uniform(-_spread, _spread) html = _pyvis_to_html(net, filtered.number_of_nodes()) # Hover effect: enlarge node + increase label size on hover. # vis.js chosen.node / chosen.label callbacks must be real JS # functions (not JSON), so they are injected via setOptions after # the network is initialised. # chosen.label checks window._coauthHiddenSet: nodes whose label # was hidden get no size multiplier (just shown at normal size); # nodes whose label was already visible get the 10x enlargement. _hover_js = """""" html = html.replace("", _hover_js + "\n") # Label overlap avoidance: hide labels that overlap a larger neighbour; # reveal the hidden label when the user hovers that node. # _hiddenSet is exposed as window._coauthHiddenSet so the chosen.label # callback above can skip the 10x multiplier for revealed-hidden labels. _coauth_overlap_js = """ """ html = html.replace("", _coauth_overlap_js + "\n") st.session_state[cache_key] = html else: html = st.session_state[cache_key] col_graph, col_stats = st.columns([3, 1]) with col_graph: st.components.v1.html(html, height=870, scrolling=False) with col_stats: _render_coauth_stats(graph) def _render_coauth_stats(graph: nx.Graph): import streamlit as st from config import COLOR_SURFACE_ELEVATED st.markdown( f'
', unsafe_allow_html=True, ) st.metric("Total Authors", graph.number_of_nodes()) st.metric("Collaboration Links", graph.number_of_edges()) communities = set( graph.nodes[n].get("community_id", 0) for n in graph.nodes() ) st.metric("Research Clusters", len(communities)) top5 = sorted( graph.nodes(data=True), key=lambda x: x[1].get("weight", 0), reverse=True, )[:5] if top5: st.markdown( "
" "Top Authors by Papers
", unsafe_allow_html=True, ) for name, data in top5: st.markdown( f"
" f"• {truncate(str(name), 22)} " f"{data.get('weight', 0)}" f"
", unsafe_allow_html=True, ) density = nx.density(graph) st.markdown( f"
" f"Density: {density:.4f}
", unsafe_allow_html=True, ) st.markdown("
", unsafe_allow_html=True) # ── B) Keyword co-occurrence network ───────────────────────────────────────── _KW_SHAPES = { "author_keyword": "dot", "mesh_descriptor": "square", "mesh_qualifier": "triangle", "chemical": "diamond", "publication_type": "star", } _KW_TYPE_COLORS = { "author_keyword": "#4E9AF1", "mesh_descriptor": "#34C78A", "mesh_qualifier": "#9B72CF", "chemical": "#F5A623", "publication_type": "#E85D5D", } def render_keyword_network( graph: nx.Graph, key_prefix: str = "keyword", ): import streamlit as st st.markdown( "### 02 — Keyword Co-occurrence Map\n" "" "Node size = frequency  |  Color = thematic cluster  |  " "Edge = co-occurrence" "", unsafe_allow_html=True, ) st.markdown("
", unsafe_allow_html=True) if graph is None or graph.number_of_nodes() == 0: st.info("No keyword data available. Run the pipeline first.") return # Shape legend legend_html = "".join( f'' f' {t.replace("_", " ").title()}' f'' for t in _KW_SHAPES ) st.markdown( f'
{legend_html}
', unsafe_allow_html=True, ) search, min_link, min_size, sel_comms, freeze = _render_controls( graph, key_prefix, f"_kw_html", lambda g: g, show_search=False, show_density=False, show_freeze=False, show_png=False, ) filtered = _filter_graph(graph, min_link, min_size, sel_comms, search) cache_key = ( f"_kw_html_{_VIZ_VERSION}_{key_prefix}_{min_link}_{min_size}_" f"{','.join(map(str, sel_comms))}_{search}_{freeze}" ) if cache_key not in st.session_state: # Keyword network: wide size range for VOSviewer-style dramatic scaling; # thin, faint edges so cluster structure reads clearly. node_sizes = _compute_node_sizes(filtered, size_min=5, size_max=90) edge_widths = _compute_edge_widths(filtered, width_min=0.5, width_max=2.5) node_weights = {n: filtered.nodes[n].get("weight", 1) for n in filtered.nodes()} def label_fn(node, data): return truncate(str(node), 30) def tooltip_fn(node, data, g): freq = data.get("weight", 0) ktype = data.get("source_type", "author_keyword") cid = data.get("community_id", 0) type_color = _KW_TYPE_COLORS.get(ktype, "#9CA3AF") nbrs = sorted( g.neighbors(node), key=lambda nb: g[node][nb].get("weight", 0), reverse=True, )[:5] cooccur_list = "".join( f"
" f"• {truncate(nb, 22)} " f"({g[node][nb].get('weight', 0)})" f"
" for nb in nbrs ) content = ( f'
' f'{node}
' f'
' f'' f'{ktype.replace("_", " ").title()}
' f'
' f'Frequency: ' f'{freq}
' f'
' f'Cluster: {cid}
' f'
' f'Top co-occurring:
{cooccur_list}
' ) return _wrap_tooltip(content) net = _build_pyvis_network( filtered, node_sizes, edge_widths, node_weights, label_fn, tooltip_fn, _default_edge_tooltip, edge_alpha=0.18, edge_roundness=0.30, font_size_boost=14, node_opacity=0.78, network_type="keyword", ) if freeze: net.toggle_physics(False) html = _pyvis_to_html(net, filtered.number_of_nodes(), network_type="keyword") st.session_state[cache_key] = html else: html = st.session_state[cache_key] col_graph, col_stats = st.columns([3, 1]) with col_graph: st.components.v1.html(html, height=870, scrolling=False) with col_stats: _render_keyword_stats(graph) def _render_keyword_stats(graph: nx.Graph): import streamlit as st from config import COLOR_SURFACE_ELEVATED st.markdown( f'
', unsafe_allow_html=True, ) st.metric("Total Keywords", graph.number_of_nodes()) communities = set( graph.nodes[n].get("community_id", 0) for n in graph.nodes() ) st.metric("Thematic Clusters", len(communities)) top_kw = max( graph.nodes(data=True), key=lambda x: x[1].get("weight", 0), default=(None, {}), ) if top_kw[0]: st.markdown( f"
" f"Most Frequent
" f"
" f"{truncate(str(top_kw[0]), 24)}
" f"
{top_kw[1].get('weight', 0)} papers
", unsafe_allow_html=True, ) st.markdown("
", unsafe_allow_html=True) # ── C) Topic landscape network ──────────────────────────────────────────────── def render_topic_network( graph: nx.Graph, papers_df=None, key_prefix: str = "topic", ): import streamlit as st st.markdown( "### 03 — Research Topic Landscape\n" "" "Node size = paper count in selected year range  |  " "Edge = shared papers between topics" "", unsafe_allow_html=True, ) st.markdown("
", unsafe_allow_html=True) if graph is None or graph.number_of_nodes() == 0: st.info("No topic data available. Run the pipeline first.") return search, min_link, min_size, sel_comms, freeze = _render_controls( graph, key_prefix, f"_topic_html", lambda g: g, ) # Year range slider (below controls per spec) year_filter = (None, None) if papers_df is not None and not papers_df.empty and "pub_year" in papers_df.columns: import pandas as pd years = papers_df["pub_year"].dropna().astype(int) if not years.empty: min_yr, max_yr = int(years.min()), int(years.max()) if min_yr < max_yr: year_filter = st.slider( "Year Range", min_value=min_yr, max_value=max_yr, value=(min_yr, max_yr), key=f"{key_prefix}_yearrange", ) filtered = _filter_graph(graph, min_link, min_size, sel_comms, search) cache_key = ( f"_topic_html_{_VIZ_VERSION}_{key_prefix}_{min_link}_{min_size}_" f"{','.join(map(str, sel_comms))}_{search}_{freeze}_{year_filter}" ) if cache_key not in st.session_state: node_sizes = _compute_node_sizes(filtered) edge_widths = _compute_edge_widths(filtered) node_weights = {n: filtered.nodes[n].get("weight", 1) for n in filtered.nodes()} def label_fn(node, data): top_words = data.get("top_words", []) if isinstance(top_words, list) and top_words: raw = [ w[0] if isinstance(w, (list, tuple)) else str(w) for w in top_words ] # Skip words that are prefix/plural variants of an already-selected # word (e.g. "mutation" and "mutations" → keep only the first seen). selected = [] for w in raw: wl = w.lower() if any( wl.startswith(s.lower()) or s.lower().startswith(wl) for s in selected ): continue selected.append(w) if len(selected) >= 3: break return ", ".join(selected) return str(data.get("label", f"Topic {node}")) def tooltip_fn(node, data, g): label = label_fn(node, data) paper_count = data.get("weight", 0) content = ( f'
' f'{label}
' f'
Papers: ' f'{paper_count}
' ) return _wrap_tooltip(content) net = _build_pyvis_network( filtered, node_sizes, edge_widths, node_weights, label_fn, tooltip_fn, _default_edge_tooltip, ) if freeze: net.toggle_physics(False) html = _pyvis_to_html(net, filtered.number_of_nodes()) st.session_state[cache_key] = html else: html = st.session_state[cache_key] col_graph, col_stats = st.columns([3, 1]) with col_graph: st.components.v1.html(html, height=870, scrolling=False) with col_stats: _render_topic_stats(graph) def _render_topic_stats(graph: nx.Graph): import streamlit as st from config import COLOR_SURFACE_ELEVATED st.markdown( f'
', unsafe_allow_html=True, ) st.metric("Topics Discovered", graph.number_of_nodes()) if graph.number_of_nodes() > 0: largest = max( graph.nodes(data=True), key=lambda x: x[1].get("weight", 0), ) label = largest[1].get("label", f"Topic {largest[0]}") st.markdown( f"
" f"Largest Topic
" f"
" f"{truncate(str(label), 28)}
" f"
{largest[1].get('weight', 0)} papers
", unsafe_allow_html=True, ) most_connected = max( graph.nodes(), key=lambda n: graph.degree(n), default=None ) if most_connected: mc_label = graph.nodes[most_connected].get("label", str(most_connected)) st.markdown( f"
" f"Most Connected
" f"
" f"{truncate(str(mc_label), 28)}
", unsafe_allow_html=True, ) st.markdown("
", unsafe_allow_html=True) # ── Shared filter helper ────────────────────────────────────────────────────── def _filter_graph( graph: nx.Graph, min_link: float, min_size: float, selected_communities: Optional[List], search_term: str = "", ) -> nx.Graph: """Apply edge weight, node size, community, and search filters.""" keep_nodes = set() for n in graph.nodes(): data = graph.nodes[n] if data.get("weight", 1) < min_size: continue if selected_communities is not None: if data.get("community_id", 0) not in selected_communities: continue if search_term: if search_term.lower() not in str(n).lower(): continue keep_nodes.add(n) sub = graph.subgraph(keep_nodes).copy() edges_to_remove = [ (u, v) for u, v in sub.edges() if sub[u][v].get("weight", 1) < min_link ] sub.remove_edges_from(edges_to_remove) return sub