I just hacked this up last night so no guarantees that it's
perfect...but here's a twisted_main_loop that's very similar in nature
and functionality to the generic_main_loop in the current urwid checkout
-- except that everything run from inside of it happens in the context
of a Twisted reactor.
It places a blocking input loop in a separate thread, but arranges for
the actual input processing to happen inside the main reactor thread so
that this shouldn't cause any real issues.
It definitely needs some refinement, but I thought I'd make it available
in case anyone has a use for it.
--
Walter Mundt
[EMAIL PROTECTED]
# Urwid Twisted-based main loop function
# (c) 2008 Walter Mundt
import threading
import sys
import os
import urwid
from twisted.internet import reactor, threads
from twisted.internet.defer import Deferred
from twisted.python.failure import Failure
class TwistedLoopManager(object):
def __init__(self, screen, topmost_widget, input_filter, unhandled_input, need_stop):
self.screen = screen
self.topmost_widget = topmost_widget
self.failure = None
if input_filter:
self.input_filter = input_filter
if unhandled_input:
self.unhandled_input = unhandled_input
self.need_stop = need_stop
def input_filter(self, k):
return k
def unhandled_input(self, k):
pass
def refresh(self):
if not self.running:
return
try:
canvas = self.topmost_widget.render(self.size, focus=True)
self.screen.draw_screen(self.size, canvas)
except urwid.main_loop.ExitMainLoop:
self.running = False
except:
self.running = False
self.failure = Failure()
def onShutdown(self):
self.running = False
if self.need_stop:
self.screen.stop()
if self.failure:
self.d.errback(self.failure)
else:
self.d.callback(True)
def run(self):
self.d = Deferred()
self.size = self.screen.get_cols_rows()
self.running = True
self.refresh()
self.inputThread = threading.Thread(target=self.inputLoop, name="inputLoop")
reactor.addSystemEventTrigger('before', 'shutdown', self.onShutdown)
self.inputThread.start()
return self.d
def inputLoop(self):
try:
while self.running:
threads.blockingCallFromThread(reactor, self.refresh)
events = None
while not events and self.running:
events = self.screen.get_input()
if self.running:
for event in events:
threads.blockingCallFromThread(reactor, self.onInput, event)
finally:
reactor.callFromThread(reactor.stop)
def onInput(self, event):
if not self.running:
return
try:
if event == 'window resize':
self.size = self.screen.get_cols_rows()
event = self.input_filter(event)
if urwid.is_mouse_event(event):
if self.topmost_widget.mouse_event(self.size, focus=True, *event):
event = None
else:
event = self.topmost_widget.keypress(self.size, event)
if event:
self.unhandled_input(event)
except urwid.main_loop.ExitMainLoop:
self.running = False
except:
self.running = False
self.failure = Failure()
def twisted_main_loop(topmost_widget, palette=[], screen=None,
handle_mouse=True, input_filter=None, unhandled_input=None):
if not screen:
import urwid.raw_display
screen = urwid.raw_display.Screen()
if palette:
screen.register_palette(palette)
handleStartStop = not screen.started
if handleStartStop:
screen.start()
needStop = True
global mgr
mgr = TwistedLoopManager(screen, topmost_widget, input_filter, unhandled_input, needStop)
if handle_mouse:
screen.set_mouse_tracking()
d = mgr.run()
d.addErrback(lambda x: True)
reactor.run()
if mgr.failure:
mgr.failure.printTraceback()
_______________________________________________
Urwid mailing list
[email protected]
http://lists.excess.org/mailman/listinfo/urwid
