test/threading.py

   1 """
   2 threading.py - S.Fourmanoit <syfou@users.sourceforge.net>, 2005
   3
   4 Test script showing how threads can be used with adesklets.
   5 It demonstrates at the same time how the 'delayed event' mechanism
   6 can be superseeded by any other policy, while preserving
   7 desklet responsiveness.
   8
   9 Two threads are created:
  10
  11 - Working thread:  catch events, and feed them to a queue
  12 - Consumer thread: dequeue events in FIFO mode, using a simple
  13                    timing mechanism to mark too old event as
  14                    expired, and use the information to build the
  15                    display, which is here only a simple trace
  16                    drawing.
  17 """
  18 import adesklets
  19 import thread
  20 from itertools import izip, count
  21 from time import time, sleep
  22
  23 class My_Events(adesklets.Events_handler):
  24     def __init__(self):
  25         adesklets.Events_handler.__init__(self)
  26         self.queue = []
  27         self.lock = thread.allocate_lock()
  28
  29     def ready(self):
  30         adesklets.window_resize(100,100)
  31         adesklets.window_reset(adesklets.WINDOW_MANAGED)
  32         adesklets.window_set_transparency(True)
  33         adesklets.context_set_color(255,255,255,255)
  34         adesklets.window_show()
  35
  36     def motion_notify(self, delayed, x, y):
  37         """
  38         Motion notify reporter
  39
  40         The main execution thread is only used to catch events,
  41         and queue things in self.queue (this is the working thread)
  42         """
  43         self.lock.acquire()
  44         self.queue.append((x, y, time()))
  45         self.lock.release()
  46
  47     def __call__(self, max_delay, sleep_time):
  48         """
  49         Routine implementing the display operations, via an
  50         asynchronous thread (this is the consumer thread).
  51         """
  52         print '\n'.join(['Display thread',
  53                         'max delay is %f seconds,' % max_delay,
  54                         'sleep time between processing is %f seconds'
  55                          % sleep_time])
  56         while 1:
  57             # Let's select the oldest valid event,
  58             # still in the given max_delay, and purge
  59             # the queue appropriatly.
  60             #
  61             self.lock.acquire()
  62             t=time()
  63             cur=None
  64             for i,cur in izip(count(),self.queue):
  65                 if (t-cur[2])<max_delay:
  66                     break
  67             if cur:
  68                 if i is not 0:
  69                     print i, 'events dropped'
  70                 del self.queue[:i+1]
  71             self.lock.release()
  72
  73             # No let's provide minimalistic visualisation
  74             #
  75             if cur:
  76                 print 'Motion Notify %s, %f seconds ago' % (str(cur[:2]),
  77                                                             t-cur[2])
  78                 adesklets.image_fill_rectangle(cur[0]-1,cur[1]-1,2,2)
  79
  80             # Then go to sleep
  81             #
  82             sleep(sleep_time)
  83
  84 # Now let's start things: main thread handling event, second thread
  85 # (see the My_Events::__call__() method) implementing the display.
  86 # Timing parameters can be modified at thread creation time for
  87 # experimentation purposes
  88 #
  89 my_events=My_Events()
  90 thread.start_new_thread(my_events,(2,.01))
  91 my_events.pause()