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