myi3status.py

#!/usr/bin/env python3

# Public Domain

import json
import os
import subprocess
import sys
import threading
import time
import urllib.request

WHITE = '#ffffff'
RED = '#ff0000'
GREEN = '#00ff00'
ORANGE = '#ffa500'

SEPARATOR_WIDTH = 19

INTERVAL = 5
IP_INTERVAL = 600
IP_SERVICE_URL = 'http://ifconfig.co/'

DF_ROOT = '/'
DF_LOW_THRESHOLD = 20 * 1024 * 1024 * 1024

LOAD_HIGH_THRESHOLD = 3
LOAD_MEDIUM_THRESHOLD = 1

DATE_FORMAT = '%Y-%m-%d %H:%M:%S'

public_ip = {'full_text': 'unknown', 'color': ORANGE, 'separator_block_width': SEPARATOR_WIDTH}

def get_ip_thread():
    global public_ip
    get_ip_request = urllib.request.Request(IP_SERVICE_URL, headers={'User-Agent': 'curl/0.0'})
    while True:
        text = ''
        color = GREEN
        try:
            text = urllib.request.urlopen(get_ip_request).read().decode('utf-8').strip()
        except KeyboardInterrupt:
            return
        except Exception as exception:
            text = type(exception).__name__
            color = RED
        public_ip = {'full_text': text, 'color': color, 'separator_block_width': SEPARATOR_WIDTH}
        time.sleep(IP_INTERVAL)

def get_df():
    avail_bytes = int(subprocess.check_output(['df', '-B1', '--output=avail', DF_ROOT]).split(b'\n')[1])
    color = GREEN
    if avail_bytes < DF_LOW_THRESHOLD:
        color = RED
    text = "{0:0.1f} GiB".format(avail_bytes / 1024 / 1024 / 1024)
    return {'full_text': text, 'color': color, 'separator_block_width': SEPARATOR_WIDTH}

def get_load():
    color = WHITE
    text = ''
    try:
        with open('/proc/loadavg', 'r') as f:
            load = float(f.read().split(' ')[0])
            if load > LOAD_HIGH_THRESHOLD:
                color = RED
            elif load > LOAD_MEDIUM_THRESHOLD:
                color = ORANGE
            text = "{0:0.2f}".format(load)
    except KeyboardInterrupt:
        sys.exit(1)
    except Exception as exception:
        text = type(exception).__name__
        color = RED

    return {'full_text': text, 'color': color, 'separator_block_width': SEPARATOR_WIDTH}

def get_time():
    return {'full_text': time.strftime(DATE_FORMAT), 'separator_block_width': SEPARATOR_WIDTH}

def write_line(d):
    sys.stdout.write(json.dumps(d));
    sys.stdout.write(',\n');
    sys.stdout.flush();

if __name__ == '__main__':
    threading.Thread(target=get_ip_thread).start()

    sys.stdout.write('{ "version": 1 }\n')
    sys.stdout.flush()
    sys.stdout.write('[\n')

    while True:
        write_line([
            public_ip,
            get_df(),
            get_load(),
            get_time(),
        ])

        try:
            time.sleep(INTERVAL)
        except KeyboardInterrupt:
            os._exit(1)