From 5b9ec7b3515f900ac304a3040aa496026b319922 Mon Sep 17 00:00:00 2001 From: devmatrix Date: Fri, 20 Feb 2026 15:01:39 +0000 Subject: [PATCH] Initial v2: Simpler, more robust fan controller --- README.md | 138 +++++++++ fan_controller.py | 534 +++++++++++++++++++++++++++++++++++ install.sh | 147 ++++++++++ requirements.txt | 4 + web_server.py | 695 ++++++++++++++++++++++++++++++++++++++++++++++ 5 files changed, 1518 insertions(+) create mode 100644 README.md create mode 100644 fan_controller.py create mode 100755 install.sh create mode 100644 requirements.txt create mode 100644 web_server.py diff --git a/README.md b/README.md new file mode 100644 index 0000000..dfaae10 --- /dev/null +++ b/README.md @@ -0,0 +1,138 @@ +# IPMI Fan Controller v2 + +A simpler, more robust fan controller for Dell T710 and compatible servers using IPMI. + +## What's Different from v1? + +- **Direct host execution** - No Docker networking complications +- **Better error recovery** - Automatically reconnects on IPMI failures +- **Simpler codebase** - Easier to debug and modify +- **Working web UI** - Clean, responsive dashboard +- **CLI testing mode** - Test without starting the web server + +## Quick Start + +### 1. Install + +```bash +cd ~/projects/fan-controller-v2 +chmod +x install.sh +sudo ./install.sh +``` + +This will: +- Install Python dependencies +- Create systemd service +- Set up config in `/etc/ipmi-fan-controller/` + +### 2. Configure + +Edit the configuration file: + +```bash +sudo nano /etc/ipmi-fan-controller/config.json +``` + +Set your IPMI credentials: +```json +{ + "host": "192.168.1.100", + "username": "root", + "password": "your-password", + "port": 623 +} +``` + +### 3. Start + +```bash +sudo systemctl start ipmi-fan-controller +``` + +Open the web UI at `http://your-server:8000` + +## CLI Testing + +Test the IPMI connection without the web server: + +```bash +python3 fan_controller.py 192.168.1.100 root password +``` + +This will: +1. Test the connection +2. Show temperatures and fan speeds +3. Try manual fan control (30% → 50% → auto) + +## Features + +### Automatic Control +- Adjusts fan speed based on CPU temperature +- Configurable fan curve (temp → speed mapping) +- Panic mode: sets fans to 100% if temp exceeds threshold + +### Manual Control +- Set any fan speed from 0-100% +- Override automatic control temporarily + +### Safety Features +- Returns to automatic control on shutdown +- Reconnects automatically if IPMI connection drops +- Panic temperature protection + +## Configuration Options + +```json +{ + "host": "192.168.1.100", // IPMI IP address + "username": "root", // IPMI username + "password": "secret", // IPMI password + "port": 623, // IPMI port (default: 623) + "enabled": false, // Start automatic control on boot + "interval": 10, // Check interval in seconds + "min_speed": 10, // Minimum fan speed (%) + "max_speed": 100, // Maximum fan speed (%) + "panic_temp": 85, // Panic mode trigger (°C) + "panic_speed": 100, // Panic mode fan speed (%) + "fan_curve": [ // Temp (°C) → Speed (%) mapping + {"temp": 30, "speed": 15}, + {"temp": 40, "speed": 25}, + {"temp": 50, "speed": 40}, + {"temp": 60, "speed": 60}, + {"temp": 70, "speed": 80}, + {"temp": 80, "speed": 100} + ] +} +``` + +## Troubleshooting + +### Connection Failed +1. Verify IPMI is enabled in BIOS/iDRAC +2. Test manually: `ipmitool -I lanplus -H -U -P mc info` +3. Check firewall allows port 623 + +### Fans Not Responding +1. Some Dell servers need 3rd party PCIe response disabled +2. Try enabling manual mode first via web UI +3. Check IPMI user has admin privileges + +### Service Won't Start +```bash +# Check logs +sudo journalctl -u ipmi-fan-controller -f + +# Check config is valid JSON +sudo python3 -c "import json; json.load(open('/etc/ipmi-fan-controller/config.json'))" +``` + +## Files + +- `fan_controller.py` - Core IPMI control logic +- `web_server.py` - FastAPI web interface +- `install.sh` - Installation script +- `requirements.txt` - Python dependencies + +## License + +MIT License - Feel free to modify and distribute. diff --git a/fan_controller.py b/fan_controller.py new file mode 100644 index 0000000..ee6f270 --- /dev/null +++ b/fan_controller.py @@ -0,0 +1,534 @@ +""" +IPMI Fan Controller v2 - Simpler, More Robust +For Dell T710 and compatible servers +""" +import subprocess +import re +import time +import json +import logging +import threading +from dataclasses import dataclass, asdict +from typing import List, Dict, Optional, Tuple +from datetime import datetime +from pathlib import Path + +# Setup logging +logging.basicConfig( + level=logging.INFO, + format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', + handlers=[ + logging.StreamHandler(), + logging.FileHandler('/tmp/ipmi-fan-controller.log') + ] +) +logger = logging.getLogger(__name__) + + +@dataclass +class FanCurvePoint: + temp: float + speed: int + + +@dataclass +class TemperatureReading: + name: str + location: str + value: float + status: str + + +@dataclass +class FanReading: + fan_id: str + fan_number: int + speed_rpm: Optional[int] + speed_percent: Optional[int] + + +class IPMIFanController: + """Simplified IPMI fan controller with robust error handling.""" + + # Default fan curve (temp C -> speed %) + DEFAULT_CURVE = [ + FanCurvePoint(30, 15), + FanCurvePoint(40, 25), + FanCurvePoint(50, 40), + FanCurvePoint(60, 60), + FanCurvePoint(70, 80), + FanCurvePoint(80, 100), + ] + + def __init__(self, host: str, username: str, password: str, port: int = 623): + self.host = host + self.username = username + self.password = password + self.port = port + self.manual_mode = False + self.last_successful_read = None + self.consecutive_failures = 0 + self.max_failures = 5 + + def _run_ipmi(self, args: List[str], timeout: int = 15) -> Tuple[bool, str]: + """Run IPMI command with error handling.""" + cmd = [ + "ipmitool", "-I", "lanplus", + "-H", self.host, + "-U", self.username, + "-P", self.password, + "-p", str(self.port) + ] + args + + try: + result = subprocess.run( + cmd, + capture_output=True, + text=True, + timeout=timeout + ) + if result.returncode == 0: + self.consecutive_failures = 0 + return True, result.stdout + else: + self.consecutive_failures += 1 + logger.warning(f"IPMI command failed: {result.stderr}") + return False, result.stderr + except subprocess.TimeoutExpired: + self.consecutive_failures += 1 + logger.error(f"IPMI command timed out after {timeout}s") + return False, "Timeout" + except Exception as e: + self.consecutive_failures += 1 + logger.error(f"IPMI command error: {e}") + return False, str(e) + + def test_connection(self) -> bool: + """Test if we can connect to the server.""" + success, _ = self._run_ipmi(["mc", "info"], timeout=10) + return success + + def enable_manual_fan_control(self) -> bool: + """Enable manual fan control mode.""" + # Dell: raw 0x30 0x30 0x01 0x00 + success, _ = self._run_ipmi(["raw", "0x30", "0x30", "0x01", "0x00"]) + if success: + self.manual_mode = True + logger.info("Manual fan control enabled") + return success + + def disable_manual_fan_control(self) -> bool: + """Return to automatic fan control.""" + # Dell: raw 0x30 0x30 0x01 0x01 + success, _ = self._run_ipmi(["raw", "0x30", "0x30", "0x01", "0x01"]) + if success: + self.manual_mode = False + logger.info("Automatic fan control restored") + return success + + def set_fan_speed(self, speed_percent: int, fan_id: str = "0xff") -> bool: + """Set fan speed (0-100%). fan_id 0xff = all fans.""" + if speed_percent < 0: + speed_percent = 0 + if speed_percent > 100: + speed_percent = 100 + + hex_speed = f"0x{speed_percent:02x}" + success, _ = self._run_ipmi([ + "raw", "0x30", "0x30", "0x02", fan_id, hex_speed + ]) + + if success: + logger.info(f"Fan speed set to {speed_percent}%") + return success + + def get_temperatures(self) -> List[TemperatureReading]: + """Get temperature readings from all sensors.""" + success, output = self._run_ipmi(["sdr", "type", "temperature"]) + if not success: + return [] + + temps = [] + for line in output.splitlines(): + # Parse: Sensor Name | 01h | ok | 3.1 | 45 degrees C + parts = [p.strip() for p in line.split("|")] + if len(parts) >= 5: + name = parts[0] + status = parts[2] if len(parts) > 2 else "unknown" + reading = parts[4] + + match = re.search(r'(\d+(?:\.\d+)?)\s+degrees\s+C', reading, re.IGNORECASE) + if match: + value = float(match.group(1)) + location = self._classify_temp_location(name) + temps.append(TemperatureReading( + name=name, + location=location, + value=value, + status=status + )) + return temps + + def get_fan_speeds(self) -> List[FanReading]: + """Get current fan speeds.""" + success, output = self._run_ipmi(["sdr", "elist", "full"]) + if not success: + return [] + + fans = [] + for line in output.splitlines(): + if "fan" in line.lower() and "rpm" in line.lower(): + parts = [p.strip() for p in line.split("|")] + if len(parts) >= 5: + name = parts[0] + reading = parts[4] + + # Extract fan number + match = re.search(r'fan\s*(\d+)', name, re.IGNORECASE) + fan_number = int(match.group(1)) if match else 0 + fan_id = f"0x{fan_number-1:02x}" if fan_number > 0 else "0x00" + + # Extract RPM + rpm_match = re.search(r'(\d+)\s*RPM', reading, re.IGNORECASE) + rpm = int(rpm_match.group(1)) if rpm_match else None + + fans.append(FanReading( + fan_id=fan_id, + fan_number=fan_number, + speed_rpm=rpm, + speed_percent=None + )) + return fans + + def _classify_temp_location(self, name: str) -> str: + """Classify temperature sensor location.""" + name_lower = name.lower() + if "cpu" in name_lower or "proc" in name_lower: + if "1" in name or "one" in name_lower: + return "cpu1" + elif "2" in name or "two" in name_lower: + return "cpu2" + return "cpu" + elif "inlet" in name_lower or "ambient" in name_lower: + return "inlet" + elif "exhaust" in name_lower: + return "exhaust" + elif "memory" in name_lower or "dimm" in name_lower: + return "memory" + return "other" + + def calculate_fan_speed(self, temps: List[TemperatureReading], + curve: Optional[List[FanCurvePoint]] = None) -> int: + """Calculate target fan speed based on temperatures.""" + if not temps: + return 50 # Default if no temps + + if curve is None: + curve = self.DEFAULT_CURVE + + # Find max CPU temperature + cpu_temps = [t for t in temps if t.location.startswith("cpu")] + if cpu_temps: + max_temp = max(t.value for t in cpu_temps) + else: + max_temp = max(t.value for t in temps) + + # Apply fan curve with linear interpolation + sorted_curve = sorted(curve, key=lambda p: p.temp) + + if max_temp <= sorted_curve[0].temp: + return sorted_curve[0].speed + if max_temp >= sorted_curve[-1].temp: + return sorted_curve[-1].speed + + for i in range(len(sorted_curve) - 1): + p1, p2 = sorted_curve[i], sorted_curve[i + 1] + if p1.temp <= max_temp <= p2.temp: + if p2.temp == p1.temp: + return p1.speed + ratio = (max_temp - p1.temp) / (p2.temp - p1.temp) + speed = p1.speed + ratio * (p2.speed - p1.speed) + return int(round(speed)) + + return sorted_curve[-1].speed + + def is_healthy(self) -> bool: + """Check if controller is working properly.""" + return self.consecutive_failures < self.max_failures + + +class FanControlService: + """Background service for automatic fan control.""" + + def __init__(self, config_path: str = "/etc/ipmi-fan-controller/config.json"): + self.config_path = config_path + self.controller: Optional[IPMIFanController] = None + self.running = False + self.thread: Optional[threading.Thread] = None + self.current_speed = 0 + self.target_speed = 0 + self.last_temps: List[TemperatureReading] = [] + self.last_fans: List[FanReading] = [] + self.lock = threading.Lock() + + # Default config + self.config = { + "host": "", + "username": "", + "password": "", + "port": 623, + "enabled": False, + "interval": 10, # seconds + "min_speed": 10, + "max_speed": 100, + "fan_curve": [ + {"temp": 30, "speed": 15}, + {"temp": 40, "speed": 25}, + {"temp": 50, "speed": 40}, + {"temp": 60, "speed": 60}, + {"temp": 70, "speed": 80}, + {"temp": 80, "speed": 100}, + ], + "panic_temp": 85, + "panic_speed": 100 + } + + self._load_config() + + def _load_config(self): + """Load configuration from file.""" + try: + if Path(self.config_path).exists(): + with open(self.config_path, 'r') as f: + loaded = json.load(f) + self.config.update(loaded) + logger.info(f"Loaded config from {self.config_path}") + except Exception as e: + logger.error(f"Failed to load config: {e}") + + def _save_config(self): + """Save configuration to file.""" + try: + Path(self.config_path).parent.mkdir(parents=True, exist_ok=True) + with open(self.config_path, 'w') as f: + json.dump(self.config, f, indent=2) + logger.info(f"Saved config to {self.config_path}") + except Exception as e: + logger.error(f"Failed to save config: {e}") + + def update_config(self, **kwargs): + """Update configuration values.""" + self.config.update(kwargs) + self._save_config() + + # Reinitialize controller if connection params changed + if any(k in kwargs for k in ['host', 'username', 'password', 'port']): + self._init_controller() + + def _init_controller(self): + """Initialize the IPMI controller.""" + if not all([self.config.get('host'), self.config.get('username'), self.config.get('password')]): + logger.warning("Missing IPMI credentials") + return False + + self.controller = IPMIFanController( + host=self.config['host'], + username=self.config['username'], + password=self.config['password'], + port=self.config.get('port', 623) + ) + + if self.controller.test_connection(): + logger.info(f"Connected to IPMI at {self.config['host']}") + return True + else: + logger.error(f"Failed to connect to IPMI at {self.config['host']}") + self.controller = None + return False + + def start(self): + """Start the fan control service.""" + if self.running: + return + + if not self._init_controller(): + logger.error("Cannot start service - IPMI connection failed") + return False + + self.running = True + self.thread = threading.Thread(target=self._control_loop, daemon=True) + self.thread.start() + logger.info("Fan control service started") + return True + + def stop(self): + """Stop the fan control service.""" + self.running = False + if self.thread: + self.thread.join(timeout=5) + + # Return to automatic control + if self.controller: + self.controller.disable_manual_fan_control() + + logger.info("Fan control service stopped") + + def _control_loop(self): + """Main control loop running in background thread.""" + # Enable manual control on startup + if self.controller: + self.controller.enable_manual_fan_control() + + while self.running: + try: + if not self.config.get('enabled', False): + time.sleep(1) + continue + + if not self.controller or not self.controller.is_healthy(): + logger.warning("Controller unhealthy, attempting reconnect...") + if not self._init_controller(): + time.sleep(30) + continue + self.controller.enable_manual_fan_control() + + # Get sensor data + temps = self.controller.get_temperatures() + fans = self.controller.get_fan_speeds() + + with self.lock: + self.last_temps = temps + self.last_fans = fans + + if not temps: + logger.warning("No temperature readings received") + time.sleep(self.config.get('interval', 10)) + continue + + # Check for panic temperature + max_temp = max((t.value for t in temps if t.location.startswith('cpu')), default=0) + if max_temp >= self.config.get('panic_temp', 85): + self.target_speed = self.config.get('panic_speed', 100) + logger.warning(f"PANIC MODE: CPU temp {max_temp}°C, setting fans to {self.target_speed}%") + else: + # Calculate target speed from curve + curve = [FanCurvePoint(p['temp'], p['speed']) for p in self.config.get('fan_curve', [])] + self.target_speed = self.controller.calculate_fan_speed(temps, curve) + + # Apply limits + self.target_speed = max(self.config.get('min_speed', 10), + min(self.config.get('max_speed', 100), self.target_speed)) + + # Apply fan speed if changed significantly (>= 5%) + if abs(self.target_speed - self.current_speed) >= 5: + if self.controller.set_fan_speed(self.target_speed): + self.current_speed = self.target_speed + logger.info(f"Fan speed adjusted to {self.target_speed}% (CPU temp: {max_temp:.1f}°C)") + + time.sleep(self.config.get('interval', 10)) + + except Exception as e: + logger.error(f"Control loop error: {e}") + time.sleep(10) + + def get_status(self) -> Dict: + """Get current status.""" + with self.lock: + return { + "running": self.running, + "enabled": self.config.get('enabled', False), + "connected": self.controller is not None and self.controller.is_healthy(), + "manual_mode": self.controller.manual_mode if self.controller else False, + "current_speed": self.current_speed, + "target_speed": self.target_speed, + "temperatures": [asdict(t) for t in self.last_temps], + "fans": [asdict(f) for f in self.last_fans], + "config": { + k: v for k, v in self.config.items() + if k != 'password' # Don't expose password + } + } + + def set_manual_speed(self, speed: int) -> bool: + """Set manual fan speed.""" + if not self.controller: + return False + + self.config['enabled'] = False + speed = max(0, min(100, speed)) + + if self.controller.set_fan_speed(speed): + self.current_speed = speed + return True + return False + + def set_auto_mode(self, enabled: bool): + """Enable or disable automatic control.""" + self.config['enabled'] = enabled + self._save_config() + + if enabled and self.controller: + self.controller.enable_manual_fan_control() + elif not enabled and self.controller: + self.controller.disable_manual_fan_control() + + +# Global service instance +_service: Optional[FanControlService] = None + + +def get_service(config_path: str = "/etc/ipmi-fan-controller/config.json") -> FanControlService: + """Get or create the global service instance.""" + global _service + if _service is None: + _service = FanControlService(config_path) + return _service + + +if __name__ == "__main__": + # Simple CLI test + import sys + + if len(sys.argv) < 4: + print("Usage: python fan_controller.py [port]") + sys.exit(1) + + host = sys.argv[1] + username = sys.argv[2] + password = sys.argv[3] + port = int(sys.argv[4]) if len(sys.argv) > 4 else 623 + + controller = IPMIFanController(host, username, password, port) + + print(f"Testing connection to {host}...") + if controller.test_connection(): + print("✓ Connected successfully") + + print("\nTemperatures:") + for temp in controller.get_temperatures(): + print(f" {temp.name}: {temp.value}°C ({temp.location})") + + print("\nFan speeds:") + for fan in controller.get_fan_speeds(): + print(f" Fan {fan.fan_number}: {fan.speed_rpm} RPM") + + print("\nEnabling manual control...") + if controller.enable_manual_fan_control(): + print("✓ Manual control enabled") + + print("\nSetting fans to 30%...") + if controller.set_fan_speed(30): + print("✓ Speed set to 30%") + time.sleep(3) + + print("\nSetting fans to 50%...") + if controller.set_fan_speed(50): + print("✓ Speed set to 50%") + time.sleep(3) + + print("\nReturning to automatic control...") + controller.disable_manual_fan_control() + print("✓ Done") + else: + print("✗ Connection failed") + sys.exit(1) diff --git a/install.sh b/install.sh new file mode 100755 index 0000000..fb49320 --- /dev/null +++ b/install.sh @@ -0,0 +1,147 @@ +#!/bin/bash +# Setup script for IPMI Fan Controller v2 + +set -e + +echo "🌬️ IPMI Fan Controller v2 - Setup" +echo "==================================" + +# Check if running as root for system-wide install +if [ "$EUID" -eq 0 ]; then + INSTALL_SYSTEM=true + INSTALL_DIR="/opt/ipmi-fan-controller" + CONFIG_DIR="/etc/ipmi-fan-controller" +else + INSTALL_SYSTEM=false + INSTALL_DIR="$HOME/.local/ipmi-fan-controller" + CONFIG_DIR="$HOME/.config/ipmi-fan-controller" + echo "⚠️ Running as user - installing to $INSTALL_DIR" + echo " (Run with sudo for system-wide install)" + echo "" +fi + +# Check dependencies +echo "📦 Checking dependencies..." + +if ! command -v python3 &> /dev/null; then + echo "❌ Python 3 is required but not installed" + exit 1 +fi + +if ! command -v ipmitool &> /dev/null; then + echo "⚠️ ipmitool not found. Installing..." + if [ "$INSTALL_SYSTEM" = true ]; then + apt-get update && apt-get install -y ipmitool + else + echo "❌ Please install ipmitool: sudo apt-get install ipmitool" + exit 1 + fi +fi + +echo "✓ Python 3: $(python3 --version)" +echo "✓ ipmitool: $(ipmitool -V)" + +# Create directories +echo "" +echo "📁 Creating directories..." +mkdir -p "$INSTALL_DIR" +mkdir -p "$CONFIG_DIR" +mkdir -p "$INSTALL_DIR/logs" + +# Copy files +echo "" +echo "📋 Installing files..." +SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" +cp "$SCRIPT_DIR/fan_controller.py" "$INSTALL_DIR/" +cp "$SCRIPT_DIR/web_server.py" "$INSTALL_DIR/" +cp "$SCRIPT_DIR/requirements.txt" "$INSTALL_DIR/" + +# Install Python dependencies +echo "" +echo "🐍 Installing Python dependencies..." +python3 -m pip install -q -r "$INSTALL_DIR/requirements.txt" + +# Create default config if not exists +if [ ! -f "$CONFIG_DIR/config.json" ]; then +echo "" +echo "⚙️ Creating default configuration..." +cat > "$CONFIG_DIR/config.json" << 'EOF' +{ + "host": "", + "username": "root", + "password": "", + "port": 623, + "enabled": false, + "interval": 10, + "min_speed": 10, + "max_speed": 100, + "fan_curve": [ + {"temp": 30, "speed": 15}, + {"temp": 40, "speed": 25}, + {"temp": 50, "speed": 40}, + {"temp": 60, "speed": 60}, + {"temp": 70, "speed": 80}, + {"temp": 80, "speed": 100} + ], + "panic_temp": 85, + "panic_speed": 100 +} +EOF +fi + +# Create systemd service (system-wide only) +if [ "$INSTALL_SYSTEM" = true ]; then +echo "" +echo "🔧 Creating systemd service..." +cat > /etc/systemd/system/ipmi-fan-controller.service << EOF +[Unit] +Description=IPMI Fan Controller v2 +After=network.target + +[Service] +Type=simple +User=root +WorkingDirectory=$INSTALL_DIR +Environment="CONFIG_PATH=$CONFIG_DIR/config.json" +ExecStart=/usr/bin/python3 $INSTALL_DIR/web_server.py +ExecStop=/usr/bin/python3 -c "import requests; requests.post('http://localhost:8000/api/shutdown')" +Restart=on-failure +RestartSec=10 + +[Install] +WantedBy=multi-user.target +EOF + +systemctl daemon-reload +systemctl enable ipmi-fan-controller.service + +echo "" +echo "✅ Installation complete!" +echo "" +echo "Next steps:" +echo " 1. Edit config: sudo nano $CONFIG_DIR/config.json" +echo " 2. Start service: sudo systemctl start ipmi-fan-controller" +echo " 3. View status: sudo systemctl status ipmi-fan-controller" +echo " 4. Open web UI: http://$(hostname -I | awk '{print $1}'):8000" +echo "" +echo "Or test from CLI:" +echo " python3 $INSTALL_DIR/fan_controller.py " + +else + +echo "" +echo "✅ User installation complete!" +echo "" +echo "Next steps:" +echo " 1. Edit config: nano $CONFIG_DIR/config.json" +echo " 2. Start manually:" +echo " CONFIG_PATH=$CONFIG_DIR/config.json python3 $INSTALL_DIR/web_server.py" +echo " 3. Open web UI: http://localhost:8000" +echo "" +echo "Or test from CLI:" +echo " python3 $INSTALL_DIR/fan_controller.py " + +fi + +echo "" +echo "📖 Configuration file: $CONFIG_DIR/config.json" diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..b4409b3 --- /dev/null +++ b/requirements.txt @@ -0,0 +1,4 @@ +fastapi==0.109.0 +uvicorn[standard]==0.27.0 +pydantic==2.5.3 +pydantic-settings==2.1.0 diff --git a/web_server.py b/web_server.py new file mode 100644 index 0000000..27ff4c4 --- /dev/null +++ b/web_server.py @@ -0,0 +1,695 @@ +""" +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)