Page MenuHomec4science
Files F66380482

leds.py
No OneTemporary
Actions

File Metadata

Created
Mon, Jun 10, 04:14

leds.py
View Options

#!/usr/bin/env python2
import collectd
import subprocess
from math import floor, ceil
import time
DEBUG = 1
SERIAL = 1
{% if ctrl_leds %}
import serial
PORT = '/dev/ttyS1'
SPEED = 9600
CORES = int(subprocess.check_output('nproc'))
LEDS_NUMBER = 8 # number of leds in a column
LEDS_MAX = 20
LEDS_NOCOL = '0 0 0 '
FAN_MAX = 200
TEMP_LOW = 20
TEMP_HIGH = 80
COLORS = [
'0 X X ', # cyan
'X 0 X ', # magenta
'X X 0 ', # yellow
'X 0 0 ', # red
'0 X 0 ', # green
'0 0 X ', # blue
]
hosts = set()
temp = {}
{% endif %}
def debug(msg):
if DEBUG:
f = open('/tmp/debug', 'a+')
f.write(str(msg) + "\n")
f.close()
#
# CPU Collector
#
def read_stat():
p = open('/proc/stat', 'r')
data = p.readlines()[0].split(' ')
p.close()
return data
def cpu():
data = read_stat()
spent = float(int(data[2]) + int(data[3]) + int(data[4]))
idle = float(data[5])
time.sleep(0.2)
data = read_stat()
spent = float(int(data[2]) + int(data[3]) + int(data[4])) - spent
idle = float(data[5]) - idle
percent = 100 * spent / (spent + idle)
return percent
def read(data=None):
vl = collectd.Values(type='cpu')
vl.plugin = 'cpu-avg'
vl.dispatch(values=[cpu()])
collectd.register_read(read)
{% if ctrl_leds %}
#
# LEDs and FAN control
#
def color(value, index):
return COLORS[0].replace('X', str(int(value)))
#return COLORS[index].replace('X', str(int(value)))
def write(vl, data=None):
global hosts, temp
line = ''
hosts.add(vl.host)
index = list(hosts).index(vl.host)
if vl.type == 'temperature' and vl.plugin_instance == 'thermal_zone0':
temp[vl.host] = float(vl.values[0])
temp_last = max(temp.values())
if temp_last > TEMP_LOW and temp_last < TEMP_HIGH:
line += 'f %d ' % (temp_last * 2 / 3 - 15)
elif temp_last > TEMP_HIGH:
line += 'f %d ' % FAN_MAX
else:
line += 'f 0 '
if vl.plugin == 'cpu-avg':
cpu = vl.values[0]
leds_on = int(ceil(LEDS_NUMBER * cpu / 100))
line += 'l%d ' % index
for i in xrange(LEDS_NUMBER - 1, -1, -1):
if i < leds_on:
line += color(LEDS_MAX, index)
else:
line += LEDS_NOCOL
if vl.type == 'load':
load1 = vl.values[0]
load1 = load1 if load1 <= CORES else CORES
leds_on = int(floor(LEDS_NUMBER / CORES * load1))
line += 'l%d ' % index
for i in xrange(LEDS_NUMBER - 1, -1, -1):
if i < leds_on:
line += color(LEDS_MAX, index)
else:
line += LEDS_NOCOL
write_serial(line)
def write_serial(line):
if line != '':
debug(line)
if SERIAL:
ser = serial.Serial(PORT, SPEED)
ser.write(line)
ser.close()
def init():
line = 'l0 ' + '0 0 0 ' * 8
write_serial(line)
line = 'l1 ' + '0 0 0 ' * 8
write_serial(line)
line = 'l2 ' + '0 0 0 ' * 8
write_serial(line)
line = 'l3 ' + '0 0 0 ' * 8
write_serial(line)
line = 'f 0 '
write_serial(line)
init()
collectd.register_write(write)
{% endif %}

Event Timeline

c4science · Help