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