io.py
153 lines
| 4.6 KiB
| text/x-python
|
PythonLexer
|
r2498 | # encoding: utf-8 | ||
""" | ||||
IO related utilities. | ||||
""" | ||||
|
r21111 | # Copyright (c) IPython Development Team. | ||
# Distributed under the terms of the Modified BSD License. | ||||
|
r2498 | |||
|
r21111 | |||
|
r21762 | import atexit | ||
|
r6653 | import os | ||
|
r2498 | import sys | ||
import tempfile | ||||
|
r26154 | from pathlib import Path | ||
|
r21298 | from warnings import warn | ||
|
r22192 | |||
from IPython.utils.decorators import undoc | ||||
|
r12223 | from .capture import CapturedIO, capture_output | ||
|
r2498 | |||
class Tee(object): | ||||
"""A class to duplicate an output stream to stdout/err. | ||||
This works in a manner very similar to the Unix 'tee' command. | ||||
When the object is closed or deleted, it closes the original file given to | ||||
it for duplication. | ||||
""" | ||||
# Inspired by: | ||||
# http://mail.python.org/pipermail/python-list/2007-May/442737.html | ||||
|
r4763 | def __init__(self, file_or_name, mode="w", channel='stdout'): | ||
|
r2498 | """Construct a new Tee object. | ||
Parameters | ||||
---------- | ||||
file_or_name : filename or open filehandle (writable) | ||||
|
r26419 | File that will be duplicated | ||
|
r2498 | mode : optional, valid mode for open(). | ||
|
r26419 | If a filename was give, open with this mode. | ||
|
r4872 | channel : str, one of ['stdout', 'stderr'] | ||
|
r2498 | """ | ||
if channel not in ['stdout', 'stderr']: | ||||
raise ValueError('Invalid channel spec %s' % channel) | ||||
|
r4872 | |||
|
r4763 | if hasattr(file_or_name, 'write') and hasattr(file_or_name, 'seek'): | ||
|
r2498 | self.file = file_or_name | ||
else: | ||||
|
r27495 | encoding = None if "b" in mode else "utf-8" | ||
|
r27494 | self.file = open(file_or_name, mode, encoding=encoding) | ||
|
r2498 | self.channel = channel | ||
self.ostream = getattr(sys, channel) | ||||
setattr(sys, channel, self) | ||||
self._closed = False | ||||
def close(self): | ||||
"""Close the file and restore the channel.""" | ||||
self.flush() | ||||
setattr(sys, self.channel, self.ostream) | ||||
self.file.close() | ||||
self._closed = True | ||||
def write(self, data): | ||||
"""Write data to both channels.""" | ||||
self.file.write(data) | ||||
self.ostream.write(data) | ||||
self.ostream.flush() | ||||
def flush(self): | ||||
"""Flush both channels.""" | ||||
self.file.flush() | ||||
self.ostream.flush() | ||||
def __del__(self): | ||||
if not self._closed: | ||||
self.close() | ||||
|
r28789 | def isatty(self): | ||
return False | ||||
|
r2498 | |||
|
r13609 | def ask_yes_no(prompt, default=None, interrupt=None): | ||
|
r2498 | """Asks a question and returns a boolean (y/n) answer. | ||
If default is given (one of 'y','n'), it is used if the user input is | ||||
|
r13609 | empty. If interrupt is given (one of 'y','n'), it is used if the user | ||
presses Ctrl-C. Otherwise the question is repeated until an answer is | ||||
given. | ||||
|
r2498 | |||
An EOF is treated as the default answer. If there is no default, an | ||||
exception is raised to prevent infinite loops. | ||||
Valid answers are: y/yes/n/no (match is not case sensitive).""" | ||||
answers = {'y':True,'n':False,'yes':True,'no':False} | ||||
ans = None | ||||
while ans not in answers.keys(): | ||||
try: | ||||
|
r13355 | ans = input(prompt+' ').lower() | ||
|
r2498 | if not ans: # response was an empty string | ||
ans = default | ||||
except KeyboardInterrupt: | ||||
|
r13609 | if interrupt: | ||
ans = interrupt | ||||
|
r23031 | print("\r") | ||
|
r2498 | except EOFError: | ||
if default in answers.keys(): | ||||
ans = default | ||||
|
r4430 | print() | ||
|
r2498 | else: | ||
raise | ||||
return answers[ans] | ||||
def temp_pyfile(src, ext='.py'): | ||||
"""Make a temporary python file, return filename and filehandle. | ||||
Parameters | ||||
---------- | ||||
src : string or list of strings (no need for ending newlines if list) | ||||
|
r26419 | Source code to be written to the file. | ||
|
r2498 | ext : optional, string | ||
|
r26419 | Extension for the generated file. | ||
|
r2498 | |||
Returns | ||||
------- | ||||
(filename, open filehandle) | ||||
|
r26419 | It is the caller's responsibility to close the open file and unlink it. | ||
|
r2498 | """ | ||
fname = tempfile.mkstemp(ext)[1] | ||||
|
r27495 | with open(Path(fname), "w", encoding="utf-8") as f: | ||
|
r25106 | f.write(src) | ||
f.flush() | ||||
return fname | ||||
|
r2498 | |||
|
r24275 | @undoc | ||
|
r2874 | def raw_print(*args, **kw): | ||
|
r24275 | """DEPRECATED: Raw print to sys.__stdout__, otherwise identical interface to print().""" | ||
warn("IPython.utils.io.raw_print has been deprecated since IPython 7.0", DeprecationWarning, stacklevel=2) | ||||
|
r4872 | |||
|
r2856 | print(*args, sep=kw.get('sep', ' '), end=kw.get('end', '\n'), | ||
file=sys.__stdout__) | ||||
sys.__stdout__.flush() | ||||
|
r24275 | @undoc | ||
|
r2874 | def raw_print_err(*args, **kw): | ||
|
r24275 | """DEPRECATED: Raw print to sys.__stderr__, otherwise identical interface to print().""" | ||
warn("IPython.utils.io.raw_print_err has been deprecated since IPython 7.0", DeprecationWarning, stacklevel=2) | ||||
|
r4872 | |||
|
r2856 | print(*args, sep=kw.get('sep', ' '), end=kw.get('end', '\n'), | ||
file=sys.__stderr__) | ||||
|
r2838 | sys.__stderr__.flush() | ||