travis-status.py

#!/usr/bin/env python
# -*- coding: utf-8 -*-

import sys

def inspect(s):
    if s is None:
        return '"None"'
    return '"' + str(s) + '"'


# CLI 
import argparse

parser = argparse.ArgumentParser(description="Display Travis Status on the PiGlow")

DEFAULT_LOG = "/tmp/travis-status.log"
parser.add_argument("-l", "--log",
    help="log file (default '" + DEFAULT_LOG + "')",
    default=DEFAULT_LOG)

parser.add_argument("-D", "--debug",
    help="debug mode",
    action='store_true')

parser.add_argument('-u', '--github-user',
    help="github user (required)",
    required=True)

parser.add_argument('-r', '--github-repo',
    help="github repo (required)",
    required=True)

parser.add_argument('-b', '--github-branch',
    help="github branch; omit to monitor latest build")

parser.add_argument('-t', '--travis-token',
    help="travis token (required); open source travis: travis login && travis token; travis pro: travis login --pro && travis token --pro",
    required=True)

DEFAULT_TRAVIS_URL = 'https://api.travis_ci.org'
parser.add_argument('-U', '--travis-url',
    help="travis url (default '" + DEFAULT_TRAVIS_URL + "'); open source travis: https://api.travis_ci.org; travis pro: https://api.travis_ci.com",
    default=DEFAULT_TRAVIS_URL)

args = parser.parse_args()


# Logging
import logging
import logging.handlers

LOG_LEVEL = logging.INFO
if args.debug:
    LOG_LEVEL = logging.DEBUG
LOG_FILENAME = args.log

logger = logging.getLogger(__name__)
logger.setLevel(LOG_LEVEL)

formatter = logging.Formatter('%(asctime)s %(levelname)-8s %(message)s')

file_handler = logging.handlers.TimedRotatingFileHandler(LOG_FILENAME, when="midnight", backupCount=3)
file_handler.setFormatter(formatter)
logger.addHandler(file_handler)

stream_handler = logging.StreamHandler(sys.stdout)
stream_handler.setFormatter(formatter)
logger.addHandler(stream_handler)

class MyLogger(object):
    def __init__(self, logger, level):
        self.logger = logger
        self.level = level

    def write(self, message):
        if message.rstrip() != "":
            self.logger.log(self.level, message.rstrip())
 
sys.stdout = MyLogger(logger, logging.INFO)
sys.stderr = MyLogger(logger, logging.ERROR)


# Debug command line options
logger.debug('log file:     ' + inspect(args.log))
logger.debug('debug:        ' + inspect(args.debug))
logger.debug('github user:  ' + inspect(args.github_user))
logger.debug('github repo:  ' + inspect(args.github_repo))
logger.debug('github branch:' + inspect(args.github_branch))
logger.debug('travis token: ' + inspect(args.travis_token))
logger.debug('travis url:   ' + inspect(args.travis_url))

 
# Travis monitoring
import signal
import atexit
import json
import requests
from time import sleep
from piglow import PiGlow
piglow = PiGlow()

def log_response_error(r):
    logger.error('Cannot get travis status.')
    logger.debug('HTTP ' + str(r.status_code) + ' ' + requests.status_codes._codes[r.status_code][0].upper())
    if r.text:
        logger.debug('Response body:')
        logger.debug(r.text)
    else:
        logger.debug('Response body empty.')

def travis_request(path):
    url = args.travis_url + path
    headers = {
        'User-Agent': 'PiGlow-Travis-Status/1.0.0',
        'Accept': 'application/vnd.travis-ci.2+json',
        'Authorization': 'token "' + args.travis_token + '"'
    }
    try:
        r = requests.get(url, headers=headers)
    except requests.exceptions.ConnectionError as e:
        logger.debug('HTTP Request failed.')
        logger.debug(e)

    if r.status_code != 200:
        log_response_error(r)
    else:
        return r.json()

def get_branch_status():
    res = travis_request('/repos/' + args.github_user + '/' + args.github_repo + '/branches/' + args.github_branch)
    if res:
        return res['branch']['state']
 
def get_current_build_status():
    res = travis_request('/repos/' + args.github_user + '/' + args.github_repo)
    if res:
        return res['repo']['last_build_state']
 
def get_status():
    if args.github_branch != None:
        return get_branch_status()
    else:
        return get_current_build_status()

def switch_off_leds():
    logger.debug('Switching off all LEDs.')
    piglow.all(0)
    sys.exit(0)

def update_leds(status):
    logger.debug('Update LEDs to status ' + inspect(status) + '.')
    piglow.all(0)
    if status == 'created':
        piglow.yellow(255)
    elif status == 'started':
        piglow.yellow(255)
    elif status == 'passed':
        piglow.green(120)
    elif status == 'failed':
        piglow.red(255)
    else:
        piglow.blue(5)

# Clear leds on exit
def cleanup(*args):
    logger.info('Exiting.')
    switch_off_leds()
    sys.exit(0)

signal.signal(signal.SIGINT, cleanup)
signal.signal(signal.SIGTERM, cleanup)

if args.github_branch != None:
    logger.info('Now monitoring Travis status of Github repo ' + args.github_user + '/' + args.github_repo + ' (' + args.github_branch + ').')
else:
    logger.info('Now monitoring Travis status of Github repo ' + args.github_user + '/' + args.github_repo + ' (current build).')

last_status = None
while True:
    status = get_status()
    if last_status != status:
        update_leds(status)
        last_status = status
    sleep(1)