openCom-Companion/sbapp/plyer/platforms/linux/cpu.py
2024-06-02 18:31:58 +02:00

117 lines
3.4 KiB
Python

'''
Module of Linux API for plyer.cpu.
'''
from os.path import join
from os import environ, listdir
from subprocess import Popen, PIPE
from sbapp.plyer.facades import CPU
from sbapp.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()