Py.Cafe

import solara
import pandas as pd
import altair as alt
from dataclasses import dataclass

# ┌──────────────────────────────────────────────────────────────────────────────┐
# │ 1. DATA LOADING: 全局只读数据初始化                                           │
# │ (这部分与 NiceGUI/Dash 版本逻辑完全一致，只读入一次，节省内存)                  │
# └──────────────────────────────────────────────────────────────────────────────┘

# 加载事实表与维度表
df_fact = pd.read_excel("FactPerformance.xlsx", sheet_name="Sheet1")
df_dimStu = pd.read_excel("DimStudents.xlsx", sheet_name="Sheet1")
df_dimCal = pd.read_excel("DimCalendar.xlsx", sheet_name="Date")
df_dimSub = pd.read_excel("DimSubjects.xlsx", sheet_name="DimSubjects")

# 关联数据（星型模型展开）
df = pd.merge(df_fact, df_dimStu[["StudentID", "GradeLevel"]], on="StudentID", how="left")
df = pd.merge(df, df_dimSub[["SubjectID", "SubjectName"]], on="SubjectID", how="left")

# 构造时间字段
df_dimCal["YearQuarterConcat"] = df_dimCal["Year"].astype(str) + " Q" + df_dimCal["QuarterNumber"].astype(str)
df_dimCal["YearMonthConcat"] = df_dimCal["Year"].astype(str) + "-" + df_dimCal["Month"].apply(lambda x: f"{x:02d}")
df = pd.merge(df, df_dimCal[["DateKey", "YearQuarterConcat", "YearMonthConcat"]], on="DateKey", how="left")

# 衍生指标与计算列
df["Weight"] = 1
df["WeightedScore"] = df["Score"] * df["Weight"]
df["PassedScore"] = df["Score"].apply(lambda x: "Pass" if x >= 55 else "Fail")

def get_grade(score):
    if score > 84: return "A"
    if score > 74: return "B"
    if score > 64: return "C"
    if score > 54: return "D"
    return "F"

df["Assessment_Grade"] = df["Score"].apply(get_grade)
# 设置有序分类，保证图表排序 A, B, C...
grade_order = ['A', 'B', 'C', 'D', 'F']
df['Assessment_Grade'] = pd.Categorical(df['Assessment_Grade'], categories=grade_order, ordered=True)


# ┌──────────────────────────────────────────────────────────────────────────────┐
# │ 2. STATE MANAGEMENT: 响应式状态管理类                                         │
# │                                                                              │
# │ ★ Solara 核心机制：                                                           │
# │   使用 solara.reactive 定义状态。当这些变量被修改时，使用它们的组件会自动刷新。     │
# └──────────────────────────────────────────────────────────────────────────────┘

class State:
    # 定义响应式变量 (初始值为 "All")
    grade = solara.reactive("All")
    level = solara.reactive("All")
    time = solara.reactive("All")
    subject = solara.reactive("All")
    
    # 视图模式：'Quarter' 或 'Month'
    view_mode = solara.reactive("Quarter")

    @staticmethod
    def reset_filters():
        """重置所有筛选器"""
        State.grade.set("All")
        State.level.set("All")
        State.time.set("All")
        State.subject.set("All")
        State.view_mode.set("Quarter")

    @staticmethod
    def get_data(ignore_grade=False, ignore_level=False, ignore_time=False, ignore_subject=False):
        """
        核心数据过滤函数（三步法中的 Filter 步骤）
        返回当前筛选条件下的数据副本。
        """
        d = df.copy()
        
        # 逐个应用筛选，除非被标记为 ignore (用于绘制该维度的分布图)
        if not ignore_grade and State.grade.value != 'All':
            d = d[d["Assessment_Grade"] == State.grade.value]
            
        if not ignore_level and State.level.value != 'All':
            d = d[d["GradeLevel"] == State.level.value]
            
        if not ignore_subject and State.subject.value != 'All':
            d = d[d["SubjectName"] == State.subject.value]
            
        # 时间筛选逻辑（支持 Q 和 Month）
        curr_time = State.time.value
        if not ignore_time and curr_time != 'All':
            if 'Q' in curr_time:
                d = d[d["YearQuarterConcat"] == curr_time]
            else:
                d = d[d["YearMonthConcat"] == curr_time]
                
        return d

# ┌──────────────────────────────────────────────────────────────────────────────┐
# │ 3. UI COMPONENTS: 视图组件                                                   │
# │                                                                              │
# │ ★ 装饰器 @solara.component：                                                  │
# │   标记这是一个 UI 组件。函数内部可以包含 Python 逻辑，返回 Solara 元素。           │
# └──────────────────────────────────────────────────────────────────────────────┘

@solara.component
def KpiCard(title, value, color_class="text-gray-800", bg_gradient=False):
    """通用的 KPI 卡片组件"""
    
    # 定义样式：根据是否需要渐变背景调整
    style = "padding: 16px; border-radius: 8px; box-shadow: 0 2px 4px rgba(0,0,0,0.1);"
    if bg_gradient:
        style += "background: linear-gradient(45deg, #6a11cb 0%, #2575fc 100%); color: white;"
    else:
        style += "background: white;"

    with solara.Column(style=style):
        solara.Text(title, style="font-size: 0.9rem; opacity: 0.8; font-weight: 500;")
        # 这里的 color_class 仅在非渐变背景下生效
        val_style = "font-size: 2rem; font-weight: bold;"
        if not bg_gradient:
            # 简单的 CSS 类模拟
            pass 
        solara.Text(value, style=val_style)

@solara.component
def KpiSection():
    """KPI 区域：获取数据并渲染4个卡片"""
    # 1. 获取全筛选数据
    d = State.get_data()
    
    # 2. 计算指标 (与之前逻辑一致)
    if d.empty:
        avg, weighted, pass_rate, perfect = "0.00", "0.00%", "0.00%", "0.0%"
    else:
        avg = f"{d['Score'].mean():.2f}"
        
        w_sum = d["Weight"].sum()
        w_val = (d["WeightedScore"].sum() / w_sum) if w_sum > 0 else 0
        # 自动转换百分比
        w_display = w_val * 100 if w_val <= 1 else w_val
        weighted = f"{w_display:.2f}%"
        
        p_rate = (d['PassedScore'] == 'Pass').mean() * 100
        pass_rate = f"{p_rate:.2f}%"
        
        target = 100 if df["Score"].max() > 1.0 else 1.0
        perf = (d['Score'] == target).mean() * 100
        perfect = f"{perf:.1f}%"

    # 3. 布局渲染
    with solara.GridFixed(columns=4, gap="16px"):
        KpiCard("Average Score", avg, bg_gradient=True)
        KpiCard("Weighted Avg", weighted, bg_gradient=True)
        KpiCard("Pass Rate", pass_rate, color_class="text-green-600") # 注意: Solara 简单样式需内联或用 extra CSS
        KpiCard("Perfect Scores", perfect, color_class="text-blue-600")

# ── 图表组件构建函数 (Altair Helpers) ──────────────────────────────────────────

def create_pie_chart(data, dimension, selected_val, title_text):
    """创建 Altair 环形图的辅助函数"""
    base = alt.Chart(data).encode(
        theta=alt.Theta("Count", stack=True)
    )
    
    # 交互：点击选择
    # 注意：Solara 中处理点击通常通过 signal 或 click handler，这里我们简化为视觉上的高亮
    # 真正的点击逻辑在 solara.FigureAltair 的 on_selection 中处理
    
    pie = base.mark_arc(innerRadius=60, outerRadius=100).encode(
        color=alt.Color(dimension),
        order=alt.Order("Count", sort="descending"),
        opacity=alt.condition(
            alt.datum[dimension] == selected_val, alt.value(1.0), alt.value(0.3)
        ) if selected_val != 'All' else alt.value(1.0),
        tooltip=[dimension, "Count"]
    )
    
    text = base.mark_text(radius=140).encode(
        text=alt.Text("Count"),
        order=alt.Order("Count", sort="descending"),
        color=alt.value("black")  
    )
    
    # 中心文字
    center_text = alt.Chart(pd.DataFrame({'text': [title_text]})).mark_text(
        size=20, fontWeight='bold'
    ).encode(text='text')

    return (pie + text + center_text).properties(height=250)

# ── 具体的图表组件 ────────────────────────────────────────────────────────────

@solara.component
def GradeDistributionChart():
    # 1. 忽略 Grade 筛选获取数据
    d = State.get_data(ignore_grade=True)
    
    if d.empty:
        solara.Text("No Data")
        return

    # 2. 聚合
    agg = d['Assessment_Grade'].value_counts().reset_index()
    agg.columns = ['Assessment_Grade', 'Count']
    
    # 3. 构建 Altair 图表
    chart = create_pie_chart(
        agg, 'Assessment_Grade', State.grade.value, f"{len(State.get_data())}\nTests"
    )
    
    # 4. 定义点击回调
    def on_click(selection):
        # selection 格式如: {'Assessment_Grade': ['A']} 或 None
        if selection and 'Assessment_Grade' in selection:
            val = selection['Assessment_Grade'][0]
            # 切换逻辑
            new_val = "All" if State.grade.value == val else val
            State.grade.set(new_val)

    # 渲染图表
    with solara.Card("Grade Distribution"):
        # 启用 Altair 的选择交互功能
        select = alt.selection_point(fields=['Assessment_Grade'], on='click', clear='dblclick')
        final_chart = chart.add_params(select)
        
        solara.FigureAltair(final_chart, on_selection=on_click)


@solara.component
def LevelDistributionChart():
    d = State.get_data(ignore_level=True)
    if d.empty: return

    agg = d.groupby('GradeLevel')['StudentID'].nunique().reset_index()
    agg.columns = ['GradeLevel', 'Count']

    chart = create_pie_chart(
        agg, 'GradeLevel', State.level.value, f"{State.get_data()['StudentID'].nunique()}\nStudents"
    )

    def on_click(selection):
        if selection and 'GradeLevel' in selection:
            val = selection['GradeLevel'][0]
            new_val = "All" if State.level.value == val else val
            State.level.set(new_val)

    with solara.Card("Level Distribution"):
        select = alt.selection_point(fields=['GradeLevel'], on='click', clear='dblclick')
        solara.FigureAltair(chart.add_params(select), on_selection=on_click)


@solara.component
def TimeTrendChart():
    # 1. 获取数据（忽略时间筛选）与上下文
    d = State.get_data(ignore_time=True)
    mode = State.view_mode.value
    current_time = State.time.value
    
    # 2. 下钻逻辑处理：如果在月视图且选了季度，只显示该季度的数据
    display_title = "Performance Over Time"
    if mode == "Month":
        if 'Q' in current_time:
            # 筛选该季度的月份
            d = d[d["YearQuarterConcat"] == current_time]
            display_title = f"Monthly Breakdown: {current_time}"
        elif '-' in current_time:
             # 如果选的是具体的月，显示该月所属季度的所有月作为上下文
             # (此处为简化，直接显示所有月，高亮选中月)
             pass

    if d.empty: return

    # 3. 聚合
    col_group = "YearQuarterConcat" if mode == "Quarter" else "YearMonthConcat"
    agg = d.groupby(col_group)["Score"].mean().reset_index()

    # 4. Altair 绘图
    base = alt.Chart(agg)
    bars = base.mark_bar().encode(
        x=alt.X(col_group, sort=None),
        y=alt.Y("Score", scale=alt.Scale(domain=[0, 100])),
        # 高亮逻辑
        opacity=alt.condition(
            alt.datum[col_group] == current_time, alt.value(1.0), alt.value(0.3)
        ) if current_time != 'All' and not (mode=='Month' and 'Q' in current_time) else alt.value(1.0),
        tooltip=[col_group, "Score"]
    )
    
    # 添加平均线
    rule = alt.Chart(pd.DataFrame({'y': [State.get_data()['Score'].mean()]})).mark_rule(color='red', strokeDash=[5,5]).encode(y='y')

    # 5. 点击交互逻辑 (含下钻)
    def on_click(selection):
        if selection and col_group in selection:
            clicked_val = selection[col_group][0]
            
            if mode == 'Quarter':
                # 下钻：点击季度 -> 切换到月视图，并设置筛选为该季度
                State.time.set(clicked_val)
                State.view_mode.set("Month")
            else:
                # 过滤：点击月 -> 切换选中/取消
                new_val = "All" if State.time.value == clicked_val else clicked_val
                State.time.set(new_val)

    with solara.Card("Time Analysis"):
        with solara.Row(justify="space-between", style="margin-bottom: 10px"):
            solara.Text(display_title, style="font-weight:bold")
            # 视图切换 Toggle
            solara.ToggleButtonsSingle(value=State.view_mode, values=["Quarter", "Month"])
            
        select = alt.selection_point(fields=[col_group], on='click', clear='dblclick')
        solara.FigureAltair((bars + rule).add_params(select), on_selection=on_click)


@solara.component
def SubjectChart():
    d = State.get_data(ignore_subject=True)
    if d.empty: return
    
    agg = d.groupby("SubjectName")["Score"].mean().reset_index().sort_values("Score", ascending=False)
    
    bars = alt.Chart(agg).mark_bar().encode(
        x=alt.X("SubjectName", sort=None),
        y="Score",
        opacity=alt.condition(
            alt.datum.SubjectName == State.subject.value, alt.value(1.0), alt.value(0.3)
        ) if State.subject.value != 'All' else alt.value(1.0)
    )
    
    def on_click(selection):
        if selection and 'SubjectName' in selection:
            val = selection['SubjectName'][0]
            new_val = "All" if State.subject.value == val else val
            State.subject.set(new_val)

    with solara.Card("Subject Performance"):
        select = alt.selection_point(fields=['SubjectName'], on='click', clear='dblclick')
        solara.FigureAltair(bars.add_params(select), on_selection=on_click)


# ┌──────────────────────────────────────────────────────────────────────────────┐
# │ 4. MAIN LAYOUT: 主界面布局                                                    │
# └──────────────────────────────────────────────────────────────────────────────┘

@solara.component
def Page():
    # 设置页面标题
    solara.Title("Solara Education Dashboard")
    
    with solara.Column(style="max-width: 1200px; margin: 0 auto; padding: 20px;"):
        
        # ── 顶部栏 ──
        with solara.Row(justify="space-between", align="center", style="margin-bottom: 20px"):
            solara.Text("Education Performance Analysis (Solara)", style="font-size: 24px; font-weight: bold;")
            
            # 显示当前筛选状态文本
            status_text = f"Filters: {State.grade.value} | {State.level.value} | {State.time.value} | {State.subject.value}"
            solara.Text(status_text, style="color: #666; font-size: 14px;")
            
            solara.Button("Reset Filters", on_click=State.reset_filters, color="grey")

        # ── 1. KPI 区域 ──
        KpiSection()
        
        solara.Text("", style="margin-top: 20px;") # Spacer

        # ── 2. 第一行图表 (Grid 布局) ──
        with solara.GridFixed(columns=2, gap="20px"):
            GradeDistributionChart()
            LevelDistributionChart()

        # ── 3. 第二行图表 ──
        with solara.GridFixed(columns=2, gap="20px", style="margin-top: 20px"):
            TimeTrendChart()
            SubjectChart()

# 启动入口 (Solara 会自动寻找 Page 组件)
# 运行命令: solara run your_script.py