Docking@HOME - Molecular Docking Platform

""" GUI Interface for Docking@HOME A modern web-based GUI using FastAPI and HTML/JavaScript for molecular docking. Integrates with AutoDock backend for real molecular docking simulations. Authors: OpenPeer AI, Riemann Computing Inc., Bleunomics, Andrew Magdy Kamal """ import os import json import asyncio from pathlib import Path from typing import Optional, List, Dict from datetime import datetime from fastapi import FastAPI, File, UploadFile, HTTPException, WebSocket, WebSocketDisconnect from fastapi.responses import HTMLResponse, FileResponse, JSONResponse from fastapi.staticfiles import StaticFiles from pydantic import BaseModel import uvicorn # Import the docking server components from .server import job_manager, initialize_server # Initialize FastAPI app app = FastAPI( title="Docking@HOME", description="Distributed Molecular Docking Platform", version="1.0.0" ) # Data models class DockingJobRequest(BaseModel): num_runs: int = 100 use_gpu: bool = True job_name: Optional[str] = None class JobStatus(BaseModel): job_id: str status: str progress: float message: str # Active WebSocket connections active_websockets: List[WebSocket] = [] # File upload directory UPLOAD_DIR = Path("uploads") UPLOAD_DIR.mkdir(exist_ok=True) RESULTS_DIR = Path("results") RESULTS_DIR.mkdir(exist_ok=True) @app.get("/", response_class=HTMLResponse) async def root(): """Serve the main GUI page""" html_content = """ Docking@HOME - Molecular Docking Platform

📤 Submit Docking Job

📊 Active Jobs

No jobs yet. Submit a docking job to get started!

📈 System Statistics

Total Jobs

Completed

Avg. Time

""" return HTMLResponse(content=html_content) @app.post("/upload") async def upload_file(file: UploadFile = File(...)): """Upload ligand or receptor file""" try: file_path = UPLOAD_DIR / file.filename with open(file_path, "wb") as f: content = await file.read() f.write(content) return {"filename": file.filename, "path": str(file_path)} except Exception as e: raise HTTPException(status_code=500, detail=str(e)) @app.post("/api/jobs") async def create_job( ligand_file: str, receptor_file: str, num_runs: int = 100, use_gpu: bool = True, job_name: Optional[str] = None ): """Create a new docking job with real AutoDock integration""" try: # Submit job to the docking server job_id = await job_manager.submit_job( ligand_file=ligand_file, receptor_file=receptor_file, num_runs=num_runs, use_gpu=use_gpu, job_name=job_name ) # Start broadcasting updates asyncio.create_task(broadcast_job_updates(job_id)) return job_manager.get_job(job_id) except Exception as e: raise HTTPException(status_code=500, detail=str(e)) @app.get("/api/jobs") async def get_jobs(): """Get all jobs""" return job_manager.get_all_jobs() @app.get("/api/jobs/{job_id}") async def get_job(job_id: str): """Get specific job""" job = job_manager.get_job(job_id) if not job: raise HTTPException(status_code=404, detail="Job not found") return job @app.get("/api/stats") async def get_stats(): """Get system statistics""" return job_manager.get_stats() @app.websocket("/ws") async def websocket_endpoint(websocket: WebSocket): """WebSocket for real-time updates""" await websocket.accept() active_websockets.append(websocket) try: while True: data = await websocket.receive_text() # Echo back or handle commands await websocket.send_json({"status": "connected"}) except WebSocketDisconnect: active_websockets.remove(websocket) except Exception as e: if websocket in active_websockets: active_websockets.remove(websocket) async def broadcast_job_updates(job_id: str): """Broadcast job progress to all connected WebSocket clients""" while True: await asyncio.sleep(0.5) # Update every 500ms job = job_manager.get_job(job_id) if not job: break # Send update to all connected clients for ws in active_websockets[:]: # Copy list to avoid modification during iteration try: await ws.send_json({ "type": "job_update", "job_id": job_id, "status": job["status"], "progress": job["progress"] }) except Exception: # Remove disconnected clients if ws in active_websockets: active_websockets.remove(ws) # Stop broadcasting if job is complete or failed if job["status"] in ["completed", "failed"]: break @app.on_event("startup") async def startup_event(): """Initialize the server on startup""" await initialize_server() def start_gui(host: str = "localhost", port: int = 8080): """Start the GUI server with AutoDock integration""" print(f""" ╔═══════════════════════════════════════════════════════════════╗ ║ Docking@HOME GUI Server ║ ║ Real AutoDock Integration with GPU Support ║ ║ ║ ║ 🌐 Server: http://{host}:{port} ║ ║ 🧬 AutoDock: Enabled (GPU acceleration supported) ║ ║ 📧 Support: andrew@bleunomics.com ║ ║ 🤗 Issues: https://huggingface.co/OpenPeerAI/DockingAtHOME ║ ║ ║ ║ Open your browser to start docking! ║ ╚═══════════════════════════════════════════════════════════════╝ """) uvicorn.run(app, host=host, port=port, log_level="info") if __name__ == "__main__": start_gui()

🧬 Docking@HOME

📤 Submit Docking Job

📊 Active Jobs

📈 System Statistics