From 0b8f5d3b226ae55e9d959833b50615c3c7367c51 Mon Sep 17 00:00:00 2001 From: MacRimi Date: Wed, 26 Nov 2025 18:00:01 +0100 Subject: [PATCH] Update AppImage --- AppImage/scripts/flask_server.py | 20 +- AppImage/scripts/hardware_monitor.py | 383 +++------------------------ 2 files changed, 48 insertions(+), 355 deletions(-) diff --git a/AppImage/scripts/flask_server.py b/AppImage/scripts/flask_server.py index a1b00d5..5a72663 100644 --- a/AppImage/scripts/flask_server.py +++ b/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: - try: - rapl_power = hardware_monitor.get_power_info() - if rapl_power: - power_meter = rapl_power - # print(f"[v0] Power meter from RAPL: {power_meter.get('watts', 0)}W") - pass - except Exception as e: - # print(f"[v0] Error getting RAPL power info: {e}") - pass + if power_meter is None: + try: + rapl_power = hardware_monitor.get_power_info() + if rapl_power: + power_meter = rapl_power + # print(f"[v0] Power meter from RAPL: {power_meter.get('watts', 0)}W") + pass + except Exception as e: + # print(f"[v0] Error getting RAPL power info: {e}") + pass return { 'temperatures': temperatures, diff --git a/AppImage/scripts/hardware_monitor.py b/AppImage/scripts/hardware_monitor.py index 7d75fe7..2c31a33 100644 --- a/AppImage/scripts/hardware_monitor.py +++ b/AppImage/scripts/hardware_monitor.py @@ -1,359 +1,72 @@ #!/usr/bin/env python3 -import json -import subprocess -import re +""" +Hardware Monitor - RAPL Power Monitoring + +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()