baxin/flex-analysis

1import os2import tempfile3from typing import Optional4 5import requests6from fastapi import FastAPI, HTTPException7from pydantic import BaseModel, Field8 9ROBOT_IP = os.getenv("ROBOT_IP", "127.0.0.1")10ROBOT_PORT = int(os.getenv("ROBOT_PORT", "31950"))11OT_API_VERSION = os.getenv("OT_API_VERSION", "3")12TIMEOUT_SEC = float(os.getenv("HTTP_TIMEOUT", "30"))13 14app = FastAPI(title="Opentrons Protocol Analyzer API")15 16class AnalyzeRequest(BaseModel):17    protocol_text: str = Field(..., description="Python protocol source code")18    filename: str = Field("protocol.py", description="Filename to send to robot-server")19 20class AnalyzeResponse(BaseModel):21    ok: bool22    status_code: int23    data: Optional[dict] = None24    error: Optional[dict] = None25 26def _post_protocol_file(filepath: str, filename: str) -> requests.Response:27    endpoint = f"http://{ROBOT_IP}:{ROBOT_PORT}/protocols"28    headers = {"Opentrons-Version": OT_API_VERSION}29    with open(filepath, "rb") as f:30        files = {"files": (filename, f, "text/x-python")}31        return requests.post(endpoint, headers=headers, files=files, timeout=TIMEOUT_SEC)32 33@app.get("/health")34def health():35    # FastAPIの生存確認（robot-serverの生存確認は別endpointに分ける）36    return {"ok": True}37 38@app.get("/robot/health")39def robot_health():40    # robot-serverが上がってるか軽く確認41    try:42        r = requests.get(f"http://{ROBOT_IP}:{ROBOT_PORT}/health", timeout=5)43        return {"ok": r.ok, "status_code": r.status_code, "text": r.text[:2000]}44    except requests.RequestException as e:45        raise HTTPException(status_code=503, detail=f"robot-server unreachable: {e}")46 47@app.post("/analyze", response_model=AnalyzeResponse)48def analyze(req: AnalyzeRequest):49    text = (req.protocol_text or "").strip()50    if not text:51        raise HTTPException(status_code=400, detail="protocol_text is empty")52 53    fd, path = tempfile.mkstemp(prefix="protocol_", suffix=".py")54    try:55        with os.fdopen(fd, "w", encoding="utf-8") as f:56            f.write(text)57 58        try:59            resp = _post_protocol_file(path, req.filename)60        except requests.RequestException as e:61            raise HTTPException(status_code=503, detail=f"robot-server request failed: {e}")62 63        # robot-serverはJSONを返す想定64        try:65            payload = resp.json()66        except ValueError:67            # JSONじゃなければそのまま返す68            raise HTTPException(status_code=502, detail=f"non-json from robot-server: {resp.text[:2000]}")69 70        if resp.status_code >= 400:71            return AnalyzeResponse(ok=False, status_code=resp.status_code, error=payload)72 73        return AnalyzeResponse(ok=True, status_code=resp.status_code, data=payload)74 75    finally:76        try:77            os.remove(path)78        except OSError:79            pass80