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
|
# Copyright 2010, 2011 Lars Wirzenius
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
import time
import ttystatus
class Messager(object):
'''Manages messages to the terminal.
This includes disabling messages, allowing notifications, and
becalming the flow of messages to avoid writing too fast. The
speed is a performance thing: writing too much message text can
slow an application down a lot (too much work for terminal
emulators), and doesn't actually help the user in any way.
'''
def __init__(self, period=None, _terminal=None):
self._period = 1.0 if period is None else period
self.enabled = True
self._cached_message = None # The latest message from caller.
self._displayed_message = None # The latest message displayed.
self._previous_write_at = 0 # When the latest message was written.
self._terminal = _terminal or ttystatus.PhysicalTerminal()
try:
self._terminal.open_tty()
except IOError:
self.enabled = False
if not self._terminal.has_capabilities():
self.enabled = False
self._area = ttystatus.AreaManager()
self._area.set_terminal(self._terminal)
def disable(self):
'''Disable all output except notifications.'''
self.enabled = False
def enable(self):
'''Enable output to happen.'''
self.enabled = True
def time_to_write(self):
'''Is it time to write now?'''
return self._now() - self._previous_write_at >= self._period
def _now(self):
'''Return current time.'''
# This is a wrapper around time.time(), for testing.
return time.time()
def get_terminal_size(self):
'''Return terminal width, height.'''
return self._terminal.get_size()
def get_max_line_length(self):
return self._area.get_max_line_length()
def write(self, message, force=False):
'''Write message to terminal.
Message may be multiple lines.
'''
if self.enabled and (force or self.time_to_write()):
if self._displayed_message is not None:
self._area.prepare_to_overwrite(self._displayed_message)
num_lines = len(message.split('\n'))
self._area.make_space(num_lines)
self._area.display(message)
self._displayed_message = message
self._previous_write_at = self._now()
self._cached_message = message
def clear(self):
'''Remove currently displayed message from terminal, if any.'''
if self._displayed_message is not None:
num_lines = len(self._displayed_message.split('\n'))
self._area.clear_area(num_lines)
self._displayed_message = None
self._cached_message = None
self._previous_write_at = 0 # Next .write() should display.
def notify(self, message, f, force=False):
'''Show a notification message string to the user.
Notifications are meant for error messages and other things
that do not belong in, say, progress bars. Whatever is currently
on the terminal is wiped, then the notification message is shown,
a new line is started, and the old message is restored.
Notifications are written even when the output is not going
to a terminal.
'''
if self.enabled or force:
self.clear()
try:
f.write(message)
f.write('\n')
f.flush()
except IOError:
# We ignore these. No point in crashing if terminal is bad.
pass
if self._cached_message is not None:
self.write(self._cached_message)
def finish(self):
'''Finalize output.'''
if self.enabled and self._cached_message is not None:
self.write(self._cached_message)
if self._cached_message:
self._terminal.write('\n')
|
