Refactory UI to grab shortcut keys, and tests to not look for triggers.

client/bin/factory_ui

 #!/usr/bin/python
 #
 # Copyright (c) 2010 The Chromium OS Authors. All rights reserved.
 # Use of this source code is governed by a BSD-style license that can be
 # found in the LICENSE file.
 
 
 # DESCRIPTION :
 #
 # This UI is intended to be used by the factory autotest suite to
@@ skipping 10 matching lines @@
 # associated keyboard shortcuts.  As tests are run, their status is
 # color-indicated to the operator -- greyed out means untested, yellow
 # means active, green passed and red failed.
 
 
 import gobject
 import gtk
 import imp
 import os
 import pango
+import signal
 import subprocess
 import sys
 import time
 
 import common
 import factory
 import factory_ui_lib as ful
 
+from gtk import gdk
+from Xlib import X
+from Xlib.display import Display as X_Display
+
 
 # These definitions are expose these classes directly into this
 # namespace, so that the test_list that is sent from the control file
 # can have cleaner syntax.  These are done in this fashion, as opposed
 # to "from factory import <class>" to work-around Python namespace
 # wackiness -- the from syntax does not work, creating new classes.
 OperatorTest = factory.OperatorTest
 InformationScreen = factory.InformationScreen
 AutomatedSequence = factory.AutomatedSequence
 AutomatedSubTest = factory.AutomatedSubTest
 AutomatedRebootSubTest = factory.AutomatedRebootSubTest
 
 
-_SEP_COLOR = gtk.gdk.color_parse('grey50')
-
 _LABEL_EN_SIZE = (170, 35)
 _LABEL_ZW_SIZE = (70, 35)
 _LABEL_EN_FONT = pango.FontDescription('courier new extra-condensed 16')
 _LABEL_ZW_FONT = pango.FontDescription('normal 12')
 _LABEL_T_SIZE = (30, 35)
 _LABEL_T_FONT = pango.FontDescription('courier new italic ultra-condensed 10')
 _LABEL_UNTESTED_FG = gtk.gdk.color_parse('grey40')
 _LABEL_TROUGH_COLOR = gtk.gdk.color_parse('grey20')
 _LABEL_STATUS_SIZE = (140, 30)
 _LABEL_STATUS_FONT = pango.FontDescription(
@@ skipping 89 matching lines @@
     def update(self, status):
         if status == ful.UNTESTED:
             return
         self.label_status.set_text(status)
         self.label_status.modify_fg(gtk.STATE_NORMAL, ful.LABEL_COLORS[status])
         self.queue_draw()
 
 
 class UiState():
 
-    def __init__(self, window, status_map, test_widget_box):
+    def __init__(self, status_map, test_widget_box):
 
         def make_empty_test_label_widget():
             label_box = gtk.EventBox()
             label_box.modify_bg(gtk.STATE_NORMAL, ful.BLACK)
             label = ful.make_label('no active test', font=_OTHER_LABEL_FONT,
                                    alignment=(0.5, 0.5))
             label_box.add(label)
             return label_box
 
         def make_automated_seq_label_widget(subtest_list):
             vbox = gtk.VBox()
             vbox.set_spacing(0)
             for subtest in subtest_list:
                 label_box = SubTestLabelBox(subtest)
                 status_map.set_label_box(subtest, label_box)
                 vbox.pack_start(status_map.lookup_label_box(subtest),
                                 False, False)
             return vbox
 
-        self._window = window
         self._status_map = status_map
         self._test_widget_box = test_widget_box
         self._empty_test_widget = make_empty_test_label_widget()
         self._active_test_widget = self._empty_test_widget
         self.active_test = None
 
         self._test_widget_box.add(self._empty_test_widget)
 
         self._automated_seq_widget_map = dict(
             (t, make_automated_seq_label_widget(t.subtest_list))
@@ skipping 11 matching lines @@
             return
         factory.log('UI active test -> %s' %
                     self._status_map.test_db.get_unique_details(test))
         self.active_test = test
         self._test_widget_box.remove(self._active_test_widget)
         active_widget = (test in self._status_map.test_db.seq_test_set
                          and self._automated_seq_widget_map[test]
                          or self._empty_test_widget)
         self._test_widget_box.add(active_widget)
         self._active_test_widget = active_widget
-        self._window.show_all()
-
-
-def make_hsep(width=1):
-    frame = gtk.EventBox()
-    frame.set_size_request(-1, width)
-    frame.modify_bg(gtk.STATE_NORMAL, _SEP_COLOR)
-    return frame
-
-
-def make_vsep(width=1):
-    frame = gtk.EventBox()
-    frame.set_size_request(width, -1)
-    frame.modify_bg(gtk.STATE_NORMAL, _SEP_COLOR)
-    return frame
+        self._test_widget_box.show_all()
 
 
 def refresh_status(status_map, ui_state):
     status_map.read_new_data()
     active_test = status_map.get_active_top_level_test()
     ui_state.set_active_test(active_test)
     return True
 
 
-def main():
+def grab_shortcut_keys(kbd_shortcut_set, control_pid):
+    disp = X_Display()
+    root = disp.screen().root
+
+    # We want to receive KeyPress events
+    root.change_attributes(event_mask = X.KeyPressMask)
+
+    keycode_map = {}
+    for shortcut in kbd_shortcut_set:
+        keysym = gdk.keyval_from_name(shortcut)
+        keycode = disp.keysym_to_keycode(keysym)
+        keycode_map[keycode] = shortcut
+        root.grab_key(keycode, X.ControlMask, 1,
+                      X.GrabModeAsync, X.GrabModeAsync)
+
+    # This flushes the XGrabKey calls to the server.
+    for x in range(0, root.display.pending_events()):
+        root.display.next_event()
+
+    def handle_xevent(src, cond, xhandle=root.display,
+                      keycode_map=keycode_map,
+                      control_pid=control_pid):
+        for i in range(0, xhandle.pending_events()):
+            xevent = xhandle.next_event()
+            if xevent.type == X.KeyPress:
+                keycode = xevent.detail
+                factory.log_shared_data('activated_kbd_shortcut',
+                                        keycode_map[keycode])
+                os.kill(control_pid, signal.SIGUSR1)
+        return True
+
+    gobject.io_add_watch(root.display, gobject.IO_IN, handle_xevent)
+
+
+def main(test_list, status_file_path, control_pid):
     '''This process is launched by the autotest suite_Factory control
-    process.  Communication with this process is an exchange of well
-    formed python expressions over stdin and stdout.  Basically
-    sending wraps arguments in a call to repr() and recv calls eval()
-    to re-generate the python data.'''
-
-    def control_recv():
-      return eval(sys.stdin.readline().rstrip())
-
-    def control_send(x):
-      print repr(x)
-      sys.stdout.flush()
-
-    # On startup, get the list of tests to run (in order) and the
-    # autotest status file path.
-    factory.log('pulling control info')
-    test_list = control_recv()
-    status_file_path = control_recv()
+    process, which should be identified by the <control pid> cmdline
+    argument.  When operators press keyboard shortcuts, the shortcut
+    value is logged with log_shared_data() and a SIGUSR1 is sent to
+    the control program.'''
 
     status_map = factory.StatusMap(test_list, status_file_path)
 
     window = gtk.Window(gtk.WINDOW_TOPLEVEL)
     window.connect('destroy', lambda _: gtk.main_quit())
     window.modify_bg(gtk.STATE_NORMAL, ful.BLACK)
 
     screen = window.get_screen()
+    if (screen is None):
+        log('ERROR: communication with the X server is not working, ' +
+            'could not find a working screen.  UI exiting.')
+        sys.exit(1)
+
     screen_size = (screen.get_width(), screen.get_height())
     window.set_size_request(*screen_size)
 
     label_trough = gtk.VBox()
     label_trough.set_spacing(0)
 
     rhs_box = gtk.EventBox()
     rhs_box.modify_bg(gtk.STATE_NORMAL, _LABEL_TROUGH_COLOR)
     rhs_box.add(label_trough)
 
     console_box = gtk.EventBox()
     console_box.set_size_request(-1, 180)
     console_box.modify_bg(gtk.STATE_NORMAL, ful.BLACK)
 
     test_widget_box = gtk.Alignment(xalign=0.5, yalign=0.5)
     test_widget_box.set_size_request(-1, -1)
 
-    ui_state = UiState(window, status_map, test_widget_box)
+    ui_state = UiState(status_map, test_widget_box)
 
     lhs_box = gtk.VBox()
     lhs_box.pack_end(console_box, False, False)
     lhs_box.pack_start(test_widget_box)
-    lhs_box.pack_start(make_hsep(3), False, False)
+    lhs_box.pack_start(ful.make_hsep(3), False, False)
 
     base_box = gtk.HBox()
     base_box.pack_end(rhs_box, False, False)
-    base_box.pack_end(make_vsep(3), False, False)
+    base_box.pack_end(ful.make_vsep(3), False, False)
     base_box.pack_start(lhs_box)
 
     window.connect('key-release-event', handle_key_release_event)
     window.add_events(gtk.gdk.KEY_RELEASE_MASK)
 
     gobject.timeout_add(_STATUS_REFRESH_MS, refresh_status,
                         status_map, ui_state)
 
     for test in test_list:
         label_box = TestLabelBox(test)
         status_map.set_label_box(test, label_box)
         label_trough.pack_start(label_box, False, False)
-        label_trough.pack_start(make_hsep(), False, False)
+        label_trough.pack_start(ful.make_hsep(), False, False)
 
     window.add(base_box)
     window.show_all()
 
+    grab_shortcut_keys(status_map.test_db.kbd_shortcut_set, control_pid)
+
     ful.hide_cursor(window.window)
 
     test_widget_allocation = test_widget_box.get_allocation()
     test_widget_size = (test_widget_allocation.width,
                         test_widget_allocation.height)
-    factory.log('test_widget_size = %s' % repr(test_widget_size))
-    control_send(test_widget_size)
+    factory.log_shared_data('test_widget_size', test_widget_size)
 
     console = Console(console_box.get_allocation())
 
     factory.log('factory_ui setup done, starting gtk.main()...')
 
     gtk.main()
 
     factory.log('factory_ui gtk.main() finished, exiting.')
 
 if __name__ == '__main__':
-    main()
+
+    if len(sys.argv) != 4:
+        print ('usage: %s <test list path> <status file path> <control pid>' %
+               sys.argv[0])
+    test_list_path, status_file_path, control_pid_str = sys.argv[1:]
+    control_pid = int(control_pid_str)
+
+    execfile(test_list_path)
+
+    main(TEST_LIST, status_file_path, control_pid)