polling_jenkins.py

# -*- coding: utf-8 -*-
import json
import os
import subprocess
import time
import urllib2

devnull = open(os.devnull, 'w')
blink1_tool_path = './blink1-tool'
jenkins_url = 'https://buildhive.cloudbees.com/view/All/job/jenkinsci/'
include_jobs = ()
exclude_jobs = ()
interval = 60 # sec


def get_jobs():
  f = urllib2.urlopen(jenkins_url + '/api/json')
  data = json.load(f)
  jobs = data['jobs']

  if include_jobs != ():
    jobs = filter(lambda x: x['name'] in include_jobs, jobs)
  if exclude_jobs != ():
    jobs = filter(lambda x: not x['name'] in exclude_jobs, jobs)

  return jobs


def str2rgb(color):
  if color == 'blue':
    return '00,00,0xFF'
  elif color == 'yellow':
    return '0xFF,0xFF,00'
  elif color == 'red':
    return '0xFF,00,00'


def exec_blink1():
  jobs = get_jobs()
  colors = [job['color'] for job in jobs]

  if 'red' in colors:
    color = 'red'
  elif 'yellow' in colors:
    color = 'yellow'
  else:
    color = 'blue'

  blink1_process = (blink1_tool_path, '--rgb', str2rgb(color))

  subprocess.check_call(blink1_process, stdout=devnull)


def start_polling():
  print 'target jobs:',
  print [x['name'] for x in get_jobs()]

  try:
    while True:
      exec_blink1()
      time.sleep(interval)
  except KeyboardInterrupt:
    subprocess.check_call((blink1_tool_path, '--off'), stdout=devnull)
    print "\nstopped polling"


if __name__ == '__main__':
  start_polling()