Spaces:
Running
Running
File size: 7,154 Bytes
99fb209 d17fdb8 5bb077e 99fb209 5bb077e 99fb209 38e4510 5bb077e d17fdb8 5bb077e 99fb209 b92ecb8 5bb077e 99fb209 5bb077e d17fdb8 5bb077e d17fdb8 5bb077e 99fb209 5bb077e 99fb209 5bb077e 38e4510 | 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 188 189 190 191 192 193 | """ADMET descriptor computation endpoints."""
from __future__ import annotations
import json
import os
import sys
import subprocess
from fastapi import APIRouter, Depends, HTTPException, Query
from pydantic import BaseModel, Field, model_validator
from app.services.auth import get_user_id
router = APIRouter(prefix="/api/admet", tags=["ADMET"])
class ADMETRequest(BaseModel):
smiles: str | None = Field(None, min_length=1, max_length=500, description="SMILES string")
name: str | None = Field(None, min_length=1, max_length=200, description="Chemical name (resolved via PubChem)")
cid: int | None = Field(None, ge=1, description="PubChem compound CID")
@model_validator(mode="after")
def _require_one(self) -> "ADMETRequest":
if not any([self.smiles, self.name, self.cid]):
raise ValueError("Provide one of: smiles, name, or cid")
return self
class ADMETResponse(BaseModel):
job_id: str | None = None
status: str = "complete"
result: dict | None = None
error: str | None = None
class ProToxRequest(BaseModel):
smiles: str | None = Field(None, min_length=1, max_length=500, description="SMILES string")
name: str | None = Field(None, min_length=1, max_length=200, description="Chemical name (resolved via PubChem)")
models: str | None = Field(None, min_length=1, max_length=400, description="Space-separated ProTox model shorthands")
@model_validator(mode="after")
def _require_one(self) -> "ProToxRequest":
if not any([self.smiles, self.name]):
raise ValueError("Provide one of: smiles or name")
return self
class SearchHit(BaseModel):
cid: int
name: str
formula: str | None = None
smiles: str | None = None
class SearchResponse(BaseModel):
query: str
results: list[SearchHit]
def _compute_in_subprocess(smiles: str) -> dict:
"""Run RDKit computation in an isolated subprocess to prevent segfaults."""
import tempfile
backend_dir = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", ".."))
with tempfile.NamedTemporaryFile(mode='w', suffix='.py', delete=False) as f:
f.write(
'import json, sys, os\n'
f'sys.path.insert(0, {backend_dir!r})\n'
'from app.tools.admet import compute_descriptors\n'
f'result = compute_descriptors({smiles!r})\n'
'print(json.dumps(result))\n'
)
script_path = f.name
try:
result = subprocess.run(
[sys.executable, script_path],
capture_output=True, text=True, timeout=30,
)
if result.returncode != 0:
err = result.stderr.strip()[-500:] if result.stderr else "unknown error"
if "Invalid SMILES" in err or "ValueError" in err:
raise ValueError(f"Invalid SMILES: {smiles}")
raise RuntimeError(f"RDKit subprocess failed: {err}")
return json.loads(result.stdout)
finally:
try:
os.unlink(script_path)
except OSError:
pass
async def _resolve_input(body: ADMETRequest) -> tuple[str, str | None, int | None]:
"""Resolve the request to (smiles, chemical_name, cid)."""
from app.tools.pubchem import cid_to_record, name_to_cid, PubChemError
if body.smiles:
return body.smiles.strip(), None, None
if body.cid:
try:
rec = await cid_to_record(body.cid)
except PubChemError as e:
raise HTTPException(status_code=404, detail=str(e))
if not rec.get("smiles"):
raise HTTPException(status_code=404, detail="PubChem has no SMILES for this CID")
return rec["smiles"], rec.get("name"), rec["cid"]
# name given: resolve name -> CID -> SMILES
try:
cid = await name_to_cid(body.name or "")
except PubChemError as e:
raise HTTPException(status_code=404, detail=str(e))
if not cid:
raise HTTPException(status_code=404, detail=f"'{body.name}' not found in PubChem")
try:
rec = await cid_to_record(cid)
except PubChemError as e:
raise HTTPException(status_code=404, detail=str(e))
if not rec.get("smiles"):
raise HTTPException(status_code=404, detail="PubChem has no SMILES for this compound")
return rec["smiles"], rec.get("name") or body.name, rec["cid"]
@router.post("/descriptors", response_model=ADMETResponse)
async def compute_descriptors(body: ADMETRequest, user_id: str | None = Depends(get_user_id)):
"""Compute molecular descriptors from SMILES / chemical name / PubChem CID.
Returns Lipinski/Veber compliance, QED score, and key properties.
"""
try:
from app.tools.admet import compute_descriptors as _compute
smiles, chemical_name, cid = await _resolve_input(body)
if os.name == "nt":
result = _compute_in_subprocess(smiles)
else:
result = _compute(smiles)
if chemical_name:
result["chemical_name"] = chemical_name
if cid is not None:
result["pubchem_cid"] = cid
return ADMETResponse(result=result)
except ValueError as e:
raise HTTPException(status_code=400, detail=str(e))
except HTTPException:
raise
except Exception as e:
raise HTTPException(status_code=500, detail=f"Descriptor computation failed: {e}")
@router.get("/search", response_model=SearchResponse)
async def search_compounds(
q: str = Query(..., min_length=2, max_length=200, description="Name fragment"),
limit: int = Query(10, ge=1, le=25),
user_id: str | None = Depends(get_user_id),
):
"""PubChem autocomplete for chemical name search."""
from app.tools.pubchem import search_suggestions
hits = await search_suggestions(q, limit)
return SearchResponse(query=q, results=hits)
@router.post("/protox", response_model=ADMETResponse)
async def predict_toxicity(body: ProToxRequest, user_id: str | None = Depends(get_user_id)):
"""ProTox 3.0 ML-based toxicity prediction (Charité).
Queries the ProTox server for acute toxicity, toxicity targets and
optional additional models (organ toxicity, Tox21 pathways, CYPs…).
The upstream server queues requests, so a run can take up to ~5 minutes.
"""
try:
from app.tools.protox import predict_toxicity as _protox, ProToxError
smiles, chemical_name, cid = await _resolve_input(
ADMETRequest(smiles=body.smiles, name=body.name, cid=None)
)
result = await _protox(
smiles=smiles,
name=chemical_name or None,
models=body.models,
)
result["chemical_name"] = chemical_name or body.name
if cid is not None:
result["pubchem_cid"] = cid
return ADMETResponse(result=result)
except ProToxError as e:
raise HTTPException(status_code=502, detail=str(e))
except ValueError as e:
raise HTTPException(status_code=400, detail=str(e))
except HTTPException:
raise
except Exception as e:
raise HTTPException(status_code=500, detail=f"ProTox prediction failed: {e}")
|