# -*- python -*-
from plugins.cpu import CPU
from plugins.cpuonline import CPUOnline
from plugins.softirqs import SoftIRQs
from common import *
class MyCPU(Entry):
def __init__(self, *args, **kwargs):
self.last_sirq_split = SoftIRQs()
cpu = CPU()
self.last_time = time.monotonic()
self.last_cpu_stat = cpu.cpu
self.last_cpu_total = sum(cpu.cpu)
self.last_cpus_total = [sum(c) for c in cpu.cpus]
self.last_cpus_stat = [[c[s] for c in cpu.cpus] for s in range(len(cpu.cpu))]
self.last_intr = cpu.intr_total
self.last_ctxt = cpu.ctxt
self.last_fork = cpu.processes
self.last_sirq = cpu.softirq_total
self.none = self.colourise(None)
self.coloured = tuple(self.colourise(i) for i in range(101))
self.show_all = True
self.show_function = 0
self.functions = [ lambda *v : self.function_cpu(CPU.idle, '', *v)
, self.function_intr
, self.function_ctxt
, self.function_fork
, self.function_proc
, self.function_sirq
]
for key in self.last_sirq_split.keys:
make = lambda key : (lambda *v : self.function_sirq_split(key, *v))
self.functions.append(make(key))
self.functions += [ lambda *v : self.function_cpu(CPU.user, '(user)', *v)
, lambda *v : self.function_cpu(CPU.nice, '(nice)', *v)
, lambda *v : self.function_cpu(CPU.system, '(system)', *v)
, lambda *v : self.function_cpu(CPU.iowait, '(iowait)', *v)
, lambda *v : self.function_cpu(CPU.irq, '(irq)', *v)
, lambda *v : self.function_cpu(CPU.softirq, '(softirq)', *v)
, lambda *v : self.function_cpu(CPU.steal, '(steal)', *v)
, lambda *v : self.function_cpu(CPU.guest, '(guest)', *v)
, lambda *v : self.function_cpu(CPU.guest_nice, '(guest nice)', *v)
, self.function_load
, self.function_task
, self.function_pid
, self.function_online
]
Entry.__init__(self, *args, **kwargs)
def action(self, col, button, x, y):
if button == LEFT_BUTTON:
self.show_all = not self.show_all
self.invalidate()
elif button == SCROLL_UP:
n = self.show_function + 1
if n < len(self.functions):
self.show_function = n
self.invalidate()
elif button == SCROLL_DOWN:
n = self.show_function - 1
if n >= 0:
self.show_function = n
self.invalidate()
def usage(self, now_stat, now_total, last_stat, last_total, idle):
total = now_total - last_total
stat = now_stat - last_stat
return None if total == 0 else (total - stat if idle else stat) * 100 / total
def colourise_(self, value):
if value is None:
return self.none
else:
return self.coloured[limited(value)]
def colourise(self, value):
if value is None:
return '--%'
elif value >= 100:
return '\033[31m100\033[0m'
colour = '39'
if value >= 5: colour = '32'
if value >= 50: colour = '33'
if value >= 90: colour = '31'
return '\033[%sm%2i\033[0m%%' % (colour, value)
def function_cpu(self, stat, name, cpu, display, tdiff):
now_cpu_stat = self.now_cpu_stat[stat]
now_cpus_stat = self.now_cpus_stat[stat]
last_cpus_stat = self.last_cpus_stat[stat]
last_cpu_stat = self.last_cpu_stat[stat]
if display:
cpu = (now_cpu_stat, self.now_cpu_total, last_cpu_stat, self.last_cpu_total)
cpu = self.colourise_(self.usage(*cpu, idle = stat == CPU.idle))
if self.show_all:
cpus = (now_cpus_stat, self.now_cpus_total, last_cpus_stat, self.last_cpus_total)
cpus = ' '.join(self.colourise_(self.usage(*c, idle = stat == CPU.idle)) for c in zip(*cpus))
cpu = 'Cpu%s: %s : %s' % (name, cpus, cpu)
else:
cpu = 'Cpu%s: %s' % (name, cpu)
return cpu
def function_intr(self, cpu, display, tdiff):
now_intr = cpu.intr_total
if display:
cpu = 'Intr: %.0fHz' % ((now_intr - self.last_intr) / tdiff)
self.last_intr = now_intr
return cpu
def function_ctxt(self, cpu, display, tdiff):
now_ctxt = cpu.ctxt
if display:
cpu = 'Ctxt: %.0fHz' % ((now_ctxt - self.last_ctxt) / tdiff)
self.last_ctxt = now_ctxt
return cpu
def function_fork(self, cpu, display, tdiff):
now_fork = cpu.processes
if display:
cpu = 'Fork: %.0fHz' % ((now_fork - self.last_fork) / tdiff)
self.last_fork = now_fork
return cpu
def function_sirq(self, cpu, display, tdiff):
now_sirq = cpu.softirq_total
if display:
cpu = 'Sirq: %.0fHz' % ((now_sirq - self.last_sirq) / tdiff)
self.last_sirq = now_sirq
return cpu
def function_proc(self, cpu, display, tdiff):
if display:
cpu = 'Proc: %irun %iio' % (cpu.procs_running, cpu.procs_blocked)
return cpu
def function_load(self, cpu, display, tdiff):
if display:
load = AverageLoad()
cpu = 'Load: %.2f %.2f %.2f' % (load.total_avg_5_min, load.total_avg_10_min, load.total_avg_15_min)
return cpu
def function_task(self, cpu, display, tdiff):
if display:
load = AverageLoad()
cpu = 'Task: %i/%i (%.0f%%)' % (load.active_tasks, load.total_tasks, load.active_tasks * 100 / load.total_tasks)
return cpu
def function_pid(self, cpu, display, tdiff):
if display:
load = AverageLoad()
cpu = 'Last PID: %i' % load.last_pid
return cpu
def function_online(self, cpu, display, tdiff):
if display:
try:
cpu, cpuonline, on, off = '', CPUOnline(), 0, 0
online, offline = cpuonline.online, cpuonline.offline
while on < len(online) and off < len(offline):
if online[on] < offline[off]:
cpu += ' \033[32m%s\033[0m' % online[on]
on += 1
else:
cpu += ' \033[31m%s\033[0m' % offline[off]
off += 1
cpu += ''.join(' \033[32m%s\033[0m' % c for c in online[on:])
cpu += ''.join(' \033[31m%s\033[0m' % c for c in offline[off:])
cpu = 'Online:%s' % cpu
except Exception as e:
cpu = str(e)
return cpu
def function_sirq_split(self, key, cpu, display, tdiff):
if display:
label = 'Sirq(%s)' % key.lower().replace('_', ' ').replace(' rx', '↓').replace(' tx', '↑')
now = self.now_sirq_split[key]
last = self.last_sirq_split[key]
hz = lambda n, l : '%0.fHz' % ((n - l) / tdiff)
n = len(now)
snow, slast = sum(now), sum(last)
anow, alast = snow / n, slast / n
each = ''
if self.show_all:
each = '%s : ' % ' '.join(hz(n, l) for n, l in zip(now, last))
else:
each = ''
cpu = '%s: %s%s(%s)' % (label, each, hz(snow, slast), hz(anow, alast))
return cpu
def function(self):
now = time.monotonic()
cpu = CPU()
self.now_sirq_split = SoftIRQs()
tdiff = now - self.last_time
self.last_time = now
display = self.show_function
self.now_cpu_stat = cpu.cpu
self.now_cpu_total = sum(cpu.cpu)
self.now_cpus_total = [sum(c) for c in cpu.cpus]
self.now_cpus_stat = [[c[s] for c in cpu.cpus] for s in range(len(cpu.cpu))]
for i in range(len(self.functions)):
if i == display:
ret = self.functions[i](cpu, True, tdiff)
else:
self.functions[i](cpu, False, tdiff)
self.last_cpus_stat = self.now_cpus_stat
self.last_cpu_stat = self.now_cpu_stat
self.last_cpus_total = self.now_cpus_total
self.last_cpu_total = self.now_cpu_total
self.last_sirq_split = self.now_sirq_split
return ret
