stress_test.py

import psutil
import time
import multiprocessing
import platform
from datetime import datetime
from multiprocessing import Value, Manager

def stress_cpu(counter):
    """Function to stress a single CPU core and count operations"""
    end_time = time.time() + 600  # Run for 600 seconds (10 minutes)
    while time.time() < end_time:
        # Perform computation and increment counter
        2**1000000
        with counter.get_lock():  # Properly lock the counter for thread-safe increment
            counter.value += 1

def get_cpu_temperature():
    """Get CPU temperature if possible"""
    try:
        if platform.system() == "Linux":
            with open("/sys/class/thermal/thermal_zone0/temp", "r") as f:
                temp = float(f.read()) / 1000.0
            return temp
        elif platform.system() == "Darwin":  # macOS
            return "Temperature monitoring not available on macOS"
        elif platform.system() == "Windows":
            return "Temperature monitoring not available on Windows"
    except:
        return "Unable to read temperature"

def format_time(seconds):
    """Format seconds into minutes and seconds"""
    minutes = int(seconds // 60)
    remaining_seconds = int(seconds % 60)
    return f"{minutes}m {remaining_seconds}s"

def main():
    # Get number of CPU cores
    cpu_count = multiprocessing.cpu_count()
    print(f"Number of CPU cores: {cpu_count}")
    
    # Create shared counter for operations
    operation_counter = Value('i', 0)  # 'i' for integer
    
    # Create processes for each CPU core
    processes = []
    start_time = time.time()
    
    print(f"Starting stress test at {datetime.now().strftime('%H:%M:%S')}")
    print("Running for 10 minutes...")
    
    # Start stress test on all cores
    for _ in range(cpu_count):
        p = multiprocessing.Process(target=stress_cpu, args=(operation_counter,))
        p.start()
        processes.append(p)
    
    # Monitor and print metrics every 15 seconds
    while time.time() - start_time < 600:
        cpu_percent = psutil.cpu_percent(interval=1, percpu=True)
        temp = get_cpu_temperature()
        with operation_counter.get_lock():  # Safely read the counter
            current_ops = operation_counter.value
        elapsed_time = time.time() - start_time
        remaining_time = 600 - elapsed_time
        
        print("\n--- Current Metrics ---")
        print(f"Time elapsed: {format_time(elapsed_time)}")
        print(f"Time remaining: {format_time(remaining_time)}")
        print(f"CPU Temperature: {temp}°C")
        print("CPU Usage per core:")
        for core, usage in enumerate(cpu_percent):
            print(f"Core {core}: {usage}%")
        print(f"Current operations: {current_ops:,}")
        if elapsed_time >= 1:  # Avoid division by zero
            print(f"Operations per minute (current rate): {int(current_ops * (60/elapsed_time)):,}")
        
        time.sleep(14)  # Wait 14 seconds before next reading (plus 1 second from cpu_percent)
    
    # Wait for all processes to complete
    for p in processes:
        p.terminate()
        p.join()
    
    # Calculate final metrics
    with operation_counter.get_lock():  # Safely read final counter
        total_operations = operation_counter.value
    operations_per_minute = int(total_operations * (60/600))  # Since we ran for 600 seconds
    
    # Final metrics
    print("\n=== Final Performance Report ===")
    print(f"Test completed at {datetime.now().strftime('%H:%M:%S')}")
    print(f"Total test duration: {format_time(time.time() - start_time)}")
    print(f"Average CPU usage: {sum(psutil.cpu_percent(percpu=True))/cpu_count:.1f}%")
    print(f"Final CPU Temperature: {get_cpu_temperature()}°C")
    print(f"Memory usage: {psutil.virtual_memory().percent}%")
    print(f"Total operations completed: {total_operations:,}")
    print(f"Operations per minute: {operations_per_minute:,}")
    print(f"Operations per minute per core: {int(operations_per_minute/cpu_count):,}")

if __name__ == "__main__":
    main()