from __future__ import with_statement
+import os
+import errno
import sys
+import time
import traceback
import functools
import contextlib
@contextlib.contextmanager
+def flock(path, timeout=-1):
+ WAIT_FOREVER = -1
+ DELAY = 0.1
+ timeSpent = 0
+
+ acquired = False
+
+ while timeSpent <= timeout or timeout == WAIT_FOREVER:
+ try:
+ fd = os.open(path, os.O_CREAT | os.O_EXCL | os.O_RDWR)
+ acquired = True
+ break
+ except OSError, e:
+ if e.errno != errno.EEXIST:
+ raise
+ time.sleep(DELAY)
+ timeSpent += DELAY
+
+ assert acquired, "Failed to grab file-lock %s within timeout %d" % (path, timeout)
+
+ try:
+ yield fd
+ finally:
+ os.unlink(path)
+
+
+@contextlib.contextmanager
def gtk_lock():
gtk.gdk.threads_enter()
try:
@autostart
+def printer_sink(format = "%s"):
+ """
+ >>> pr = printer_sink("%r")
+ >>> pr.send("Hello")
+ 'Hello'
+ >>> pr.send("5")
+ '5'
+ >>> pr.send(5)
+ 5
+ >>> p = printer_sink()
+ >>> p.send("Hello")
+ Hello
+ >>> p.send("World")
+ World
+ >>> # p.throw(RuntimeError, "Goodbye")
+ >>> # p.send("Meh")
+ >>> # p.close()
+ """
+ while True:
+ item = yield
+ print format % (item, )
+
+
+@autostart
def null_sink():
"""
Good for uses like with cochain to pick up any slack
"""
>>> p = printer_sink()
>>> cm = comap(lambda x: x+1, p)
- >>> cm.send(0)
+ >>> cm.send((0, ))
1
- >>> cm.send(1.0)
+ >>> cm.send((1.0, ))
2.0
- >>> cm.send(-2)
+ >>> cm.send((-2, ))
-1
- >>> # cm.throw(RuntimeError, "Goodbye")
- >>> # cm.send(0)
- >>> # cm.send(1.0)
- >>> # cm.close()
"""
while True:
try:
target.throw(e.__class__, e.message)
+def _flush_queue(queue):
+ while not queue.empty():
+ yield queue.get()
+
+
@autostart
def queue_sink(queue):
"""
item = yield
queue.put((None, item))
except StandardError, e:
- queue.put((e.__class__, e.message))
+ queue.put((e.__class__, str(e)))
except GeneratorExit:
queue.put((GeneratorExit, None))
raise
def nonqueue_source(queue, target):
- """
- >>> q = Queue.Queue()
- >>> for i in [
- ... (None, 'Hello'),
- ... (None, 'World'),
- ... (GeneratorExit, None),
- ... ]:
- ... q.put(i)
- >>> qs = queue_source(q, printer_sink())
- Hello
- """
isDone = False
while not isDone:
item = queue.get()
}
widgetTree.signal_autoconnect(callbackMapping)
- def request_credentials(self, parentWindow = None):
+ def request_credentials(self,
+ parentWindow = None,
+ defaultCredentials = ("", "")
+ ):
"""
@note UI Thread
"""
self._serviceCombo.hide()
self._serviceList.clear()
+ self._usernameEntry.set_text(defaultCredentials[0])
+ self._passwordEntry.set_text(defaultCredentials[1])
+
try:
self._dialog.set_transient_for(parentWindow)
self._dialog.set_default_response(gtk.RESPONSE_OK)
parentWindow = self._parentWindow
self._serviceList.clear()
- for serviceIdserviceName in services.iteritems():
+ for serviceIdserviceName in services:
self._serviceList.append(serviceIdserviceName)
self._serviceCombo.set_active(0)
self._serviceCombo.show()
self._dialog.response(gtk.RESPONSE_CANCEL)
+def safecall(f, errorDisplay=None, default=None, exception=Exception):
+ '''
+ Returns modified f. When the modified f is called and throws an
+ exception, the default value is returned
+ '''
+ def _safecall(*args, **argv):
+ try:
+ return f(*args,**argv)
+ except exception, e:
+ if errorDisplay is not None:
+ errorDisplay.push_exception(e)
+ return default
+ return _safecall
+
+
class ErrorDisplay(object):
def __init__(self, widgetTree):
else:
self.__show_message(message)
- def push_exception_with_lock(self, exception = None):
+ def push_exception_with_lock(self, exception = None, stacklevel=3):
with gtk_lock():
- self.push_exception(exception)
+ self.push_exception(exception, stacklevel=stacklevel)
- def push_exception(self, exception = None):
+ def push_exception(self, exception = None, stacklevel=2):
if exception is None:
userMessage = str(sys.exc_value)
warningMessage = str(traceback.format_exc())
userMessage = str(exception)
warningMessage = str(exception)
self.push_message(userMessage)
- warnings.warn(warningMessage, stacklevel=3)
+ warnings.warn(warningMessage, stacklevel=stacklevel)
def pop_message(self):
if 0 < len(self.__messages):
self._errorDisplay.push_exception()
+class TapOrHold(object):
+
+ def __init__(self, widget):
+ self._widget = widget
+ self._isTap = True
+ self._isPointerInside = True
+ self._holdTimeoutId = None
+ self._tapTimeoutId = None
+ self._taps = 0
+
+ self._bpeId = None
+ self._breId = None
+ self._eneId = None
+ self._lneId = None
+
+ def enable(self):
+ self._bpeId = self._widget.connect("button-press-event", self._on_button_press)
+ self._breId = self._widget.connect("button-release-event", self._on_button_release)
+ self._eneId = self._widget.connect("enter-notify-event", self._on_enter)
+ self._lneId = self._widget.connect("leave-notify-event", self._on_leave)
+
+ def disable(self):
+ self._widget.disconnect(self._bpeId)
+ self._widget.disconnect(self._breId)
+ self._widget.disconnect(self._eneId)
+ self._widget.disconnect(self._lneId)
+
+ def on_tap(self, taps):
+ print "TAP", taps
+
+ def on_hold(self, taps):
+ print "HOLD", taps
+
+ def on_holding(self):
+ print "HOLDING"
+
+ def on_cancel(self):
+ print "CANCEL"
+
+ def _on_button_press(self, *args):
+ # Hack to handle weird notebook behavior
+ self._isPointerInside = True
+ self._isTap = True
+
+ if self._tapTimeoutId is not None:
+ gobject.source_remove(self._tapTimeoutId)
+ self._tapTimeoutId = None
+
+ # Handle double taps
+ if self._holdTimeoutId is None:
+ self._tapTimeoutId = None
+
+ self._taps = 1
+ self._holdTimeoutId = gobject.timeout_add(1000, self._on_hold_timeout)
+ else:
+ self._taps = 2
+
+ def _on_button_release(self, *args):
+ assert self._tapTimeoutId is None
+ # Handle release after timeout if user hasn't double-clicked
+ self._tapTimeoutId = gobject.timeout_add(100, self._on_tap_timeout)
+
+ def _on_actual_press(self, *args):
+ if self._holdTimeoutId is not None:
+ gobject.source_remove(self._holdTimeoutId)
+ self._holdTimeoutId = None
+
+ if self._isPointerInside:
+ if self._isTap:
+ self.on_tap(self._taps)
+ else:
+ self.on_hold(self._taps)
+ else:
+ self.on_cancel()
+
+ def _on_tap_timeout(self, *args):
+ self._tapTimeoutId = None
+ self._on_actual_press()
+ return False
+
+ def _on_hold_timeout(self, *args):
+ self._holdTimeoutId = None
+ self._isTap = False
+ self.on_holding()
+ return False
+
+ def _on_enter(self, *args):
+ self._isPointerInside = True
+
+ def _on_leave(self, *args):
+ self._isPointerInside = False
+
+
if __name__ == "__main__":
if False:
import datetime
projects / gc-dialer / blobdiff