Upload gradio_app.py with huggingface_hub

gradio_app.py

@@ -0,0 +1,163 @@
+import gradio as gr
+import json
+from pathlib import Path
+import importlib.util
+from wearable_anomaly_detector import WearableAnomalyDetector
+
+BASE_DIR = Path(__file__).parent
+MODEL_DIR = BASE_DIR / "checkpoints" / "phase2" / "exp_factor_balanced"
+LLM_INPUT_DIR = BASE_DIR / "demo_llm_inputs"
+LLM_MANIFEST = LLM_INPUT_DIR / "manifest.json"
+PATCHAD_CASE_DIR = BASE_DIR / "demo_patchad_cases"
+PATCHAD_CASE_MANIFEST = PATCHAD_CASE_DIR / "manifest.json"
+
+SAMPLES = {
+    "示例: 正常": BASE_DIR / "test_data" / "example_window.json",
+    "示例: 短期异常": BASE_DIR / "data_storage" / "users" / "demo_anomaly.jsonl",
+    "示例: 长期异常": BASE_DIR / "data_storage" / "users" / "demo_pattern.jsonl",
+    "示例: 缺失数据": BASE_DIR / "data_storage" / "users" / "demo_missing.jsonl",
+}
+
+LLM_CASES = {}
+if LLM_MANIFEST.exists():
+    manifest = json.loads(LLM_MANIFEST.read_text(encoding="utf-8"))
+    for item in manifest:
+        display = item.get("title") or item.get("case_id")
+        file_name = item.get("file")
+        if display and file_name:
+            LLM_CASES[display] = LLM_INPUT_DIR / file_name
+
+PATCHAD_CASES = {}
+if PATCHAD_CASE_MANIFEST.exists():
+    manifest = json.loads(PATCHAD_CASE_MANIFEST.read_text(encoding="utf-8"))
+    for item in manifest:
+        display = item.get("title")
+        file_name = item.get("file")
+        if display and file_name:
+            PATCHAD_CASES[display] = PATCHAD_CASE_DIR / file_name
+
+formatter_spec = importlib.util.spec_from_file_location(
+    "formatter", BASE_DIR / "utils" / "formatter.py"
+)
+formatter_module = importlib.util.module_from_spec(formatter_spec)
+formatter_spec.loader.exec_module(formatter_module)
+AnomalyFormatter = formatter_module.AnomalyFormatter
+
+detector = WearableAnomalyDetector(model_dir=MODEL_DIR, device="cpu")
+formatter = AnomalyFormatter()
+
+
+def load_sample(path: Path):
+    if path.suffix == ".jsonl":
+        data = [json.loads(line) for line in path.read_text(encoding="utf-8").splitlines() if line.strip()]
+    else:
+        data = json.loads(path.read_text(encoding="utf-8"))
+        if isinstance(data, dict):
+            data = data.get("records") or data.get("data") or [data]
+    if not isinstance(data, list):
+        raise ValueError("样例文件需包含数组")
+    return data
+
+
+def infer(sample_name: str):
+    window = load_sample(SAMPLES[sample_name])
+    realtime = detector.detect_realtime(window, update_baseline=False, return_details=True)
+    baseline_info = {
+        "baseline_mean": window[0]["features"].get("baseline_hrv_mean", 75.0),
+        "baseline_std": window[0]["features"].get("baseline_hrv_std", 5.0),
+        "current_value": sum(pt["features"].get("hrv_rmssd", 0) for pt in window) / len(window),
+        "deviation_pct": 0.0,
+    }
+    text = formatter.format_for_llm(realtime, baseline_info=baseline_info)
+    return json.dumps(realtime, ensure_ascii=False, indent=2), text
+
+
+def show_llm_input(case_name: str):
+    if not LLM_CASES:
+        return {}, "当前未提供LLM输入示例。"
+    path = LLM_CASES[case_name]
+    data = json.loads(path.read_text(encoding="utf-8"))
+    messages = data.get("messages", [])
+    system_text = ""
+    user_text = ""
+    if messages:
+        system_msg = next((m for m in messages if m.get("role") == "system"), {})
+        user_msg = next((m for m in messages if m.get("role") == "user"), {})
+        system_text = system_msg.get("content", "")
+        user_text = user_msg.get("content", "")
+    display_text = "## 系统提示\n"
+    display_text += (system_text or "（无）")
+    display_text += "\n\n---\n\n## 用户输入（Markdown）\n"
+    display_text += (user_text or "（无）")
+    return data, display_text
+
+
+def show_patchad_case(case_name: str):
+    if not PATCHAD_CASES:
+        return {}, {}, "当前未提供 PatchTrAD 示例。"
+    path = PATCHAD_CASES[case_name]
+    data = json.loads(path.read_text(encoding="utf-8"))
+    bundle = data.get("case_bundle", {})
+    summary = {
+        "sample": data.get("sample"),
+        "mode": data.get("mode"),
+        "precheck": data.get("precheck"),
+        "validation": bundle.get("validation"),
+    }
+    case_json = bundle.get("case", {})
+    llm_text = bundle.get("llm_input", "（无）")
+    return summary, case_json, llm_text
+
+
+realtime_demo = gr.Interface(
+    fn=infer,
+    inputs=gr.Dropdown(choices=list(SAMPLES.keys()), value="示例: 正常", label="选择测试数据"),
+    outputs=[
+        gr.JSON(label="模型输出"),
+        gr.Markdown(label="LLM 文本"),
+    ],
+    title="Wearable Anomaly Detector Demo",
+    description="选择预置数据（正常/短期异常/长期异常/缺失数据）即可查看时序模型输出及格式化LLM文本。",
+)
+
+llm_input_demo = gr.Interface(
+    fn=show_llm_input,
+    inputs=gr.Dropdown(
+        choices=list(LLM_CASES.keys()) or ["暂无示例"],
+        value=list(LLM_CASES.keys())[0] if LLM_CASES else "暂无示例",
+        label="选择LLM输入示例",
+    ),
+    outputs=[
+        gr.JSON(label="messages JSON"),
+        gr.Markdown(label="完整用户输入"),
+    ],
+    title="标准化LLM输入示例",
+    description="直接展示系统提示+用户输入的完整Markdown，便于在Hugging Face页面选择典型案例查看。",
+)
+
+demo = gr.TabbedInterface(
+    [
+        realtime_demo,
+        llm_input_demo,
+        gr.Interface(
+            fn=show_patchad_case,
+            inputs=gr.Dropdown(
+                choices=list(PATCHAD_CASES.keys()) or ["暂无示例"],
+                value=list(PATCHAD_CASES.keys())[0] if PATCHAD_CASES else "暂无示例",
+                label="选择 PatchTrAD 案例",
+            ),
+            outputs=[
+                gr.JSON(label="摘要 / 预筛信息"),
+                gr.JSON(label="Case JSON"),
+                gr.Markdown(label="LLM 输入"),
+            ],
+            title="PatchTrAD + build_case 案例",
+            description="从预置示例中选择模式A/B与不同数据样本，查看完整 case 与 LLM 输入。",
+        ),
+    ],
+    ["实时窗口检测", "LLM输入示例", "PatchTrAD案例"],
+)
+
+if __name__ == "__main__":
+    demo.launch()
+