USB WebKit Buildbot monitor

bbmonitor

#!/usr/bin/python2
#
# Buildbot monitor
# Author: Leandro A. F. Pereira <leandro@profusion.mobi>
#
# Copyright (C) 2010 ProFUSION Embedded Systems
#

import urllib
import BeautifulSoup
import time
try:
  import usb
  has_usb = True
except ImportError:
  has_usb = False


class Buildbot(object):
  CACHE_TIME = 60 * 5


  def __init__(self, slave_url):
    self._slave_url = slave_url
    self._cached_recent_builds_time = 0
    self._cached_recent_builds = []


  def recent_builds(self):
    current_time = time.time()
    if current_time - self._cached_recent_builds_time <= Buildbot.CACHE_TIME:
      return self._cached_recent_builds

    separator_char = '&' if '?' in self._slave_url else '?'
    slave_url = '%s%ctime=%s' % (self._slave_url, separator_char, time.time())
    try:
        info = urllib.urlopen(slave_url).read()
    except:
	info = None
    if not info:
      return self._cached_recent_builds

    soup = BeautifulSoup.BeautifulSoup(info)
    if not soup:
      return self._cached_recent_builds

    build_info = soup.find('table', {'class': 'info'})
    if not build_info:
      return self._cached_recent_builds

    recent_builds = []
    for line in build_info.findAll('tr')[1:]:
      status = line.findAll('td')
      if not status or len(status) < 6:
        continue

      date = str(status[0].string)
      rev = str(status[1].string)
      state = str(status[2].string)
      build_no = str(status[4].find('a').string)
      info = str(status[5].string)

      recent_builds.append((date, rev, state, build_no, info))

    self._cached_recent_builds = recent_builds
    self._cached_recent_builds_time = current_time
    
    return recent_builds


  def last_green_revision(self):
    for build in self.recent_builds():
      if build[2] == 'success':
        return build
    return None


  def __str__(self):
    return self._slave_url


class WebKitBuildbot(Buildbot):
  def __init__(self, builder):
    Buildbot.__init__(self, 'http://build.webkit.org/buildslaves/%s' % builder)
    self._builder = builder


  def __str__(self):
    return self._builder


class BuildbotMonitor(object):
  def __init__(self, bot_type, bot_names):
    self._bots = [bot_type(bot) for bot in bot_names]


  def will_poll_slaves(self):
    pass


  def polled_slave(self, slave, status):
    pass


  def done_polling_slaves(self):
    pass


  def loop(self):
    while True:
      self.will_poll_slaves()
      for bot in self._bots:
        self.polled_slave(bot, bot.recent_builds()[0])
      self.done_polling_slaves()


class ConsoleBuildbotMonitor(BuildbotMonitor):
  NORMAL_COLOR = '\033[0;m'
  CLEAR_SCREEN = '\033[H\033[2J'
  SUCCESS_COLOR = '\033[42;33;1m'
  FAILURE_COLOR = '\033[41;33;1m'


  def __init__(self, bot_type, bot_names):
    BuildbotMonitor.__init__(self, bot_type, bot_names)

    largest_slave_name = max(max(len(bot) for bot in bot_names),
                             len('Slave Name'))
    self._format_string = "%%-%ds %%s    %%s    %s   %%s" % \
         (largest_slave_name + 5, ConsoleBuildbotMonitor.NORMAL_COLOR)
    self._header = self._format_string % ('Slave Name', '', 'Status', 'Revision')


  def will_poll_slaves(self):
    print(ConsoleBuildbotMonitor.CLEAR_SCREEN + self._header)


  def polled_slave(self, slave, status):
    last_rev = status[1] if status else '---'
    if not status or status[2] != 'success':
      color = ConsoleBuildbotMonitor.FAILURE_COLOR
      status = 'Failure'
    else:
      color = ConsoleBuildbotMonitor.SUCCESS_COLOR
      status = 'Success'

    print(self._format_string % (slave, color, status, last_rev))


  def done_polling_slaves(self):
    time.sleep(15)


if has_usb:
  # USB code slightly based on http://code.google.com/p/usbmailnotifier/
  # Gadget can be bought at http://www.dealextreme.com/details.dx/sku.27062
  class USBDevice(object):
    def __init__(self, vendor, product):
      self.handle = None
      for bus in usb.busses():
        for device in bus.devices:
          if (device.idVendor, device.idProduct) == (vendor, product):
            self.device = device
            self.configuration = self.device.configurations[0]
            self.interface = self.configuration.interfaces[0][0]
            return
      raise IOError("device not found (vendor id 0x%x, product id 0x%x)" % (vendor, product))


    def open(self):
      if self.handle:
        self.handle = None
      try:
        self.handle = self.device.open()
        self.handle.detachKernelDriver(0)
        self.handle.detachKernelDriver(1)
        self.handle.setConfiguration(self.configuration)
        self.handle.claimInterface(self.interface)
        self.handle.setAltInterface(self.interface)
        return True
      except:
        return False


  class USBBuildbotMonitor(BuildbotMonitor):
    BUILDING_COLOR = 5
    FAILURE_COLOR = 2
    SUCCESS_COLOR = 3


    USB_GADGET_VENDOR_ID = 0x1294
    USB_GADGET_PRODUCT_ID = 0x1320


    def __init__(self, bot_type, bot_names):
      BuildbotMonitor.__init__(self, bot_type, bot_names)
      self.initDevice()

    def initDevice(self):
      self._device = USBDevice(USBBuildbotMonitor.USB_GADGET_VENDOR_ID,
                               USBBuildbotMonitor.USB_GADGET_PRODUCT_ID)
      self._device.open()
      self._device.handle.reset()

    def set_color(self, color):
      try:
        self._device.handle.interruptWrite(0x02, (color, 0, 0, 0, 0), 1000)
      except:
	print "Failed to set new color on USB gadget, retrying."
        self._device = None
        self.initDevice()
	self.set_color(color)


    def polled_slave(self, slave, status):
      if not status or status[2] != 'success':
        self.set_color(USBBuildbotMonitor.FAILURE_COLOR)
      else:
        self.set_color(USBBuildbotMonitor.SUCCESS_COLOR)


    def done_polling_slaves(self):
      time.sleep(15)


if __name__ == '__main__':
  bots = ['efl-linux-slave-4']
  monitor = None
  if has_usb:
    try:
      monitor = USBBuildbotMonitor(WebKitBuildbot, bots)
    except IOError as err:
      print("Could not use USB gadget: %s" % err)
  if not monitor:
    monitor = ConsoleBuildbotMonitor(WebKitBuildbot, bots)
  monitor.loop()