Now able to display the list of radio programs
[watersofshiloah] / src / util / go_utils.py
1 #!/usr/bin/env python
2
3 from __future__ import with_statement
4
5 import time
6 import functools
7 import threading
8 import Queue
9 import logging
10
11 import gobject
12
13 import algorithms
14 import misc
15
16
17 _moduleLogger = logging.getLogger(__name__)
18
19
20 def make_idler(func):
21         """
22         Decorator that makes a generator-function into a function that will continue execution on next call
23         """
24         a = []
25
26         @functools.wraps(func)
27         def decorated_func(*args, **kwds):
28                 if not a:
29                         a.append(func(*args, **kwds))
30                 try:
31                         a[0].next()
32                         return True
33                 except StopIteration:
34                         del a[:]
35                         return False
36
37         return decorated_func
38
39
40 def async(func):
41         """
42         Make a function mainloop friendly. the function will be called at the
43         next mainloop idle state.
44
45         >>> import misc
46         >>> misc.validate_decorator(async)
47         """
48
49         @functools.wraps(func)
50         def new_function(*args, **kwargs):
51
52                 def async_function():
53                         func(*args, **kwargs)
54                         return False
55
56                 gobject.idle_add(async_function)
57
58         return new_function
59
60
61 class Async(object):
62
63         def __init__(self, func, once = True):
64                 self.__func = func
65                 self.__idleId = None
66                 self.__once = once
67
68         def start(self):
69                 assert self.__idleId is None
70                 if self.__once:
71                         self.__idleId = gobject.idle_add(self._on_once)
72                 else:
73                         self.__idleId = gobject.idle_add(self.__func)
74
75         def is_running(self):
76                 return self.__idleId is not None
77
78         def cancel(self):
79                 if self.__idleId is not None:
80                         gobject.source_remove(self.__idleId)
81                         self.__idleId = None
82
83         def __call__(self):
84                 return self.start()
85
86         @misc.log_exception(_moduleLogger)
87         def _on_once(self):
88                 self.cancel()
89                 try:
90                         self.__func()
91                 except Exception:
92                         pass
93                 return False
94
95
96 class Timeout(object):
97
98         def __init__(self, func):
99                 self.__func = func
100                 self.__timeoutId = None
101
102         def start(self, **kwds):
103                 assert self.__timeoutId is None
104
105                 assert len(kwds) == 1
106                 timeoutInSeconds = kwds["seconds"]
107                 assert 0 <= timeoutInSeconds
108                 if timeoutInSeconds == 0:
109                         self.__timeoutId = gobject.idle_add(self._on_once)
110                 else:
111                         self.__timeoutId = timeout_add_seconds(timeoutInSeconds, self._on_once)
112
113         def is_running(self):
114                 return self.__timeoutId is not None
115
116         def cancel(self):
117                 if self.__timeoutId is not None:
118                         gobject.source_remove(self.__timeoutId)
119                         self.__timeoutId = None
120
121         def __call__(self, **kwds):
122                 return self.start(**kwds)
123
124         @misc.log_exception(_moduleLogger)
125         def _on_once(self):
126                 self.cancel()
127                 try:
128                         self.__func()
129                 except Exception:
130                         pass
131                 return False
132
133
134 _QUEUE_EMPTY = object()
135
136
137 class AsyncPool(object):
138
139         def __init__(self):
140                 self.__workQueue = Queue.Queue()
141                 self.__thread = threading.Thread(
142                         name = type(self).__name__,
143                         target = self.__consume_queue,
144                 )
145                 self.__isRunning = True
146
147         def start(self):
148                 self.__thread.start()
149
150         def stop(self):
151                 self.__isRunning = False
152                 for _ in algorithms.itr_available(self.__workQueue):
153                         pass # eat up queue to cut down dumb work
154                 self.__workQueue.put(_QUEUE_EMPTY)
155
156         def clear_tasks(self):
157                 for _ in algorithms.itr_available(self.__workQueue):
158                         pass # eat up queue to cut down dumb work
159
160         def add_task(self, func, args, kwds, on_success, on_error):
161                 task = func, args, kwds, on_success, on_error
162                 self.__workQueue.put(task)
163
164         @misc.log_exception(_moduleLogger)
165         def __trampoline_callback(self, on_success, on_error, isError, result):
166                 if not self.__isRunning:
167                         if isError:
168                                 _moduleLogger.error("Masking: %s" % (result, ))
169                         isError = True
170                         result = StopIteration("Cancelling all callbacks")
171                 callback = on_success if not isError else on_error
172                 try:
173                         callback(result)
174                 except Exception:
175                         _moduleLogger.exception("Callback errored")
176                 return False
177
178         @misc.log_exception(_moduleLogger)
179         def __consume_queue(self):
180                 while True:
181                         task = self.__workQueue.get()
182                         if task is _QUEUE_EMPTY:
183                                 break
184                         func, args, kwds, on_success, on_error = task
185
186                         try:
187                                 result = func(*args, **kwds)
188                                 isError = False
189                         except Exception, e:
190                                 _moduleLogger.error("Error, passing it back to the main thread")
191                                 result = e
192                                 isError = True
193                         self.__workQueue.task_done()
194
195                         gobject.idle_add(self.__trampoline_callback, on_success, on_error, isError, result)
196                 _moduleLogger.debug("Shutting down worker thread")
197
198
199 class AsyncLinearExecution(object):
200
201         def __init__(self, pool, func):
202                 self._pool = pool
203                 self._func = func
204                 self._run = None
205
206         def start(self, *args, **kwds):
207                 assert self._run is None
208                 self._run = self._func(*args, **kwds)
209                 trampoline, args, kwds = self._run.send(None) # priming the function
210                 self._pool.add_task(
211                         trampoline,
212                         args,
213                         kwds,
214                         self.on_success,
215                         self.on_error,
216                 )
217
218         @misc.log_exception(_moduleLogger)
219         def on_success(self, result):
220                 _moduleLogger.debug("Processing success for: %r", self._func)
221                 try:
222                         trampoline, args, kwds = self._run.send(result)
223                 except StopIteration, e:
224                         pass
225                 else:
226                         self._pool.add_task(
227                                 trampoline,
228                                 args,
229                                 kwds,
230                                 self.on_success,
231                                 self.on_error,
232                         )
233
234         @misc.log_exception(_moduleLogger)
235         def on_error(self, error):
236                 _moduleLogger.debug("Processing error for: %r", self._func)
237                 try:
238                         trampoline, args, kwds = self._run.throw(error)
239                 except StopIteration, e:
240                         pass
241                 else:
242                         self._pool.add_task(
243                                 trampoline,
244                                 args,
245                                 kwds,
246                                 self.on_success,
247                                 self.on_error,
248                         )
249
250
251 def throttled(minDelay, queue):
252         """
253         Throttle the calls to a function by queueing all the calls that happen
254         before the minimum delay
255
256         >>> import misc
257         >>> import Queue
258         >>> misc.validate_decorator(throttled(0, Queue.Queue()))
259         """
260
261         def actual_decorator(func):
262
263                 lastCallTime = [None]
264
265                 def process_queue():
266                         if 0 < len(queue):
267                                 func, args, kwargs = queue.pop(0)
268                                 lastCallTime[0] = time.time() * 1000
269                                 func(*args, **kwargs)
270                         return False
271
272                 @functools.wraps(func)
273                 def new_function(*args, **kwargs):
274                         now = time.time() * 1000
275                         if (
276                                 lastCallTime[0] is None or
277                                 (now - lastCallTime >= minDelay)
278                         ):
279                                 lastCallTime[0] = now
280                                 func(*args, **kwargs)
281                         else:
282                                 queue.append((func, args, kwargs))
283                                 lastCallDelta = now - lastCallTime[0]
284                                 processQueueTimeout = int(minDelay * len(queue) - lastCallDelta)
285                                 gobject.timeout_add(processQueueTimeout, process_queue)
286
287                 return new_function
288
289         return actual_decorator
290
291
292 def _old_timeout_add_seconds(timeout, callback):
293         return gobject.timeout_add(timeout * 1000, callback)
294
295
296 def _timeout_add_seconds(timeout, callback):
297         return gobject.timeout_add_seconds(timeout, callback)
298
299
300 try:
301         gobject.timeout_add_seconds
302         timeout_add_seconds = _timeout_add_seconds
303 except AttributeError:
304         timeout_add_seconds = _old_timeout_add_seconds