page.py
348 lines
| 11.5 KiB
| text/x-python
|
PythonLexer
|
r2205 | # encoding: utf-8 | ||
""" | ||||
Paging capabilities for IPython.core | ||||
Notes | ||||
----- | ||||
|
r19387 | For now this uses IPython hooks, so it can't be in IPython.utils. If we can get | ||
|
r2205 | rid of that dependency, we could move it there. | ||
----- | ||||
""" | ||||
|
r19387 | # Copyright (c) IPython Development Team. | ||
# Distributed under the terms of the Modified BSD License. | ||||
|
r2205 | |||
import os | ||||
|
r25239 | import io | ||
|
r2205 | import re | ||
import sys | ||||
|
r2498 | import tempfile | ||
|
r25239 | import subprocess | ||
|
r2205 | |||
|
r4923 | from io import UnsupportedOperation | ||
|
r26023 | from pathlib import Path | ||
|
r4923 | |||
|
r10581 | from IPython import get_ipython | ||
|
r25632 | from IPython.display import display | ||
|
r2205 | from IPython.core.error import TryNext | ||
|
r2498 | from IPython.utils.data import chop | ||
|
r2908 | from IPython.utils.process import system | ||
|
r2498 | from IPython.utils.terminal import get_terminal_size | ||
|
r8966 | from IPython.utils import py3compat | ||
|
r2205 | |||
|
r19387 | def display_page(strng, start=0, screen_lines=25): | ||
"""Just display, no paging. screen_lines is ignored.""" | ||||
if isinstance(strng, dict): | ||||
data = strng | ||||
else: | ||||
if start: | ||||
strng = u'\n'.join(strng.splitlines()[start:]) | ||||
|
r22463 | data = { 'text/plain': strng } | ||
|
r19387 | display(data, raw=True) | ||
def as_hook(page_func): | ||||
"""Wrap a pager func to strip the `self` arg | ||||
|
r27288 | |||
|
r19387 | so it can be called as a hook. | ||
""" | ||||
return lambda self, *args, **kwargs: page_func(*args, **kwargs) | ||||
|
r2205 | |||
esc_re = re.compile(r"(\x1b[^m]+m)") | ||||
|
r2498 | def page_dumb(strng, start=0, screen_lines=25): | ||
|
r2205 | """Very dumb 'pager' in Python, for when nothing else works. | ||
Only moves forward, same interface as page(), except for pager_cmd and | ||||
|
r22463 | mode. | ||
""" | ||||
if isinstance(strng, dict): | ||||
strng = strng.get('text/plain', '') | ||||
|
r2205 | out_ln = strng.splitlines()[start:] | ||
screens = chop(out_ln,screen_lines-1) | ||||
if len(screens) == 1: | ||||
|
r22192 | print(os.linesep.join(screens[0])) | ||
|
r2205 | else: | ||
last_escape = "" | ||||
for scr in screens[0:-1]: | ||||
hunk = os.linesep.join(scr) | ||||
|
r22192 | print(last_escape + hunk) | ||
|
r2205 | if not page_more(): | ||
return | ||||
esc_list = esc_re.findall(hunk) | ||||
if len(esc_list) > 0: | ||||
last_escape = esc_list[-1] | ||||
|
r22192 | print(last_escape + os.linesep.join(screens[-1])) | ||
|
r2205 | |||
|
r9388 | def _detect_screen_size(screen_lines_def): | ||
|
r4920 | """Attempt to work out the number of lines on the screen. | ||
|
r8966 | |||
|
r4920 | This is called by page(). It can raise an error (e.g. when run in the | ||
test suite), so it's separated out so it can easily be called in a try block. | ||||
""" | ||||
|
r4921 | TERM = os.environ.get('TERM',None) | ||
|
r9388 | if not((TERM=='xterm' or TERM=='xterm-color') and sys.platform != 'sunos5'): | ||
|
r4897 | # curses causes problems on many terminals other than xterm, and | ||
# some termios calls lock up on Sun OS5. | ||||
|
r9388 | return screen_lines_def | ||
try: | ||||
|
r4897 | import termios | ||
import curses | ||||
|
r9388 | except ImportError: | ||
return screen_lines_def | ||||
# There is a bug in curses, where *sometimes* it fails to properly | ||||
# initialize, and then after the endwin() call is made, the | ||||
# terminal is left in an unusable state. Rather than trying to | ||||
|
r25146 | # check every time for this (by requesting and comparing termios | ||
|
r9388 | # flags each time), we just save the initial terminal state and | ||
# unconditionally reset it every time. It's cheaper than making | ||||
# the checks. | ||||
|
r21030 | try: | ||
term_flags = termios.tcgetattr(sys.stdout) | ||||
except termios.error as err: | ||||
|
r21033 | # can fail on Linux 2.6, pager_page will catch the TypeError | ||
|
r25833 | raise TypeError('termios error: {0}'.format(err)) from err | ||
|
r9388 | |||
try: | ||||
|
r4897 | scr = curses.initscr() | ||
|
r9388 | except AttributeError: | ||
# Curses on Solaris may not be complete, so we can't use it there | ||||
|
r4897 | return screen_lines_def | ||
|
r9388 | |||
screen_lines_real,screen_cols = scr.getmaxyx() | ||||
curses.endwin() | ||||
# Restore terminal state in case endwin() didn't. | ||||
termios.tcsetattr(sys.stdout,termios.TCSANOW,term_flags) | ||||
# Now we have what we needed: the screen size in rows/columns | ||||
return screen_lines_real | ||||
|
r28756 | # print('***Screen size:',screen_lines_real,'lines x', | ||
# screen_cols,'columns.') # dbg | ||||
|
r2498 | |||
|
r19387 | def pager_page(strng, start=0, screen_lines=0, pager_cmd=None): | ||
|
r16665 | """Display a string, piping through a pager after a certain length. | ||
|
r27288 | |||
|
r16665 | strng can be a mime-bundle dict, supplying multiple representations, | ||
keyed by mime-type. | ||||
|
r2205 | |||
The screen_lines parameter specifies the number of *usable* lines of your | ||||
terminal screen (total lines minus lines you need to reserve to show other | ||||
information). | ||||
If you set screen_lines to a number <=0, page() will try to auto-determine | ||||
your screen size and will only use up to (screen_size+screen_lines) for | ||||
printing, paging after that. That is, if you want auto-detection but need | ||||
to reserve the bottom 3 lines of the screen, use screen_lines = -3, and for | ||||
auto-detection without any lines reserved simply use screen_lines = 0. | ||||
If a string won't fit in the allowed lines, it is sent through the | ||||
specified pager command. If none given, look for PAGER in the environment, | ||||
and ultimately default to less. | ||||
If no system pager works, the string is sent through a 'dumb pager' | ||||
written in python, very simplistic. | ||||
""" | ||||
|
r16586 | |||
# for compatibility with mime-bundle form: | ||||
if isinstance(strng, dict): | ||||
strng = strng['text/plain'] | ||||
|
r2205 | |||
# Ugly kludge, but calling curses.initscr() flat out crashes in emacs | ||||
TERM = os.environ.get('TERM','dumb') | ||||
if TERM in ['dumb','emacs'] and os.name != 'nt': | ||||
|
r7817 | print(strng) | ||
|
r2205 | return | ||
# chop off the topmost part of the string we don't want to see | ||||
|
r3739 | str_lines = strng.splitlines()[start:] | ||
|
r2205 | str_toprint = os.linesep.join(str_lines) | ||
num_newlines = len(str_lines) | ||||
len_str = len(str_toprint) | ||||
# Dumb heuristics to guesstimate number of on-screen lines the string | ||||
# takes. Very basic, but good enough for docstrings in reasonable | ||||
# terminals. If someone later feels like refining it, it's not hard. | ||||
numlines = max(num_newlines,int(len_str/80)+1) | ||||
|
r2498 | screen_lines_def = get_terminal_size()[1] | ||
|
r2205 | |||
# auto-determine screen size | ||||
if screen_lines <= 0: | ||||
|
r4897 | try: | ||
|
r9388 | screen_lines += _detect_screen_size(screen_lines_def) | ||
|
r4923 | except (TypeError, UnsupportedOperation): | ||
|
r22192 | print(str_toprint) | ||
|
r4897 | return | ||
|
r2205 | |||
|
r28756 | # print('numlines',numlines,'screenlines',screen_lines) # dbg | ||
|
r2205 | if numlines <= screen_lines : | ||
|
r28756 | # print('*** normal print') # dbg | ||
|
r22192 | print(str_toprint) | ||
|
r2205 | else: | ||
# Try to open pager and default to internal one if that fails. | ||||
# All failure modes are tagged as 'retval=1', to match the return | ||||
# value of a failed system command. If any intermediate attempt | ||||
# sets retval to 1, at the end we resort to our own page_dumb() pager. | ||||
pager_cmd = get_pager_cmd(pager_cmd) | ||||
pager_cmd += ' ' + get_pager_start(pager_cmd,start) | ||||
if os.name == 'nt': | ||||
if pager_cmd.startswith('type'): | ||||
# The default WinXP 'type' command is failing on complex strings. | ||||
retval = 1 | ||||
else: | ||||
|
r15372 | fd, tmpname = tempfile.mkstemp('.txt') | ||
|
r26024 | tmppath = Path(tmpname) | ||
|
r15372 | try: | ||
os.close(fd) | ||||
|
r27495 | with tmppath.open("wt", encoding="utf-8") as tmpfile: | ||
|
r15372 | tmpfile.write(strng) | ||
|
r26024 | cmd = "%s < %s" % (pager_cmd, tmppath) | ||
|
r15372 | # tmpfile needs to be closed for windows | ||
if os.system(cmd): | ||||
retval = 1 | ||||
else: | ||||
retval = None | ||||
finally: | ||||
|
r26024 | Path.unlink(tmppath) | ||
|
r2205 | else: | ||
try: | ||||
retval = None | ||||
|
r25239 | # Emulate os.popen, but redirect stderr | ||
|
r27495 | proc = subprocess.Popen( | ||
pager_cmd, | ||||
shell=True, | ||||
stdin=subprocess.PIPE, | ||||
stderr=subprocess.DEVNULL, | ||||
) | ||||
pager = os._wrap_close( | ||||
io.TextIOWrapper(proc.stdin, encoding="utf-8"), proc | ||||
) | ||||
|
r8966 | try: | ||
pager_encoding = pager.encoding or sys.stdout.encoding | ||||
|
r23666 | pager.write(strng) | ||
|
r8966 | finally: | ||
retval = pager.close() | ||||
|
r7787 | except IOError as msg: # broken pipe when user quits | ||
|
r8966 | if msg.args == (32, 'Broken pipe'): | ||
|
r2205 | retval = None | ||
else: | ||||
retval = 1 | ||||
except OSError: | ||||
# Other strange problems, sometimes seen in Win2k/cygwin | ||||
retval = 1 | ||||
if retval is not None: | ||||
page_dumb(strng,screen_lines=screen_lines) | ||||
|
r2498 | |||
|
r19387 | def page(data, start=0, screen_lines=0, pager_cmd=None): | ||
"""Display content in a pager, piping through a pager after a certain length. | ||||
|
r27288 | |||
|
r19387 | data can be a mime-bundle dict, supplying multiple representations, | ||
keyed by mime-type, or text. | ||||
|
r27288 | |||
|
r19387 | Pager is dispatched via the `show_in_pager` IPython hook. | ||
If no hook is registered, `pager_page` will be used. | ||||
""" | ||||
# Some routines may auto-compute start offsets incorrectly and pass a | ||||
# negative value. Offset to 0 for robustness. | ||||
start = max(0, start) | ||||
# first, try the hook | ||||
ip = get_ipython() | ||||
if ip: | ||||
try: | ||||
ip.hooks.show_in_pager(data, start=start, screen_lines=screen_lines) | ||||
return | ||||
except TryNext: | ||||
pass | ||||
# fallback on default pager | ||||
return pager_page(data, start, screen_lines, pager_cmd) | ||||
|
r2498 | def page_file(fname, start=0, pager_cmd=None): | ||
|
r2205 | """Page a file, using an optional pager command and starting line. | ||
""" | ||||
pager_cmd = get_pager_cmd(pager_cmd) | ||||
pager_cmd += ' ' + get_pager_start(pager_cmd,start) | ||||
try: | ||||
if os.environ['TERM'] in ['emacs','dumb']: | ||||
raise EnvironmentError | ||||
|
r2908 | system(pager_cmd + ' ' + fname) | ||
|
r2205 | except: | ||
try: | ||||
if start > 0: | ||||
start -= 1 | ||||
|
r27495 | page(open(fname, encoding="utf-8").read(), start) | ||
|
r2205 | except: | ||
|
r7817 | print('Unable to show file',repr(fname)) | ||
|
r2205 | |||
|
r2498 | def get_pager_cmd(pager_cmd=None): | ||
"""Return a pager command. | ||||
|
r2205 | |||
|
r2498 | Makes some attempts at finding an OS-correct one. | ||
""" | ||||
|
r2205 | if os.name == 'posix': | ||
|
r22263 | default_pager_cmd = 'less -R' # -R for color control sequences | ||
|
r2205 | elif os.name in ['nt','dos']: | ||
default_pager_cmd = 'type' | ||||
if pager_cmd is None: | ||||
try: | ||||
pager_cmd = os.environ['PAGER'] | ||||
except: | ||||
pager_cmd = default_pager_cmd | ||||
|
r22100 | |||
|
r22272 | if pager_cmd == 'less' and '-r' not in os.environ.get('LESS', '').lower(): | ||
|
r22262 | pager_cmd += ' -R' | ||
|
r22100 | |||
|
r2205 | return pager_cmd | ||
|
r2498 | |||
def get_pager_start(pager, start): | ||||
|
r2205 | """Return the string for paging files with an offset. | ||
This is the '+N' argument which less and more (under Unix) accept. | ||||
""" | ||||
if pager in ['less','more']: | ||||
if start: | ||||
start_string = '+' + str(start) | ||||
else: | ||||
start_string = '' | ||||
else: | ||||
start_string = '' | ||||
return start_string | ||||
|
r2498 | |||
# (X)emacs on win32 doesn't like to be bypassed with msvcrt.getch() | ||||
|
r2205 | if os.name == 'nt' and os.environ.get('TERM','dumb') != 'emacs': | ||
import msvcrt | ||||
def page_more(): | ||||
""" Smart pausing between pages | ||||
@return: True if need print more lines, False if quit | ||||
""" | ||||
|
r22192 | sys.stdout.write('---Return to continue, q to quit--- ') | ||
|
r12260 | ans = msvcrt.getwch() | ||
|
r2205 | if ans in ("q", "Q"): | ||
result = False | ||||
else: | ||||
result = True | ||||
|
r22192 | sys.stdout.write("\b"*37 + " "*37 + "\b"*37) | ||
|
r2205 | return result | ||
else: | ||||
def page_more(): | ||||
|
r13355 | ans = py3compat.input('---Return to continue, q to quit--- ') | ||
|
r2205 | if ans.lower().startswith('q'): | ||
return False | ||||
else: | ||||
return True | ||||