Upload gpu_monitoring.py with huggingface_hub

gpu_monitoring.py

@@ -0,0 +1,600 @@
+#!/usr/bin/env python3
+"""
+GPU Detection and Monitoring Module
+
+Handles GPU detection, hardware monitoring, and data collection for AMD GPUs.
+Supports multiple GPU models and provides a unified interface for monitoring.
+"""
+
+import os
+import glob
+import time
+import logging
+import json
+import sqlite3
+from typing import Dict, List, Optional, Tuple, Any
+from dataclasses import dataclass, asdict
+from pathlib import Path
+
+logger = logging.getLogger(__name__)
+
+
+@dataclass
+class GPUInfo:
+    """GPU information data structure."""
+    name: str
+    vendor: str
+    device_id: str
+    bus_id: str
+    hwmon_path: str
+    temp_sensor: str
+    fan_control: str
+    power_sensor: str
+    memory_total: int = 0
+    memory_used: int = 0
+
+
+@dataclass
+class GPUStatus:
+    """Current GPU status data structure."""
+    timestamp: float
+    temperature: float
+    load: float
+    fan_speed: int
+    fan_pwm: int
+    power_draw: float
+    memory_used: int
+    memory_total: int
+    core_clock: int
+    memory_clock: int
+    voltage: float = 0.0
+    efficiency: float = 0.0
+
+
+class GPUHardwareDetector:
+    """Detects and identifies GPU hardware."""
+    
+    def __init__(self):
+        self.gpus = []
+        self.detected_gpus = []
+        
+    def detect_amd_gpus(self) -> List[GPUInfo]:
+        """Detect AMD GPUs in the system."""
+        logger.info("Detecting AMD GPUs...")
+        
+        gpus = []
+        
+        # Look for AMD GPUs in sysfs
+        for card_path in glob.glob("/sys/class/drm/card*"):
+            try:
+                # Check if this is an AMD GPU
+                device_path = os.path.join(card_path, "device")
+                vendor_file = os.path.join(device_path, "vendor")
+                
+                if not os.path.exists(vendor_file):
+                    continue
+                    
+                with open(vendor_file, 'r') as f:
+                    vendor_id = f.read().strip()
+                
+                if vendor_id != "0x1002":  # AMD vendor ID
+                    continue
+                
+                # Get device information
+                device_id_file = os.path.join(device_path, "device")
+                with open(device_id_file, 'r') as f:
+                    device_id = f.read().strip()
+                
+                # Get GPU name
+                gpu_name = self._get_gpu_name(device_id)
+                
+                # Find hwmon path
+                hwmon_path = self._find_hwmon_path(device_path)
+                
+                if hwmon_path:
+                    gpu_info = GPUInfo(
+                        name=gpu_name,
+                        vendor="AMD",
+                        device_id=device_id,
+                        bus_id=os.path.basename(card_path),
+                        hwmon_path=hwmon_path,
+                        temp_sensor=os.path.join(hwmon_path, "temp1_input"),
+                        fan_control=os.path.join(hwmon_path, "pwm1"),
+                        power_sensor=os.path.join(hwmon_path, "power1_input")
+                    )
+                    gpus.append(gpu_info)
+                    logger.info(f"Found AMD GPU: {gpu_name} ({device_id})")
+                    
+            except Exception as e:
+                logger.warning(f"Error detecting GPU at {card_path}: {e}")
+                continue
+        
+        self.detected_gpus = gpus
+        return gpus
+    
+    def _get_gpu_name(self, device_id: str) -> str:
+        """Get human-readable GPU name from device ID."""
+        gpu_names = {
+            "0x73bf": "Radeon Pro VII",
+            "0x73ff": "Radeon Pro VII",
+            "0x7310": "Radeon Pro VII",
+            "0x7340": "Radeon Pro VII",
+            "0x73a0": "Radeon Pro VII",
+            "0x73b0": "Radeon Pro VII",
+            "0x73c0": "Radeon Pro VII",
+            "0x73d0": "Radeon Pro VII",
+            "0x73e0": "Radeon Pro VII",
+            "0x73f0": "Radeon Pro VII",
+            # Add more GPU mappings as needed
+        }
+        
+        return gpu_names.get(device_id, f"AMD GPU {device_id}")
+    
+    def _find_hwmon_path(self, device_path: str) -> Optional[str]:
+        """Find the hwmon path for a GPU device."""
+        hwmon_base = os.path.join(device_path, "hwmon")
+        
+        if not os.path.exists(hwmon_base):
+            return None
+        
+        try:
+            hwmons = os.listdir(hwmon_base)
+            if hwmons:
+                return os.path.join(hwmon_base, hwmons[0])
+        except Exception as e:
+            logger.warning(f"Error finding hwmon for {device_path}: {e}")
+        
+        return None
+
+
+class GPUDataCollector:
+    """Collects GPU monitoring data."""
+    
+    def __init__(self, gpu_info: GPUInfo):
+        self.gpu_info = gpu_info
+        self.last_status = None
+        
+    def read_temperature(self) -> Optional[float]:
+        """Read GPU temperature in Celsius."""
+        try:
+            if os.path.exists(self.gpu_info.temp_sensor):
+                with open(self.gpu_info.temp_sensor, 'r') as f:
+                    temp_millic = int(f.read().strip())
+                    return temp_millic / 1000.0
+        except Exception as e:
+            logger.debug(f"Error reading temperature: {e}")
+        return None
+    
+    def read_fan_speed(self) -> Optional[int]:
+        """Read fan speed in RPM."""
+        fan_speed_file = self.gpu_info.fan_control.replace("pwm1", "fan1_input")
+        try:
+            if os.path.exists(fan_speed_file):
+                with open(fan_speed_file, 'r') as f:
+                    return int(f.read().strip())
+        except Exception as e:
+            logger.debug(f"Error reading fan speed: {e}")
+        return None
+    
+    def read_fan_pwm(self) -> Optional[int]:
+        """Read fan PWM value (0-255)."""
+        try:
+            if os.path.exists(self.gpu_info.fan_control):
+                with open(self.gpu_info.fan_control, 'r') as f:
+                    return int(f.read().strip())
+        except Exception as e:
+            logger.debug(f"Error reading fan PWM: {e}")
+        return None
+    
+    def read_power_draw(self) -> Optional[float]:
+        """Read power draw in watts."""
+        try:
+            if os.path.exists(self.gpu_info.power_sensor):
+                with open(self.gpu_info.power_sensor, 'r') as f:
+                    power_microw = int(f.read().strip())
+                    return power_microw / 1000000.0
+        except Exception as e:
+            logger.debug(f"Error reading power: {e}")
+        return None
+    
+    def read_memory_info(self) -> Tuple[Optional[int], Optional[int]]:
+        """Read VRAM usage in MB."""
+        try:
+            device_path = os.path.dirname(os.path.dirname(self.gpu_info.hwmon_path))
+            vram_used_file = os.path.join(device_path, "mem_info_vram_used")
+            vram_total_file = os.path.join(device_path, "mem_info_vram_total")
+            
+            if os.path.exists(vram_used_file) and os.path.exists(vram_total_file):
+                with open(vram_used_file, 'r') as f:
+                    used = int(f.read().strip()) // (1024 * 1024)
+                with open(vram_total_file, 'r') as f:
+                    total = int(f.read().strip()) // (1024 * 1024)
+                return used, total
+        except Exception as e:
+            logger.debug(f"Error reading memory info: {e}")
+        return None, None
+    
+    def read_gpu_load(self) -> Optional[float]:
+        """Read GPU load percentage."""
+        try:
+            device_path = os.path.dirname(os.path.dirname(self.gpu_info.hwmon_path))
+            load_file = os.path.join(device_path, "gpu_busy_percent")
+            
+            if os.path.exists(load_file):
+                with open(load_file, 'r') as f:
+                    return float(f.read().strip())
+        except Exception as e:
+            logger.debug(f"Error reading GPU load: {e}")
+        return None
+    
+    def read_clocks(self) -> Tuple[Optional[int], Optional[int]]:
+        """Read core and memory clocks in MHz."""
+        try:
+            device_path = os.path.dirname(os.path.dirname(self.gpu_info.hwmon_path))
+            sclk_file = os.path.join(device_path, "pp_dpm_sclk")
+            mclk_file = os.path.join(device_path, "pp_dpm_mclk")
+            
+            core_clock = self._parse_clock_file(sclk_file)
+            memory_clock = self._parse_clock_file(mclk_file)
+            
+            return core_clock, memory_clock
+        except Exception as e:
+            logger.debug(f"Error reading clocks: {e}")
+        return None, None
+    
+    def _parse_clock_file(self, clock_file: str) -> Optional[int]:
+        """Parse clock file to extract current clock speed."""
+        try:
+            if os.path.exists(clock_file):
+                with open(clock_file, 'r') as f:
+                    for line in f:
+                        if '*' in line:  # Active clock
+                            parts = line.strip().split(':')
+                            if len(parts) >= 2:
+                                clock_str = parts[1].strip().split()[0]
+                                return int(clock_str.replace('Mhz', ''))
+        except Exception as e:
+            logger.debug(f"Error parsing clock file {clock_file}: {e}")
+        return None
+    
+    def collect_status(self) -> Optional[GPUStatus]:
+        """Collect all GPU status information."""
+        try:
+            # Read all sensors
+            temp = self.read_temperature()
+            fan_speed = self.read_fan_speed()
+            fan_pwm = self.read_fan_pwm()
+            power = self.read_power_draw()
+            mem_used, mem_total = self.read_memory_info()
+            load = self.read_gpu_load()
+            core_clock, mem_clock = self.read_clocks()
+            
+            # Calculate efficiency if we have power and load data
+            efficiency = 0.0
+            if power and load and power > 0:
+                efficiency = load / power
+            
+            status = GPUStatus(
+                timestamp=time.time(),
+                temperature=temp or 0.0,
+                load=load or 0.0,
+                fan_speed=fan_speed or 0,
+                fan_pwm=fan_pwm or 0,
+                power_draw=power or 0.0,
+                memory_used=mem_used or 0,
+                memory_total=mem_total or 0,
+                core_clock=core_clock or 0,
+                memory_clock=mem_clock or 0,
+                voltage=0.0,  # Not implemented yet
+                efficiency=efficiency
+            )
+            
+            self.last_status = status
+            return status
+            
+        except Exception as e:
+            logger.error(f"Error collecting GPU status: {e}")
+            return None
+
+
+class GPUDataManager:
+    """Manages GPU data storage and retrieval."""
+    
+    def __init__(self, db_path: str = "gpu_monitoring.db"):
+        self.db_path = db_path
+        self.init_database()
+    
+    def init_database(self):
+        """Initialize the SQLite database."""
+        try:
+            with sqlite3.connect(self.db_path) as conn:
+                cursor = conn.cursor()
+                
+                # Create tables
+                cursor.execute('''
+                    CREATE TABLE IF NOT EXISTS gpu_status (
+                        id INTEGER PRIMARY KEY AUTOINCREMENT,
+                        timestamp REAL,
+                        gpu_name TEXT,
+                        temperature REAL,
+                        load REAL,
+                        fan_speed INTEGER,
+                        fan_pwm INTEGER,
+                        power_draw REAL,
+                        memory_used INTEGER,
+                        memory_total INTEGER,
+                        core_clock INTEGER,
+                        memory_clock INTEGER,
+                        voltage REAL,
+                        efficiency REAL
+                    )
+                ''')
+                
+                cursor.execute('''
+                    CREATE TABLE IF NOT EXISTS gpu_info (
+                        id INTEGER PRIMARY KEY AUTOINCREMENT,
+                        gpu_name TEXT UNIQUE,
+                        vendor TEXT,
+                        device_id TEXT,
+                        bus_id TEXT,
+                        hwmon_path TEXT,
+                        detected_at REAL
+                    )
+                ''')
+                
+                cursor.execute('''
+                    CREATE TABLE IF NOT EXISTS alerts (
+                        id INTEGER PRIMARY KEY AUTOINCREMENT,
+                        timestamp REAL,
+                        gpu_name TEXT,
+                        alert_type TEXT,
+                        message TEXT,
+                        value REAL,
+                        threshold REAL
+                    )
+                ''')
+                
+                conn.commit()
+                logger.info("Database initialized successfully")
+                
+        except Exception as e:
+            logger.error(f"Error initializing database: {e}")
+    
+    def save_gpu_info(self, gpu_info: GPUInfo):
+        """Save GPU information to database."""
+        try:
+            with sqlite3.connect(self.db_path) as conn:
+                cursor = conn.cursor()
+                
+                cursor.execute('''
+                    INSERT OR REPLACE INTO gpu_info 
+                    (gpu_name, vendor, device_id, bus_id, hwmon_path, detected_at)
+                    VALUES (?, ?, ?, ?, ?, ?)
+                ''', (gpu_info.name, gpu_info.vendor, gpu_info.device_id, 
+                      gpu_info.bus_id, gpu_info.hwmon_path, time.time()))
+                
+                conn.commit()
+                logger.debug(f"Saved GPU info: {gpu_info.name}")
+                
+        except Exception as e:
+            logger.error(f"Error saving GPU info: {e}")
+    
+    def save_status(self, gpu_name: str, status: GPUStatus):
+        """Save GPU status to database."""
+        try:
+            with sqlite3.connect(self.db_path) as conn:
+                cursor = conn.cursor()
+                
+                cursor.execute('''
+                    INSERT INTO gpu_status 
+                    (timestamp, gpu_name, temperature, load, fan_speed, fan_pwm,
+                     power_draw, memory_used, memory_total, core_clock, memory_clock, 
+                     voltage, efficiency)
+                    VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+                ''', (status.timestamp, gpu_name, status.temperature, status.load,
+                      status.fan_speed, status.fan_pwm, status.power_draw,
+                      status.memory_used, status.memory_total, status.core_clock,
+                      status.memory_clock, status.voltage, status.efficiency))
+                
+                conn.commit()
+                
+        except Exception as e:
+            logger.error(f"Error saving status: {e}")
+    
+    def get_recent_status(self, gpu_name: str, limit: int = 100) -> List[Dict[str, Any]]:
+        """Get recent status entries for a GPU."""
+        try:
+            with sqlite3.connect(self.db_path) as conn:
+                cursor = conn.cursor()
+                
+                cursor.execute('''
+                    SELECT timestamp, temperature, load, fan_speed, fan_pwm,
+                           power_draw, memory_used, memory_total, core_clock, 
+                           memory_clock, voltage, efficiency
+                    FROM gpu_status 
+                    WHERE gpu_name = ? 
+                    ORDER BY timestamp DESC 
+                    LIMIT ?
+                ''', (gpu_name, limit))
+                
+                rows = cursor.fetchall()
+                
+                # Convert to list of dictionaries
+                columns = ['timestamp', 'temperature', 'load', 'fan_speed', 'fan_pwm',
+                          'power_draw', 'memory_used', 'memory_total', 'core_clock',
+                          'memory_clock', 'voltage', 'efficiency']
+                
+                return [dict(zip(columns, row)) for row in rows]
+                
+        except Exception as e:
+            logger.error(f"Error getting recent status: {e}")
+            return []
+    
+    def get_gpu_info(self, gpu_name: Optional[str] = None) -> List[Dict[str, Any]]:
+        """Get GPU information from database."""
+        try:
+            with sqlite3.connect(self.db_path) as conn:
+                cursor = conn.cursor()
+                
+                if gpu_name:
+                    cursor.execute('SELECT * FROM gpu_info WHERE gpu_name = ?', (gpu_name,))
+                else:
+                    cursor.execute('SELECT * FROM gpu_info')
+                
+                rows = cursor.fetchall()
+                
+                if not rows:
+                    return []
+                
+                columns = [desc[0] for desc in cursor.description]
+                return [dict(zip(columns, row)) for row in rows]
+                
+        except Exception as e:
+            logger.error(f"Error getting GPU info: {e}")
+            return []
+    
+    def cleanup_old_data(self, days: int = 7):
+        """Remove data older than specified days."""
+        try:
+            cutoff_time = time.time() - (days * 24 * 3600)
+            
+            with sqlite3.connect(self.db_path) as conn:
+                cursor = conn.cursor()
+                
+                cursor.execute('DELETE FROM gpu_status WHERE timestamp < ?', (cutoff_time,))
+                cursor.execute('DELETE FROM alerts WHERE timestamp < ?', (cutoff_time,))
+                
+                conn.commit()
+                logger.info(f"Cleaned up data older than {days} days")
+                
+        except Exception as e:
+            logger.error(f"Error cleaning up old data: {e}")
+
+
+class GPUManager:
+    """Main GPU management class."""
+    
+    def __init__(self):
+        self.detector = GPUHardwareDetector()
+        self.data_manager = GPUDataManager()
+        self.collectors = {}
+        
+    def initialize(self):
+        """Initialize GPU detection and data collection."""
+        logger.info("Initializing GPU manager...")
+        
+        # Detect GPUs
+        gpus = self.detector.detect_amd_gpus()
+        
+        if not gpus:
+            logger.warning("No AMD GPUs detected")
+            return False
+        
+        # Initialize collectors for each GPU
+        for gpu in gpus:
+            self.collectors[gpu.name] = GPUDataCollector(gpu)
+            self.data_manager.save_gpu_info(gpu)
+            logger.info(f"Initialized collector for {gpu.name}")
+        
+        logger.info(f"Successfully initialized {len(gpus)} GPU(s)")
+        return True
+    
+    def get_status(self, gpu_name: Optional[str] = None) -> Dict[str, Optional[GPUStatus]]:
+        """Get current status for specified GPU or all GPUs."""
+        results = {}
+        
+        if gpu_name:
+            if gpu_name in self.collectors:
+                status = self.collectors[gpu_name].collect_status()
+                results[gpu_name] = status
+            else:
+                results[gpu_name] = None
+        else:
+            for name, collector in self.collectors.items():
+                status = collector.collect_status()
+                results[name] = status
+        
+        # Save to database
+        for name, status in results.items():
+            if status:
+                self.data_manager.save_status(name, status)
+        
+        return results
+    
+    def get_gpu_list(self) -> List[str]:
+        """Get list of detected GPU names."""
+        return list(self.collectors.keys())
+    
+    def get_gpu_info(self, gpu_name: Optional[str] = None) -> List[Dict[str, Any]]:
+        """Get GPU information."""
+        return self.data_manager.get_gpu_info(gpu_name)
+    
+    def get_historical_data(self, gpu_name: str, hours: int = 24) -> List[Dict[str, Any]]:
+        """Get historical data for a GPU."""
+        cutoff_time = time.time() - (hours * 3600)
+        
+        try:
+            with sqlite3.connect(self.data_manager.db_path) as conn:
+                cursor = conn.cursor()
+                
+                cursor.execute('''
+                    SELECT timestamp, temperature, load, fan_speed, fan_pwm,
+                           power_draw, memory_used, memory_total, core_clock, 
+                           memory_clock, voltage, efficiency
+                    FROM gpu_status 
+                    WHERE gpu_name = ? AND timestamp >= ?
+                    ORDER BY timestamp ASC
+                ''', (gpu_name, cutoff_time))
+                
+                rows = cursor.fetchall()
+                
+                if not rows:
+                    return []
+                
+                columns = ['timestamp', 'temperature', 'load', 'fan_speed', 'fan_pwm',
+                          'power_draw', 'memory_used', 'memory_total', 'core_clock',
+                          'memory_clock', 'voltage', 'efficiency']
+                
+                return [dict(zip(columns, row)) for row in rows]
+                
+        except Exception as e:
+            logger.error(f"Error getting historical data: {e}")
+            return []
+
+
+# Example usage and testing
+if __name__ == "__main__":
+    import sys
+    
+    # Setup logging
+    logging.basicConfig(
+        level=logging.INFO,
+        format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
+    )
+    
+    # Test GPU detection and monitoring
+    manager = GPUManager()
+    
+    if manager.initialize():
+        print("GPU detection successful!")
+        
+        # Get GPU list
+        gpus = manager.get_gpu_list()
+        print(f"Detected GPUs: {gpus}")
+        
+        # Get current status
+        status = manager.get_status()
+        for gpu_name, gpu_status in status.items():
+            if gpu_status:
+                print(f"\n{gpu_name} Status:")
+                print(f"  Temperature: {gpu_status.temperature}°C")
+                print(f"  Load: {gpu_status.load}%")
+                print(f"  Fan Speed: {gpu_status.fan_speed} RPM")
+                print(f"  Fan PWM: {gpu_status.fan_pwm}")
+                print(f"  Power: {gpu_status.power_draw}W")
+                print(f"  Memory: {gpu_status.memory_used}/{gpu_status.memory_total} MB")
+                print(f"  Core Clock: {gpu_status.core_clock} MHz")
+                print(f"  Memory Clock: {gpu_status.memory_clock} MHz")
+    else:
+        print("GPU detection failed!")
+        sys.exit(1)