"""Render the prescription comparison window with synthetic, offline data.""" from __future__ import annotations import os from copy import deepcopy from pathlib import Path from typing import Any os.environ.setdefault("QT_QPA_PLATFORM", "offscreen") from PySide6.QtWidgets import QApplication from doctor_workstation.ui.dialogs import issued_prescription_ai as ai from doctor_workstation.ui.theme import apply_theme def example_batch() -> dict[str, Any]: """Three prescriptions that actually differ, so every state of the page has something to draw.""" doctor = {"生地黄": 16, "天花粉": 15, "干石斛": 20, "醋五味子": 6, "生麦冬": 12, "茯苓": 10, "红参片": 6, "生牡丹皮": 10} candidates = { "qwen": {"生地黄": 15, "干石斛": 12, "醋五味子": 6, "生麦冬": 12, "茯苓": 15, "麸炒白术": 12, "丹参": 15}, "openai": {"生地黄": 12, "醋五味子": 6, "生麦冬": 10, "茯苓": 12, "生白芍": 5, "炒酸枣仁": 6}, } reports = { "qwen": {"summary": "界面演示数据：对照药味组成、剂量及用法，辅助医生逐项复核。", "diagnosis": "消渴病，气阴两虚兼血热。兼证需结合舌脉资料核实。", "analysis": "阅读药方对比页，查看同名药材的剂量差异、仅医生方药味与仅候选方药味。", "risk_assessment": [{"level": "high", "label": "血压数据缺失，影响补气药安全性评估"}, {"level": "medium", "label": "肝肾功能具体指标未提供"}], "missing_information": ["甲状腺功能及眼底检查报告缺失", "舌象与脉诊仅有附件，无文本记录"]}, "openai": {"summary": "界面演示数据：两份候选方独立生成，引用编号可回查原始资料。", "diagnosis": "已记录气阴两虚证；辨证依据需补充。", "analysis": "候选方以益气养阴为主，安神药味为本模型新增，需要医师确认。", "risk_assessment": [{"level": "high", "label": "缺少当前用药记录，无法排除配伍风险"}], "missing_information": ["舌象与脉诊仅有附件，无文本记录", "近期体重变化及 BMI 数据缺失"]}, } models = {} for key, doses in candidates.items(): herbs, rows = [], [] for name in sorted(set(doctor) | set(doses), key=lambda item: (item not in doses, item)): common = {"name": name, "unit": "g", "dose_basis": "per_dose", "formula_type": "主方"} left, right = doctor.get(name), doses.get(name) if right is not None: herbs.append({**common, "dosage": right}) rows.append({ **common, "key": name, "doctor": {**common, "dosage": left} if left is not None else None, "candidate": {**common, "dosage": right, "source_rows": [len(herbs) - 1]} if right is not None else None, "doctor_dosage": left, "candidate_dosage": right, "match_type": "candidate_only" if left is None else "doctor_only" if right is None else "matched", "contribution": min(left, right) / max(left, right) if left and right else None, }) matched = sum(row["match_type"] == "matched" for row in rows) denominator = len(doctor) + len(herbs) models[key] = { "status": "success", "report_id": 10 if key == "qwen" else 11, "candidate": {"status": "available_for_review", "prescription_name": "候选药方 · 界面示例", "herbs": herbs, "dose_basis": "per_dose", "prescription_type": "浓缩水丸", "usage_instruction": "服法由医生复核后确认。", "usage_days": 7, "times_per_day": 2, "rationale": "这是用于检查界面排版的模拟药方，不对应真实患者。", "risk_warnings": ["药味与剂量差异需要逐项复核。"]}, "comparison": {"status": "comparable", "score": 200 * sum(row["contribution"] or 0 for row in rows) / denominator, "herb_score": 200 * matched / denominator, "matched_count": matched, "doctor_count": len(doctor), "candidate_count": len(herbs), "rows": rows, "reason": "药味与剂量可比；服法与疗程需单独复核。", "usage_differences": []}, "report": reports[key], "coverage": {"status": "incomplete", "complete": False, "files": [{"file_id": f"a{index:04d}", "type": "image" if index % 3 else "document", "status": "processed", "transmitted": True, "version_verified": True} for index in range(20 if key == "qwen" else 17)] + [{"file_id": f"z{index:04d}", "type": "document", "status": "restricted", "transmitted": False, "version_verified": False, "reason": "FILE_UNAVAILABLE_OR_UNSUPPORTED"} for index in range(2)]}, "progress": {"stage_label": "处理完成", "elapsed_seconds": 135 if key == "qwen" else 591, "attempt": 1}, "usage": {"total_calls": 3, "calls": [ {"stage": "text:0", "ok": True, "latency_ms": 3120, "file_count": 0, "usage": {"completion_tokens": 980}, "error_code": ""}, {"stage": "files:0", "ok": True, "latency_ms": 22400, "file_count": 20 if key == "qwen" else 17, "usage": {"completion_tokens": 2140}, "error_code": ""}, {"stage": "final", "ok": True, "latency_ms": 18800, "file_count": 0, "usage": {"completion_tokens": 5617}, "error_code": ""}]}, "review": {"status": "viewed", "comment": ""}, "algorithm_version": "prescription-soft-dice-v1.1.0", "prompt_version": "manual-prescription-required-candidate-v4", } return {"id": 40, "prescription_id": 7556, "patient_id": 1391, "diagnosis_id": 1391, "prescription_revision": 1, "status": "success", "validity": "current", "comparison_type": "non_independent", "models": models, "coverage_status": "partial", "created_at": "2026-09-10 15:29:00", "cutoff_at": "2026-09-10 15:29:00", "source_summary": {"diagnoses_count": 2, "attachment_count": 22, "video_calls_count": 4, "chat_messages_count": 137, "source_record_count": 31}, "doctor_snapshot": {"patient": {"name": "张卫君", "gender": 2, "gender_label": "女", "age": 58}, "diagnosis": {"clinical_diagnosis": "2型糖尿病 消渴病 · 气阴两虚兼血热", "chief_complaint": "咳嗽反复1月余"}, "prescription": {"herbs": [{"name": name, "dosage": dose, "unit": "g", "dose_basis": "per_dose", "formula_type": "主方"} for name, dose in doctor.items()], "prescription_type": "浓缩水丸", "dose_count": 1, "usage_instruction": "每日1剂，水煎分服。"}}, "missing": [{"code": "TRANSCRIPT_NOT_VERIFIED_COMPLETE", "critical": True}, {"code": "TRANSCRIPT_NOT_VERIFIED_COMPLETE", "critical": True}, {"code": "ARCHIVE_SYNC_WATERMARK_UNAVAILABLE", "critical": False}]} def example_statistics() -> dict[str, Any]: """Doctor-level shape the statistics window renders; synthetic, but structurally complete.""" def doctor(identifier: int, name: str, totals: tuple[int, int, int], qwen: tuple[int, float | None], openai: tuple[int, float | None], review: tuple[int, int]) -> dict[str, Any]: total, patients, paired = totals models = {} for key, (eligible, mean) in (("qwen", qwen), ("openai", openai)): share = (0.34, 0.38, 0.18, 0.07, 0.03) models[key] = {"eligible_count": eligible, "mean": mean, "median": None if mean is None else round(mean - 1.4, 1), "excluded_reasons": {"SOURCE_HISTORY_VERSIONS_UNAVAILABLE": max(0, total - eligible - 2), "incomplete_coverage": min(2, max(0, total - eligible))}, "distribution": {name: round(eligible * fraction) for name, fraction in zip( ("[0,20)", "[20,40)", "[40,60)", "[60,80)", "[80,100]"), share, strict=True)}} evaluated, qualified = review return {"doctor_id": identifier, "doctor_name": name, "total_count": total, "patient_count": patients, "paired_count": paired, "models": models, "review": {"evaluated_count": evaluated, "qualified_count": qualified, "qualified_rate": None if not evaluated else round(100 * qualified / evaluated, 1)}} doctors = [doctor(26, "何医生", (18, 14, 9), (12, 21.4), (9, 18.9), (6, 4)), doctor(31, "李医生", (11, 9, 4), (7, 26.8), (5, 24.1), (3, 1)), doctor(44, "王医生", (6, 5, 1), (2, 15.2), (0, None), (0, 0))] return {"total_count": sum(item["total_count"] for item in doctors), "patient_count": sum(item["patient_count"] for item in doctors), "doctors": doctors} class PreviewRepository: def __init__(self) -> None: self.batch = example_batch() def list_prescription_ai_reports(self, **_params: Any) -> dict[str, Any]: return {"enabled": True, "lists": deepcopy(self.history()), "count": len(self.history())} def history(self) -> list[dict[str, Any]]: """The current batch plus the earlier ones it supersedes, newest first.""" older = [ {"id": 39, "created_at": "2026-09-10 15:18:00", "status": "success", "validity": "superseded", "comparison_type": "non_independent", "models": {"qwen": {"comparison": {"status": "comparable", "score": 48.9}, "algorithm_version": "prescription-soft-dice-v1.0.1", "prompt_version": "v3"}, "openai": {"comparison": {"status": "comparable", "score": 41.7}, "algorithm_version": "prescription-soft-dice-v1.0.1", "prompt_version": "v3"}}}, {"id": 38, "created_at": "2026-09-10 14:05:00", "status": "success", "validity": "superseded", "models": {"qwen": {"comparison": {"status": "comparable", "score": 33.4}, "algorithm_version": "prescription-soft-dice-v1.0.1", "prompt_version": "v3"}, "openai": {"comparison": {"status": "comparable", "score": 31.1}, "algorithm_version": "prescription-soft-dice-v1.0.1", "prompt_version": "v3"}}}, {"id": 37, "created_at": "2026-09-10 13:25:00", "status": "failed", "validity": "superseded", "models": {"qwen": {"error_message": "模型返回未通过校验", "algorithm_version": "prescription-soft-dice-v1.0.1"}, "openai": {"comparison": {"status": "not_comparable"}}}}, {"id": 36, "created_at": "2026-09-10 12:34:00", "status": "success", "validity": "superseded", "models": {"qwen": {"comparison": {"status": "not_comparable"}, "algorithm_version": "prescription-soft-dice-v1.0.0"}, "openai": {"comparison": {"status": "not_comparable"}}}}, ] return [self.batch, *older] def get_prescription_ai_report(self, _batch_id: int) -> dict[str, Any]: return deepcopy(self.batch) def prescription_ai_statistics(self, *_args: Any, **_params: Any) -> dict[str, Any]: return deepcopy(example_statistics()) def main() -> None: application = QApplication.instance() or QApplication([]) apply_theme(application) output = Path(__file__).resolve().parents[1] / "artifacts" / "issued_prescription_redesign" output.mkdir(parents=True, exist_ok=True) def immediate(function: Any, **callbacks: Any) -> None: result = function() if callbacks.get("on_success"): callbacks["on_success"](result) if callbacks.get("on_finished"): callbacks["on_finished"]() ai.run_async = immediate repository = PreviewRepository() dialog = ai.IssuedPrescriptionAiDialog(repository, ["*"], prescription_id=7556, current_user={"name": "张医生"}) dialog.show() names = {dialog.tabs.tabText(index): index for index in range(dialog.tabs.count())} for filename, width, height, tab in ( ("prescription-1440.png", 1440, 940, "对比总览"), ("prescription-1280.png", 1280, 860, "对比总览"), ("prescription-1024.png", 1024, 700, "对比总览"), ("prescription-940.png", 940, 640, "对比总览"), ("original-1280.png", 1280, 860, "原方记录"), ("analysis-1280.png", 1280, 860, "完整报告"), ("candidate-1280.png", 1280, 860, "方义与用法"), ("per-herb-1440.png", 1440, 940, "候选与逐味"), ("sources-1440.png", 1440, 940, "资料与缺口"), ("history-1440.png", 1440, 940, "历史与趋势"), ("pipeline-1440.png", 1440, 940, "处理进度"), ): dialog.resize(width, height) dialog.tabs.setCurrentIndex(names[tab]) for _ in range(4): application.processEvents() assert (dialog.width(), dialog.height()) == (width, height), (filename, dialog.size()) assert dialog.grab().save(str(output / filename)) print(filename, "window", width, height, "workspace", dialog.tabs.width(), dialog.tabs.height()) dialog.tabs.setCurrentIndex(names["完整报告"]) dialog._set_report_mode("differences") application.processEvents() assert dialog.grab().save(str(output / "report-differences-1440.png")) dialog._set_report_mode("both") dialog.tabs.setCurrentIndex(names["对比总览"]) dialog.resize(1280, 860) for _ in range(4): application.processEvents() dialog.review_model.setCurrentIndex(1) application.processEvents() assert dialog.grab().save(str(output / "openai-1280.png")) dialog.review_model.setCurrentIndex(0) repository.batch["status"] = "running" repository.batch["models"]["openai"].update(status="running", candidate=None, comparison=None, report=None) dialog.refresh() application.processEvents() assert dialog.grab().save(str(output / "partial-1280.png")) repository.batch.update(validity="superseded", status="success") dialog.refresh() application.processEvents() assert dialog.grab().save(str(output / "historical-1280.png")) repository.batch.update(validity="current", status="running") repository.batch["models"]["qwen"].update(status="running", candidate=None, comparison=None, report=None) dialog.refresh() application.processEvents() assert dialog.grab().save(str(output / "pending-1280.png")) # A failed model must state the reason and offer its own retry on the card itself. repository.batch.update(validity="current", status="partial") repository.batch["models"]["qwen"].update( status="failed", error_code="upstream_timeout", error_message="上游模型超时，未返回结果", candidate=None, comparison=None, report=None, retry_count=1, max_retry=3) repository.batch["models"]["openai"] = deepcopy(example_batch()["models"]["openai"]) dialog.refresh() application.processEvents() assert dialog.grab().save(str(output / "failed-1280.png")) # The light theme is the same window with the other palette; capture it once. dialog.resize(1440, 940) dialog._switch_theme() for _ in range(12): application.processEvents() dialog.repaint() assert dialog.grab().save(str(output / "light-1440.png")) dialog._switch_theme() for _ in range(4): application.processEvents() dialog.close() statistics = ai.PrescriptionAiStatisticsDialog(repository, ["*"]) statistics.resize(1240, 880) statistics.show() for _ in range(4): application.processEvents() assert statistics.grab().save(str(output / "statistics-1240.png")) statistics.close() print("statistics-1240.png", statistics.panel.doctors.rowCount(), "doctors") print(output) if __name__ == "__main__": main()