""" Web API for IPMI Fan Controller v2 """ import asyncio import json import logging from contextlib import asynccontextmanager from pathlib import Path from typing import Optional, List, Dict from fastapi import FastAPI, HTTPException, BackgroundTasks from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import HTMLResponse, FileResponse from fastapi.staticfiles import StaticFiles from pydantic import BaseModel, Field from fan_controller import get_service, FanControlService, IPMIFanController logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) # Pydantic models class ConfigUpdate(BaseModel): host: Optional[str] = None username: Optional[str] = None password: Optional[str] = None port: Optional[int] = 623 enabled: Optional[bool] = None interval: Optional[int] = Field(None, ge=5, le=300) min_speed: Optional[int] = Field(None, ge=0, le=100) max_speed: Optional[int] = Field(None, ge=0, le=100) panic_temp: Optional[float] = Field(None, ge=50, le=100) panic_speed: Optional[int] = Field(None, ge=0, le=100) class FanCurvePoint(BaseModel): temp: float = Field(..., ge=0, le=100) speed: int = Field(..., ge=0, le=100) class FanCurveUpdate(BaseModel): points: List[FanCurvePoint] class ManualSpeedRequest(BaseModel): speed: int = Field(..., ge=0, le=100) class StatusResponse(BaseModel): running: bool enabled: bool connected: bool manual_mode: bool current_speed: int target_speed: int temperatures: List[Dict] fans: List[Dict] # Create static directory and HTML STATIC_DIR = Path(__file__).parent / "static" STATIC_DIR.mkdir(exist_ok=True) # Create the HTML dashboard DASHBOARD_HTML = ''' IPMI Fan Controller v2

🌬️ IPMI Fan Controller v2

Dell T710 & Compatible Servers

📊 Current Status

Connection
-
Control Mode
-
Current Speed
-
Max Temp
-

🎛️ Quick Controls

⚙️ Configuration

📈 Fan Curve

Temp (°C) → Speed (%)

📝 Logs

Ready...
''' # Write the HTML file (STATIC_DIR / "index.html").write_text(DASHBOARD_HTML) # FastAPI app @asynccontextmanager async def lifespan(app: FastAPI): """Application lifespan handler.""" service = get_service() # Auto-start if configured if service.config.get('enabled') and service.config.get('host'): service.start() yield service.stop() app = FastAPI( title="IPMI Fan Controller v2", description="Simple, robust fan control for Dell servers", version="2.0.0", lifespan=lifespan ) app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) @app.get("/", response_class=HTMLResponse) async def root(): """Serve the dashboard.""" return HTMLResponse(content=DASHBOARD_HTML) @app.get("/api/status", response_model=StatusResponse) async def get_status(): """Get current controller status.""" service = get_service() return service.get_status() @app.post("/api/test") async def test_connection(): """Test IPMI connection.""" service = get_service() if not service.controller: if not service._init_controller(): return {"success": False, "error": "Failed to initialize controller"} success = service.controller.test_connection() return {"success": success, "error": None if success else "Connection failed"} @app.post("/api/config") async def update_config(update: ConfigUpdate): """Update configuration.""" service = get_service() updates = {k: v for k, v in update.model_dump().items() if v is not None} if not updates: return {"success": False, "error": "No valid updates provided"} # Require password if host/username changed and no new password provided if ('host' in updates or 'username' in updates) and 'password' not in updates: if not service.config.get('password'): return {"success": False, "error": "Password required when setting host/username"} try: service.update_config(**updates) return {"success": True} except Exception as e: return {"success": False, "error": str(e)} @app.post("/api/config/curve") async def update_curve(curve: FanCurveUpdate): """Update fan curve.""" service = get_service() points = [{"temp": p.temp, "speed": p.speed} for p in curve.points] service.update_config(fan_curve=points) return {"success": True} @app.post("/api/control/auto") async def set_auto_control(data: dict): """Enable/disable automatic control.""" service = get_service() enabled = data.get('enabled', False) if enabled and not service.config.get('host'): return {"success": False, "error": "IPMI host not configured"} service.set_auto_mode(enabled) if enabled and not service.running: if not service.start(): return {"success": False, "error": "Failed to start service"} return {"success": True} @app.post("/api/control/manual") async def set_manual_control(req: ManualSpeedRequest): """Set manual fan speed.""" service = get_service() if not service.controller: if not service._init_controller(): return {"success": False, "error": "Failed to connect to server"} if service.set_manual_speed(req.speed): return {"success": True} return {"success": False, "error": "Failed to set fan speed"} @app.post("/api/shutdown") async def shutdown(): """Return fans to automatic control and stop service.""" service = get_service() service.stop() return {"success": True, "message": "Service stopped, fans returned to automatic control"} if __name__ == "__main__": import uvicorn uvicorn.run(app, host="0.0.0.0", port=8000)