Skip to content

Instantly share code, notes, and snippets.

Embed
What would you like to do?
#!/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)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment