mirror of
https://github.com/RYDE-WORK/lnp_ml.git
synced 2026-10-01 13:25:46 +08:00
141 lines
3.2 KiB
Python
141 lines
3.2 KiB
Python
"""Streamlit 前端共享配置与工具。
|
||
"""
|
||
|
||
import os
|
||
|
||
import httpx
|
||
import streamlit as st
|
||
|
||
# ============ API ============
|
||
|
||
API_URL = os.environ.get("API_URL", "http://localhost:8000")
|
||
|
||
|
||
# ============ 选项与标签 ============
|
||
|
||
AVAILABLE_ORGANS = [
|
||
"liver",
|
||
"spleen",
|
||
"lung",
|
||
"heart",
|
||
"kidney",
|
||
"muscle",
|
||
"lymph_nodes",
|
||
]
|
||
|
||
ORGAN_LABELS = {
|
||
"liver": "肝脏 (Liver)",
|
||
"spleen": "脾脏 (Spleen)",
|
||
"lung": "肺 (Lung)",
|
||
"heart": "心脏 (Heart)",
|
||
"kidney": "肾脏 (Kidney)",
|
||
"muscle": "肌肉 (Muscle)",
|
||
"lymph_nodes": "淋巴结 (Lymph Nodes)",
|
||
}
|
||
|
||
AVAILABLE_ROUTES = [
|
||
"intravenous",
|
||
"intramuscular",
|
||
]
|
||
|
||
ROUTE_LABELS = {
|
||
"intravenous": "静脉注射 (Intravenous)",
|
||
"intramuscular": "肌肉注射 (Intramuscular)",
|
||
}
|
||
|
||
HELPER_LIPID_OPTIONS = ["DOPE", "DSPC"]
|
||
|
||
# PDI 分类标签
|
||
PDI_CLASS_LABELS = {
|
||
0: "<0.2 (优)",
|
||
1: "≥0.2 (欠佳)",
|
||
}
|
||
|
||
EE_CLASS_LABELS = {
|
||
0: "<50% (低)",
|
||
1: "50-80% (中)",
|
||
2: ">80% (高)",
|
||
}
|
||
|
||
TOXIC_CLASS_LABELS = {
|
||
0: "无毒 ✓",
|
||
1: "有毒 ⚠",
|
||
}
|
||
|
||
|
||
# ============ 功能模式 ============
|
||
|
||
MODE_OPTIMIZE = "optimize"
|
||
MODE_SCREEN = "screen"
|
||
|
||
MODE_LABELS = {
|
||
MODE_OPTIMIZE: "配方优选",
|
||
MODE_SCREEN: "批量筛选",
|
||
}
|
||
|
||
# ============ 访问口令 ============
|
||
|
||
def check_password() -> bool:
|
||
"""校验访问口令，未设置 APP_PASSWORD 时直接放行，内网访问不受影响。"""
|
||
expected = os.environ.get("APP_PASSWORD")
|
||
if not expected:
|
||
return True
|
||
if st.session_state.get("_authed"):
|
||
return True
|
||
|
||
gate = st.empty()
|
||
with gate.container():
|
||
st.markdown("### LNP 配方优化")
|
||
pwd = st.text_input("🔒 访问口令", type="password")
|
||
if pwd and pwd != expected:
|
||
st.error("口令不正确")
|
||
|
||
if pwd == expected:
|
||
st.session_state["_authed"] = True
|
||
gate.empty()
|
||
return True
|
||
return False
|
||
|
||
|
||
def require_auth() -> None:
|
||
"""口令未通过时终止页面渲染。应用只有 app.py 一个入口，在其顶部调用一次即可。"""
|
||
if not check_password():
|
||
st.stop()
|
||
|
||
|
||
# ============ API 客户端 ============
|
||
|
||
def check_api_status(timeout: float = 5.0) -> bool:
|
||
"""检查 API 是否在线。"""
|
||
try:
|
||
with httpx.Client(timeout=timeout) as client:
|
||
return client.get(f"{API_URL}/").status_code == 200
|
||
except httpx.HTTPError:
|
||
return False
|
||
|
||
|
||
def render_api_status(api_online: bool) -> None:
|
||
"""在侧边栏渲染 API 状态与模型能力，两种功能模式共用。"""
|
||
if not api_online:
|
||
st.error("🔴 API 服务离线")
|
||
st.info(
|
||
f"请先启动 API 服务:\n```\nuvicorn app.api:app --port 8010\n```\n"
|
||
f"当前 API_URL: {API_URL}"
|
||
)
|
||
return
|
||
|
||
st.success("🟢 API 服务在线")
|
||
try:
|
||
with httpx.Client(timeout=5) as client:
|
||
info = client.get(f"{API_URL}/").json()
|
||
except httpx.HTTPError:
|
||
return
|
||
|
||
caps = [
|
||
name
|
||
for name, key in (("MoE", "use_moe"), ("LLM", "use_llm"), ("RAG", "use_rag"))
|
||
if info.get(key)
|
||
]
|
||
st.caption(
|
||
f"模型: {' + '.join(caps) if caps else '仅 backbone'}｜{info.get('device', '?')}"
|
||
) |