""" IPMI Controller Web Server Advanced web interface with dark mode, fan groups, multiple curves """ import asyncio import json import logging import hashlib import secrets from contextlib import asynccontextmanager from pathlib import Path from typing import Optional, List, Dict from datetime import datetime, timedelta from fastapi import FastAPI, HTTPException, Depends, Request from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import HTMLResponse, JSONResponse, Response from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials from fastapi.staticfiles import StaticFiles from pydantic import BaseModel, Field import sys sys.path.insert(0, str(Path(__file__).parent)) from fan_controller import get_service, IPMIControllerService logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) security = HTTPBearer(auto_error=False) # Data directories DATA_DIR = Path("/app/data") if Path("/app/data").exists() else Path(__file__).parent / "data" DATA_DIR.mkdir(exist_ok=True) CONFIG_FILE = DATA_DIR / "config.json" USERS_FILE = DATA_DIR / "users.json" # Pydantic models class UserLogin(BaseModel): username: str password: str class ChangePassword(BaseModel): current_password: str new_password: str class SetupRequest(BaseModel): admin_username: str admin_password: str ipmi_host: str ipmi_username: str ipmi_password: str ipmi_port: int = 623 http_sensor_enabled: Optional[bool] = False http_sensor_url: Optional[str] = None http_sensor_timeout: Optional[int] = 10 class IPMIConfig(BaseModel): host: str username: str password: Optional[str] = None port: int = 623 class HTTPConfig(BaseModel): enabled: bool = False url: Optional[str] = None timeout: int = 10 class FanSettings(BaseModel): enabled: Optional[bool] = None poll_interval: Optional[int] = Field(None, ge=5, le=300) fan_update_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) panic_on_no_data: Optional[bool] = None no_data_timeout: Optional[int] = Field(None, ge=10, le=300) primary_sensor: Optional[str] = None sensor_preference: Optional[str] = None theme: Optional[str] = None class FanCurvePoint(BaseModel): temp: float = Field(..., ge=0, le=100) speed: int = Field(..., ge=0, le=100) class FanCurveCreate(BaseModel): name: str points: List[FanCurvePoint] sensor_source: str = "cpu" applies_to: str = "all" class FanConfig(BaseModel): fan_id: str name: Optional[str] = None group: Optional[str] = None curve: Optional[str] = None class FanGroupCreate(BaseModel): name: str fans: List[str] curve: str = "Default" class ManualSpeedRequest(BaseModel): speed: int = Field(..., ge=0, le=100) fan_id: Optional[str] = "0xff" group: Optional[str] = None class IdentifyRequest(BaseModel): fan_id: str # User Manager class UserManager: def __init__(self): self.users_file = USERS_FILE self._users = {} self._sessions = {} self._load() def _load(self): if self.users_file.exists(): try: with open(self.users_file) as f: data = json.load(f) self._users = data.get('users', {}) except Exception as e: logger.error(f"Failed to load users: {e}") def _save(self): with open(self.users_file, 'w') as f: json.dump({'users': self._users}, f) def _hash(self, password: str) -> str: return hashlib.sha256(password.encode()).hexdigest() def verify(self, username: str, password: str) -> bool: return username in self._users and self._users[username] == self._hash(password) def create(self, username: str, password: str) -> bool: if username in self._users: return False self._users[username] = self._hash(password) self._save() return True def change_password(self, username: str, current: str, new: str) -> bool: if not self.verify(username, current): return False self._users[username] = self._hash(new) self._save() return True def create_token(self, username: str) -> str: token = secrets.token_urlsafe(32) self._sessions[token] = (username, datetime.utcnow() + timedelta(days=7)) return token def verify_token(self, token: str) -> Optional[str]: if token not in self._sessions: return None username, expiry = self._sessions[token] if datetime.utcnow() > expiry: del self._sessions[token] return None return username def is_setup_complete(self) -> bool: return len(self._users) > 0 user_manager = UserManager() async def get_current_user(credentials: HTTPAuthorizationCredentials = Depends(security)) -> str: if not credentials: raise HTTPException(status_code=401, detail="Not authenticated") username = user_manager.verify_token(credentials.credentials) if not username: raise HTTPException(status_code=401, detail="Invalid or expired token") return username # CSS Themes THEMES = { "dark": """ :root { --bg-primary: #0f0f1e; --bg-secondary: #1a1a2e; --bg-card: rgba(255,255,255,0.05); --text-primary: #ffffff; --text-secondary: #a0a0a0; --accent-primary: #2196f3; --accent-success: #4caf50; --accent-warning: #ff9800; --accent-danger: #f44336; --border: rgba(255,255,255,0.1); } """, "light": """ :root { --bg-primary: #f5f5f5; --bg-secondary: #ffffff; --bg-card: #ffffff; --text-primary: #333333; --text-secondary: #666666; --accent-primary: #1976d2; --accent-success: #388e3c; --accent-warning: #f57c00; --accent-danger: #d32f2f; --border: #e0e0e0; } body { background: var(--bg-primary) !important; color: var(--text-primary) !important; } .card { background: var(--bg-card) !important; border-color: var(--border) !important; } input, select, textarea { background: #f0f0f0 !important; color: #333 !important; border-color: #ccc !important; } """ } # HTML Template def get_html(theme="dark"): theme_css = THEMES.get(theme, THEMES["dark"]) return f''' IPMI Controller

IPMI Controller

IPMI
-
Mode
-
Max Temp
-
Fan Speed
-
Sensors
-

Quick Controls

Temperatures

Loading...

Fans

Loading...

IPMI Configuration

HTTP Sensor (lm-sensors over HTTP)

Fan Control Settings

Fan Groups Management

Loading fan groups...

Individual Fan Configuration

Connect to IPMI to see fans

Quick Actions

Fan Curves

System Logs

''' LOGIN_HTML = ''' Login - IPMI Controller

IPMI Controller

''' with open('/home/devmatrix/.openclaw/workspace/setup_html.txt', 'r') as f: SETUP_HTML = f.read().strip().replace("SETUP_HTML = '''", "").replace("'''", "") # FastAPI app @asynccontextmanager async def lifespan(app: FastAPI): DATA_DIR.mkdir(parents=True, exist_ok=True) # Auto-start fan control if enabled in config and setup is complete if user_manager.is_setup_complete(): try: service = get_service(str(CONFIG_FILE)) if service.config.get('enabled', False): logger.info("Auto-starting fan control (enabled in config)") if not service.running: service.start() except Exception as e: logger.error(f"Auto-start failed: {e}") yield app = FastAPI( title="IPMI Controller", description="Advanced fan control for Dell servers", version="3.0.0", lifespan=lifespan ) app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) # Static files for icons app.mount("/icons", StaticFiles(directory=str(Path(__file__).parent / "static" / "icons")), name="icons") # Routes @app.get("/favicon.ico") async def favicon(): """Fan icon as favicon.""" icon_path = Path(__file__).parent / "static" / "icons" / "favicon.svg" return Response(content=icon_path.read_text(), media_type="image/svg+xml") @app.get("/") async def root(request: Request): """Main dashboard.""" if not user_manager.is_setup_complete(): return HTMLResponse(content=SETUP_HTML) # Get theme preference from query or default to dark theme = request.query_params.get('theme', 'dark') return HTMLResponse(content=get_html(theme)) @app.get("/login") async def login_page(): """Login page.""" if not user_manager.is_setup_complete(): return HTMLResponse(content=SETUP_HTML) return HTMLResponse(content=LOGIN_HTML) # Auth API @app.post("/api/auth/login") async def api_login(credentials: UserLogin): if not user_manager.verify(credentials.username, credentials.password): return {"success": False, "error": "Invalid credentials"} token = user_manager.create_token(credentials.username) return {"success": True, "token": token} @app.post("/api/auth/change-password") async def api_change_password(data: ChangePassword, username: str = Depends(get_current_user)): if not user_manager.change_password(username, data.current_password, data.new_password): return {"success": False, "error": "Current password is incorrect"} return {"success": True} # Setup test endpoints (no auth required for setup) @app.post("/api/setup/test-ipmi") async def api_setup_test_ipmi(data: dict): """Test IPMI connection during setup.""" from fan_controller import IPMIFanController try: controller = IPMIFanController( host=data.get('host'), username=data.get('username'), password=data.get('password'), port=data.get('port', 623) ) if controller.test_connection(): return {"success": True} return {"success": False, "error": "Connection failed"} except Exception as e: return {"success": False, "error": str(e)} @app.post("/api/setup/test-http") async def api_setup_test_http(data: dict): """Test HTTP sensor during setup.""" from fan_controller import HTTPSensorClient try: client = HTTPSensorClient(url=data.get('url'), timeout=10) temps = client.fetch_sensors() return {"success": True, "sensors": len(temps)} except Exception as e: return {"success": False, "error": str(e)} @app.post("/api/setup") async def api_setup(data: SetupRequest): if user_manager.is_setup_complete(): return {"success": False, "error": "Setup already completed"} if not user_manager.create(data.admin_username, data.admin_password): return {"success": False, "error": "Failed to create user"} service = get_service(str(CONFIG_FILE)) updates = { "ipmi_host": data.ipmi_host, "ipmi_username": data.ipmi_username, "ipmi_password": data.ipmi_password, "ipmi_port": data.ipmi_port } # Add HTTP config if provided if hasattr(data, 'http_sensor_enabled') and data.http_sensor_enabled: updates["http_sensor_enabled"] = True updates["http_sensor_url"] = getattr(data, 'http_sensor_url', '') updates["http_sensor_timeout"] = getattr(data, 'http_sensor_timeout', 10) service.update_config(**updates) token = user_manager.create_token(data.admin_username) return {"success": True, "token": token} # Status API @app.get("/api/status") async def api_status(username: str = Depends(get_current_user)): service = get_service(str(CONFIG_FILE)) return service.get_status() @app.post("/api/test") async def api_test(username: str = Depends(get_current_user)): service = get_service(str(CONFIG_FILE)) if not service.controller: if not service._init_controller(): return {"success": False, "error": "Failed to connect - check config"} success = service.controller.test_connection() return {"success": success, "error": None if success else "Connection failed"} # Control API @app.post("/api/control/auto") async def api_control_auto(data: dict, username: str = Depends(get_current_user)): service = get_service(str(CONFIG_FILE)) service.set_auto_mode(data.get('enabled', False)) if data.get('enabled') and not service.running: if not service.start(): return {"success": False, "error": "Failed to start"} return {"success": True} @app.post("/api/control/manual") async def api_control_manual(req: ManualSpeedRequest, username: str = Depends(get_current_user)): service = get_service(str(CONFIG_FILE)) if not service.controller: if not service._init_controller(): return {"success": False, "error": "Not connected"} # Handle group speed setting if req.group: groups = service.config.get('fan_groups', {}) if req.group not in groups: return {"success": False, "error": f"Group '{req.group}' not found"} fan_ids = groups[req.group].get('fan_ids', []) success = True for fan_id in fan_ids: if not service.set_manual_speed(req.speed, fan_id): success = False return {"success": success} # Handle individual fan or all fans if service.set_manual_speed(req.speed, req.fan_id): return {"success": True} return {"success": False, "error": "Failed"} # Config API @app.post("/api/config/ipmi") async def api_config_ipmi(data: IPMIConfig, username: str = Depends(get_current_user)): service = get_service(str(CONFIG_FILE)) updates = { "ipmi_host": data.host, "ipmi_username": data.username, "ipmi_port": data.port } if data.password: updates["ipmi_password"] = data.password service.update_config(**updates) return {"success": True} @app.post("/api/config/http") async def api_config_http(data: HTTPConfig, username: str = Depends(get_current_user)): service = get_service(str(CONFIG_FILE)) service.update_config( http_sensor_enabled=data.enabled, http_sensor_url=data.url, http_sensor_timeout=data.timeout ) return {"success": True} @app.post("/api/config/settings") async def api_config_settings(data: FanSettings, username: str = Depends(get_current_user)): service = get_service(str(CONFIG_FILE)) updates = {k: v for k, v in data.model_dump().items() if v is not None} service.update_config(**updates) return {"success": True} # Fan Curves API @app.get("/api/curves") async def api_list_curves(username: str = Depends(get_current_user)): """List all fan curves.""" service = get_service(str(CONFIG_FILE)) return { "curves": service.config.get('fan_curves', {}), "active": service.config.get('active_curve', 'Balanced') } @app.post("/api/curves") async def api_create_curve(data: dict, username: str = Depends(get_current_user)): """Create or update a fan curve.""" service = get_service(str(CONFIG_FILE)) name = data.get('name') if not name: return {"success": False, "error": "Name required"} curves = service.config.get('fan_curves', {}) curves[name] = { "points": data.get('points', []), "sensor_source": data.get('sensor_source', 'cpu'), "applies_to": data.get('applies_to', 'all'), "group": data.get('group') or None } service.update_config(fan_curves=curves) return {"success": True} @app.post("/api/curves/active") async def api_set_active_curve(data: dict, username: str = Depends(get_current_user)): """Set active fan curve.""" service = get_service(str(CONFIG_FILE)) name = data.get('name') curves = service.config.get('fan_curves', {}) if name not in curves: return {"success": False, "error": "Curve not found"} service.update_config(active_curve=name) return {"success": True} @app.delete("/api/curves/{name}") async def api_delete_curve(name: str, username: str = Depends(get_current_user)): """Delete a fan curve.""" service = get_service(str(CONFIG_FILE)) curves = service.config.get('fan_curves', {}) if name in curves: del curves[name] service.update_config(fan_curves=curves) return {"success": True} # Fan API @app.post("/api/fans/identify") async def api_identify_fan(req: IdentifyRequest, username: str = Depends(get_current_user)): service = get_service(str(CONFIG_FILE)) if service.identify_fan(req.fan_id): return {"success": True} return {"success": False, "error": "Failed"} @app.post("/api/fans/stop-identify") async def api_stop_identify(username: str = Depends(get_current_user)): service = get_service(str(CONFIG_FILE)) service.stop_identify() return {"success": True} # Fan Groups API @app.post("/api/fans/groups") async def api_save_group(data: dict, username: str = Depends(get_current_user)): service = get_service(str(CONFIG_FILE)) name = data.get('name') fan_ids = data.get('fan_ids', []) if not name: return {"success": False, "error": "Group name required"} if 'fan_groups' not in service.config: service.config['fan_groups'] = {} service.config['fan_groups'][name] = {'fan_ids': fan_ids} service._save_config() return {"success": True} @app.delete("/api/fans/groups") async def api_delete_group(data: dict, username: str = Depends(get_current_user)): service = get_service(str(CONFIG_FILE)) name = data.get('name') if 'fan_groups' in service.config and name in service.config['fan_groups']: del service.config['fan_groups'][name] service._save_config() return {"success": True} return {"success": False, "error": "Group not found"} # Public API (no auth required - for external integrations) @app.get("/api/public/status") async def api_public_status(): """Public status endpoint for integrations.""" service = get_service(str(CONFIG_FILE)) status = service.get_status() # Return limited public data return { "temperatures": status.get("temperatures", []), "fans": status.get("fans", []), "current_speeds": status.get("current_speeds", {}), "connected": status.get("connected", False), "enabled": status.get("enabled", False) } @app.get("/api/public/temperatures") async def api_public_temps(): """Public temperatures endpoint.""" service = get_service(str(CONFIG_FILE)) return {"temperatures": service.get_status().get("temperatures", [])} @app.get("/api/public/fans") async def api_public_fans(): """Public fans endpoint.""" service = get_service(str(CONFIG_FILE)) return {"fans": service.get_status().get("fans", [])} if __name__ == "__main__": import uvicorn uvicorn.run(app, host="0.0.0.0", port=8000)