baxin/ot_analysis

1import os2import tempfile3import requests4import gradio as gr5 6ROBOT_IP = os.getenv("ROBOT_IP", "localhost")7ROBOT_PORT = int(os.getenv("ROBOT_PORT", "31950"))8OT_API_VERSION = os.getenv("OT_API_VERSION", "3")9TIMEOUT_SEC = float(os.getenv("HTTP_TIMEOUT", "30"))10 11def upload_protocol_file(filepath: str) -> str:12    endpoint = f"http://{ROBOT_IP}:{ROBOT_PORT}/protocols"13    headers = {"Opentrons-Version": OT_API_VERSION}14    filename = os.path.basename(filepath)15 16    try:17        with open(filepath, "rb") as f:18            files = {"files": (filename, f, "text/x-python")}19            resp = requests.post(endpoint, headers=headers, files=files, timeout=TIMEOUT_SEC)20    except requests.RequestException as e:21        return f"HTTP error: {e}"22 23    try:24        data = resp.json()25    except ValueError:26        return f"Non-JSON response (status={resp.status_code}):\n{resp.text[:1000]}"27 28    if resp.status_code >= 400:29        if isinstance(data, dict) and data.get("errors"):30            err = data["errors"][0]31            return (32                f"Upload failed (HTTP {resp.status_code})\n"33                f"id: {err.get('id')}\n"34                f"code: {err.get('errorCode')}\n"35                f"detail: {err.get('detail')}"36            )37        return f"Upload failed (HTTP {resp.status_code}):\n{data}"38 39    if isinstance(data, dict) and "data" in data:40        d = data["data"]41        summaries = d.get("analysisSummaries") or []42        return (43            "✅ success\n"44            f"protocol_id: {d.get('id')}\n"45            f"analysis_id: {(summaries[0].get('id') if summaries else None)}\n"46            f"analysis_status: {(summaries[0].get('status') if summaries else None)}"47        )48 49    return f"Unexpected response:\n{data}"50 51def send_message(text, history):52    text = (text or "").strip()53    history = history or []54 55    if not text:56        return history57 58    # user message59    history.append({"role": "user", "content": text})60 61    fd, path = tempfile.mkstemp(prefix="protocol_", suffix=".py")62    try:63        with os.fdopen(fd, "w", encoding="utf-8") as f:64            f.write(text)65        result = upload_protocol_file(path)66    finally:67        try:68            os.remove(path)69        except OSError:70            pass71 72    # assistant message73    history.append({"role": "assistant", "content": result})74    return history75 76with gr.Blocks() as app:77    gr.Markdown("## Opentrons Protocol Analyzer")78    textbox = gr.Textbox(lines=12, label="Paste protocol Python code")79    send_button = gr.Button(value="Analyze")80 81    # 旧Gradioでも動きやすい：余計な引数を渡さない82    chatbot = gr.Chatbot(label="Results")83 84    clear_button = gr.ClearButton([textbox, chatbot])85 86    send_button.click(send_message, [textbox, chatbot], [chatbot])87    textbox.submit(send_message, [textbox, chatbot], [chatbot])88 89if __name__ == "__main__":90    app.launch(server_name="0.0.0.0", server_port=int(os.getenv("PORT", "7860")))91