#!/usr/bin/env python3 """ Web Interface for GPU Monitoring Provides a web-based dashboard for remote GPU monitoring with real-time charts, historical data, and fan control capabilities. """ from flask import Flask, render_template, jsonify, request, redirect, url_for from flask_cors import CORS import json import time import logging from datetime import datetime, timedelta from typing import Dict, List, Any, Optional from gpu_monitoring import GPUManager, GPUStatus from gpu_fan_controller import FanController, FanMode, ProfileType logger = logging.getLogger(__name__) app = Flask(__name__) CORS(app, origins="*") # Allow all origins for local development class WebGPUManager: """Web interface manager for GPU monitoring.""" def __init__(self): self.gpu_manager = GPUManager() self.fan_controller = FanController() self.config = self.load_config() # Initialize components self.gpu_manager.initialize() self.fan_controller.initialize() def load_config(self) -> Dict: """Load web interface configuration.""" try: with open('config/monitoring.json', 'r') as f: config = json.load(f) return config.get('web', {}) except: return { 'enabled': True, 'host': '0.0.0.0', 'port': 5000, 'debug': False } def get_current_status(self) -> Dict[str, Any]: """Get current GPU status.""" status_dict = self.gpu_manager.get_status() fan_status = self.fan_controller.get_status() result = { 'timestamp': time.time(), 'gpus': {}, 'fan_control': { 'mode': fan_status.mode.value if fan_status else 'unknown', 'profile': fan_status.profile if fan_status else 'unknown', 'current_pwm': fan_status.current_pwm if fan_status else 0, 'temperature': fan_status.temperature if fan_status else 0.0 } } for gpu_name, gpu_status in status_dict.items(): if gpu_status: result['gpus'][gpu_name] = { 'temperature': gpu_status.temperature, 'load': gpu_status.load, 'fan_speed': gpu_status.fan_speed, 'fan_pwm': gpu_status.fan_pwm, 'power_draw': gpu_status.power_draw, 'memory_used': gpu_status.memory_used, 'memory_total': gpu_status.memory_total, 'core_clock': gpu_status.core_clock, 'memory_clock': gpu_status.memory_clock, 'voltage': gpu_status.voltage, 'efficiency': gpu_status.efficiency } return result def get_historical_data(self, gpu_name: str, hours: int = 24) -> List[Dict[str, Any]]: """Get historical data for a GPU.""" return self.gpu_manager.get_historical_data(gpu_name, hours) def get_gpu_list(self) -> List[str]: """Get list of available GPUs.""" return self.gpu_manager.get_gpu_list() def get_fan_profiles(self) -> Dict[str, Any]: """Get available fan profiles.""" profiles = self.fan_controller.get_profiles() result = {} for name, profile in profiles.items(): result[name] = { 'name': profile.name, 'type': profile.profile_type.value, 'description': profile.description, 'curve': profile.curve, 'safety': profile.safety, 'enabled': profile.enabled } return result def set_fan_profile(self, profile_name: str) -> bool: """Set fan profile.""" return self.fan_controller.set_profile(profile_name) def set_fan_mode(self, mode: str) -> bool: """Set fan mode.""" try: fan_mode = FanMode(mode) self.fan_controller.set_mode(fan_mode) return True except: return False def set_manual_pwm(self, pwm: int) -> bool: """Set manual PWM.""" if 0 <= pwm <= 255: self.fan_controller.set_manual_pwm(pwm) return True return False # Initialize web manager web_manager = WebGPUManager() @app.route('/') def index(): """Main dashboard page.""" return render_template('index.html') @app.route('/api/status') def api_status(): """API endpoint for current status.""" try: status = web_manager.get_current_status() return jsonify(status) except Exception as e: logger.error(f"Error getting status: {e}") return jsonify({'error': str(e)}), 500 @app.route('/api/gpus') def api_gpus(): """API endpoint for GPU list.""" try: gpus = web_manager.get_gpu_list() return jsonify({'gpus': gpus}) except Exception as e: logger.error(f"Error getting GPU list: {e}") return jsonify({'error': str(e)}), 500 @app.route('/api/history/') def api_history(gpu_name): """API endpoint for historical data.""" try: hours = request.args.get('hours', 24, type=int) data = web_manager.get_historical_data(gpu_name, hours) return jsonify({'data': data}) except Exception as e: logger.error(f"Error getting historical data: {e}") return jsonify({'error': str(e)}), 500 @app.route('/api/fan/profiles') def api_fan_profiles(): """API endpoint for fan profiles.""" try: profiles = web_manager.get_fan_profiles() return jsonify({'profiles': profiles}) except Exception as e: logger.error(f"Error getting fan profiles: {e}") return jsonify({'error': str(e)}), 500 @app.route('/api/fan/profile', methods=['POST']) def api_set_fan_profile(): """API endpoint to set fan profile.""" try: data = request.get_json() profile_name = data.get('profile') if not profile_name: return jsonify({'error': 'Profile name required'}), 400 success = web_manager.set_fan_profile(profile_name) if success: return jsonify({'success': True, 'message': f'Set profile to {profile_name}'}) else: return jsonify({'error': f'Failed to set profile {profile_name}'}), 400 except Exception as e: logger.error(f"Error setting fan profile: {e}") return jsonify({'error': str(e)}), 500 @app.route('/api/fan/mode', methods=['POST']) def api_set_fan_mode(): """API endpoint to set fan mode.""" try: data = request.get_json() mode = data.get('mode') if not mode: return jsonify({'error': 'Mode required'}), 400 success = web_manager.set_fan_mode(mode) if success: return jsonify({'success': True, 'message': f'Set mode to {mode}'}) else: return jsonify({'error': f'Failed to set mode {mode}'}), 400 except Exception as e: logger.error(f"Error setting fan mode: {e}") return jsonify({'error': str(e)}), 500 @app.route('/api/fan/manual', methods=['POST']) def api_set_manual_pwm(): """API endpoint to set manual PWM.""" try: data = request.get_json() pwm = data.get('pwm') if pwm is None: return jsonify({'error': 'PWM value required'}), 400 success = web_manager.set_manual_pwm(pwm) if success: return jsonify({'success': True, 'message': f'Set manual PWM to {pwm}'}) else: return jsonify({'error': f'Invalid PWM value: {pwm}'}), 400 except Exception as e: logger.error(f"Error setting manual PWM: {e}") return jsonify({'error': str(e)}), 500 @app.route('/api/alerts') def api_alerts(): """API endpoint for alerts.""" try: # Get recent alerts from database # This would need to be implemented in the GPUDataManager alerts = [] # Placeholder return jsonify({'alerts': alerts}) except Exception as e: logger.error(f"Error getting alerts: {e}") return jsonify({'error': str(e)}), 500 @app.route('/api/system') def api_system(): """API endpoint for system information.""" try: import psutil system_info = { 'cpu_count': psutil.cpu_count(), 'cpu_percent': psutil.cpu_percent(interval=1), 'memory': { 'total': psutil.virtual_memory().total // (1024**3), # GB 'available': psutil.virtual_memory().available // (1024**3), # GB 'percent': psutil.virtual_memory().percent }, 'disk': { 'total': psutil.disk_usage('/').total // (1024**3), # GB 'free': psutil.disk_usage('/').free // (1024**3), # GB 'percent': (psutil.disk_usage('/').used / psutil.disk_usage('/').total) * 100 }, 'uptime': time.time() - psutil.boot_time() } return jsonify(system_info) except Exception as e: logger.error(f"Error getting system info: {e}") return jsonify({'error': str(e)}), 500 @app.errorhandler(404) def not_found(error): """Handle 404 errors.""" return jsonify({'error': 'Not found'}), 404 @app.errorhandler(500) def internal_error(error): """Handle 500 errors.""" return jsonify({'error': 'Internal server error'}), 500 def create_templates(): """Create HTML templates directory and files.""" templates_dir = Path('templates') static_dir = Path('static') templates_dir.mkdir(exist_ok=True) static_dir.mkdir(exist_ok=True) # Create main HTML template index_html = """ GPU Monitoring Dashboard

GPU Monitoring Dashboard

Connecting...

Real-time Status

Fan Control

Fan Mode: --
Current Profile: --
Current PWM: --%

Temperature History

""" with open(templates_dir / 'index.html', 'w') as f: f.write(index_html) if __name__ == '__main__': # Create templates if they don't exist create_templates() # Get configuration config = web_manager.config # Start Flask app app.run( host=config.get('host', '0.0.0.0'), port=config.get('port', 5000), debug=config.get('debug', False) )