lerningOs / app.py
YVES28's picture
Update app.py
4dae44c verified
Raw History Blame Contribute Delete
17.6 kB
"""
12D AI Assessment Engine V2.2 - Gradio SDK entry point for Hugging Face Spaces.
Uses demo.launch() to trigger ZeroGPU startup detection.
Multi-portal UI integrated: Parent (with Interactive Scale/Assessment), Doctor, and School portals.
"""
import os
import sys
import time
import logging
# -------------------------------------------------------------------
# 环境路径与目录初始化
# -------------------------------------------------------------------
_backend_dir = os.path.join(os.path.dirname(__file__), "backend")
sys.path.insert(0, _backend_dir)
_data_dir = os.path.join(os.path.dirname(__file__), "data")
os.makedirs(_data_dir, exist_ok=True)
import gradio as gr
import spaces
# ZeroGPU placeholder
@spaces.GPU(duration=5)
def _zerogpu_placeholder():
pass
# -------------------------------------------------------------------
# 模拟 12D 评估量表题库数据 (Assessment Scale Item Bank)
# -------------------------------------------------------------------
SCALE_ITEMS = [
{
"id": "q1",
"dimension": "专注力 (Attention)",
"question": "1. 孩子在完成日常作业或游戏任务时,容易被无关刺激干扰吗?",
"options": ["从不", "偶有", "经常", "总是"],
},
{
"id": "q2",
"dimension": "工作记忆 (Working Memory)",
"question": "2. 给孩子连续下达3个简单口头指令(如:关灯、拿书、坐下),能否准确执行?",
"options": ["完全不能", "能完成1项", "能完成2项", "全部准确完成"],
},
{
"id": "q3",
"dimension": "抑制控制 (Inhibitory Control)",
"question": "3. 在需要轮流等待或遵守规则的场景下,孩子能否控制冲动?",
"options": ["极难控制", "需要提醒", "基本可以", "完全没问题"],
},
{
"id": "q4",
"dimension": "空间想象 (Spatial Cognition)",
"question": "4. 孩子拼图或进行积木搭建时,对空间位置变换的感知表现如何?",
"options": ["较困难", "一般", "良好", "极优秀"],
},
]
# -------------------------------------------------------------------
# 构建 Gradio 多端交互界面
# -------------------------------------------------------------------
with gr.Blocks(title="12D AI Assessment Engine V2.2", theme=gr.themes.Soft()) as demo:
gr.Markdown(
"""
# 💡 12D AI Assessment Engine V2.2
**LearningOS V2.0 Enterprise Edition** — 智能化儿童认知能力多端评估与诊断平台
---
"""
)
with gr.Tabs():
# ==================== 1. 👶 家长端 Portal (含评估量表) ====================
with gr.Tab("👶 家长端 (Parent Portal)"):
gr.Markdown("### 👶 家长工作台 — 儿童档案与 12D 认知能力评估量表")
with gr.Row():
# 左侧:档案管理
with gr.Column(scale=1):
p_name = gr.Textbox(label="孩子姓名/昵称", value="小敏")
p_age = gr.Number(label="年龄", value=7, precision=0)
p_gender = gr.Dropdown(
label="性别", choices=["男", "女", "其他"], value="男"
)
p_create_btn = gr.Button("➕ 建立儿童档案", variant="primary")
p_create_output = gr.JSON(label="档案创建结果")
# 右侧:量表测评答题区
with gr.Column(scale=2):
p_child_id = gr.Textbox(label="儿童 ID", value="child_demo_888")
p_start_assess_btn = gr.Button(
"🚀 加载/发起 12D 评估量表", variant="secondary"
)
# 评估量表交互区域
with gr.Group(visible=False) as scale_group:
gr.Markdown("#### 📋 12D AI 认知能力快速筛查量表")
q1_radio = gr.Radio(
choices=SCALE_ITEMS[0]["options"],
label=f"【{SCALE_ITEMS[0]['dimension']}】 {SCALE_ITEMS[0]['question']}",
)
q2_radio = gr.Radio(
choices=SCALE_ITEMS[1]["options"],
label=f"【{SCALE_ITEMS[1]['dimension']}】 {SCALE_ITEMS[1]['question']}",
)
q3_radio = gr.Radio(
choices=SCALE_ITEMS[2]["options"],
label=f"【{SCALE_ITEMS[2]['dimension']}】 {SCALE_ITEMS[2]['question']}",
)
q4_radio = gr.Radio(
choices=SCALE_ITEMS[3]["options"],
label=f"【{SCALE_ITEMS[3]['dimension']}】 {SCALE_ITEMS[3]['question']}",
)
p_submit_answers_btn = gr.Button(
"📤 提交量表答案并计算分析", variant="primary"
)
p_assess_output = gr.JSON(label="评估 Session & 答题结果")
p_report_output = gr.JSON(label="📊 智能生成 12D 认知测评报告")
# 逻辑响应函数
def _parent_create_child(name, age, gender):
return {
"status": "success",
"message": f"成功为 {name} ({age}岁, {gender}) 建立档案",
"child_id": f"child_{name}_888",
}
def _parent_start_assess(child_id):
if not child_id:
return gr.update(visible=False), {"error": "请输入有效的儿童 ID"}
session_data = {
"status": "active",
"session_id": "sess_12d_999",
"child_id": child_id,
"total_items": len(SCALE_ITEMS),
"message": "量表加载成功,请在下方完成评估量表勾选。",
}
return gr.update(visible=True), session_data
def _parent_submit_scale(child_id, a1, a2, a3, a4):
score_map = {
"从不": 100, "偶有": 80, "经常": 60, "总是": 40,
"完全不能": 40, "能完成1项": 60, "能完成2项": 80, "全部准确完成": 100,
"极难控制": 40, "需要提醒": 60, "基本可以": 80, "完全没问题": 100,
"较困难": 40, "一般": 60, "良好": 80, "极优秀": 100,
}
s1 = score_map.get(a1, 70)
s2 = score_map.get(a2, 70)
s3 = score_map.get(a3, 70)
s4 = score_map.get(a4, 70)
avg_score = round((s1 + s2 + s3 + s4) / 4, 1)
report_data = {
"child_id": child_id or "child_demo_888",
"overall_score": avg_score,
"dimension_scores": {
"专注力": s1,
"工作记忆": s2,
"抑制控制": s3,
"空间想象": s4,
},
"strengths": [
k
for k, v in {
"专注力": s1,
"工作记忆": s2,
"抑制控制": s3,
"空间想象": s4,
}.items()
if v >= 80
],
"weaknesses": [
k
for k, v in {
"专注力": s1,
"工作记忆": s2,
"抑制控制": s3,
"空间想象": s4,
}.items()
if v < 70
],
"ai_suggestions": f"综合得分 {avg_score} 分。"
+ (
"建议多安排专注力训练游戏及户外有氧运动。"
if avg_score < 80
else "各项能力表现优异,保持当前引导策略!"
),
}
return {
"status": "completed",
"submitted_answers": {"q1": a1, "q2": a2, "q3": a3, "q4": a4},
}, report_data
p_create_btn.click(
_parent_create_child,
inputs=[p_name, p_age, p_gender],
outputs=p_create_output,
)
p_start_assess_btn.click(
_parent_start_assess,
inputs=[p_child_id],
outputs=[scale_group, p_assess_output],
)
p_submit_answers_btn.click(
_parent_submit_scale,
inputs=[p_child_id, q1_radio, q2_radio, q3_radio, q4_radio],
outputs=[p_assess_output, p_report_output],
)
# ==================== 2. 👨‍⚕️ 医生端 Portal ====================
with gr.Tab("👨‍⚕️ 医生端 (Doctor Portal)"):
gr.Markdown("### 👨‍⚕️ 医生临床诊断工作台 — 深入认知量表与危险因子诊断")
with gr.Row():
with gr.Column():
d_child_id = gr.Textbox(label="患者/儿童 ID", value="child_demo_888")
d_adhd_score = gr.Slider(0, 10, value=2, label="ADHD 倾向评估分 (0-10)")
d_asd_score = gr.Slider(0, 10, value=1, label="ASD 倾向评估分 (0-10)")
d_notes = gr.TextArea(
label="临床诊断意见", placeholder="录入医生专业评语或干预方案..."
)
d_submit_btn = gr.Button("💾 提交临床诊断报告", variant="primary")
d_submit_output = gr.JSON(label="诊断录入结果")
with gr.Column():
d_twin_btn = gr.Button(
"🧬 提取临床级数字孪生 (Diagnostic Twin)", variant="secondary"
)
d_twin_output = gr.JSON(label="深度多维认知矩阵")
def _doctor_submit_diag(child_id, adhd, asd, notes):
return {
"status": "success",
"patient_id": child_id,
"adhd_risk_index": adhd / 10.0,
"asd_risk_index": asd / 10.0,
"clinical_notes": notes or "未填写评语",
"message": "临床数据已保存至云端电子病历系统",
}
def _doctor_get_twin(child_id):
return {
"child_id": child_id,
"raw_cognitive_matrix": {
"working_memory": 92,
"inhibitory_control": 78,
"cognitive_flexibility": 85,
"processing_speed": 88,
},
"neurological_risk": "Low Risk",
}
d_submit_btn.click(
_doctor_submit_diag,
inputs=[d_child_id, d_adhd_score, d_asd_score, d_notes],
outputs=d_submit_output,
)
d_twin_btn.click(
_doctor_get_twin, inputs=[d_child_id], outputs=d_twin_output
)
# ==================== 3. 🏫 学校端 Portal ====================
with gr.Tab("🏫 学校端 (School Portal)"):
gr.Markdown("### 🏫 学校管理工作台 — 班级认知分布与群体预警")
with gr.Row():
with gr.Column():
s_school_name = gr.Textbox(label="学校名称", value="第一实验小学")
s_class_name = gr.Textbox(label="班级", value="三年级 (2) 班")
s_batch_btn = gr.Button("📦 一键导入班级学生名单", variant="primary")
s_batch_output = gr.JSON(label="导入结果")
with gr.Column():
s_class_id = gr.Textbox(label="班级 ID", value="class_302")
s_analytics_btn = gr.Button(
"📈 生成班级群体认知分析", variant="secondary"
)
s_analytics_output = gr.JSON(label="班级报告与预警名单")
def _school_batch_import(school, cls):
return {
"status": "success",
"school": school,
"class": cls,
"imported_students": 45,
}
def _school_get_analytics(cls_id):
return {
"class_id": cls_id,
"total_students": 45,
"class_average_score": 81.5,
"dimension_averages": {"逻辑": 83, "记忆": 80, "专注": 77},
"attention_warning_list": ["学生_03", "学生_17", "学生_29"],
}
s_batch_btn.click(
_school_batch_import,
inputs=[s_school_name, s_class_name],
outputs=s_batch_output,
)
s_analytics_btn.click(
_school_get_analytics,
inputs=[s_class_id],
outputs=s_analytics_output,
)
# ==================== 4. ⚙️ 系统与 API 状态 ====================
with gr.Tab("⚙️ 系统与 API 文档"):
gr.Markdown(
"""
## 📌 系统访问与 Demo 凭据
* **Swagger UI (OpenAPI 接口文档)**: [`/docs`](/docs)
* **Health check (健康检查)**: [`/health`](/health)
* **演示管理员账号:** `admin` | **密码:** `Admin123456!`
"""
)
status_box = gr.Textbox(label="System Status", interactive=False)
check_btn = gr.Button("Check System Status")
def check_status():
return "Engine: 12D AI Assessment Engine V2.2 | Status: Operational | API Docs: /docs | Health: /health"
check_btn.click(fn=check_status, outputs=status_box)
# -------------------------------------------------------------------
# 启动 Gradio 前端服务器
# -------------------------------------------------------------------
demo.launch(
server_name="0.0.0.0",
server_port=7860,
prevent_thread_lock=True,
quiet=True,
ssr_mode=False,
app_kwargs={"docs_url": "/docs", "redoc_url": "/redoc"},
)
# -------------------------------------------------------------------
# FastAPI 路由注入与后台逻辑
# -------------------------------------------------------------------
from app.api.v1 import tenant, assessment, report, analytics, auth, child
from app.api.v1.endpoints import billing_audit_demo, offline_sync, compliance
from app.api.v2.endpoints import multimodal_stream
from app.core.config import settings
from fastapi.staticfiles import StaticFiles
app = demo.app
app.include_router(auth.router, prefix="/api/v1")
app.include_router(tenant.router, prefix="/api/v1")
app.include_router(child.router, prefix="/api/v1")
app.include_router(assessment.router, prefix="/api/v1")
app.include_router(report.router, prefix="/api/v1")
app.include_router(analytics.router, prefix="/api/v1")
app.include_router(billing_audit_demo.router, prefix="/api/v1")
app.include_router(offline_sync.router, prefix="/api/v1")
app.include_router(compliance.router, prefix="/api/v1")
app.include_router(multimodal_stream.router, prefix="/api/v2")
_static_dir = os.path.join(os.path.dirname(__file__), "backend", "static")
os.makedirs(_static_dir, exist_ok=True)
app.mount("/wasm-static", StaticFiles(directory=_static_dir), name="wasm-static")
@app.get("/health")
def health_check():
return {
"status": "healthy",
"engine": "12D AI Assessment Engine V2.2",
"version": settings.APP_VERSION,
}
@app.get("/api")
def api_info():
return {
"service": "12D AI Assessment Engine V2.2",
"docs": "/docs",
"health": "/health",
}
# -------------------------------------------------------------------
# 数据库与 Seed 初始化逻辑
# -------------------------------------------------------------------
logging.basicConfig(
level=logging.INFO, format="%(asctime)s | %(levelname)s | %(name)s | %(message)s"
)
_logger = logging.getLogger("main")
try:
from app.db.session import engine
from app.models.base import Base
from app.models.auth import User, Role, Permission, Organization
from app.models.child import Child, Guardian
from app.models.assessment import (
AssessmentItem,
AssessmentSession,
AssessmentResponse,
)
from app.models.report import AssessmentReport
from app.models.tenant import Tenant
from app.core.audit_logger import AuditLog
Base.metadata.create_all(bind=engine)
_logger.info("Database tables created/verified.")
from scripts.seed_database import seed_data
seed_data()
_logger.info("Demo data seeded successfully.")
except Exception as e:
_logger.error(f"Startup initialization error: {e}")
# 主进程保持运行状态
if __name__ == "__main__":
try:
while True:
time.sleep(3600)
except KeyboardInterrupt:
demo.close()