| """
|
| MolGate — Module 5 本機面板後端
|
| - 提供靜態 molgate_module5_panel.html
|
| - GET /api/module5/status?session=…
|
| - POST /api/module5/run-engine-a
|
| - POST /api/module5/run-engine-b
|
|
|
| 使用方式（在 MolGate 目錄）:
|
| py -3 molgate_module5_server.py
|
| 瀏覽器:
|
| http://127.0.0.1:8767/?session=你的16位hex&target=COX2
|
| """
|
|
|
| from __future__ import annotations
|
|
|
| import json
|
| import re
|
| import subprocess
|
| import sys
|
| import urllib.parse
|
| from http.server import BaseHTTPRequestHandler, HTTPServer
|
| from pathlib import Path
|
| from typing import Any
|
|
|
| _UI = Path(__file__).resolve().parents[1]
|
| if str(_UI) not in sys.path:
|
| sys.path.insert(0, str(_UI))
|
| from molgate_ui.console_utf8 import subprocess_env
|
| from molgate_ui.llm_proxy import llm_chat_json, llm_chat_stream, llm_public_config
|
| from molgate_ui.ai_keys import ensure_api_keys_loaded
|
| from molgate_ui.paths import MASTER_INDEX, load_env_file, resolve_vina_bin
|
| from molgate_ui.server_paths import (
|
| _PKG_ROOT,
|
| _SESSIONS,
|
| UI_M5_PANEL as _PANEL,
|
| MOLGATE_JS as _SHARED_JS,
|
| read_static_asset,
|
| send_cors_headers,
|
| )
|
| _SESSION_ID_RE = re.compile(r"^[a-fA-F0-9]{16}$")
|
|
|
|
|
| def _read_json(p: Path) -> dict[str, Any] | None:
|
| if not p.is_file():
|
| return None
|
| try:
|
| return json.loads(p.read_text(encoding="utf-8"))
|
| except (json.JSONDecodeError, OSError):
|
| return None
|
|
|
|
|
| def _safe_session_dir(session_id: str) -> Path | None:
|
| sid = (session_id or "").strip()
|
| if not sid or not _SESSION_ID_RE.match(sid):
|
| return None
|
| p = (_SESSIONS / sid).resolve()
|
| try:
|
| p.relative_to(_SESSIONS.resolve())
|
| except ValueError:
|
| return None
|
| return p if p.is_dir() else None
|
|
|
|
|
| def _run_subprocess(argv: list[str], cwd: Path) -> tuple[int, str]:
|
| try:
|
| r = subprocess.run(
|
| argv,
|
| cwd=str(cwd),
|
| capture_output=True,
|
| text=True,
|
| encoding="utf-8",
|
| errors="replace",
|
| timeout=3600,
|
| env=subprocess_env(),
|
| )
|
| except subprocess.TimeoutExpired:
|
| return -1, "subprocess timeout (3600s)"
|
| except OSError as e:
|
| return -1, str(e)
|
| out = (r.stdout or "") + ("\n" if r.stdout and r.stderr else "") + (r.stderr or "")
|
| return r.returncode, out.strip() or f"(exit {r.returncode})"
|
|
|
|
|
| def _resolve_session_drug(sd: Path, target: str, dr: dict[str, Any] | None) -> str:
|
| if dr and dr.get("target_drug"):
|
| return str(dr["target_drug"]).strip()
|
| for fname in ("session_index.json", "stage_03_index.json"):
|
| j = _read_json(sd / fname)
|
| if j and j.get("target_drug"):
|
| return str(j["target_drug"]).strip()
|
| tgt = (target or "").strip().upper()
|
| if tgt and MASTER_INDEX.is_file():
|
| try:
|
| entries = json.loads(MASTER_INDEX.read_text(encoding="utf-8"))
|
| entry = entries.get(tgt) or entries.get(target.strip())
|
| if isinstance(entry, dict) and entry.get("target_drug"):
|
| return str(entry["target_drug"]).strip()
|
| except (json.JSONDecodeError, OSError):
|
| pass
|
| return ""
|
|
|
|
|
| def build_m5_status(session_id: str) -> dict[str, Any]:
|
| err: str | None = None
|
| sd = _safe_session_dir(session_id)
|
| if not session_id.strip():
|
| err = "請提供 session（16 位 hex）"
|
| elif not sd:
|
| err = f"找不到 session：{session_id!r}"
|
|
|
| base: dict[str, Any] = {
|
| "ok": sd is not None,
|
| "error": err,
|
| "session_id": session_id.strip() if session_id else "",
|
| "session_dir_abs": str(sd.resolve()) if sd else "",
|
| "target": "",
|
| "pdb_id": "",
|
| "drug": "",
|
| "engine_a": {"complete": False, "summary": "", "status": "", "raw24": None, "raw25": None, "raw26": None},
|
| "engine_b": {"complete": False, "summary": "", "status": "", "raw27": None},
|
| "docking_result": None,
|
| "final_report": None,
|
| "vina_bin": "",
|
| "vina_ready": False,
|
| }
|
| vina = resolve_vina_bin()
|
| base["vina_bin"] = vina or ""
|
| base["vina_ready"] = bool(vina)
|
| if not sd:
|
| return base
|
|
|
| meta = _read_json(sd / "session_meta.json")
|
| dr = _read_json(sd / "docking_result.json")
|
| fr = _read_json(sd / "final_report.json")
|
| s24 = _read_json(sd / "stage_24_vina.json")
|
| s25 = _read_json(sd / "stage_25_qc.json")
|
| s26 = _read_json(sd / "stage_26_result.json")
|
| s27 = _read_json(sd / "stage_27_review.json")
|
|
|
| target = (dr or {}).get("target") or (meta or {}).get("target") or ""
|
| pdb_id = (dr or {}).get("pdb_id") or (meta or {}).get("pdb_id") or ""
|
|
|
| base["target"] = str(target).strip()
|
| base["pdb_id"] = str(pdb_id).strip().upper()
|
| base["drug"] = _resolve_session_drug(sd, base["target"], dr) if sd else ""
|
|
|
| base["docking_result"] = dr
|
| base["final_report"] = fr
|
| base["engine_a"]["raw24"] = s24
|
| base["engine_a"]["raw25"] = s25
|
| base["engine_a"]["raw26"] = s26
|
| base["engine_b"]["raw27"] = s27
|
|
|
| if dr and dr.get("ready_for_review"):
|
| aff = dr.get("best_affinity")
|
| tier = dr.get("confidence_tier", "")
|
| qc = "PASS" if dr.get("qc_pass") else "WARN"
|
| base["engine_a"]["summary"] = f"{aff} kcal/mol · tier {tier} · {qc}"
|
| base["engine_a"]["status"] = "PASS" if dr.get("qc_pass") else "WARN"
|
| base["engine_a"]["complete"] = True
|
|
|
| if s27:
|
| dec = str(s27.get("decision") or s27.get("status") or "").upper()
|
| base["engine_b"]["summary"] = dec or "—"
|
| base["engine_b"]["status"] = dec or ""
|
| base["engine_b"]["complete"] = dec == "ACCEPT"
|
|
|
| return base
|
|
|
|
|
| class Handler(BaseHTTPRequestHandler):
|
| def log_message(self, fmt: str, *args: Any) -> None:
|
| sys.stderr.write(f"[molgate-m5] {args[0]}\n")
|
|
|
| def _send(self, code: int, body: bytes, content_type: str) -> None:
|
| self.send_response(code)
|
| self.send_header("Content-Type", content_type)
|
| self.send_header("Content-Length", str(len(body)))
|
| send_cors_headers(self)
|
| self.end_headers()
|
| self.wfile.write(body)
|
|
|
| def do_OPTIONS(self) -> None:
|
| self.send_response(204)
|
| send_cors_headers(self)
|
| self.end_headers()
|
|
|
| def _send_json(self, code: int, obj: Any) -> None:
|
| data = json.dumps(obj, ensure_ascii=False).encode("utf-8")
|
| self._send(code, data, "application/json; charset=utf-8")
|
|
|
| def do_GET(self) -> None:
|
| parsed = urllib.parse.urlparse(self.path)
|
| path = parsed.path
|
|
|
| if path in ("/", "/molgate_module5_panel.html"):
|
| if not _PANEL.is_file():
|
| self._send(404, b"panel html missing", "text/plain; charset=utf-8")
|
| return
|
| self._send(200, _PANEL.read_bytes(), "text/html; charset=utf-8")
|
| return
|
|
|
| static = read_static_asset(path)
|
| if static:
|
| body, ctype = static
|
| self._send(200, body, ctype)
|
| return
|
|
|
| if path in ("/api/module5/status", "/api/status"):
|
| qs = urllib.parse.parse_qs(parsed.query)
|
| sid = (qs.get("session") or [""])[0]
|
| self._send_json(200, build_m5_status(sid))
|
| return
|
|
|
| if path == "/api/llm/config":
|
| self._send_json(200, llm_public_config())
|
| return
|
|
|
| self._send(404, b"not found", "text/plain; charset=utf-8")
|
|
|
| def _handle_llm_chat(self, payload: dict[str, Any]) -> None:
|
| stream = bool(payload.get("stream", True))
|
| try:
|
| if not stream:
|
| code, body = llm_chat_json(payload)
|
| self._send_json(code, body)
|
| return
|
| except ValueError as e:
|
| self._send_json(400, {"ok": False, "error": str(e)})
|
| return
|
|
|
| self.send_response(200)
|
| self.send_header("Content-Type", "text/event-stream; charset=utf-8")
|
| self.send_header("Cache-Control", "no-cache")
|
| self.send_header("X-Accel-Buffering", "no")
|
| send_cors_headers(self)
|
| self.end_headers()
|
| try:
|
| for chunk in llm_chat_stream(payload):
|
| self.wfile.write(chunk)
|
| self.wfile.flush()
|
| except ValueError as e:
|
| err = json.dumps({"error": str(e)}, ensure_ascii=False).encode("utf-8")
|
| self.wfile.write(b"data: " + err + b"\n\n")
|
| self.wfile.flush()
|
|
|
| def do_POST(self) -> None:
|
| parsed = urllib.parse.urlparse(self.path)
|
| length = int(self.headers.get("Content-Length") or 0)
|
| raw = self.rfile.read(length) if length else b"{}"
|
| try:
|
| payload = json.loads(raw.decode("utf-8"))
|
| except json.JSONDecodeError:
|
| self._send_json(400, {"ok": False, "error": "invalid JSON"})
|
| return
|
|
|
| if parsed.path in ("/api/llm/chat", "/api/ollama/chat"):
|
| self._handle_llm_chat(payload)
|
| return
|
|
|
| session_id = (payload.get("session") or "").strip()
|
| sd = _safe_session_dir(session_id)
|
| if not sd:
|
| self._send_json(400, {"ok": False, "error": "invalid or missing session id"})
|
| return
|
|
|
| if parsed.path == "/api/module5/run-engine-a":
|
| target = (payload.get("target") or "").strip()
|
| vina_bin = (payload.get("vina_bin") or "").strip() or (resolve_vina_bin() or "")
|
| if not vina_bin:
|
| status = build_m5_status(session_id)
|
| self._send_json(
|
| 200,
|
| {
|
| "ok": False,
|
| "returncode": 1,
|
| "log": (
|
| "找不到 AutoDock Vina。請將 vina.exe 放到 moltgate_UI/tools/vina/，"
|
| "或設定 MOLGATE_VINA_BIN，或用 UI 指定 --vina-bin 路徑。"
|
| ),
|
| "status": status,
|
| },
|
| )
|
| return
|
| argv = [
|
| sys.executable,
|
| str(_PKG_ROOT / "molgate_module5_engineA.py"),
|
| "--session-dir",
|
| str(sd),
|
| "--vina-bin",
|
| vina_bin,
|
| ]
|
| if target:
|
| argv.extend(["--target", target])
|
| code, output = _run_subprocess(argv, _PKG_ROOT)
|
| status = build_m5_status(session_id)
|
| self._send_json(200, {"ok": code == 0, "returncode": code, "log": output, "status": status})
|
| return
|
|
|
| if parsed.path == "/api/module5/run-engine-b":
|
| decision = str(payload.get("decision") or "").strip().upper()
|
| note = str(payload.get("note") or "").strip()
|
| retry_module = str(payload.get("retry_module") or "").strip()
|
| if decision not in ("ACCEPT", "RETRY", "FLAG"):
|
| self._send_json(400, {"ok": False, "error": "decision must be ACCEPT/RETRY/FLAG"})
|
| return
|
| argv = [
|
| sys.executable,
|
| str(_PKG_ROOT / "molgate_module5_engineB.py"),
|
| "--session-dir",
|
| str(sd),
|
| "--decision",
|
| decision,
|
| ]
|
| if note:
|
| argv.extend(["--note", note])
|
| if decision == "RETRY" and retry_module in ("1", "2", "3", "4", "5"):
|
| argv.extend(["--retry-module", retry_module])
|
| code, output = _run_subprocess(argv, _PKG_ROOT)
|
| status = build_m5_status(session_id)
|
| self._send_json(200, {"ok": code == 0, "returncode": code, "log": output, "status": status})
|
| return
|
|
|
| self._send(404, b"not found", "text/plain; charset=utf-8")
|
|
|
|
|
| def main() -> None:
|
| import argparse
|
|
|
| load_env_file()
|
| ensure_api_keys_loaded()
|
| ap = argparse.ArgumentParser(description="MolGate Module 5 panel server")
|
| ap.add_argument("--host", default="127.0.0.1")
|
| ap.add_argument("--port", type=int, default=8767)
|
| args = ap.parse_args()
|
| httpd = HTTPServer((args.host, args.port), Handler)
|
| print(f"MolGate Module 5 panel → http://{args.host}:{args.port}/")
|
| print("範例：http://127.0.0.1:8767/?session=你的session&target=COX2")
|
| print("按 Ctrl+C 結束")
|
| try:
|
| httpd.serve_forever()
|
| except KeyboardInterrupt:
|
| print("\n已停止")
|
|
|
|
|
| if __name__ == "__main__":
|
| main()
|
|
|
|
|
