-
Notifications
You must be signed in to change notification settings - Fork 72
/
Copy pathtui.py
363 lines (309 loc) · 11.8 KB
/
tui.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
# -*- coding: utf-8 -*-
#
# Copyright 2016 Ahmed Nazmy
#
# Meta
__license__ = "AGPLv3"
__author__ = 'Ahmed Nazmy <[email protected]>'
import urwid
import aker
import signal
import logging
import os
from popup import SimplePopupLauncher
class Listing(urwid.ListBox):
"""
Base class to handle listbox actions
"""
def __init__(self, items=None):
self.search = Search()
self.search.update_text("Type to search:\n")
self._items = []
if items is not None:
for item in items:
listitem = MenuItem("%s" % (item))
self._items.append(
urwid.AttrMap(
listitem,
'body',
focus_map='SSH_focus'))
super(Listing, self).__init__(urwid.SimpleFocusListWalker(self._items))
def updatelist(self, items):
self.empty()
for item in items:
self.add_item(item)
def add_item(self, item):
listitem = MenuItem("%s" % (item))
self.body.append(
urwid.AttrMap(
listitem,
'body',
focus_map='SSH_focus'))
def empty(self):
del self.body[:] # clear listbox
def get_selected(self):
return self.focus
def get_box(self):
self.search.clear()
return urwid.Frame(urwid.AttrWrap(self, 'body'), header=self.search)
class HostList(Listing):
"""
Class to handle hosts screen actions,
keypresses for now.
"""
def __init__(self, hosts=None):
super(HostList, self).__init__(hosts)
def keypress(self, size, key):
if (key == 'enter') or (key == 'right'):
urwid.emit_signal(
self,
'connect',
self.focus.original_widget.get_caption())
key = None
elif key == 'esc':
if self.search.get_edit_text() == "":
key = 'left'
else:
self.search.clear()
key = None
# Unless its arrow keys send keypress to search box,
# implies emitting EditBox "change" signal
elif key not in ['right', 'down', 'up', 'left', 'page up', 'page down']:
self.search.keypress((10,), key)
return super(HostList, self).keypress(size, key)
class HostGroupList(Listing):
"""
Class to handle hostgroups screen actions,
keypresses for now.
"""
def __init__(self, hostgroups=None):
super(HostGroupList, self).__init__(hostgroups)
def keypress(self, size, key):
if (key == 'enter') or (key == 'right'):
# emit signal to call hostgroup_chosen_handler with MenuItem caption,
# caption is group name showing on screen
if self.focus is not None:
urwid.emit_signal(
self,
'group_chosen',
self.focus.original_widget.get_caption())
key = None
elif key == 'esc':
self.search.clear()
key = None
# Unless its arrow keys send keypress to search box,
# implies emitting EditBox "change" signal
elif key not in ['right', 'down', 'up', 'left', 'page up', 'page down']:
self.search.keypress((10,), key)
return super(HostGroupList, self).keypress(size, key)
class Header(urwid.Columns):
def __init__(self, text):
self.text = text
self.header_widget = urwid.Text(self.text, align='left')
self.popup = SimplePopupLauncher()
self.popup_padding = urwid.Padding(self.popup, 'right', 20)
self.popup_map = urwid.AttrMap(self.popup_padding, 'indicator')
self.header_map = urwid.AttrMap(self.header_widget, 'head')
super(Header, self).__init__([self.header_map, self.popup_map])
def update_text(self, text):
self.text = text
self.header_map.original_widget.set_text(self.text)
def popup_message(self, message):
logging.debug("TUI: popup message is {0}".format(message))
self.popup.message = str(message)
self.popup.open_pop_up()
class Footer(urwid.AttrMap):
def __init__(self, text):
self.footer_text = urwid.Text(text, align='center')
super(Footer, self).__init__(self.footer_text, 'foot')
class Search(urwid.Edit):
def __init__(self):
super(Search, self).__init__()
def update_text(self, caption):
self.set_caption(caption)
def clear(self):
self.set_edit_text("")
class MenuItem(urwid.Text):
def __init__(self, caption):
self.caption = caption
urwid.Text.__init__(self, self.caption)
def keypress(self, size, key):
return key
def selectable(self):
return True
def get_caption(self):
return str(self.caption)
class Window(object):
"""
Where all the Tui magic happens,
handles creating urwid widgets and
user interactions
"""
def __init__(self, aker_core):
self.aker = aker_core
self.user = self.aker.user
self.current_hostgroup = ""
self.set_palette()
def set_palette(self):
self.palette = [
('body', 'black', 'light gray'), # Normal Text
('focus', 'light green', 'black', 'standout'), # Focus
('head', 'white', 'dark gray', 'standout'), # Header
('foot', 'light gray', 'dark gray'), # Footer Separator
('key', 'light green', 'dark gray', 'bold'),
('title', 'white', 'black', 'bold'),
('popup', 'white', 'dark red'),
('msg', 'yellow', 'dark gray'),
('SSH', 'dark blue', 'light gray', 'underline'),
('SSH_focus', 'light green', 'dark blue', 'standout')] # Focus
def draw(self):
self.header_text = [
('key', "Aker"), " ",
('msg', "User:"),
('key', "%s" % self.user.name), " "]
self.footer_text = [
('msg', "Move:"),
('key', "Up"), ",",
('key', "Down"), ",",
('key', "Left"), ",",
('key', "Right"), ",",
('key', "PgUp"), ",",
('key', "PgDn"), ",",
('msg', "Select:"),
('key', "Enter"), " ",
('msg', "Refresh:"),
('key', "F5"), " ",
('msg', "Quit:"),
('key', "F9"), " ",
('msg', "By:"),
('key', "Ahmed Nazmy")]
# Define widgets
self.header = Header(self.header_text)
self.footer = Footer(self.footer_text)
self.hostgrouplist = HostGroupList(list(self.user.hostgroups.keys()))
self.hostlist = HostList(list(self.user.allowed_ssh_hosts.keys()))
self.topframe = urwid.Frame(
self.hostgrouplist.get_box(),
header=self.header,
footer=self.footer)
self.screen = urwid.raw_display.Screen()
# Register signals
urwid.register_signal(HostList, ['connect'])
urwid.register_signal(HostGroupList, ['group_chosen'])
# Connect signals
urwid.connect_signal(
self.hostgrouplist.search,
'change',
self.group_search_handler)
urwid.connect_signal(
self.hostgrouplist,
'group_chosen',
self.group_chosen_handler)
urwid.connect_signal(
self.hostlist.search,
'change',
self.host_search_handler)
urwid.connect_signal(
self.hostlist,
'connect',
self.host_chosen_handler)
self.loop = urwid.MainLoop(
self.topframe,
palette=self.palette,
unhandled_input=self._input_handler,
screen=self.screen,
pop_ups=True)
def _input_handler(self, key):
if not urwid.is_mouse_event(key):
if key == 'f5':
self.update_lists()
elif key == 'f9':
logging.info(
"TUI: User {0} logging out of Aker".format(
self.user.name))
raise urwid.ExitMainLoop()
elif key == 'left':
# For now if its not hostgroup window left should bring it up
if self.topframe.get_body() != self.hostgrouplist.get_box():
self.current_hostgroup = ""
self.hostlist.empty()
self.header.update_text(self.header_text)
self.topframe.set_body(self.hostgrouplist.get_box())
else:
logging.debug(
"TUI: User {0} unhandled input : {1}".format(
self.user.name, key))
def group_search_handler(self, search, search_text):
logging.debug(
"TUI: Group search handler called with text {0}".format(search_text))
matchinghostgroups = []
for hostgroup in self.user.hostgroups.keys():
if search_text in hostgroup:
logging.debug(
"TUI: hostgroup {1} matches search text {0}".format(
search_text, hostgroup))
matchinghostgroups.append(hostgroup)
self.hostgrouplist.updatelist(matchinghostgroups)
def host_search_handler(self, search, search_text):
logging.debug(
"TUI: Host search handler called with text {0}".format(search_text))
matchinghosts = []
for host in self.user.hostgroups[self.current_hostgroup].hosts:
if search_text in host:
logging.debug(
"TUI: host {1} matches search text {0}".format(
search_text, host))
matchinghosts.append(host)
self.hostlist.updatelist(sorted(matchinghosts))
def group_chosen_handler(self, hostgroup):
logging.debug(
"TUI: user %s chose hostgroup %s " %
(self.user.name, hostgroup))
self.current_hostgroup = hostgroup
self.hostlist.empty()
matchinghosts = []
for host in self.user.hostgroups[self.current_hostgroup].hosts:
logging.debug(
"TUI: host {1} is in hostgroup {0}, adding".format(
hostgroup, host))
matchinghosts.append(host)
self.hostlist.updatelist(sorted(matchinghosts))
header_text = [
('key', "Aker"), " ",
('msg', "User:"),
('key', "%s" % self.user.name), " ",
('msg', "HostGroup:"),
('key', "%s" % self.current_hostgroup)]
self.header.update_text(header_text)
self.topframe.set_body(self.hostlist.get_box())
def host_chosen_handler(self, choice):
host = choice
logging.debug("TUI: user %s chose server %s " % (self.user.name, host))
self.aker.init_connection(self.user.allowed_ssh_hosts[host])
def update_lists(self):
logging.info(
"TUI: Refreshing entries for user {0}".format(
self.aker.user.name))
self.aker.user.refresh_allowed_hosts(False)
self.hostgrouplist.empty()
for hostgroup in self.user.hostgroups.keys():
self.hostgrouplist.add_item(hostgroup)
if self.current_hostgroup != "":
self.hostlist.empty()
for host in self.user.hostgroups[self.current_hostgroup].hosts:
self.hostlist.add_item(host)
self.header.popup_message("Entries Refreshed")
def start(self):
logging.debug("TUI: tui started")
self.loop.run()
def stop(self):
logging.debug(u"TUI: tui stopped")
raise urwid.ExitMainLoop()
def pause(self):
logging.debug("TUI: tui paused")
self.loop.screen.stop()
urwid.emit_signal(self.loop.screen, urwid.display_common.INPUT_DESCRIPTORS_CHANGED)
def restore(self):
logging.debug("TUI restored")
self.loop.screen.start()
urwid.emit_signal(self.loop.screen, urwid.display_common.INPUT_DESCRIPTORS_CHANGED)