<reponame>mphirke/plyer<filename>plyer/platforms/linux/cpu.py<gh_stars>1000+
'''
Module of Linux API for plyer.cpu.
'''

from os.path import join
from os import environ, listdir
from subprocess import Popen, PIPE
from plyer.facades import CPU
from plyer.utils import whereis_exe


class LinuxCPU(CPU):
    '''
    Implementation of Linux CPU API.
    '''

    def _sockets(self):
        # physical CPU sockets (or slots) on motherboard
        sockets = []  # list of CPU ids from kernel

        # open Linux kernel data file for CPU
        with open('/proc/cpuinfo', 'rb') as fle:
            lines = fle.readlines()

        # go through the lines and obtain physical CPU ids
        for line in lines:
            line = line.decode('utf-8')
            if 'physical id' not in line:
                continue
            cpuid = line.split(':')[1].strip()
            sockets.append(cpuid)

        # total sockets is the length of unique CPU ids from kernel
        sockets = len(set(sockets))
        return sockets

    def _physical(self):
        # cores
        physical = []  # list of CPU ids from kernel

        # open Linux kernel data file for CPU
        with open('/proc/cpuinfo', 'rb') as fle:
            lines = fle.readlines()

        # go through the lines and obtain CPU core ids
        for line in lines:
            line = line.decode('utf-8')
            if 'core id' not in line:
                continue
            cpuid = line.split(':')[1].strip()
            physical.append(cpuid)

        # total cores (socket * core per socket)
        # is the length of unique CPU core ids from kernel
        physical = len(set(physical))
        return physical

    def _logical(self):
        # cores * threads
        logical = None
        old_lang = environ.get('LANG', '')
        environ['LANG'] = 'C'

        _logical = Popen(['nproc', '--all'], stdout=PIPE)
        output = _logical.communicate()[0].decode('utf-8').strip()
        if output:
            logical = int(output)

        environ['LANG'] = old_lang
        return logical

    def _cache(self):
        values = {key: 0 for key in ('L1', 'L2', 'L3')}
        cpu_path = join('/sys', 'devices', 'system', 'cpu')

        # get present cores from kernel device
        with open(join(cpu_path, 'present')) as fle:
            present = fle.read().decode('utf-8')
        present = present.strip().split('-')

        if len(present) == 2:
            present = range(int(present[1]) + 1)
        else:
            present = [present[0]]

        cores = ['cpu{}'.format(i) for i in present]
        for core in cores:
            indicies = [
                # get 'indexN' files from 'cache' folder assuming
                # the filename is in range index0 to index99
                # in case a wild 'index_whatevercontent' file appears
                fle for fle in listdir(join(cpu_path, core, 'cache'))
                if fle.startswith('index') and len(fle) <= len('index') + 2
            ]

            for index in indicies:
                index_type = join(cpu_path, core, 'cache', index, 'level')
                with open(index_type, 'rb') as fle:
                    cache_level = fle.read().decode('utf-8').strip()
                values['L{}'.format(cache_level)] += 1
        return values

    @staticmethod
    def _numa():
        return


def instance():
    '''
    Instance for facade proxy.
    '''
    import sys
    if whereis_exe('nproc'):
        return LinuxCPU()
    sys.stderr.write("nproc not found.")
    return CPU()

'''
Module of Linux API for plyer.cpu.
'''

from os.path import join
from os import environ, listdir
from subprocess import Popen, PIPE
from plyer.facades import CPU
from plyer.utils import whereis_exe


class LinuxCPU(CPU):
    '''
    Implementation of Linux CPU API.
    '''

    def _sockets(self):
        # physical CPU sockets (or slots) on motherboard
        sockets = []  # list of CPU ids from kernel

        # open Linux kernel data file for CPU
        with open('/proc/cpuinfo', 'rb') as fle:
            lines = fle.readlines()

        # go through the lines and obtain physical CPU ids
        for line in lines:
            line = line.decode('utf-8')
            if 'physical id' not in line:
                continue
            cpuid = line.split(':')[1].strip()
            sockets.append(cpuid)

        # total sockets is the length of unique CPU ids from kernel
        sockets = len(set(sockets))
        return sockets

    def _physical(self):
        # cores
        physical = []  # list of CPU ids from kernel

        # open Linux kernel data file for CPU
        with open('/proc/cpuinfo', 'rb') as fle:
            lines = fle.readlines()

        # go through the lines and obtain CPU core ids
        for line in lines:
            line = line.decode('utf-8')
            if 'core id' not in line:
                continue
            cpuid = line.split(':')[1].strip()
            physical.append(cpuid)

        # total cores (socket * core per socket)
        # is the length of unique CPU core ids from kernel
        physical = len(set(physical))
        return physical

    def _logical(self):
        # cores * threads
        logical = None
        old_lang = environ.get('LANG', '')
        environ['LANG'] = 'C'

        _logical = Popen(['nproc', '--all'], stdout=PIPE)
        output = _logical.communicate()[0].decode('utf-8').strip()
        if output:
            logical = int(output)

        environ['LANG'] = old_lang
        return logical

    def _cache(self):
        values = {key: 0 for key in ('L1', 'L2', 'L3')}
        cpu_path = join('/sys', 'devices', 'system', 'cpu')

        # get present cores from kernel device
        with open(join(cpu_path, 'present')) as fle:
            present = fle.read().decode('utf-8')
        present = present.strip().split('-')

        if len(present) == 2:
            present = range(int(present[1]) + 1)
        else:
            present = [present[0]]

        cores = ['cpu{}'.format(i) for i in present]
        for core in cores:
            indicies = [
                # get 'indexN' files from 'cache' folder assuming
                # the filename is in range index0 to index99
                # in case a wild 'index_whatevercontent' file appears
                fle for fle in listdir(join(cpu_path, core, 'cache'))
                if fle.startswith('index') and len(fle) <= len('index') + 2
            ]

            for index in indicies:
                index_type = join(cpu_path, core, 'cache', index, 'level')
                with open(index_type, 'rb') as fle:
                    cache_level = fle.read().decode('utf-8').strip()
                values['L{}'.format(cache_level)] += 1
        return values

    @staticmethod
    def _numa():
        return


def instance():
    '''
    Instance for facade proxy.
    '''
    import sys
    if whereis_exe('nproc'):
        return LinuxCPU()
    sys.stderr.write("nproc not found.")
    return CPU()
Module of Linux API for plyer.cpu. Implementation of Linux CPU API. # physical CPU sockets (or slots) on motherboard # list of CPU ids from kernel # open Linux kernel data file for CPU # go through the lines and obtain physical CPU ids # total sockets is the length of unique CPU ids from kernel # cores # list of CPU ids from kernel # open Linux kernel data file for CPU # go through the lines and obtain CPU core ids # total cores (socket * core per socket) # is the length of unique CPU core ids from kernel # cores * threads # get present cores from kernel device # get 'indexN' files from 'cache' folder assuming # the filename is in range index0 to index99 # in case a wild 'index_whatevercontent' file appears Instance for facade proxy.