"""Rendering helpers for the Experimental ALS Therapy Landscape tab.
Loads data/landscape/landscape.json and renders a single "ALS Therapeutic Pipeline by
Clinical Trial Phase" wheel: mechanism groups are angular SECTORS, the three trial phases
are concentric RINGS (inner = Phase 1, outer = Phase 3), and each compound is a dot inside
its sector×ring cell (hover shows its name). Dots go grey when the compound has no
recruiting/active trial in the current view. Two dropdowns filter the wheel (recruitment
status + trial phase); a Mechanism dropdown surfaces a group's compounds, each with its
pipeline stage, mechanism confidence, and a trials table.
"""
from __future__ import annotations
import json
import html
from config import LANDSCAPE_PATH
_INSUFFICIENT = "Insufficient evidence"
_STATUS_BADGE = {
"recruiting": ("#00B894", "Recruiting"), "active": ("#0984E3", "Active"),
"completed": ("#636E72", "Completed"), "terminated": ("#D63031", "Terminated"),
"other": ("#B2BEC3", "Unknown"),
}
# Recruitment-status filter for the pipeline wheel.
STATUS_FILTER_OPTIONS = ["All trials", "Recruiting", "Not recruiting"]
# Sentinel for the mechanism filter meaning "don't narrow the wheel to one group".
ALL_MECHANISMS = "All mechanisms"
# Pipeline-wheel rings, inner → outer. Phase 3 and Expanded Access (EAP) share the outer
# ring. These double as the "Trial phase" checkbox options. (Phase 4 / NA are never placed.)
PHASE_RINGS = ["Phase 1", "Phase 2", "Phase 3 / EAP"]
# Grey used for compounds with no recruiting/active trial in the current view.
_INACTIVE_COLOR = "#B8BFC7"
def _phase_buckets(phase: str) -> set:
"""Map a ClinicalTrials.gov phase string to wheel rings (Phase 3 + Expanded Access merge)."""
p = (phase or "").upper()
b = set()
if "PHASE1" in p: # also catches EARLY_PHASE1
b.add("Phase 1")
if "PHASE2" in p:
b.add("Phase 2")
if "PHASE3" in p or "EXPANDED" in p or "ACCESS" in p:
b.add("Phase 3 / EAP")
return b or {"Not applicable"} # PHASE4 / NA / blank are not placed on the wheel
def _top_ring(trials: list[dict]) -> str | None:
"""Most advanced wheel ring (PHASE_RINGS order) present across a compound's trials."""
present: set = set()
for tr in trials:
present |= _phase_buckets(tr.get("phase", ""))
for ring in reversed(PHASE_RINGS):
if ring in present:
return ring
return None
def load_landscape() -> dict | None:
if not LANDSCAPE_PATH.exists():
return None
try:
return json.loads(LANDSCAPE_PATH.read_text())
except Exception:
return None
# ── "ALS Therapeutic Pipeline by Clinical Trial Phase" wheel (inline SVG) ──────
# Magazine-style wheel: mechanism groups are equal angular SECTORS, the three
# trial phases are concentric RINGS (inner=Phase 1, outer=Phase 3), and each
# compound is a dot inside its sector×ring cell (hover shows its name). Driven by
# real landscape data; the app's 12 mechanism classes are mapped to 8 display groups.
def _all_compounds(landscape: dict) -> list[dict]:
"""Distinct compounds (therapies) across all classes + unclassified, deduped by name.
A therapy is multi-label (appears under each class it acts through), but every copy
carries the same `mechanisms` and `trials`, so keeping the first is sufficient.
"""
seen: dict[str, dict] = {}
for c in landscape.get("classifications", []):
for t in c.get("therapies", []):
seen.setdefault(t["name"], t)
for t in landscape.get("unclassified", []):
seen.setdefault(t["name"], t)
return list(seen.values())
def _primary_class(therapy: dict) -> str:
"""The compound's dominant mechanism class (primary role, else highest confidence)."""
mechs = therapy.get("mechanisms") or []
if not mechs:
return _INSUFFICIENT
pool = [m for m in mechs if m.get("role") == "primary"] or mechs
best = max(pool, key=lambda m: m.get("confidence", 0) or 0)
return best.get("class", _INSUFFICIENT)
def _filter_trials(trials: list[dict], status_filter: str, phases: list | None = None) -> list[dict]:
"""Filter a compound's trials by recruitment status and the selected wheel rings.
`phases` is a list of PHASE_RINGS labels (None = all rings). Trials that map only to a
non-ring bucket (Phase 4 / NA) are always dropped — the wheel is Phase 1–3/EAP only.
"""
out = trials
if status_filter == "Recruiting":
out = [tr for tr in out if tr.get("status_group") == "recruiting"]
elif status_filter == "Not recruiting":
out = [tr for tr in out if tr.get("status_group") != "recruiting"]
selected = set(phases) if phases else set(PHASE_RINGS)
out = [tr for tr in out if _phase_buckets(tr.get("phase", "")) & selected]
return list(out)
def _compound_active(trials: list[dict]) -> bool:
"""True if any trial is recruiting or active-not-recruiting (drives dot coloring)."""
return any(tr.get("status_group") in ("recruiting", "active") for tr in trials)
# Eight display groups (clockwise from top) and their colors, matching the
# reference infographic. The app's finer 12-class taxonomy maps down to these.
GROUP_ORDER = [
"Neuroinflammation / Immunity",
"RNA / Gene Targeting",
"Neuroprotection / Cell Survival",
"Protein Homeostasis / TDP-43 Pathology",
"Metabolic / Mitochondrial Function",
"Neuromuscular Function",
"Stem Cell / Regenerative",
"Others / Multiple",
]
GROUP_COLORS = {
"Neuroinflammation / Immunity": "#4C6FB1",
"RNA / Gene Targeting": "#E4586E",
"Neuroprotection / Cell Survival": "#26B6A6",
"Protein Homeostasis / TDP-43 Pathology": "#EFAA3A",
"Metabolic / Mitochondrial Function": "#9B7EC8",
"Neuromuscular Function": "#EE7B4E",
"Stem Cell / Regenerative": "#5BB56A",
"Others / Multiple": "#F4CE14", # yellow (grey is reserved for inactive compounds)
}
_CLASS_TO_GROUP = {
"TDP-43 proteinopathy": "Protein Homeostasis / TDP-43 Pathology",
"Proteostasis / autophagy": "Protein Homeostasis / TDP-43 Pathology",
"SOD1": "RNA / Gene Targeting",
"C9orf72": "RNA / Gene Targeting",
"FUS": "RNA / Gene Targeting",
"RNA metabolism": "RNA / Gene Targeting",
"Neuroinflammation": "Neuroinflammation / Immunity",
"Oxidative stress": "Neuroprotection / Cell Survival",
"Glutamate excitotoxicity": "Neuroprotection / Cell Survival",
"Mitochondrial dysfunction": "Metabolic / Mitochondrial Function",
"Neurotrophic / regenerative": "Stem Cell / Regenerative",
"Symptomatic / Other": "Others / Multiple",
_INSUFFICIENT: "Others / Multiple",
}
_DOT_CAP = 12 # max dots drawn per sector×ring cell (real counts live in hover/summary)
def _group_of(therapy: dict) -> str:
return _CLASS_TO_GROUP.get(_primary_class(therapy), "Others / Multiple")
def _pipeline_grid(landscape: dict | None, status_filter: str, phases: list,
mech_filter: str = ALL_MECHANISMS) -> dict:
"""{group: {ring: [(compound name, is_active)]}} over the filtered, ring-placed compounds.
`phases` is the list of selected PHASE_RINGS; `mech_filter` other than ALL_MECHANISMS
narrows the wheel to a single mechanism group. Each compound lands in the most advanced
selected ring it has a trial in.
"""
grid = {g: {r: [] for r in PHASE_RINGS} for g in GROUP_ORDER}
if landscape:
for t in _all_compounds(landscape):
group = _group_of(t)
if mech_filter != ALL_MECHANISMS and group != mech_filter:
continue
trials = _filter_trials(t.get("trials", []), status_filter, phases)
if not trials:
continue
ring = _top_ring(trials)
if not ring:
continue
grid[group][ring].append((t["name"], _compound_active(trials)))
return grid
def _polar_xy(cx: float, cy: float, r: float, ang_deg: float) -> tuple:
"""Angle 0 = top (12 o'clock), increasing clockwise, in screen coords."""
import math
t = math.radians(ang_deg - 90.0)
return cx + r * math.cos(t), cy + r * math.sin(t)
def _annular_sector_path(cx, cy, r_in, r_out, a0, a1) -> str:
large = 1 if (a1 - a0) % 360 > 180 else 0
x0o, y0o = _polar_xy(cx, cy, r_out, a0)
x1o, y1o = _polar_xy(cx, cy, r_out, a1)
x1i, y1i = _polar_xy(cx, cy, r_in, a1)
x0i, y0i = _polar_xy(cx, cy, r_in, a0)
return (f"M {x0o:.1f} {y0o:.1f} A {r_out:.1f} {r_out:.1f} 0 {large} 1 {x1o:.1f} {y1o:.1f} "
f"L {x1i:.1f} {y1i:.1f} A {r_in:.1f} {r_in:.1f} 0 {large} 0 {x0i:.1f} {y0i:.1f} Z")
def _cell_dots(cx, cy, r_in, r_out, a0, a1, cells, col) -> str:
"""Lay up to _DOT_CAP compound dots on a jittered grid inside one annular cell.
`cells` is a list of (name, is_active); active dots take the group `col`, inactive grey.
"""
import math
n = min(len(cells), _DOT_CAP)
if n == 0:
return ""
cols = 4 if n > 6 else max(1, min(n, 3))
rows = max(1, math.ceil(n / cols))
rr0, rr1 = r_in + 13, r_out - 13
aa0, aa1 = a0 + 3.0, a1 - 3.0
out = []
for k, (name, active) in enumerate(cells[:n]):
row, col_i = divmod(k, cols)
in_row = min(cols, n - row * cols)
fr = (row + 0.5) / rows
fa = (col_i + 0.5) / in_row
jr = ((k * 37) % 7 - 3) * 1.1
ja = ((k * 53) % 5 - 2) * 0.6
r = rr0 + fr * (rr1 - rr0) + jr
a = aa0 + fa * (aa1 - aa0) + ja
x, y = _polar_xy(cx, cy, r, a)
fill = col if active else _INACTIVE_COLOR
out.append(f'{html.escape(name)}')
return "".join(out)
def build_pipeline_svg(landscape: dict | None, status_filter: str = "All trials",
phases: list | None = None,
mech_filter: str = ALL_MECHANISMS) -> str:
"""Inline-SVG 'ALS Therapeutic Pipeline by Phase' infographic (see section header).
`phases` (list of PHASE_RINGS labels; None = all) drives BOTH the filter and the geometry:
one concentric ring is drawn per selected phase, inner → outer in PHASE_RINGS order.
"""
rings = [r for r in PHASE_RINGS if r in set(phases)] if phases else list(PHASE_RINGS)
grid = _pipeline_grid(landscape, status_filter, rings, mech_filter)
groups = [g for g in GROUP_ORDER if any(grid[g][r] for r in rings)]
ph_tot = {r: sum(len(grid[g][r]) for g in groups) for r in rings}
total = sum(ph_tot.values())
if not groups or total == 0 or not rings:
sel = ", ".join(rings) if rings else "no phases"
return ('
'
f'No trials match “{html.escape(status_filter)} · {html.escape(sel)} · '
f'{html.escape(mech_filter)}”.
')
W = 720
cx = cy = W / 2
r_hub = 78
r_out = 338
nr = len(rings)
t = (r_out - r_hub) / nr # ring thickness recomputed for the number of selected phases
n = len(groups)
seg = 360.0 / n
gap_a = 1.4
mx, my_t, my_b = 116, 46, 70 # margins so perimeter labels aren't clipped
svg = [f'")
# ── side panels (legend + summary) ──
legend_rows = "".join(
f'
'
f'{html.escape(g)}'
f'{sum(len(grid[g][r]) for r in rings)}
'
for g in groups)
legend_rows += (
'
'
f'Inactive — no recruiting/active trial
')
legend = (
'
'
'
Mechanism of Action
'
f'{legend_rows}
')
def _row(lbl, val, sub, strong=False):
w = "800" if strong else "600"
return (f'
'
f'{lbl}'
f'{val}'
f'{sub}
')
pct = lambda v: f"{round(100 * v / total)}%" if total else "0%"
summary = (
'
'
'
Pipeline Summary
'
+ "".join(_row(r, ph_tot[r], pct(ph_tot[r])) for r in rings)
+ _row("Total", total, "Candidates", strong=True)
+ f'
')
subtitle = " | ".join(
f'{"Inner" if i == 0 else "Outer" if i == nr - 1 else "Middle"} ring: {html.escape(r)}'
for i, r in enumerate(rings))
return (
'
'
'
'
'
ALS Therapeutic Pipeline by Clinical Trial Phase
'
f'
{subtitle}
'
'
'
f'
{"".join(svg)}
'
f'
{legend}{summary}
'
'
')
# ── Mechanism → compound drill-down (drives the detail panel + trials table) ───
def group_names(landscape: dict | None) -> list[str]:
"""Display groups that have at least one compound, in canonical GROUP_ORDER."""
if not landscape:
return []
present = {_group_of(t) for t in _all_compounds(landscape)}
return [g for g in GROUP_ORDER if g in present]
def mechanism_filter_options(landscape: dict | None) -> list[str]:
"""Mechanism dropdown choices: 'All mechanisms' plus every group that has a compound."""
return [ALL_MECHANISMS] + group_names(landscape)
def compounds_of_group(landscape: dict | None, group: str,
status_filter: str = "All trials", phases: list | None = None) -> list[dict]:
"""Compounds in `group` (or all groups when group is ALL_MECHANISMS) with ≥1 passing trial."""
out = []
for t in _all_compounds(landscape or {}):
if group and group != ALL_MECHANISMS and _group_of(t) != group:
continue
if not _filter_trials(t.get("trials", []), status_filter, phases):
continue
out.append(t)
return out
def compound_labels(landscape: dict | None, group: str,
status_filter: str = "All trials", phases: list | None = None) -> list[str]:
"""Dropdown labels, e.g. 'Tofersen — 3 trials, 1 recruiting'."""
labels = []
for t in compounds_of_group(landscape, group, status_filter, phases):
c = t["trial_counts"]
rec = f", {c['recruiting']} recruiting" if c["recruiting"] else ""
labels.append(f"{t['name']} — {c['total']} trial{'s' if c['total'] != 1 else ''}{rec}")
return labels
def _compound_by_label(landscape: dict | None, label: str) -> dict | None:
name = label.split(" — ")[0] if label else ""
if not name:
return None
for t in _all_compounds(landscape or {}):
if t["name"] == name:
return t
return None
def compound_detail_md(landscape: dict | None, label: str) -> str:
if not landscape:
return "*Select a compound to see its pipeline stage and mechanisms.*"
t = _compound_by_label(landscape, label)
if not t:
return "*Select a compound above.*"
stage = _top_ring(t["trials"]) or "Preclinical / not applicable"
header = (f"### {t['name']}\n{t['modality']} · Target: {t['target']} · "
f"Pipeline stage: {stage}")
if t.get("aliases"):
header += f"\nAlso: {', '.join(t['aliases'][:6])}"
mechs = t.get("mechanisms") or []
if not mechs:
return header + "\n\n**Mechanism of action:** _Not established from the available evidence._"
lines = [header, "\n**Mechanism(s) of action:**"]
for m in mechs:
conf = int(round(m["confidence"] * 100))
ev = f" — {m['evidence']}" if m.get("evidence") else ""
lines.append(f"- **{m['class']}** · _{m['role']}, {conf}% confidence_{ev}")
return "\n".join(lines)
def compound_trials_html(landscape: dict | None, label: str) -> str:
if not landscape:
return ""
t = _compound_by_label(landscape, label)
if not t:
return ""
rows = []
for tr in t["trials"]:
color, badge_label = _STATUS_BADGE.get(tr["status_group"], _STATUS_BADGE["other"])
badge = (f'{badge_label}')
phase = html.escape((tr.get("phase") or "—").replace("PHASE", "Ph"))
title = html.escape(tr.get("title", "")[:110])
nct = html.escape(tr.get("nct_id", ""))
url = html.escape(tr.get("url", ""))
sponsor = html.escape((tr.get("sponsor") or "")[:40])
rows.append(
f'