paper_lifecycle / app.py
elfsong
Smooth hype-cycle curve, hover-only labels, surface emerging topics
ace9bfd
Raw
History Blame Contribute Delete
7.37 kB
"""Paper Lifecycle — Gartner hype-cycle visualization of arXiv CS topics.
每个季度一张"炒作周期"快照(累积到该季度),滑块拖动即可查看不同季度。
数据来自 src/lifecycle_quarterly.py 生成的 lifecycle_quarterly.json
(本地优先;HF Space 上从数据集仓库拉取)。
"""
import json
import os
from pathlib import Path
import numpy as np
import gradio as gr
import plotly.graph_objects as go
DATA_FILE = "lifecycle_quarterly.json"
HF_LIFECYCLE_REPO = os.getenv("HF_LIFECYCLE_REPO", "Elfsong/arxiv_cs_lifecycle")
def _load_data() -> dict:
p = Path(__file__).resolve().parent / DATA_FILE
if p.exists():
return json.loads(p.read_text())
from huggingface_hub import hf_hub_download
local = hf_hub_download(HF_LIFECYCLE_REPO, DATA_FILE,
repo_type="dataset", token=os.getenv("HF_TOKEN"))
return json.loads(Path(local).read_text())
DATA = _load_data()
QUARTERS = [q for q in DATA["quarters"] if DATA["snapshots"][q]["topics"]]
PHASES = DATA["phase_order"]
PHASE_COLOR = {
"Innovation Trigger": "#3b82f6",
"Peak of Inflated Expectations": "#ef4444",
"Trough of Disillusionment": "#8b5cf6",
"Slope of Enlightenment": "#f59e0b",
"Plateau of Productivity": "#22c55e",
}
PHASE_SHORT = {
"Innovation Trigger": "Innovation\nTrigger",
"Peak of Inflated Expectations": "Peak of Inflated\nExpectations",
"Trough of Disillusionment": "Trough of\nDisillusionment",
"Slope of Enlightenment": "Slope of\nEnlightenment",
"Plateau of Productivity": "Plateau of\nProductivity",
}
PHASE_X = {
"Innovation Trigger": (3, 13),
"Peak of Inflated Expectations": (15, 28),
"Trough of Disillusionment": (37, 52),
"Slope of Enlightenment": (58, 78),
"Plateau of Productivity": (82, 97),
}
def _curve_y(x):
"""平滑解析曲线:一个早期高斯"炒作峰" + 一条升向高原的 sigmoid。"""
peak = 82.0 * np.exp(-(((x - 18.0) / 8.5) ** 2))
plateau = 62.0 / (1.0 + np.exp(-(x - 60.0) / 6.5))
return 6.0 + peak + plateau
def _jit(s, lo, hi):
h = (hash(s) % 1000) / 1000.0
return lo + (hi - lo) * h
def _display_phase(t):
# 选作 emerging 的 topic 摆到 Innovation Trigger 区
return "Innovation Trigger" if t.get("emerging") else t["phase"]
def _select(snap, top_n):
"""top-N 成熟 topic(按累计计数) + 全部 emerging(始终显示)。"""
mature = [t for t in snap["topics"] if not t.get("emerging")][:int(top_n)]
emerging = [t for t in snap["topics"] if t.get("emerging")]
return mature + emerging
def build_figure(quarter: str, top_n: int):
snap = DATA["snapshots"][quarter]
topics = _select(snap, top_n)
fig = go.Figure()
# 1) 平滑曲线背景(解析函数密采样)
xs = np.linspace(0, 100, 400)
fig.add_trace(go.Scatter(
x=xs, y=_curve_y(xs), mode="lines",
line=dict(color="#cbd5e1", width=3, shape="spline"),
hoverinfo="skip", showlegend=False))
# 2) 相位底色 + 标签
for ph in PHASES:
lo, hi = PHASE_X[ph]
fig.add_vrect(x0=lo - 2.5, x1=hi + 2.5, fillcolor=PHASE_COLOR[ph],
opacity=0.05, line_width=0, layer="below")
fig.add_annotation(x=(lo + hi) / 2, y=-9, text=PHASE_SHORT[ph],
showarrow=False, align="center",
font=dict(size=10, color=PHASE_COLOR[ph]))
# 3) 按"显示相位"分组铺开(emerging → Innovation 区)
by_phase = {ph: [] for ph in PHASES}
for t in topics:
by_phase[_display_phase(t)].append(t)
max_cnt = max((t["total_count"] for t in topics), default=1)
for ph, items in by_phase.items():
if not items:
continue
items.sort(key=lambda r: r["current_avg"])
lo, hi = PHASE_X[ph]
n = len(items)
xpos, ypos, sizes, hovers = [], [], [], []
for i, t in enumerate(items):
x = lo + (hi - lo) * (i + 0.5) / n + _jit(t["topic"], -1.2, 1.2)
x = min(max(x, 1), 99)
xpos.append(x)
ypos.append(float(_curve_y(x)) + _jit(t["topic"] + "y", -3.5, 3.5))
sizes.append(10 + 34 * (t["total_count"] / max_cnt) ** 0.5)
tag = " 🌱 emerging" if t.get("emerging") else ""
hovers.append(
f"<b>{t['topic']}</b>{tag}<br>"
f"phase: {t['phase']}<br>"
f"papers: {t['total_count']}<br>"
f"recent share: {t.get('recent_fraction', '?')}<br>"
f"decline ratio: {t['decline_ratio']} slope: {t['slope']}<br>"
f"peak: {t['peak_quarter']}<extra></extra>")
fig.add_trace(go.Scatter(
x=xpos, y=ypos, mode="markers", name=ph,
marker=dict(size=sizes, color=PHASE_COLOR[ph], opacity=0.78,
line=dict(width=1, color="white")),
hovertemplate="%{hovertext}", hovertext=hovers))
fig.update_layout(
title=dict(text=f"<b>arXiv CS — Topic Hype Cycle · {quarter}</b>"
f" <span style='font-size:13px;color:#64748b'>"
f"{snap['n_papers']:,} papers · {len(topics)} topics</span>",
x=0.5, xanchor="center"),
xaxis=dict(range=[-3, 103], showgrid=False, zeroline=False,
showticklabels=False, title=""),
yaxis=dict(range=[-16, 108], showgrid=False, zeroline=False,
showticklabels=False, title="Expectations →"),
plot_bgcolor="white", height=620,
margin=dict(l=20, r=20, t=60, b=30),
hoverlabel=dict(bgcolor="white", font_size=12),
legend=dict(orientation="h", yanchor="bottom", y=1.02,
xanchor="center", x=0.5, font=dict(size=10)))
return fig
def update(idx: int, top_n: int):
q = QUARTERS[int(idx)]
snap = DATA["snapshots"][q]
shown = _select(snap, top_n)
# 图例计数按"显示相位"统计,和图一致
from collections import Counter
dc = Counter(_display_phase(t) for t in shown)
md = (f"### 📅 Snapshot: **{q}** | {snap['n_papers']:,} papers cumulative\n"
+ " ".join(f"<span style='color:{PHASE_COLOR[p]}'>●</span> "
f"{p.split(' ')[0]}: **{dc.get(p, 0)}**" for p in PHASES))
return build_figure(q, int(top_n)), md
with gr.Blocks(title="Paper Lifecycle") as demo:
gr.Markdown("# 🔄 arXiv CS Topic Lifecycle — Gartner Hype Cycle\n"
"拖动滑块查看不同季度(累积)的研究主题炒作周期。"
"点大小=累计论文数,颜色=阶段;🌱 为新兴主题(置于 Innovation 区)。"
"**鼠标悬停查看主题与指标。**")
with gr.Row():
idx = gr.Slider(0, len(QUARTERS) - 1, value=len(QUARTERS) - 1, step=1,
label=f"Quarter snapshot (0 = {QUARTERS[0]} … "
f"{len(QUARTERS)-1} = {QUARTERS[-1]})")
topn = gr.Slider(10, 120, value=50, step=5, label="Top-N topics")
info = gr.Markdown()
plot = gr.Plot()
idx.change(update, [idx, topn], [plot, info])
topn.change(update, [idx, topn], [plot, info])
demo.load(update, [idx, topn], [plot, info])
if __name__ == "__main__":
demo.launch(theme=gr.themes.Soft())