基于Taipy GUI构建带筛选的仪表盘及适配SQLite3、SPARQL的技术求助

一、CSV数据加载与单日评估结果实时展示 #

1. CSV数据加载处理 #

针对字符串为主的数据，加载时明确指定列类型，避免自动类型转换异常：

import pandas as pd
from taipy.gui import Gui, State

# 加载CSV文件，指定核心列类型为字符串
df = pd.read_csv("assessment_data.csv", dtype={"实体ID": str, "实体名称": str})
# 统一日期格式为日期类型（便于后续筛选）
df["评估日期"] = pd.to_datetime(df["评估日期"]).dt.date

2. 实体/日期筛选与实时更新 #

通过Taipy的组件回调实现筛选逻辑，筛选后数据实时同步到仪表盘表格：

# 初始化筛选参数：默认选中最新评估日期
selected_entity = None
selected_date = df["评估日期"].max()
filtered_df = df[df["评估日期"] == selected_date]

def on_filter_change(state: State):
    # 基于选中的实体和日期过滤数据
    temp_df = state.df.copy()
    if state.selected_date:
        temp_df = temp_df[temp_df["评估日期"] == state.selected_date]
    if state.selected_entity:
        temp_df = temp_df[temp_df["实体名称"] == state.selected_entity]
    state.filtered_df = temp_df

# 仪表盘页面布局
dashboard_page = """
# 评估结果仪表盘
<|layout|columns=1 2|
<|筛选区|
<|实体选择|dropdown|lov={list(df["实体名称"].unique())}|value={selected_entity}|on_change=on_filter_change|>
<|日期选择|date|value={selected_date}|on_change=on_filter_change|>
|>
<|单日评估结果|table|data={filtered_df}|columns=["实体名称", "评估日期", "指标1", "指标2", "指标3", "指标4", "指标5"]|>
|>
"""

二、分页评估信息收集表单 #

利用Taipy的页面导航实现分页表单，支持评估信息录入与保存：

# 表单临时存储数据
form_data = {
    "实体名称": "",
    "评估日期": pd.Timestamp.today().date(),
    "指标1": "",
    "指标2": "",
    "指标3": "",
    "指标4": "",
    "指标5": ""
}

def save_assessment(state: State):
    # 将表单数据追加到主数据框
    new_row = pd.DataFrame([state.form_data])
    state.df = pd.concat([state.df, new_row], ignore_index=True)
    # 重置指标输入框
    for k in state.form_data:
        if k.startswith("指标"):
            state.form_data[k] = ""

# 表单页面布局
form_page = """
# 评估信息收集
<|实体名称|input|value={form_data["实体名称"]}|>
<|评估日期|date|value={form_data["评估日期"]}|>
<|指标1|input|value={form_data["指标1"]}|>
<|指标2|input|value={form_data["指标2"]}|>
<|指标3|input|value={form_data["指标3"]}|>
<|指标4|input|value={form_data["指标4"]}|>
<|指标5|input|value={form_data["指标5"]}|>
<|保存评估|button|on_action=save_assessment|>
"""

# 全局导航页面
root_page = """
<|menu|label=导航|
<|仪表盘|page=dashboard_page|>
<|评估表单|page=form_page|>
|>
"""

# 启动应用
Gui(pages={"/": root_page, "dashboard_page": dashboard_page, "form_page": form_page}).run()

三、SQLite3数据源适配 #

替换CSV为SQLite存储，支持数据持久化读写：

import sqlite3

# 初始化SQLite数据库表
def init_sqlite_db():
    conn = sqlite3.connect("assessment.db")
    cursor = conn.cursor()
    cursor.execute("""
    CREATE TABLE IF NOT EXISTS assessments (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        entity_name TEXT,
        assess_date DATE,
        metric1 TEXT,
        metric2 TEXT,
        metric3 TEXT,
        metric4 TEXT,
        metric5 TEXT
    )
    """)
    conn.commit()
    conn.close()

# 从SQLite读取数据转为DataFrame
def load_from_sqlite():
    conn = sqlite3.connect("assessment.db")
    df = pd.read_sql("SELECT * FROM assessments", conn)
    df["assess_date"] = pd.to_datetime(df["assess_date"]).dt.date
    conn.close()
    return df

# 将表单数据写入SQLite
def save_to_sqlite(form_data):
    conn = sqlite3.connect("assessment.db")
    cursor = conn.cursor()
    cursor.execute("""
    INSERT INTO assessments (entity_name, assess_date, metric1, metric2, metric3, metric4, metric5)
    VALUES (?, ?, ?, ?, ?, ?, ?)
    """, (
        form_data["实体名称"],
        form_data["评估日期"],
        form_data["指标1"],
        form_data["指标2"],
        form_data["指标3"],
        form_data["指标4"],
        form_data["指标5"]
    ))
    conn.commit()
    conn.close()

# 修改保存函数适配SQLite
def save_assessment(state: State):
    save_to_sqlite(state.form_data)
    state.df = load_from_sqlite()  # 更新数据框
    # 重置指标输入框
    for k in state.form_data:
        if k.startswith("指标"):
            state.form_data[k] = ""

# 初始化数据库并加载数据
init_sqlite_db()
df = load_from_sqlite()

四、SPARQL查询数据源适配 #

使用SPARQLWrapper库查询RDF数据，转换为DataFrame格式：

from SPARQLWrapper import SPARQLWrapper, JSON

def load_from_sparql(endpoint_url):
    sparql = SPARQLWrapper(endpoint_url)
    # 替换为实际SPARQL查询语句
    sparql.setQuery("""
    PREFIX ex: <http://example.org/assessment/>
    SELECT ?entityName ?assessDate ?metric1 ?metric2 ?metric3 ?metric4 ?metric5
    WHERE {
        ?assessment ex:entityName ?entityName ;
                    ex:assessDate ?assessDate ;
                    ex:metric1 ?metric1 ;
                    ex:metric2 ?metric2 ;
                    ex:metric3 ?metric3 ;
                    ex:metric4 ?metric4 ;
                    ex:metric5 ?metric5 .
    }
    """)
    sparql.setReturnFormat(JSON)
    results = sparql.query().convert()
    
    # 转换查询结果为DataFrame
    data_rows = []
    for result in results["results"]["bindings"]:
        data_rows.append({
            "实体名称": result["entityName"]["value"],
            "评估日期": pd.to_datetime(result["assessDate"]["value"]).dt.date,
            "指标1": result["metric1"]["value"],
            "指标2": result["metric2"]["value"],
            "指标3": result["metric3"]["value"],
            "指标4": result["metric4"]["value"],
            "指标5": result["metric5"]["value"]
        })
    return pd.DataFrame(data_rows)

# 加载SPARQL数据示例
df = load_from_sparql("http://your-sparql-endpoint-url")

内容的提问来源于stack exchange，提问作者Rym Guerbi Michaut