# -*- 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