#!/usr/bin/env python3
"""
π DroneAgent - HF Spaces Application
FastAPI application for HuggingFace Spaces deployment
"""
import os
import asyncio
import uvicorn
from pathlib import Path
from fastapi import FastAPI, HTTPException, Request
from fastapi.responses import HTMLResponse, FileResponse
from fastapi.staticfiles import StaticFiles
from fastapi.middleware.cors import CORSMiddleware
from contextlib import asynccontextmanager
# Import our DroneAgent modules
from src.services.drone_agent import DroneAgent
from src.core.websocket_server import start_websocket_server
# Global drone agent instance
drone_agent = None
@asynccontextmanager
async def lifespan(app: FastAPI):
"""Application lifespan manager."""
global drone_agent
print("π Starting DroneAgent v3.0 for HF Spaces")
print("π Initializing GPS coordinate systems...")
print("π§ Loading AI knowledge base...")
# Initialize DroneAgent first
try:
drone_agent = DroneAgent()
print("β
DroneAgent v3.0 initialized successfully")
print("π§ Gemini 2.0 Flash AI Integration Ready")
print("π‘ Real-Time Flight Management Ready")
print("πΊοΈ 6-Decimal GPS Precision Ready")
except Exception as e:
print(f"β οΈ DroneAgent initialization warning: {e}")
drone_agent = None
# Start WebSocket server for real-time notifications
print("π‘ Starting WebSocket notification server...")
try:
# Start WebSocket server in background
asyncio.create_task(start_websocket_server())
print("β
WebSocket server started successfully on port 8001")
except Exception as e:
print(f"β οΈ WebSocket server warning: {e}")
# Create necessary directories
Path("downloads").mkdir(exist_ok=True)
Path("chat_testing").mkdir(exist_ok=True)
print("π DroneAgent v3.0 fully operational!")
print("π Ready to accept chat requests and mission planning")
yield
print("π Shutting down DroneAgent v3.0")
# Initialize FastAPI with lifespan
app = FastAPI(
title="π DroneAgent",
description="Professional AI-Powered Drone Mission Planner with Real-Time Flight Management",
version="3.0.0",
lifespan=lifespan
)
# Add CORS middleware
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
@app.get("/", response_class=HTMLResponse)
async def root():
"""Serve the main DroneAgent interface."""
return """
π DroneAgent
π DroneAgent v3.0
Professional AI-Powered Drone Mission Planner
β¨ Key Features:
- π§ Gemini 2.0 Flash AI Integration
- π‘ Real-Time Flight Management
- πΊοΈ 6-Decimal GPS Precision
- π Vietnamese Geocoding Support
- π± WebSocket Notifications
- βΈοΈ Dynamic Mission Control
π API Status:
π Access Points:
"""
@app.get("/api/status")
async def api_status():
"""Get API status information."""
return {
"name": "DroneAgent",
"version": "3.0.0",
"status": "operational" if drone_agent else "initializing",
"drone_agent_loaded": drone_agent is not None,
"websocket_enabled": True,
"features": [
"π§ Gemini 2.0 Flash AI Integration",
"π‘ Real-Time Flight Management",
"πΊοΈ 6-Decimal GPS Precision",
"π Vietnamese Geocoding Support",
"π± WebSocket Notifications",
"βΈοΈ Dynamic Mission Control"
]
}
@app.get("/chat", response_class=HTMLResponse)
async def chat_interface():
"""Serve the chat interface."""
try:
if os.path.exists("chatbot_enhanced.html"):
return FileResponse("chatbot_enhanced.html", media_type="text/html")
elif os.path.exists("chatbot.html"):
return FileResponse("chatbot.html", media_type="text/html")
except:
pass
return """
π DroneAgent Chat
π DroneAgent Chat Interface
Use the API endpoints directly:
"""
# Include our DroneAgent API routes
try:
from src.api.routes import router as api_router
app.include_router(api_router)
print("β
API routes loaded successfully")
except Exception as e:
print(f"β οΈ Could not load API routes: {e}")
# Add basic chat endpoint if routes fail to load
from src.api.models import ChatMessage
@app.post("/chat")
async def fallback_chat(message: ChatMessage):
"""Fallback chat endpoint when main routes fail."""
if drone_agent:
try:
response = await drone_agent.process_message(message.message)
return response
except Exception as e:
return {
"response": f"Error processing message: {str(e)}",
"error": True,
"is_guidance": True
}
else:
return {
"response": "DroneAgent is initializing. Please try again in a moment.",
"error": True,
"is_guidance": True
}
if __name__ == "__main__":
port = int(os.environ.get('PORT', 7860)) # HF Spaces default port
print(f"π Starting DroneAgent on port {port}")
uvicorn.run(app, host="0.0.0.0", port=port, log_level="info")