Skip to content

measure_device_usage

measure_device_usage

measure_cpu_usage_per_core(interval, logger, plot=False)

Decorator to measure CPU usage per core.

Parameters:

Name Type Description Default
interval int

Interval between measurements in seconds.

 required
logger logging.Logger

Logger object for logging the results.

 required
plot bool

Flag indicating whether to plot the usage data. Defaults to False.

 False

Returns:

Name Type Description
function

Decorator function.
Source code in redesign_pipeline/utils/measure_device_usage.py 
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
def measure_cpu_usage_per_core(interval,logger,plot=False):
    """Decorator to measure CPU usage per core.

    Args:
        interval (int): Interval between measurements in seconds.
        logger (logging.Logger): Logger object for logging the results.
        plot (bool, optional): Flag indicating whether to plot the usage data. Defaults to False.

    Returns:
        function: Decorator function.
    """

    cpu_use = []
    memory_use = []

    def decorator(func):
        def wrapper(*args, **kwargs):

            def run_func():
                memory_start = psutil.virtual_memory().available
                battery = psutil.sensors_battery()
                plugged_start = battery.power_plugged

                result = func(*args, **kwargs)


                memory_end = psutil.virtual_memory().available
                battery = psutil.sensors_battery()
                plugged_end = battery.power_plugged

                logger.info(f"Available memory before: {memory_start/(1024*1024*1024):.2f}GB, after: {memory_end/(1024*1024*1024):.2f}GB")
                if plugged_start != plugged_end:
                    if plugged_end:
                        logger.info("Battery was plugged in during the execution")
                    else:
                        logger.info("Battery was unplugged during the execution")
                elif plugged_end:
                    logger.info("Power is connected")
                else:
                    logger.info("Power is disconnected")

            @run_until_finished(run_func)
            def get_usages():
                cpu_usages = psutil.cpu_percent(interval=None, percpu=True)
                memory_usage = psutil.virtual_memory().available

                if plot:
                    cpu_use.append(cpu_usages)
                    memory_use.append(memory_usage/(1024*1024*1024))

                time.sleep(interval)

                logger.info(''.join( [f"Core{i}: {core}%  " for i,core in zip(range(len(cpu_usages)), cpu_usages)] ))
                logger.info(f"\nMemory: {memory_usage/(1024*1024*1024):.2f}GB")

            logger.info("CPU usage every %d seconds:" % interval)
            threads = get_usages()
            for thread in threads:
                thread.join()

            if plot:

                os.makedirs('results', exist_ok=True)
                usage_folder = os.path.join('results', 'usage')
                os.makedirs(usage_folder, exist_ok=True)
                store_timestamp = time.strftime("%Y%m%d-%H%M%S")

                plt.cla()
                plt.clf()
                plt.bar(list(range(len(cpu_use[0]))),height=np.mean(cpu_use,axis=0), yerr = np.std(cpu_use,axis=0))
                plt.ylim(top = 100) 
                plt.title('CPU core usage summary')
                plt.xlabel('CPU core')
                plt.ylabel('Usage %')
                plt.savefig(os.path.join(usage_folder, f"cpu_usage_{store_timestamp}.jpg"))

                plt.cla()
                plt.clf()
                plt.plot(memory_use)
                plt.title('Memory usage summary')
                plt.xlabel(f'Time {interval}s')
                plt.ylabel('Usage GB')
                plt.savefig(os.path.join(usage_folder, f"memory_usage_{store_timestamp}.jpg"))


        return wrapper
    return decorator

run_until_finished(dependent_func)

Decorator that runs a function and a dependent function concurrently.

Parameters:

Name Type Description Default
dependent_func function

The dependent function to be executed.

 required

Returns:

Name Type Description
function

Decorated function.
Source code in redesign_pipeline/utils/measure_device_usage.py 
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
def run_until_finished(dependent_func):
    """Decorator that runs a function and a dependent function concurrently.

    Args:
        dependent_func (function): The dependent function to be executed.

    Returns:
        function: Decorated function.
    """
    def run_in_thread(func):
        def wrapper(*args, **kwargs):
            stop_event = threading.Event()

            # Define a new function that calls the original function and sets the stop event
            def run_func():
                while not stop_event.is_set():
                    func(*args, **kwargs)

            # Create a new thread to run the dependent function until the stop event is set
            def run_dependent():
                dependent_func()
                stop_event.set()

            # Start both threads
            func_thread = threading.Thread(target=run_func)
            dependent_thread = threading.Thread(target=run_dependent)
            func_thread.start()
            dependent_thread.start()

            # Return the threads so they can be joined later
            return func_thread, dependent_thread

        return wrapper
    return run_in_thread