Monitor netlink kernel events for cpu hotplug

netlink.py

# Inspired by https://github.com/OpenViX/enigma2/blob/master/lib/python/Components/Netlink.py

import os
import socket
from collections import OrderedDict


NETLINK_KOBJECT_UEVENT = 15


class KernelEventMonitor(object):

    def __init__(self):
        self.recieved_events = OrderedDict()
        self.socket = socket.socket(
            socket.AF_NETLINK, socket.SOCK_DGRAM, NETLINK_KOBJECT_UEVENT)

    def start(self):
        self.socket.bind((os.getpid(), -1))

    def stop(self):
        self.socket.close()

    def __enter__(self):
        self.start()
        return self

    def __exit__(self, exc_type, exc_value, traceback):
        self.stop()

    def __iter__(self):
        while True:
            for item in monitor.next_events():
                yield item

    def next_events(self):
        data = self.socket.recv(16384)
        event = {}
        for item in data.split(b'\x00'):
            if not item:
                # check if we have an event and if we already received it
                if event and event['SEQNUM'] not in self.recieved_events:
                    self.recieved_events[event['SEQNUM']] = None
                    if (len(self.recieved_events) > 100):
                        self.recieved_events.popitem(last=False)
                    yield event
                event = {}
            else:
                try:
                    k, v = item.split(b'=', 1)
                    event[k.decode('ascii')] = v.decode('ascii')
                except ValueError:
                    pass

if __name__ == '__main__':
    with KernelEventMonitor() as monitor:
        for event in monitor:
            if event['SUBSYSTEM'] == 'cpu':
                print('CPU %s went %s' % (event['DEVPATH'], event['ACTION']))