anurag-raapid's picture
Upload app, Docker, staging parquet only (no venv)
4b988e0 verified
Raw
History Blame Contribute Delete
66.9 kB
#!/usr/bin/env python3
"""
Clinical Graph Explorer (FalkorDB)
================================
Dash + dash-cytoscape app for exploring clinical subgraphs via FalkorDB.
Data source: FalkorDB graph ``clinical_graph`` (localhost:6380 by default).
Category metadata is enriched from local ``graph_nodes.parquet``.
Usage:
python3 app.py
Open http://127.0.0.1:8050 in a browser.
"""
import os
import time
import dash
import dash_cytoscape as cyto
import redis.exceptions
from dash import Input, Output, State, ctx, dcc, html
import graph_backend as gb
from tui_definitions import format_tui_list, tui_codes
build_subgraph = gb.build_subgraph
build_path_subgraph = gb.build_path_subgraph
find_paths_between_auis = gb.find_paths_between_auis
_extract_filter_options = gb._extract_filter_options
_aui_passes_filters = gb._aui_passes_filters
filter_to_seed_component = gb.filter_to_seed_component
fetch_aui_filter_meta = gb.fetch_aui_filter_meta
# Pre-filled in the explore box; subgraph loads automatically on first page visit.
DEFAULT_SEED_AUI = "A27828238"
def _format_status(msg: str):
"""Highlight warnings (large graph trim, caps) in the status bar."""
if "⚠" in msg:
return html.Span(
msg,
style={"color": "#B7950B", "fontWeight": "600"},
title=msg,
)
return msg
def _stamp_path_query_rev(elements: list, rev: int) -> list:
"""Force a fresh element-store payload so Cytoscape refreshes on repeat path queries."""
stamped = []
for el in elements:
stamped.append({**el, "data": {**el["data"], "_query_rev": rev}})
return stamped
def _append_class(el: dict, class_name: str) -> dict:
existing = (el.get("classes") or "").split()
if class_name not in existing:
existing.append(class_name)
return {**el, "classes": " ".join(existing)}
def _apply_path_focus_highlight(
nodes: list[dict], edges: list[dict], focus_id: str | None
) -> tuple[list[dict], list[dict]]:
"""Dim the graph except the focused node and path-step neighbors (original colors kept)."""
if not focus_id:
return nodes, edges
neighbor_ids: set[str] = set()
for el in nodes:
if el["data"].get("id") == focus_id:
stored = el["data"].get("nav_neighbor_ids")
if stored:
neighbor_ids = set(stored)
break
if not neighbor_ids:
for el in edges:
d = el["data"]
src, tgt = d.get("source", ""), d.get("target", "")
if src == focus_id:
neighbor_ids.add(tgt)
elif tgt == focus_id:
neighbor_ids.add(src)
highlighted_nodes = []
for el in nodes:
nid = el["data"].get("id", "")
if nid == focus_id:
highlighted_nodes.append(_append_class(el, "path-focus"))
elif nid in neighbor_ids:
highlighted_nodes.append(_append_class(el, "path-neighbor"))
else:
highlighted_nodes.append(_append_class(el, "path-dimmed"))
highlighted_edges = []
for el in edges:
d = el["data"]
src, tgt = d.get("source", ""), d.get("target", "")
links_step = (src == focus_id and tgt in neighbor_ids) or (
tgt == focus_id and src in neighbor_ids
)
if links_step:
highlighted_edges.append(el)
else:
highlighted_edges.append(_append_class(el, "path-dimmed"))
return highlighted_nodes, highlighted_edges
# ─────────────────────────────────────────────────────────────────────────────
# Cytoscape stylesheet
# ─────────────────────────────────────────────────────────────────────────────
STYLESHEET = [
# ── AUI node (default) β€” light fill, black label
{
"selector": ".aui-node",
"style": {
"shape": "ellipse",
"background-color": "#85C1E9",
"border-color": "#2E86AB",
"border-width": 2,
"label": "data(label)",
"color": "#000000",
"font-size": "7px",
"text-wrap": "wrap",
"text-max-width": "88px",
"text-valign": "center",
"text-halign": "center",
"width": 62,
"height": 62,
},
},
# ── Seed AUI (start node, highlighted red)
{
"selector": ".seed-node",
"style": {
"background-color": "#F1948A",
"border-color": "#C0392B",
"border-width": 4,
"width": 75,
"height": 75,
"font-size": "8px",
"font-weight": "bold",
"color": "#000000",
},
},
# ── CUI node
{
"selector": ".cui-node",
"style": {
"shape": "diamond",
"background-color": "#F8C471",
"border-color": "#D68910",
"border-width": 2,
"label": "data(label)",
"color": "#000000",
"font-size": "7px",
"text-wrap": "wrap",
"text-max-width": "88px",
"text-valign": "center",
"text-halign": "center",
"width": 76,
"height": 76,
},
},
# ── R1 MRHIER hierarchy edges (solid blue arrow)
{
"selector": ".mrhier-edge",
"style": {
"line-color": "#2980B9",
"target-arrow-color": "#2980B9",
"target-arrow-shape": "triangle",
"curve-style": "bezier",
"width": 2,
"label": "data(label)",
"font-size": "7px",
"color": "#2980B9",
"text-rotation": "autorotate",
"text-margin-y": 14,
"text-background-color": "#FFFFFF",
"text-background-opacity": 0.92,
"text-background-padding": "2px",
"text-background-shape": "roundrectangle",
},
},
# ── R2 AUI-CUI membership edges (dashed green arrow)
{
"selector": ".aui-cui-edge",
"style": {
"line-color": "#27AE60",
"target-arrow-color": "#27AE60",
"target-arrow-shape": "triangle",
"curve-style": "bezier",
"width": 1.5,
"line-style": "dashed",
"line-dash-pattern": [6, 3],
},
},
# ── R3 Same-CUI sibling edges (dotted purple, undirected)
{
"selector": ".same-cui-edge",
"style": {
"line-color": "#8E44AD",
"target-arrow-shape": "none",
"curve-style": "bezier",
"width": 1.5,
"line-style": "dotted",
},
},
# ── R4 CUI hierarchy edges (thick dark-orange solid arrow)
{
"selector": ".cui-cui-edge",
"style": {
"line-color": "#CA6F1E",
"target-arrow-color": "#CA6F1E",
"target-arrow-shape": "triangle",
"curve-style": "bezier",
"width": 3,
"label": "data(label)",
"font-size": "8px",
"color": "#CA6F1E",
"text-rotation": "autorotate",
"text-margin-y": 16,
"text-background-color": "#FFFFFF",
"text-background-opacity": 0.92,
"text-background-padding": "2px",
"text-background-shape": "roundrectangle",
},
},
# ── Parent-only CUI node (lighter shade β€” not in core subgraph)
{
"selector": ".cui-parent-node",
"style": {
"background-color": "#F5CBA7",
"border-color": "#E67E22",
"border-style": "dashed",
"border-width": 2,
"color": "#000000",
"opacity": 0.85,
},
},
# ── R5 MRREL semantic edges (teal solid arrow, RELA label)
{
"selector": ".mrrel-edge",
"style": {
"line-color": "#16A085",
"target-arrow-color": "#16A085",
"target-arrow-shape": "triangle",
"curve-style": "bezier",
"width": 2,
"label": "data(label)",
"font-size": "7px",
"color": "#117A65",
"text-rotation": "autorotate",
"text-margin-y": 14,
"text-background-color": "#FFFFFF",
"text-background-opacity": 0.92,
"text-background-padding": "2px",
"text-background-shape": "roundrectangle",
},
},
# ── Hide labels when toggled off
{
"selector": ".hide-labels",
"style": {"label": ""},
},
# ── Path finder: start / end / intermediate AUIs
{
"selector": ".path-endpoint-start",
"style": {
"background-color": "#7DCEA0",
"border-color": "#27AE60",
"border-width": 4,
"width": 72,
"height": 72,
"color": "#000000",
},
},
{
"selector": ".path-endpoint-end",
"style": {
"background-color": "#C39BD3",
"border-color": "#8E44AD",
"border-width": 4,
"width": 72,
"height": 72,
"color": "#000000",
},
},
{
"selector": ".path-node",
"style": {
"background-color": "#7FB3D5",
"border-color": "#3498DB",
"color": "#000000",
},
},
{
"selector": ".path-edge",
"style": {
"width": 4,
"line-color": "#E74C3C",
"target-arrow-color": "#E74C3C",
"z-index": 10,
},
},
# ── Path explorer: backdrop dim β€” only focus + step neighbors stay bright
{
"selector": ".path-focus",
"style": {
"border-color": "#F39C12",
"border-width": 6,
"z-index": 999,
"opacity": 1,
},
},
{
"selector": ".path-neighbor",
"style": {
"border-color": "#1ABC9C",
"border-width": 5,
"z-index": 50,
"opacity": 1,
},
},
{
"selector": ".path-dimmed",
"style": {
"opacity": 0.2,
},
},
# ── Selected node / edge highlight
{
"selector": ":selected",
"style": {
"background-color": "#F7DC6F",
"border-color": "#F39C12",
"border-width": 4,
"color": "#000000",
"line-color": "#F39C12",
"target-arrow-color": "#F39C12",
},
},
]
# ─────────────────────────────────────────────────────────────────────────────
# Dash App
# ─────────────────────────────────────────────────────────────────────────────
cyto.load_extra_layouts()
app = dash.Dash(__name__, title="Clinical Graph Explorer", assets_folder="assets")
server = app.server
app.config.suppress_callback_exceptions = True
LAYOUTS = [
"cose-bilkent",
"cose",
"breadthfirst",
"dagre",
"circle",
"concentric",
"grid",
"random",
]
def _spacing_scale(spacing: int) -> float:
"""Non-linear scale: 1 β‰ˆ compact, 5 β‰ˆ wide, 10 β‰ˆ very spread out (~4Γ— level 5)."""
s = max(1, min(10, int(spacing or 4)))
return 0.75 + 0.48 * (s**1.38)
def _layout_config(
layout_name: str, spacing: int, fit_to_view: bool, node_count: int = 0
) -> dict:
"""
Cytoscape layout options tuned for readable, non-overlapping views.
spacing 1–10 scales edge length and repulsion (non-linear). fit_to_view=False
(default) leaves the layout at natural size so you can pan/zoom instead of
squeezing everything into the viewport.
"""
spacing = max(1, min(10, int(spacing or 4)))
scale = _spacing_scale(spacing)
fit = bool(fit_to_view)
pad = int(24 + spacing * 14)
n = max(0, int(node_count or 0))
large = n > 500
huge = n > 1200
common = {
# Animating the layout is the dominant browser cost on big graphs.
# Only animate small graphs; compute large layouts in one shot.
"animate": n <= 200,
"randomize": False,
"fit": fit,
"padding": pad,
}
extra_sparse = spacing >= 7
if layout_name == "cose-bilkent":
cfg = {
"name": "cose-bilkent",
**common,
"idealEdgeLength": int(50 * scale),
"nodeRepulsion": int(4500 * scale * (1.15 if extra_sparse else 1.0)),
"gravity": max(0.03, 0.28 / scale),
"numIter": (
700 if huge else (1400 if large else (3500 if extra_sparse else 2500))
),
"nodeDimensionsIncludeLabels": True,
"tilingPaddingVertical": int(16 * scale),
"tilingPaddingHorizontal": int(16 * scale),
"edgeElasticity": max(0.15, 0.45 / (scale**0.35)),
}
if extra_sparse:
cfg["tile"] = False
return cfg
if layout_name == "cose":
return {
"name": "cose",
**common,
"idealEdgeLength": int(60 * scale),
"nodeRepulsion": int(15000 * scale),
"nodeOverlap": int(22 * scale),
"spacingFactor": scale * 1.1,
"nodeDimensionsIncludeLabels": True,
}
if layout_name == "grid":
return {
"name": "grid",
**common,
"spacingFactor": scale * 1.6,
"condense": False,
}
if layout_name == "circle":
return {"name": "circle", **common, "spacingFactor": scale * 1.4}
if layout_name == "concentric":
return {
"name": "concentric",
**common,
"minNodeSpacing": int(45 * scale),
"nodeDimensionsIncludeLabels": True,
}
if layout_name == "breadthfirst":
return {"name": "breadthfirst", **common, "spacingFactor": scale * 1.3}
if layout_name == "dagre":
return {
"name": "dagre",
**common,
"nodeSep": int(45 * scale),
"rankSep": int(60 * scale),
"edgeSep": int(22 * scale),
}
return {"name": layout_name, **common}
# ── Helper widgets ─────────────────────────────────────────────────────────
def _node_legend(color: str, is_diamond: bool, label: str):
shape_style = {
"display": "inline-block",
"width": "14px",
"height": "14px",
"background": color,
"marginRight": "7px",
"verticalAlign": "middle",
"borderRadius": "50%" if not is_diamond else "0",
"transform": "rotate(45deg)" if is_diamond else "none",
}
return html.Div(
[html.Span(style=shape_style), html.Span(label)],
style={"marginBottom": "5px", "fontSize": "12px"},
)
def _edge_legend(color: str, line_style: str, label: str):
line_css = {"solid": "solid", "dashed": "dashed", "dotted": "dotted"}[line_style]
bar_style = {
"display": "inline-block",
"width": "28px",
"height": "0",
"borderTop": f"3px {line_css} {color}",
"marginRight": "8px",
"verticalAlign": "middle",
}
return html.Div(
[html.Span(style=bar_style), html.Span(label)],
style={"marginBottom": "5px", "fontSize": "12px"},
)
# ─────────────────────────────────────────────────────────────────────────────
# Layout
# ─────────────────────────────────────────────────────────────────────────────
app.layout = html.Div(
[
# ── Title bar ─────────────────────────────────────────────────────────
html.Div(
[
html.H2(
"Clinical Graph Explorer",
style={"margin": "0", "color": "#ECF0F1", "fontSize": "20px"},
),
html.P(
"Clinical subgraph explorer Β· "
"FalkorDB (HIER_ISA + MRREL) on port 6380 Β· category from parquet",
style={"margin": "0", "color": "#95A5A6", "fontSize": "11px"},
),
],
style={
"background": "#2C3E50",
"padding": "10px 20px",
"borderBottom": "3px solid #3498DB",
},
),
# ── Main body ─────────────────────────────────────────────────────────
html.Div(
[
# ── Left sidebar ──────────────────────────────────────────────
html.Div(
[
# Search
html.Label(
"Search by AUI or CUI",
style={
"fontWeight": "bold",
"fontSize": "13px",
"display": "block",
"marginBottom": "6px",
},
),
dcc.Input(
id="seed-input",
type="text",
value=DEFAULT_SEED_AUI,
placeholder="e.g. A0016612 or C0000119",
debounce=False,
style={
"width": "100%",
"height": "42px",
"minHeight": "42px",
"padding": "10px 8px",
"marginBottom": "6px",
"borderRadius": "4px",
"border": "1px solid #BDC3C7",
"fontFamily": "monospace",
"fontSize": "13px",
"boxSizing": "border-box",
"lineHeight": "1.35",
},
),
html.Button(
"Explore β†’",
id="explore-btn",
n_clicks=0,
style={
"width": "100%",
"padding": "9px",
"marginBottom": "12px",
"background": "#3498DB",
"color": "#fff",
"border": "none",
"borderRadius": "4px",
"cursor": "pointer",
"fontSize": "14px",
"fontWeight": "bold",
},
),
# Legend first (visible on first visit)
html.Div(
[
html.Div(
"Legend",
style={
"fontWeight": "bold",
"fontSize": "13px",
"marginBottom": "8px",
"color": "#2C3E50",
},
),
html.B("Nodes", style={"fontSize": "12px"}),
html.Div(style={"marginBottom": "4px"}),
_node_legend("#F1948A", False, "Seed AUI"),
_node_legend("#85C1E9", False, "AUI node"),
html.Div(style={"marginBottom": "6px"}),
html.B("Edges", style={"fontSize": "12px"}),
html.Div(style={"marginBottom": "4px"}),
_edge_legend(
"#2980B9", "solid", "R1 MRHIER (AUI→PAUI)"
),
_edge_legend("#27AE60", "dashed", "R2 AUI→CUI"),
_edge_legend("#8E44AD", "dotted", "R3 Same-CUI"),
_edge_legend(
"#16A085", "solid", "R4 Semantic (edge_type)"
),
],
style={
"background": "#FFFFFF",
"border": "1px solid #D5DBDB",
"borderRadius": "6px",
"padding": "10px",
"marginBottom": "14px",
"boxShadow": "0 1px 3px rgba(0,0,0,0.06)",
},
),
html.Hr(style={"margin": "0 0 14px 0"}),
html.Label(
"Graph Mode",
style={
"fontWeight": "bold",
"fontSize": "13px",
"display": "block",
"marginBottom": "6px",
},
),
dcc.Checklist(
id="cui-layer-toggle",
options=[
{
"label": " Show CUI layer (diamond nodes + R2–R3)",
"value": "on",
}
],
value=[],
labelStyle={"display": "block", "lineHeight": "2"},
style={"fontSize": "12px", "marginBottom": "6px"},
),
html.Div(
"Off (default): AUI-only graph. Hop depth uses MRHIER + semantic edges. CUI parent hierarchy is not in this dataset.",
style={
"fontSize": "10px",
"color": "#7F8C8D",
"marginBottom": "14px",
},
),
html.Label(
"Relation Types",
style={
"fontWeight": "bold",
"fontSize": "13px",
"display": "block",
"marginBottom": "6px",
},
),
dcc.Checklist(
id="edge-toggles",
options=[
{
"label": " R1 MRHIER hierarchy (AUI β†’ PAUI)",
"value": "mrhier",
},
{
"label": " R2 AUI β†’ CUI membership",
"value": "aui_cui",
},
{
"label": " R3 Same-CUI siblings (AUI ↔ AUI)",
"value": "same_cui",
},
{
"label": " R4 Semantic relations (AUI β†’ AUI, edge_type)",
"value": "mrrel",
},
],
value=["mrhier", "mrrel"],
labelStyle={"display": "block", "lineHeight": "2"},
style={"fontSize": "12px", "marginBottom": "14px"},
),
html.Hr(style={"margin": "0 0 14px 0"}),
html.Label(
"Path Finder (AUI β†’ AUI)",
style={
"fontWeight": "bold",
"fontSize": "13px",
"display": "block",
"marginBottom": "6px",
},
),
html.Label(
"Start AUI",
style={"fontSize": "11px", "color": "#566573"},
),
dcc.Input(
id="path-aui-start",
type="text",
placeholder="e.g. A0016587",
debounce=False,
style={
"width": "100%",
"height": "42px",
"minHeight": "42px",
"padding": "10px 10px",
"marginBottom": "8px",
"borderRadius": "4px",
"border": "1px solid #BDC3C7",
"fontFamily": "monospace",
"fontSize": "13px",
"boxSizing": "border-box",
"lineHeight": "1.35",
},
),
html.Label(
"End AUI",
style={"fontSize": "11px", "color": "#566573"},
),
dcc.Input(
id="path-aui-end",
type="text",
placeholder="e.g. A0016627",
debounce=False,
style={
"width": "100%",
"height": "42px",
"minHeight": "42px",
"padding": "10px 10px",
"marginBottom": "8px",
"borderRadius": "4px",
"border": "1px solid #BDC3C7",
"fontFamily": "monospace",
"fontSize": "13px",
"boxSizing": "border-box",
"lineHeight": "1.35",
},
),
dcc.RadioItems(
id="path-edge-mode",
options=[
{
"label": " Undirected (either direction)",
"value": "undirected",
},
{
"label": " Directed (AUI→PAUI, AUI1→AUI2 only)",
"value": "directed",
},
],
value="undirected",
labelStyle={
"display": "block",
"lineHeight": "1.8",
"fontSize": "12px",
},
style={"marginBottom": "8px"},
),
html.Button(
"Find paths β†’",
id="find-paths-btn",
n_clicks=0,
style={
"width": "100%",
"padding": "9px",
"marginBottom": "6px",
"background": "#16A085",
"color": "#fff",
"border": "none",
"borderRadius": "4px",
"cursor": "pointer",
"fontSize": "14px",
"fontWeight": "bold",
},
),
html.Button(
"Clear highlight",
id="path-clear-highlight-btn",
n_clicks=0,
style={
"width": "100%",
"padding": "7px",
"marginBottom": "6px",
"background": "#95A5A6",
"color": "#fff",
"border": "none",
"borderRadius": "4px",
"cursor": "pointer",
"fontSize": "12px",
},
),
html.Div(
"MRHIER + semantic edges. Up to 50 paths. Directed mode only counts paths where every edge follows its arrow. "
"After Find paths, click a node β€” it and path-step neighbors stay bright; the rest of the graph fades. Click empty graph or Clear highlight to reset.",
style={
"fontSize": "10px",
"color": "#7F8C8D",
"marginBottom": "14px",
},
),
html.Hr(style={"margin": "0 0 14px 0"}),
html.Label(
"Filters (after Explore)",
style={
"fontWeight": "bold",
"fontSize": "13px",
"display": "block",
"marginBottom": "8px",
},
),
html.Label(
"Category",
style={"fontSize": "11px", "color": "#566573"},
),
dcc.Dropdown(
id="category-filter",
options=[],
value=[],
multi=True,
placeholder="All categories",
style={"fontSize": "12px", "marginBottom": "8px"},
),
html.Label(
"SAB β€” vocabulary",
style={"fontSize": "11px", "color": "#566573"},
),
dcc.Dropdown(
id="sab-filter",
options=[],
value=[],
multi=True,
placeholder="All vocabularies",
style={"fontSize": "12px", "marginBottom": "8px"},
),
html.Label(
"TUI β€” semantic type",
style={"fontSize": "11px", "color": "#566573"},
),
dcc.Dropdown(
id="tui-filter",
options=[],
value=[],
multi=True,
placeholder="All semantic types",
style={"fontSize": "12px", "marginBottom": "8px"},
),
html.Label(
"TTY β€” term type",
style={"fontSize": "11px", "color": "#566573"},
),
dcc.Dropdown(
id="tty-filter",
options=[],
value=[],
multi=True,
placeholder="All term types",
style={"fontSize": "12px", "marginBottom": "8px"},
),
html.Div(
"Filters apply to AUI nodes and their edges (instant).",
style={
"fontSize": "10px",
"color": "#7F8C8D",
"marginBottom": "14px",
},
),
dcc.Checklist(
id="label-toggle",
options=[
{"label": " Show node & edge labels", "value": "on"}
],
value=["on"],
style={"fontSize": "12px", "marginBottom": "14px"},
),
html.Hr(style={"margin": "0 0 14px 0"}),
html.Label(
"Graph Layout",
style={
"fontWeight": "bold",
"fontSize": "13px",
"display": "block",
"marginBottom": "6px",
},
),
dcc.Dropdown(
id="layout-select",
options=[
{
"label": (
"Cose bilkent (sparse, recommended)"
if name == "cose-bilkent"
else name.capitalize()
),
"value": name,
}
for name in LAYOUTS
],
value="cose-bilkent",
clearable=False,
style={"fontSize": "13px", "marginBottom": "10px"},
),
html.Label(
"Node spacing",
style={
"fontWeight": "bold",
"fontSize": "13px",
"display": "block",
"marginBottom": "4px",
},
),
dcc.Slider(
id="spacing-slider",
min=1,
max=10,
step=1,
value=4,
marks={
1: "Tight",
5: "Wide",
10: "Max",
},
),
html.Div(
id="spacing-display",
style={
"fontSize": "11px",
"color": "#7F8C8D",
"marginTop": "4px",
"marginBottom": "8px",
},
),
dcc.Checklist(
id="fit-view-toggle",
options=[
{
"label": " Fit entire graph in view (may overlap nodes)",
"value": "on",
}
],
value=[],
style={"fontSize": "12px", "marginBottom": "14px"},
),
html.P(
"With fit off, pan (drag) and scroll to zoom. Higher spacing spreads nodes apart.",
style={
"fontSize": "10px",
"color": "#95A5A6",
"marginTop": "0",
"marginBottom": "14px",
"fontStyle": "italic",
},
),
html.Hr(style={"margin": "0 0 14px 0"}),
html.Label(
"Hop Depth",
style={
"fontWeight": "bold",
"fontSize": "13px",
"display": "block",
"marginBottom": "4px",
},
),
dcc.Slider(
id="depth-slider",
min=1,
max=15,
step=1,
value=2,
marks={i: str(i) for i in range(1, 16)},
),
html.Div(
id="depth-display",
style={
"fontSize": "11px",
"color": "#7F8C8D",
"marginBottom": "8px",
"marginTop": "4px",
},
),
html.P(
"Double-click a node to re-explore from it.",
style={
"fontSize": "10px",
"color": "#95A5A6",
"marginTop": "4px",
"fontStyle": "italic",
},
),
],
style={
"width": "320px",
"minWidth": "320px",
"padding": "16px",
"background": "#F8F9FA",
"borderRight": "1px solid #DEE2E6",
"overflowY": "auto",
"display": "flex",
"flexDirection": "column",
},
),
# ── Right panel (graph + node info) ───────────────────────────
html.Div(
[
# Status bar
html.Div(
id="status-bar",
children=(
f"Loading default graph for {DEFAULT_SEED_AUI} …"
),
style={
"padding": "7px 14px",
"background": "#EBF5FB",
"borderBottom": "1px solid #AED6F1",
"fontSize": "12px",
"color": "#1A5276",
"fontFamily": "monospace",
"minHeight": "30px",
},
),
# Loading wrapper around cytoscape
dcc.Loading(
id="graph-loading",
type="circle",
children=cyto.Cytoscape(
id="graph",
elements=[],
layout=_layout_config("cose-bilkent", 4, False),
style={
"width": "100%",
"height": "calc(100vh - 230px)",
"background": "#FAFAFA",
},
stylesheet=STYLESHEET,
minZoom=0.03,
maxZoom=8,
boxSelectionEnabled=True,
clearOnUnhover=False,
responsive=True,
),
),
# Node / edge info panel
html.Div(
id="node-info",
children=[
html.P(
"Hover or click a node or edge for details. Double-click a node to re-explore.",
style={
"color": "#95A5A6",
"fontStyle": "italic",
"margin": "0",
},
)
],
style={
"padding": "8px 14px",
"background": "#FDFEFE",
"borderTop": "1px solid #DEE2E6",
"minHeight": "64px",
"maxHeight": "90px",
"overflowY": "auto",
"fontSize": "12px",
"fontFamily": "monospace",
},
),
],
style={
"flex": "1",
"display": "flex",
"flexDirection": "column",
"overflow": "hidden",
},
),
],
style={
"display": "flex",
"height": "calc(100vh - 52px)",
"overflow": "hidden",
},
),
dcc.Store(id="element-store", storage_type="memory"),
dcc.Store(id="filter-options-store", storage_type="memory"),
dcc.Store(id="path-search-rev", data=0, storage_type="memory"),
dcc.Store(id="path-focus-store", storage_type="memory"),
dcc.Store(id="inspector-focus-store", storage_type="memory"),
dcc.Store(id="last-tap-store", storage_type="memory"),
],
style={
"fontFamily": "Segoe UI, Arial, sans-serif",
"margin": "0",
"padding": "0",
"height": "100vh",
"overflow": "hidden",
},
)
# ─────────────────────────────────────────────────────────────────────────────
# Callbacks
# ─────────────────────────────────────────────────────────────────────────────
@app.callback(
Output("element-store", "data"),
Output("status-bar", "children"),
Output("filter-options-store", "data"),
Output("path-focus-store", "data"),
Input("explore-btn", "n_clicks"),
State("seed-input", "value"),
State("depth-slider", "value"),
prevent_initial_call=False,
)
def fetch_subgraph(n_clicks: int, seed: str, depth: int):
"""Query FalkorDB and store ALL elements in the Store."""
if not seed or not seed.strip():
return (
dash.no_update,
"⚠ Please enter an AUI or CUI identifier.",
dash.no_update,
None,
)
t0 = time.time()
try:
elements, msg = build_subgraph(
seed,
depth or 1,
show_mrhier=True,
show_aui_cui=True,
show_same_cui=True,
show_cui_cui=False,
show_mrrel=True,
show_cui_layer=True,
)
except redis.exceptions.ResponseError as exc:
if "timed out" not in str(exc).lower():
raise
time.sleep(0.5)
elements, msg = build_subgraph(
seed,
depth or 1,
show_mrhier=True,
show_aui_cui=True,
show_same_cui=True,
show_cui_cui=False,
show_mrrel=True,
show_cui_layer=True,
)
msg = f"{msg} (retried after DB timeout)"
elapsed = time.time() - t0
full_msg = f"{msg} | Query: {elapsed:.2f}s"
return (
elements,
_format_status(full_msg),
_extract_filter_options(elements),
None,
)
@app.callback(
Output("element-store", "data", allow_duplicate=True),
Output("status-bar", "children", allow_duplicate=True),
Output("filter-options-store", "data", allow_duplicate=True),
Output("path-search-rev", "data"),
Output("path-focus-store", "data", allow_duplicate=True),
Input("find-paths-btn", "n_clicks"),
State("path-aui-start", "value"),
State("path-aui-end", "value"),
State("path-edge-mode", "value"),
State("path-search-rev", "data"),
prevent_initial_call=True,
)
def fetch_paths(
n_clicks: int,
aui_start: str,
aui_end: str,
path_edge_mode: str | None,
prev_rev: int | None,
):
"""Find all simple paths between two AUIs and show their union subgraph."""
if not n_clicks:
raise dash.exceptions.PreventUpdate
if not aui_start or not str(aui_start).strip():
return (
dash.no_update,
"⚠ Enter a start AUI.",
dash.no_update,
dash.no_update,
None,
)
if not aui_end or not str(aui_end).strip():
return (
dash.no_update,
"⚠ Enter an end AUI.",
dash.no_update,
dash.no_update,
None,
)
directed = path_edge_mode == "directed"
rev = (prev_rev or 0) + 1
t0 = time.time()
print(
f"β–Ά Path search {aui_start.strip().upper()} β†’ {aui_end.strip().upper()} "
f"({'directed' if directed else 'undirected'}) …",
flush=True,
)
paths, path_msg = find_paths_between_auis(aui_start, aui_end, directed=directed)
elapsed = time.time() - t0
print(f"β–Ά Path search done ({elapsed:.2f}s): {path_msg[:120]}", flush=True)
if not paths:
return [], _format_status(f"{path_msg} | {elapsed:.2f}s"), [], rev, None
elements, viz_msg = build_path_subgraph(
aui_start.strip().upper(),
aui_end.strip().upper(),
paths,
directed=directed,
)
elements = _stamp_path_query_rev(elements, rev)
full_msg = f"{path_msg} | {viz_msg} | Query: {elapsed:.2f}s"
return (
elements,
_format_status(full_msg),
_extract_filter_options(elements),
rev,
None,
)
@app.callback(
Output("path-focus-store", "data", allow_duplicate=True),
Input("graph", "tapNodeData"),
State("element-store", "data"),
prevent_initial_call=True,
)
def path_focus_on_tap(node_data: dict | None, all_elements: list | None):
"""In path-finder view, single-click highlights immediate neighbors for step-by-step navigation."""
if not node_data or not node_data.get("id"):
return dash.no_update
is_path_view = bool(
all_elements
and any(
el.get("data", {}).get("path_role")
for el in all_elements
if "source" not in el.get("data", {})
)
)
if not is_path_view:
return None
return {"focus_id": node_data["id"]}
@app.callback(
Output("path-focus-store", "data", allow_duplicate=True),
Input("path-clear-highlight-btn", "n_clicks"),
prevent_initial_call=True,
)
def path_focus_clear_btn(n_clicks: int):
"""Clear path-step highlight (sidebar button or canvas tap via path_focus_bridge.js)."""
if not n_clicks:
return dash.no_update
return None
@app.callback(
Output("sab-filter", "options"),
Output("sab-filter", "value"),
Output("tui-filter", "options"),
Output("tui-filter", "value"),
Output("tty-filter", "options"),
Output("tty-filter", "value"),
Output("category-filter", "options"),
Output("category-filter", "value"),
Input("filter-options-store", "data"),
prevent_initial_call=True,
)
def refresh_filter_dropdowns(opts: dict | None):
if not opts:
return [], [], [], [], [], [], [], []
return (
opts.get("sab", []),
[],
opts.get("tui", []),
[],
opts.get("tty", []),
[],
opts.get("category", []),
[],
)
@app.callback(
Output("graph", "elements"),
Output("graph", "layout"),
Input("element-store", "data"),
Input("path-search-rev", "data"),
Input("edge-toggles", "value"),
Input("cui-layer-toggle", "value"),
Input("layout-select", "value"),
Input("spacing-slider", "value"),
Input("fit-view-toggle", "value"),
Input("sab-filter", "value"),
Input("tui-filter", "value"),
Input("tty-filter", "value"),
Input("category-filter", "value"),
Input("label-toggle", "value"),
Input("path-focus-store", "data"),
prevent_initial_call=True,
)
def update_graph(
all_elements: list | None,
_path_search_rev: int | None,
edge_toggles: list | None,
cui_layer_toggle: list | None,
layout_name: str,
spacing: int,
fit_view_toggle: list | None,
sab_filter: list | None,
tui_filter: list | None,
tty_filter: list | None,
category_filter: list | None,
label_toggle: list | None,
path_focus: dict | None,
):
"""
Filter stored elements by toggled edge types, SAB/TUI/TTY, and labels.
Toggle changes are instant (no re-query).
"""
fit_to_view = "on" in (fit_view_toggle or [])
if not all_elements:
return [], _layout_config(layout_name, spacing, fit_to_view, 0)
shown = set(edge_toggles or [])
show_cui_layer = "on" in (cui_layer_toggle or [])
if not show_cui_layer:
shown -= {"aui_cui", "same_cui", "cui_cui"}
sab_keep = set(sab_filter or [])
tui_keep = set(tui_filter or [])
tty_keep = set(tty_filter or [])
category_keep = set(category_filter or [])
filters_on = bool(sab_keep or tui_keep or tty_keep or category_keep)
show_labels = "on" in (label_toggle or [])
nodes = [el for el in all_elements if "source" not in el["data"]]
edges = [el for el in all_elements if "source" in el["data"]]
if not show_cui_layer:
nodes = [el for el in nodes if el["data"].get("node_type") != "CUI"]
aui_meta: dict[str, dict] = {}
aui_ids = [
el["data"]["aui"]
for el in nodes
if el["data"].get("node_type") == "AUI" and el["data"].get("aui")
]
# Path-finder results already carry SAB/TUI/TTY on nodes β€” skip heavy mrrel scan.
is_path_view = any(
el["data"].get("path_role")
for el in nodes
if el["data"].get("node_type") == "AUI"
)
if is_path_view:
for el in nodes:
d = el["data"]
if d.get("node_type") != "AUI" or not d.get("aui"):
continue
aui_meta[el["data"]["id"]] = {
"sab": d.get("sab") or "",
"tty": d.get("tty") or "",
"tuis": tui_codes(d.get("tui") or ""),
"category": d.get("category") or "",
}
else:
for el in nodes:
d = el["data"]
if d.get("node_type") != "AUI" or not d.get("aui"):
continue
aui_meta[d["id"]] = {
"sab": d.get("sab") or "",
"tty": d.get("tty") or "",
"tuis": tui_codes(d.get("tui") or ""),
"category": d.get("category") or "",
}
def edge_visible(el: dict) -> bool:
et = el["data"].get("edge_type", "")
if et == "mrhier" and "mrhier" not in shown:
return False
if et == "aui_cui" and "aui_cui" not in shown:
return False
if et == "same_cui" and "same_cui" not in shown:
return False
if et == "cui_cui" and "cui_cui" not in shown:
return False
if et == "mrrel" and "mrrel" not in shown:
return False
return True
visible_edges = [el for el in edges if edge_visible(el)]
if filters_on:
filtered_edges = []
for el in visible_edges:
d = el["data"]
et = d.get("edge_type", "")
if et == "cui_cui":
filtered_edges.append(el)
continue
if et in ("mrhier", "mrrel"):
if _aui_passes_filters(
d["source"], aui_meta, sab_keep, tui_keep, tty_keep, category_keep
):
filtered_edges.append(el)
continue
src = d.get("source", "")
if et in ("aui_cui", "same_cui") and _aui_passes_filters(
src, aui_meta, sab_keep, tui_keep, tty_keep, category_keep
):
filtered_edges.append(el)
visible_edges = filtered_edges
# Explore view: keep only the seed-connected component. Path view keeps the full path union.
if not is_path_view:
nodes, visible_edges = filter_to_seed_component(nodes, visible_edges)
if is_path_view and path_focus and path_focus.get("focus_id"):
focus_id = path_focus["focus_id"]
if any(el["data"].get("id") == focus_id for el in nodes):
nodes, visible_edges = _apply_path_focus_highlight(
nodes, visible_edges, focus_id
)
filtered = nodes + visible_edges
if not show_labels:
filtered = [
{**el, "classes": (el.get("classes", "") + " hide-labels").strip()}
for el in filtered
]
node_count = len(nodes)
layout_cfg = _layout_config(layout_name, spacing, fit_to_view, node_count)
# Re-running the layout is the expensive browser-side step. Only do it when
# the node set or layout settings change. Edge toggles, label toggle and
# path-focus only restyle/hide existing elements, so we keep current node
# positions (return no_update for layout) β€” instant and non-jarring.
relayout_inputs = {
"element-store",
"path-search-rev",
"cui-layer-toggle",
"sab-filter",
"tui-filter",
"tty-filter",
"category-filter",
"layout-select",
"spacing-slider",
"fit-view-toggle",
}
triggered_ids = {t["prop_id"].split(".")[0] for t in (ctx.triggered or [])}
should_relayout = (not triggered_ids) or bool(triggered_ids & relayout_inputs)
return filtered, (layout_cfg if should_relayout else dash.no_update)
def _info_panel_placeholder():
return html.P(
"Hover or click a node or edge for details. Double-click a node to re-explore.",
style={"color": "#95A5A6", "fontStyle": "italic", "margin": "0"},
)
def _build_edge_info(edge_data: dict) -> html.Div:
et = edge_data.get("edge_type", "")
if et == "mrrel":
return html.Div(
[
html.Span("MRREL ", style={"color": "#16A085", "fontWeight": "bold"}),
html.Span(f"RELA: {edge_data.get('rela', '')}"),
html.Span(" Β· "),
html.Span(f"REL: {edge_data.get('rel', '')}"),
html.Span(" Β· "),
html.Span(f"SAB: {edge_data.get('sab', '')}"),
html.Br(),
html.Span(
f"{edge_data.get('source', '')} β†’ {edge_data.get('target', '')}"
),
]
)
if et == "mrhier":
return html.Div(
[
html.Span("MRHIER ", style={"color": "#2980B9", "fontWeight": "bold"}),
html.Span(f"RELA: {edge_data.get('label', '')}"),
html.Span(" Β· "),
html.Span(f"SAB: {edge_data.get('sab', '')}"),
]
)
if et == "cui_cui":
return html.Div(
[
html.Span(
"CUI hierarchy ",
style={"color": "#CA6F1E", "fontWeight": "bold"},
),
html.Span(
f"{edge_data.get('source', '')} β†’ {edge_data.get('target', '')}"
),
]
)
return html.Div(
[
html.Span(f"Edge: {et} "),
html.Span(f"{edge_data.get('source', '')} β†’ {edge_data.get('target', '')}"),
]
)
def _build_node_info(data: dict) -> html.Div:
if data.get("node_type") == "AUI":
path_hint = []
if data.get("path_role"):
path_hint = [
html.Br(),
html.Span(
"Path view: clicked node + path-step neighbors stay bright; others fade. Click empty canvas or Clear highlight to reset.",
style={"color": "#7F8C8D", "fontStyle": "italic"},
),
]
return html.Div(
[
html.Span(
f"AUI: {data.get('aui', '')}",
style={"color": "#2E86AB", "fontWeight": "bold"},
),
html.Span(" Β· "),
html.Span(f"CUI: {data.get('cui', '')}"),
html.Span(" Β· "),
html.Span(f"SAB: {data.get('sab', '')}"),
html.Span(" Β· "),
html.Span(f"Category: {data.get('category', '')}"),
html.Span(" Β· "),
html.Span(
f"TUI: {data.get('tui_label') or format_tui_list(data.get('tui', ''))}"
),
html.Span(" Β· "),
html.Span(f"TTY: {data.get('tty', '')}"),
html.Br(),
html.Span(
f"STR: {data.get('str', '')}",
style={"color": "#2C3E50"},
),
*path_hint,
]
)
return html.Div(
[
html.Span(
f"CUI: {data.get('cui', '')}",
style={"color": "#E67E22", "fontWeight": "bold"},
),
html.Span(" Β· "),
html.Span(
f"TUI: {data.get('tui_label') or format_tui_list(data.get('tui', ''))}"
),
html.Br(),
html.Span(
f"Preferred term: {data.get('str', '')}",
style={"color": "#2C3E50"},
),
]
)
def _selection_from_graph_props(
triggered_prop: str,
node_hover: dict | None,
edge_hover: dict | None,
node_tap: dict | None,
edge_tap: dict | None,
) -> dict | None:
"""Pick node/edge payload from dash-cytoscape hover/tap props."""
if triggered_prop == "graph.mouseoverEdgeData":
return edge_hover
if triggered_prop == "graph.mouseoverNodeData":
return node_hover
if triggered_prop == "graph.tapEdgeData":
return edge_tap
if triggered_prop == "graph.tapNodeData":
return node_tap
return None
@app.callback(
Output("node-info", "children"),
Input("inspector-focus-store", "data"),
Input("graph", "mouseoverNodeData"),
Input("graph", "mouseoverEdgeData"),
Input("graph", "tapNodeData"),
Input("graph", "tapEdgeData"),
prevent_initial_call=True,
)
def show_selection_info(
inspector_data: dict | None,
node_hover: dict | None,
edge_hover: dict | None,
node_tap: dict | None,
edge_tap: dict | None,
):
"""Inspector on hover or click only β€” does not reload the graph."""
if not ctx.triggered:
return _info_panel_placeholder()
triggered_prop = ctx.triggered[0]["prop_id"]
if triggered_prop == "inspector-focus-store.data":
data = inspector_data
else:
data = _selection_from_graph_props(
triggered_prop, node_hover, edge_hover, node_tap, edge_tap
)
if not data:
return dash.no_update
if data.get("edge_type"):
return _build_edge_info(data)
return _build_node_info(data)
@app.callback(
Output("depth-display", "children"),
Input("depth-slider", "value"),
)
def update_depth_label(val: int):
return (
f"Traverse {val} hop{'s' if val != 1 else ''} from seed "
f"(MRHIER + semantic edges, undirected)"
)
@app.callback(
Output("spacing-display", "children"),
Input("spacing-slider", "value"),
)
def update_spacing_label(val: int):
labels = {
1: "Compact β€” nodes may touch on dense graphs",
2: "Slightly spread",
3: "Balanced separation",
4: "Wide β€” good default for medium graphs",
5: "Sparse β€” large graphs; pan/zoom to explore",
6: "Very sparse",
7: "Extra sparse (disables layout tiling)",
8: "Ultra sparse β€” huge canvas, more layout time",
9: "Maximum spread for dense subgraphs",
10: "Max sparsity β€” slowest layout, widest separation",
}
return labels.get(val, labels[4])
_DOUBLE_TAP_MS = 450
@app.callback(
Output("seed-input", "value"),
Output("last-tap-store", "data"),
Output("element-store", "data", allow_duplicate=True),
Output("status-bar", "children", allow_duplicate=True),
Output("filter-options-store", "data", allow_duplicate=True),
Input("graph", "tapNodeData"),
State("last-tap-store", "data"),
State("depth-slider", "value"),
prevent_initial_call=True,
)
def handle_node_tap(
node_data: dict | None,
last_tap: dict | None,
depth: int,
):
"""
Detect double-tap on a node and re-run Explore (same as Explore button).
dash-cytoscape has no native dbltap output; two taps <450ms on the same node.
"""
if not node_data:
return (
dash.no_update,
dash.no_update,
dash.no_update,
dash.no_update,
dash.no_update,
)
now_ms = int(time.time() * 1000)
nid = node_data.get("id")
if (
last_tap
and last_tap.get("id") == nid
and (now_ms - int(last_tap.get("t", 0))) < _DOUBLE_TAP_MS
):
seed = (
node_data.get("aui", "")
if node_data.get("node_type") == "AUI"
else node_data.get("cui", "")
)
if not seed:
return (
dash.no_update,
None,
dash.no_update,
dash.no_update,
dash.no_update,
)
t0 = time.time()
elements, msg = build_subgraph(
seed,
depth or 1,
show_mrhier=True,
show_aui_cui=True,
show_same_cui=True,
show_cui_cui=False,
show_mrrel=True,
show_cui_layer=True,
)
elapsed = time.time() - t0
full_msg = f"{msg} | Re-explore: {elapsed:.2f}s"
return (
seed,
None,
elements,
_format_status(full_msg),
_extract_filter_options(elements),
)
return (
dash.no_update,
{"id": nid, "t": now_ms},
dash.no_update,
dash.no_update,
dash.no_update,
)
# ── Enter key in input triggers explore ───────────────────────────────────────
app.clientside_callback(
"""
function(n_keydown, n_clicks) {
return n_clicks + 1;
}
""",
Output("explore-btn", "n_clicks", allow_duplicate=True),
Input("seed-input", "n_submit"),
State("explore-btn", "n_clicks"),
prevent_initial_call=True,
)
# ─────────────────────────────────────────────────────────────────────────────
# Entry point
# ─────────────────────────────────────────────────────────────────────────────
# if __name__ == "__main__":
# print("β–Ά Starting Dash server at http://0.0.0.0:8050 …", flush=True)
# app.run(debug=False, host="0.0.0.0", port=8050)
# ─────────────────────────────────────────────────────────────────────────────
# Entry point
# ─────────────────────────────────────────────────────────────────────────────
if __name__ == "__main__":
port = int(os.environ.get("PORT", "8050"))
print(f"β–Ά Starting Dash server at http://0.0.0.0:{port} …", flush=True)
# Single worker: path/explore callbacks share one FalkorDB connection (lock-safe).
app.run(debug=False, host="0.0.0.0", port=port, threaded=False)