Monitoring CPU Events With Python

I decided to carry out a little experiment in Python - watching for CPU events.  By CPU event, I'm referring to a change in utilization my application might find interesting - large spikes for example.  To do this, we need to monitor the CPU usage.  However, my earlier approach at doing this wasn't exactly optimal.  For one thing, it only works on Linux.  This excellent explanation pointed me in a new direction.  So I've adapted it for my purposes of monitoring CPU events.

If I'm able to monitor CPU usage, I'll need something that'll run periodically in the background, checking for changes.  My basic need is this - I have a main application class I want notified when the CPU load meets a given threshold.  This should be relatively straightforward, especially since we've got a handy times() function that'll give us everything we need.  Here is an example of what I came up with.

from threading import Thread
from time import sleep, time
from os import times

class CPUMonitor(Thread):
    
    def __init__(self, frequency=1, threshold=10):
        super(CPUMonitor, self).__init__()
        self.daemon = True
        self.frequency = frequency
        self.threshold = threshold
        self.used, self.elapsed = self.cputime
        self.cache = 0.0
        self.start()
        
    def __repr__(self):
        return '%.2f %%'%self.utilization
                   
    def run(self):
        while True:
            self.events()
            sleep(self.frequency)
            
    def events(self):
        if self.utilization >= self.threshold:
            self.jump()
            
    def jump(self):
        pass
            
    @property
    def cputime(self):
        cputime = times()
        return sum(cputime[0:4]), cputime[4]
        
    @property
    def utilization(self):
        used, elapsed = self.cputime
        try:
            result = (used-self.used) / (elapsed-self.elapsed) * 100
        except ZeroDivisionError:
            result = self.cache
        self.used = used
        self.elapsed = elapsed
        self.cache = result
        return result
        
class App(CPUMonitor):
    def __init__(self):
        super(App, self).__init__()
        self.power = 1000
        
        while True:
            try:
                print 'APP: Computing with %s...' % self.power
                10**self.power
                sleep(0.1)
                self.power += 1000
            except KeyboardInterrupt:
                break
                        
    def jump(self):
        print 'CPU: Jumped - %s' % self
        self.power = 1000
        
if __name__ == '__main__':
    
    app = App()
    

The basic idea is this - when the CPU utilization reaches 10%, my application is notified, and can adjust accordingly.  The CPUMonitor class is meant to extend any application class I might come up with.

CPUMonitor is a thread that runs in the background.  By default, it checks for CPU load changes every second.  If the threshold is matched, the application is notified by calling jump().  Obviously the application needs to provide a jump() implementation. 

In my very simple example scenario, App extends CPUMonitor.  So when the App class is first instantiated, the CPU monitor runs behind the scenes.  Jump is only called if the resources are being over-utilized.  The great thing about this is that I decide what over utilization is.  Maybe 25% is perfectly acceptable to the operating system, but maybe my application doesn't think so.  This value, along with the polling frequency can be altered on the fly.

Try giving this a go, it shouldn't get past 12% utilization or so.  You could also play around with the frequency and threshold settings.  I've only implemented one event.  It wouldn't be too difficult to extend this to, say, trigger a changed by X event.

Python CPU Usage

I recently needed to obtain the CPU usage in a Pythonic-way. After some searching, I found this example. I've adapted it to fit my needs and to fit a more general purpose usage. Here is my version of the example.

#Python CPU usage example.

import time

class CPUsage:
   def __init__(self, interval=0.1, percentage=True):
       self.interval=interval
       self.percentage=percentage
       self.result=self.compute()
      
   def get_time(self):
       stat_file=file("/proc/stat", "r")
       time_list=stat_file.readline().split(" ")[2:6]
       stat_file.close()
       for i in range(len(time_list))  :
           time_list[i]=int(time_list[i])
       return time_list
  
   def delta_time(self):
       x=self.get_time()
       time.sleep(self.interval)
       y=self.get_time()
       for i in range(len(x)):
           y[i]-=x[i]
       return y   

   def compute(self):
       t=self.delta_time()
       if self.percentage:
           result=100-(t[len(t)-1]*100.00/sum(t))
       else:
           result=sum(t)
       return result
  
   def __repr__(self):
       return str(self.result)
  
if __name__ == "__main__":
   print CPUsage()

Here, we have a class called CPUsage. This class is capable of getting the CPU usage as a percentage or in milliseconds. This is specified by the percentage constructor parameter. The interval constructor parameter specifies the length of time we are measuring. This defaults to 0.1 seconds. Finally, the constructor invokes the compute() method to store the result.

The get_time() method will return a list of CPU numbers needed to compute the number of milliseconds the CPU has been in use for the interval we are measuring.

The delta_time() method will accept a time list returned by the get_time() method. It will then return the delta time based on the same list format.

The compute() method will calculate the percentage of CPU usage or simply return the CPU time.

To me, this makes for a more object-oriented approach. This class can also be modified to suite a different usage scenario.