Update AppImage

AppImage/scripts/flask_server.py

@@ -2702,16 +2702,16 @@ def get_temperature_info():
         # print(f"[v0] Error getting temperature info: {e}")
         pass
 
-        if power_meter is None:
+    if power_meter is None:
-            try:
+        try:
-                rapl_power = hardware_monitor.get_power_info()
+            rapl_power = hardware_monitor.get_power_info()
-                if rapl_power:
+            if rapl_power:
-                    power_meter = rapl_power
+                power_meter = rapl_power
-                    # print(f"[v0] Power meter from RAPL: {power_meter.get('watts', 0)}W")
+                # print(f"[v0] Power meter from RAPL: {power_meter.get('watts', 0)}W")
-                    pass
+                pass
-            except Exception as e:
+        except Exception as e:
-                # print(f"[v0] Error getting RAPL power info: {e}")
+            # print(f"[v0] Error getting RAPL power info: {e}")
-                pass   
+            pass   
     
     return {
         'temperatures': temperatures,

AppImage/scripts/hardware_monitor.py

@@ -1,359 +1,72 @@
 #!/usr/bin/env python3
-import json
+"""
-import subprocess
+Hardware Monitor - RAPL Power Monitoring
-import re
+
+This module provides CPU power consumption monitoring using Intel RAPL
+(Running Average Power Limit) interface when IPMI is not available.
+
+Only contains get_power_info() - all other hardware monitoring is handled
+by flask_server.py to avoid code duplication.
+"""
+
 import os
 import time
-from typing import Dict, List, Any, Optional
+from typing import Dict, Any, Optional
 
+# Global variable to store previous energy reading for power calculation
 _last_energy_reading = {'energy_uj': None, 'timestamp': None}
 
-def run_command(cmd: List[str]) -> str:
-    """Run a command and return its output."""
-    try:
-        result = subprocess.run(cmd, capture_output=True, text=True, timeout=5)
-        return result.stdout
-    except Exception:
-        return ""
-
-def get_nvidia_gpu_info() -> List[Dict[str, Any]]:
-    """Get detailed NVIDIA GPU information using nvidia-smi."""
-    gpus = []
-    
-    # Check if nvidia-smi is available
-    if not os.path.exists('/usr/bin/nvidia-smi'):
-        return gpus
-    
-    try:
-        # Query all GPU metrics at once
-        query_fields = [
-            'index',
-            'name',
-            'driver_version',
-            'memory.total',
-            'memory.used',
-            'memory.free',
-            'temperature.gpu',
-            'utilization.gpu',
-            'utilization.memory',
-            'power.draw',
-            'power.limit',
-            'clocks.current.graphics',
-            'clocks.current.memory',
-            'pcie.link.gen.current',
-            'pcie.link.width.current'
-        ]
-        
-        cmd = ['nvidia-smi', '--query-gpu=' + ','.join(query_fields), '--format=csv,noheader,nounits']
-        output = run_command(cmd)
-        
-        if not output:
-            return gpus
-        
-        for line in output.strip().split('\n'):
-            if not line:
-                continue
-                
-            values = [v.strip() for v in line.split(',')]
-            if len(values) < len(query_fields):
-                continue
-            
-            gpu_info = {
-                'index': values[0],
-                'name': values[1],
-                'driver_version': values[2],
-                'memory_total': f"{values[3]} MiB",
-                'memory_used': f"{values[4]} MiB",
-                'memory_free': f"{values[5]} MiB",
-                'temperature': values[6],
-                'utilization_gpu': values[7],
-                'utilization_memory': values[8],
-                'power_draw': f"{values[9]} W",
-                'power_limit': f"{values[10]} W",
-                'clock_graphics': f"{values[11]} MHz",
-                'clock_memory': f"{values[12]} MHz",
-                'pcie_gen': values[13],
-                'pcie_width': f"x{values[14]}"
-            }
-            
-            # Get CUDA version if available
-            cuda_output = run_command(['nvidia-smi', '--query-gpu=compute_cap', '--format=csv,noheader', '-i', values[0]])
-            if cuda_output:
-                gpu_info['compute_capability'] = cuda_output.strip()
-            
-            gpus.append(gpu_info)
-            
-    except Exception as e:
-        print(f"Error getting NVIDIA GPU info: {e}", file=sys.stderr)
-    
-    return gpus
-
-def get_amd_gpu_info() -> List[Dict[str, Any]]:
-    """Get AMD GPU information using rocm-smi."""
-    gpus = []
-    
-    # Check if rocm-smi is available
-    if not os.path.exists('/opt/rocm/bin/rocm-smi'):
-        return gpus
-    
-    try:
-        # Get basic GPU info
-        output = run_command(['/opt/rocm/bin/rocm-smi', '--showid', '--showtemp', '--showuse', '--showmeminfo', 'vram'])
-        
-        if not output:
-            return gpus
-        
-        # Parse rocm-smi output (format varies, this is a basic parser)
-        current_gpu = None
-        for line in output.split('\n'):
-            if 'GPU[' in line:
-                if current_gpu:
-                    gpus.append(current_gpu)
-                current_gpu = {'index': line.split('[')[1].split(']')[0]}
-            elif current_gpu:
-                if 'Temperature' in line:
-                    temp_match = re.search(r'(\d+\.?\d*)', line)
-                    if temp_match:
-                        current_gpu['temperature'] = temp_match.group(1)
-                elif 'GPU use' in line:
-                    use_match = re.search(r'(\d+)%', line)
-                    if use_match:
-                        current_gpu['utilization_gpu'] = use_match.group(1)
-                elif 'VRAM' in line:
-                    mem_match = re.search(r'(\d+)MB / (\d+)MB', line)
-                    if mem_match:
-                        current_gpu['memory_used'] = f"{mem_match.group(1)} MiB"
-                        current_gpu['memory_total'] = f"{mem_match.group(2)} MiB"
-        
-        if current_gpu:
-            gpus.append(current_gpu)
-            
-    except Exception as e:
-        print(f"Error getting AMD GPU info: {e}", file=sys.stderr)
-    
-    return gpus
-
-def get_temperatures() -> List[Dict[str, Any]]:
-    """Get temperature readings from sensors."""
-    temps = []
-    output = run_command(['sensors', '-A', '-u'])
-    
-    current_adapter = None
-    current_sensor = None
-    
-    for line in output.split('\n'):
-        line = line.strip()
-        if not line:
-            continue
-            
-        if line.endswith(':') and not line.startswith(' '):
-            current_adapter = line[:-1]
-        elif '_input:' in line and current_adapter:
-            parts = line.split(':')
-            if len(parts) == 2:
-                sensor_name = parts[0].replace('_input', '').replace('_', ' ').title()
-                try:
-                    temp_value = float(parts[1].strip())
-                    temps.append({
-                        'name': sensor_name,
-                        'current': round(temp_value, 1),
-                        'adapter': current_adapter
-                    })
-                except ValueError:
-                    pass
-    
-    return temps
-
-def get_fans() -> List[Dict[str, Any]]:
-    """Get fan speed readings."""
-    fans = []
-    output = run_command(['sensors', '-A', '-u'])
-    
-    current_adapter = None
-    
-    for line in output.split('\n'):
-        line = line.strip()
-        if not line:
-            continue
-            
-        if line.endswith(':') and not line.startswith(' '):
-            current_adapter = line[:-1]
-        elif 'fan' in line.lower() and '_input:' in line and current_adapter:
-            parts = line.split(':')
-            if len(parts) == 2:
-                fan_name = parts[0].replace('_input', '').replace('_', ' ').title()
-                try:
-                    speed = float(parts[1].strip())
-                    fans.append({
-                        'name': fan_name,
-                        'speed': int(speed),
-                        'unit': 'RPM'
-                    })
-                except ValueError:
-                    pass
-    
-    return fans
-
-def get_network_cards() -> List[Dict[str, Any]]:
-    """Get network interface information."""
-    cards = []
-    output = run_command(['ip', '-o', 'link', 'show'])
-    
-    for line in output.split('\n'):
-        if not line or 'lo:' in line:
-            continue
-            
-        parts = line.split()
-        if len(parts) >= 2:
-            name = parts[1].rstrip(':')
-            state = 'UP' if 'UP' in line else 'DOWN'
-            
-            # Get interface type
-            iface_type = 'Unknown'
-            if 'ether' in line:
-                iface_type = 'Ethernet'
-            elif 'wlan' in name or 'wifi' in name:
-                iface_type = 'WiFi'
-            
-            # Try to get speed
-            speed = None
-            speed_output = run_command(['ethtool', name])
-            speed_match = re.search(r'Speed: (\d+\w+)', speed_output)
-            if speed_match:
-                speed = speed_match.group(1)
-            
-            cards.append({
-                'name': name,
-                'type': iface_type,
-                'status': state,
-                'speed': speed
-            })
-    
-    return cards
-
-def get_storage_devices() -> List[Dict[str, Any]]:
-    """Get storage device information."""
-    devices = []
-    output = run_command(['lsblk', '-d', '-o', 'NAME,TYPE,SIZE,MODEL', '-n'])
-    
-    for line in output.split('\n'):
-        if not line:
-            continue
-            
-        parts = line.split(None, 3)
-        if len(parts) >= 3:
-            name = parts[0]
-            dev_type = parts[1]
-            size = parts[2]
-            model = parts[3] if len(parts) > 3 else 'Unknown'
-            
-            if dev_type in ['disk', 'nvme']:
-                devices.append({
-                    'name': name,
-                    'type': dev_type,
-                    'size': size,
-                    'model': model.strip()
-                })
-    
-    return devices
-
-def get_pci_devices() -> List[Dict[str, Any]]:
-    """Get PCI device information including GPUs."""
-    devices = []
-    output = run_command(['lspci', '-vmm'])
-    
-    current_device = {}
-    
-    for line in output.split('\n'):
-        line = line.strip()
-        
-        if not line:
-            if current_device:
-                devices.append(current_device)
-                current_device = {}
-            continue
-        
-        if ':' in line:
-            key, value = line.split(':', 1)
-            key = key.strip().lower().replace(' ', '_')
-            value = value.strip()
-            current_device[key] = value
-    
-    if current_device:
-        devices.append(current_device)
-    
-    # Enhance GPU devices with monitoring data
-    nvidia_gpus = get_nvidia_gpu_info()
-    amd_gpus = get_amd_gpu_info()
-    
-    nvidia_idx = 0
-    amd_idx = 0
-    
-    for device in devices:
-        # Check if it's a GPU
-        device_class = device.get('class', '').lower()
-        vendor = device.get('vendor', '').lower()
-        
-        if 'vga' in device_class or 'display' in device_class or '3d' in device_class:
-            device['type'] = 'GPU'
-            
-            # Add NVIDIA GPU monitoring data
-            if 'nvidia' in vendor and nvidia_idx < len(nvidia_gpus):
-                gpu_data = nvidia_gpus[nvidia_idx]
-                device['gpu_memory'] = gpu_data.get('memory_total')
-                device['gpu_driver_version'] = gpu_data.get('driver_version')
-                device['gpu_compute_capability'] = gpu_data.get('compute_capability')
-                device['gpu_power_draw'] = gpu_data.get('power_draw')
-                device['gpu_temperature'] = float(gpu_data.get('temperature', 0))
-                device['gpu_utilization'] = float(gpu_data.get('utilization_gpu', 0))
-                device['gpu_memory_used'] = gpu_data.get('memory_used')
-                device['gpu_memory_total'] = gpu_data.get('memory_total')
-                device['gpu_clock_speed'] = gpu_data.get('clock_graphics')
-                device['gpu_memory_clock'] = gpu_data.get('clock_memory')
-                nvidia_idx += 1
-            
-            # Add AMD GPU monitoring data
-            elif 'amd' in vendor and amd_idx < len(amd_gpus):
-                gpu_data = amd_gpus[amd_idx]
-                device['gpu_temperature'] = float(gpu_data.get('temperature', 0))
-                device['gpu_utilization'] = float(gpu_data.get('utilization_gpu', 0))
-                device['gpu_memory_used'] = gpu_data.get('memory_used')
-                device['gpu_memory_total'] = gpu_data.get('memory_total')
-                amd_idx += 1
-        elif 'network' in device_class or 'ethernet' in device_class:
-            device['type'] = 'Network'
-        elif 'storage' in device_class or 'sata' in device_class or 'nvme' in device_class:
-            device['type'] = 'Storage'
-        else:
-            device['type'] = 'Other'
-    
-    return devices
 
 def get_power_info() -> Optional[Dict[str, Any]]:
-    """Get power consumption information if available."""
+    """
+    Get CPU power consumption using Intel RAPL interface.
+    
+    This function measures power consumption by reading energy counters
+    from /sys/class/powercap/intel-rapl interfaces and calculating
+    the power draw based on the change in energy over time.
+    
+    Used as fallback when IPMI power monitoring is not available.
+    
+    Returns:
+        dict: Power meter information with 'name', 'watts', and 'adapter' keys
+              or None if RAPL interface is unavailable
+              
+    Example:
+        {
+            'name': 'CPU Power',
+            'watts': 45.32,
+            'adapter': 'Intel RAPL (CPU only)'
+        }
+    """
     global _last_energy_reading
     
     rapl_path = '/sys/class/powercap/intel-rapl/intel-rapl:0/energy_uj'
     
     if os.path.exists(rapl_path):
         try:
+            # Read current energy value in microjoules
             with open(rapl_path, 'r') as f:
                 current_energy_uj = int(f.read().strip())
             current_time = time.time()
             
             watts = 0.0
             
+            # Calculate power if we have a previous reading
             if _last_energy_reading['energy_uj'] is not None and _last_energy_reading['timestamp'] is not None:
                 time_diff = current_time - _last_energy_reading['timestamp']
                 if time_diff > 0:
                     energy_diff = current_energy_uj - _last_energy_reading['energy_uj']
+                    # Handle counter overflow (wraps around at max value)
                     if energy_diff < 0:
                         energy_diff = current_energy_uj
+                    # Power (W) = Energy (µJ) / time (s) / 1,000,000
                     watts = round((energy_diff / time_diff) / 1000000, 2)
             
+            # Store current reading for next calculation
             _last_energy_reading['energy_uj'] = current_energy_uj
             _last_energy_reading['timestamp'] = current_time
             
+            # Detect CPU vendor for display purposes
             cpu_vendor = 'CPU'
             try:
                 with open('/proc/cpuinfo', 'r') as f:
@@ -374,23 +87,3 @@ def get_power_info() -> Optional[Dict[str, Any]]:
             pass
     
     return None
-
-def main():
-    """Main function to gather all hardware information."""
-    data = {
-        'temperatures': get_temperatures(),
-        'fans': get_fans(),
-        'network_cards': get_network_cards(),
-        'storage_devices': get_storage_devices(),
-        'pci_devices': get_pci_devices(),
-    }
-    
-    power_info = get_power_info()
-    if power_info:
-        data['power_meter'] = power_info
-    
-    print(json.dumps(data, indent=2))
-
-if __name__ == '__main__':
-    import sys
-    main()