Created
September 29, 2014 12:35
-
-
Save drott/f800633b9b9847e4d45e to your computer and use it in GitHub Desktop.
USB WebKit Buildbot monitor
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
#!/usr/bin/python2 | |
# | |
# Buildbot monitor | |
# Author: Leandro A. F. Pereira <[email protected]> | |
# | |
# 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() |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment