1
2
3
4
5
6
7
8
9
10
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
45
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
|
'''
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()
|
