1#!/usr/bin/env python 2"""Demo of using urwid with Python 3.4's asyncio. 3 4This code works on older Python 3.x if you install `asyncio` from PyPI, and 5even Python 2 if you install `trollius`! 6""" 7from __future__ import print_function 8 9try: 10 import asyncio 11except ImportError: 12 import trollius as asyncio 13 14from datetime import datetime 15import sys 16import weakref 17 18import urwid 19from urwid.raw_display import Screen 20from urwid.display_common import BaseScreen 21 22import logging 23logging.basicConfig() 24 25loop = asyncio.get_event_loop() 26 27 28# ----------------------------------------------------------------------------- 29# General-purpose setup code 30 31def build_widgets(): 32 input1 = urwid.Edit('What is your name? ') 33 input2 = urwid.Edit('What is your quest? ') 34 input3 = urwid.Edit('What is the capital of Assyria? ') 35 inputs = [input1, input2, input3] 36 37 def update_clock(widget_ref): 38 widget = widget_ref() 39 if not widget: 40 # widget is dead; the main loop must've been destroyed 41 return 42 43 widget.set_text(datetime.now().isoformat()) 44 45 # Schedule us to update the clock again in one second 46 loop.call_later(1, update_clock, widget_ref) 47 48 clock = urwid.Text('') 49 update_clock(weakref.ref(clock)) 50 51 return urwid.Filler(urwid.Pile([clock] + inputs), 'top') 52 53 54def unhandled(key): 55 if key == 'ctrl c': 56 raise urwid.ExitMainLoop 57 58 59# ----------------------------------------------------------------------------- 60# Demo 1 61 62def demo1(): 63 """Plain old urwid app. Just happens to be run atop asyncio as the event 64 loop. 65 66 Note that the clock is updated using the asyncio loop directly, not via any 67 of urwid's facilities. 68 """ 69 main_widget = build_widgets() 70 71 urwid_loop = urwid.MainLoop( 72 main_widget, 73 event_loop=urwid.AsyncioEventLoop(loop=loop), 74 unhandled_input=unhandled, 75 ) 76 urwid_loop.run() 77 78 79# ----------------------------------------------------------------------------- 80# Demo 2 81 82class AsyncScreen(Screen): 83 """An urwid screen that speaks to an asyncio stream, rather than polling 84 file descriptors. 85 86 This is fairly limited; it can't, for example, determine the size of the 87 remote screen. Fixing that depends on the nature of the stream. 88 """ 89 def __init__(self, reader, writer, encoding="utf-8"): 90 self.reader = reader 91 self.writer = writer 92 self.encoding = encoding 93 94 Screen.__init__(self, None, None) 95 96 _pending_task = None 97 98 def write(self, data): 99 self.writer.write(data.encode(self.encoding)) 100 101 def flush(self): 102 pass 103 104 def hook_event_loop(self, event_loop, callback): 105 # Wait on the reader's read coro, and when there's data to read, call 106 # the callback and then wait again 107 def pump_reader(fut=None): 108 if fut is None: 109 # First call, do nothing 110 pass 111 elif fut.cancelled(): 112 # This is in response to an earlier .read() call, so don't 113 # schedule another one! 114 return 115 elif fut.exception(): 116 pass 117 else: 118 try: 119 self.parse_input( 120 event_loop, callback, bytearray(fut.result())) 121 except urwid.ExitMainLoop: 122 # This will immediately close the transport and thus the 123 # connection, which in turn calls connection_lost, which 124 # stops the screen and the loop 125 self.writer.abort() 126 127 # create_task() schedules a coroutine without using `yield from` or 128 # `await`, which are syntax errors in Pythons before 3.5 129 self._pending_task = event_loop._loop.create_task( 130 self.reader.read(1024)) 131 self._pending_task.add_done_callback(pump_reader) 132 133 pump_reader() 134 135 def unhook_event_loop(self, event_loop): 136 if self._pending_task: 137 self._pending_task.cancel() 138 del self._pending_task 139 140 141class UrwidProtocol(asyncio.Protocol): 142 def connection_made(self, transport): 143 print("Got a client!") 144 self.transport = transport 145 146 # StreamReader is super convenient here; it has a regular method on our 147 # end (feed_data), and a coroutine on the other end that will 148 # faux-block until there's data to be read. We could also just call a 149 # method directly on the screen, but this keeps the screen somewhat 150 # separate from the protocol. 151 self.reader = asyncio.StreamReader(loop=loop) 152 screen = AsyncScreen(self.reader, transport) 153 154 main_widget = build_widgets() 155 self.urwid_loop = urwid.MainLoop( 156 main_widget, 157 event_loop=urwid.AsyncioEventLoop(loop=loop), 158 screen=screen, 159 unhandled_input=unhandled, 160 ) 161 162 self.urwid_loop.start() 163 164 def data_received(self, data): 165 self.reader.feed_data(data) 166 167 def connection_lost(self, exc): 168 print("Lost a client...") 169 self.reader.feed_eof() 170 self.urwid_loop.stop() 171 172 173def demo2(): 174 """Urwid app served over the network to multiple clients at once, using an 175 asyncio Protocol. 176 """ 177 coro = loop.create_server(UrwidProtocol, port=12345) 178 loop.run_until_complete(coro) 179 print("OK, good to go! Try this in another terminal (or two):") 180 print() 181 print(" socat TCP:127.0.0.1:12345 STDIN,rawer") 182 print() 183 loop.run_forever() 184 185 186if __name__ == '__main__': 187 if len(sys.argv) == 2: 188 which = sys.argv[1] 189 else: 190 which = None 191 192 if which == '1': 193 demo1() 194 elif which == '2': 195 demo2() 196 else: 197 print("Please run me with an argument of either 1 or 2.") 198 sys.exit(1) 199