/usr/lib/python2.7/dist-packages/ttystatus/messager.py is in python-ttystatus 0.32-1.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
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')
|