Spaces:
Runtime error
Runtime error
File size: 6,046 Bytes
89d9c28 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 | from fastapi import FastAPI, Request, HTTPException
from fastapi.exceptions import RequestValidationError
from fastapi.responses import JSONResponse
from pydantic import BaseModel
from typing import List
from pathlib import Path
from collections import OrderedDict
from rdkit import Chem
from DeepPurpose.utils import data_process
from DeepPurpose import DTI as models
import traceback
import csv
import torch
# --------------------------------------------------
# Project paths (absolute, derived from this file)
# --------------------------------------------------
PROJECT_ROOT = Path(__file__).resolve().parents[2]
MODEL_DIR = PROJECT_ROOT / "models" / "affinity"
TARGET_SEQUENCE_FILE = PROJECT_ROOT / "models" / "affinity" / "target_sequence.txt"
app = FastAPI(title="EGFR DeepPurpose REINVENT Adapter")
# ---------- Request models ----------
class PredictRequest(BaseModel):
smiles: List[str]
# ---------- Helpers ----------
def load_default_egfr_sequence(sequence_path: Path) -> str:
if not sequence_path.exists():
raise FileNotFoundError(f"Target sequence file not found: {sequence_path}")
seq = sequence_path.read_text(encoding="utf-8").strip()
if not seq:
raise RuntimeError(f"Target sequence file is empty: {sequence_path}")
return seq
def canon(smi: str) -> str:
try:
m = Chem.MolFromSmiles(smi)
return Chem.MolToSmiles(m, isomericSmiles=True) if m is not None else smi
except Exception:
return smi
# ---------- Load model and target ----------
seq = load_default_egfr_sequence(TARGET_SEQUENCE_FILE)
model = models.model_pretrained(path_dir=str(MODEL_DIR))
# ---------- Simple cache ----------
CACHE = OrderedDict()
CACHE_MAX = 50000
def cache_get(k: str):
if k in CACHE:
CACHE.move_to_end(k)
return CACHE[k]
return None
def cache_put(k: str, v: float):
CACHE[k] = v
CACHE.move_to_end(k)
if len(CACHE) > CACHE_MAX:
CACHE.popitem(last=False)
# ---------- Shared prediction core ----------
def predict_smiles_list(smiles_in: List[str]) -> List[float]:
if not smiles_in:
return []
out = [None] * len(smiles_in)
to_compute = []
idx_map = []
for i, smi in enumerate(smiles_in):
k = canon(smi)
v = cache_get(k)
if v is None:
to_compute.append(smi)
idx_map.append((i, k))
else:
out[i] = float(v)
if to_compute:
X_target = [seq] * len(to_compute)
y_dummy = [0.0] * len(to_compute)
ret = data_process(
to_compute,
X_target,
y_dummy,
drug_encoding="Morgan",
target_encoding="AAC",
split_method="no_split",
)
X_pred = ret[0] if isinstance(ret, (tuple, list)) else ret
with torch.inference_mode():
preds = model.predict(X_pred)
for j, p in enumerate(preds):
i, k = idx_map[j]
val = float(p)
out[i] = val
cache_put(k, val)
return [0.0 if v is None else float(v) for v in out]
# ---------- Helpful debugging ----------
@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request: Request, exc: RequestValidationError):
body = await request.body()
print("\n=== 422 VALIDATION ERROR ===")
print("PATH:", request.url.path)
print("ERRORS:", exc.errors())
print("BODY:", body.decode("utf-8", errors="replace"))
print("=== END 422 ===\n")
return JSONResponse(status_code=422, content={"detail": exc.errors()})
# ---------- Legacy endpoint ----------
@app.post("/predict")
def predict(req: PredictRequest):
try:
preds = predict_smiles_list(req.smiles or [])
return {"pred_pAff_mean": preds}
except Exception as e:
traceback.print_exc()
raise HTTPException(status_code=500, detail=str(e))
# ---------- REINVENT-compatible endpoint ----------
@app.post("/reinvent_predict")
async def reinvent_predict(request: Request):
try:
payload = await request.json()
# REINVENT in your logs sends {"smiles": [...]}
if isinstance(payload, dict) and "smiles" in payload:
smiles = payload.get("smiles") or []
if not isinstance(smiles, list):
raise HTTPException(status_code=422, detail="Field 'smiles' must be a list.")
preds = predict_smiles_list(smiles)
return {"pred_pAff_mean": preds}
# Optional compatibility with list-of-items payload
if isinstance(payload, list):
smiles = []
query_ids = []
for item in payload:
if not isinstance(item, dict):
raise HTTPException(status_code=422, detail="Each list item must be an object.")
if "input_string" not in item or "query_id" not in item:
raise HTTPException(
status_code=422,
detail="Each item must contain 'input_string' and 'query_id'."
)
smiles.append(item["input_string"])
query_ids.append(str(item["query_id"]))
preds = predict_smiles_list(smiles)
successes = [
{"query_id": qid, "output_value": float(pred)}
for qid, pred in zip(query_ids, preds)
]
return {"output": {"successes_list": successes}}
raise HTTPException(
status_code=422,
detail="Unsupported request body. Expected either {'smiles': [...]} or a list of {'input_string','query_id'} items."
)
except HTTPException:
raise
except Exception as e:
traceback.print_exc()
raise HTTPException(status_code=500, detail=str(e))
# ---------- Health ----------
@app.get("/health")
def health():
return {
"status": "ok",
"model_dir": str(MODEL_DIR),
"target_sequence_file": str(TARGET_SEQUENCE_FILE),
"seq_len": len(seq),
} |