Spaces:
Paused
Paused
File size: 2,850 Bytes
9b48f64 |
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 |
import os
import tempfile
from typing import Optional
import requests
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field
ROBOT_IP = os.getenv("ROBOT_IP", "127.0.0.1")
ROBOT_PORT = int(os.getenv("ROBOT_PORT", "31950"))
OT_API_VERSION = os.getenv("OT_API_VERSION", "3")
TIMEOUT_SEC = float(os.getenv("HTTP_TIMEOUT", "30"))
app = FastAPI(title="Opentrons Protocol Analyzer API")
class AnalyzeRequest(BaseModel):
protocol_text: str = Field(..., description="Python protocol source code")
filename: str = Field("protocol.py", description="Filename to send to robot-server")
class AnalyzeResponse(BaseModel):
ok: bool
status_code: int
data: Optional[dict] = None
error: Optional[dict] = None
def _post_protocol_file(filepath: str, filename: str) -> requests.Response:
endpoint = f"http://{ROBOT_IP}:{ROBOT_PORT}/protocols"
headers = {"Opentrons-Version": OT_API_VERSION}
with open(filepath, "rb") as f:
files = {"files": (filename, f, "text/x-python")}
return requests.post(endpoint, headers=headers, files=files, timeout=TIMEOUT_SEC)
@app.get("/health")
def health():
# FastAPIの生存確認（robot-serverの生存確認は別endpointに分ける）
return {"ok": True}
@app.get("/robot/health")
def robot_health():
# robot-serverが上がってるか軽く確認
try:
r = requests.get(f"http://{ROBOT_IP}:{ROBOT_PORT}/health", timeout=5)
return {"ok": r.ok, "status_code": r.status_code, "text": r.text[:2000]}
except requests.RequestException as e:
raise HTTPException(status_code=503, detail=f"robot-server unreachable: {e}")
@app.post("/analyze", response_model=AnalyzeResponse)
def analyze(req: AnalyzeRequest):
text = (req.protocol_text or "").strip()
if not text:
raise HTTPException(status_code=400, detail="protocol_text is empty")
fd, path = tempfile.mkstemp(prefix="protocol_", suffix=".py")
try:
with os.fdopen(fd, "w", encoding="utf-8") as f:
f.write(text)
try:
resp = _post_protocol_file(path, req.filename)
except requests.RequestException as e:
raise HTTPException(status_code=503, detail=f"robot-server request failed: {e}")
# robot-serverはJSONを返す想定
try:
payload = resp.json()
except ValueError:
# JSONじゃなければそのまま返す
raise HTTPException(status_code=502, detail=f"non-json from robot-server: {resp.text[:2000]}")
if resp.status_code >= 400:
return AnalyzeResponse(ok=False, status_code=resp.status_code, error=payload)
return AnalyzeResponse(ok=True, status_code=resp.status_code, data=payload)
finally:
try:
os.remove(path)
except OSError:
pass
|